dht enhancements

* create(filename@<distribute-volume-name>-<subvol-name>), will
  create the file in corresponding subvol of dht.
* same for unlink()

same logic can be extended to other fops if there is a need

Signed-off-by: Amar Tumballi <amar@gluster.com>
Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

BUG: 1200 (enhance distribute with hooks so lot of debugging can be done online)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=1200

3 files changed, 143 insertions, 49 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index be9c2b64afc..cafc534ad4d 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -168,9 +168,9 @@ unlock:
 					local->loc.path);
 				goto selfheal;
 			}
-			
+
 			dht_layout_set (this, local->inode, layout);
-			
+
 			if (local->ia_ino) {
 				local->stbuf.ia_ino = local->ia_ino;
                                 local->stbuf.ia_gen = local->ia_gen;
@@ -810,23 +810,26 @@ dht_lookup (call_frame_t *frame, xlator_t *this,
 			"Out of memory");
 		goto err;
 	}
-
-        ret = loc_dup (loc, &local->loc);
-        if (ret == -1) {
-                op_errno = errno;
-                gf_log (this->name, GF_LOG_DEBUG,
-                        "copying location failed for path=%s",
-                        loc->path);
-                goto err;
+        if (!dht_filter_loc_subvol_key (this, loc, &local->loc,
+                                        &hashed_subvol)) {
+                ret = loc_dup (loc, &local->loc);
+                if (ret == -1) {
+                        op_errno = errno;
+                        gf_log (this->name, GF_LOG_DEBUG,
+                                "copying location failed for path=%s",
+                                loc->path);
+                        goto err;
+                }
         }
-	
+
 	if (xattr_req) {
 		local->xattr_req = dict_ref (xattr_req);
 	} else {
 		local->xattr_req = dict_new ();
 	}
 
-	hashed_subvol = dht_subvol_get_hashed (this, loc);
+        if (!hashed_subvol)
+                hashed_subvol = dht_subvol_get_hashed (this, loc);
 	cached_subvol = dht_subvol_get_cached (this, loc->inode);
 
 	local->cached_subvol = cached_subvol;
@@ -870,7 +873,7 @@ dht_lookup (call_frame_t *frame, xlator_t *this,
 			
 			STACK_WIND (frame, dht_revalidate_cbk,
 				    subvol, subvol->fops->lookup,
-				    loc, local->xattr_req);
+				    &local->loc, local->xattr_req);
 
 			if (!--call_cnt)
 				break;
@@ -2953,6 +2956,17 @@ dht_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc)
 	VALIDATE_OR_GOTO (this, err);
 	VALIDATE_OR_GOTO (loc, err);
 
+        if (dht_filter_loc_subvol_key (this, loc, &local->loc,
+                                       &cached_subvol)) {
+                gf_log (this->name, GF_LOG_NORMAL,
+                        "unlinking %s on %s (given path %s)",
+                        local->loc.path, cached_subvol->name, loc->path);
+                STACK_WIND (frame, dht_unlink_cbk,
+                            cached_subvol, cached_subvol->fops->unlink,
+                            &local->loc);
+                goto done;
+        }
+
 	cached_subvol = dht_subvol_get_cached (this, loc->inode);
 	if (!cached_subvol) {
 		gf_log (this->name, GF_LOG_DEBUG,
@@ -2993,7 +3007,7 @@ dht_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc)
                 STACK_WIND (frame, dht_unlink_cbk,
                             cached_subvol, cached_subvol->fops->unlink, loc);
         }
-
+done:
 	return 0;
 err:
 	op_errno = (op_errno == -1) ? errno : op_errno;
@@ -3252,14 +3266,16 @@ dht_create (call_frame_t *frame, xlator_t *this,
 		goto err;
 	}
 
-	subvol = dht_subvol_get_hashed (this, loc);
-	if (!subvol) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"no subvolume in layout for path=%s",
-			loc->path);
-		op_errno = ENOENT;
-		goto err;
-	}
+        if (dht_filter_loc_subvol_key (this, loc, &local->loc,
+                                       &subvol)) {
+                gf_log (this->name, GF_LOG_NORMAL,
+                        "creating %s on %s (got create on %s)",
+                        local->loc.path, subvol->name, loc->path);
+                STACK_WIND (frame, dht_create_cbk,
+                            subvol, subvol->fops->create,
+                            &local->loc, flags, mode, fd);
+                goto done;
+        }
 
         ret = loc_dup (loc, &local->loc);
         if (ret == -1) {
@@ -3268,6 +3284,14 @@ dht_create (call_frame_t *frame, xlator_t *this,
                         "Out of memory");
                 goto err;
         }
+        subvol = dht_subvol_get_hashed (this, loc);
+        if (!subvol) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "no subvolume in layout for path=%s",
+                        loc->path);
+                op_errno = ENOENT;
+                goto err;
+        }
 
         if (!dht_is_subvol_filled (this, subvol)) {
                 gf_log (this->name, GF_LOG_TRACE,
@@ -3275,34 +3299,33 @@ dht_create (call_frame_t *frame, xlator_t *this,
                 STACK_WIND (frame, dht_create_cbk,
                             subvol, subvol->fops->create,
                             loc, flags, mode, fd);
-        } else {
-                /* Choose the minimum filled volume, and create the 
-                   files there */
-                /* TODO */
-                avail_subvol = dht_free_disk_available_subvol (this, subvol);
-                if (avail_subvol != subvol) {
-                        local->fd = fd_ref (fd);
-                        local->flags = flags;
-                        local->mode = mode;
-
-                        local->cached_subvol = avail_subvol;
-                        local->hashed_subvol = subvol;
-                        gf_log (this->name, GF_LOG_TRACE,
-                                "creating %s on %s (link at %s)", loc->path, 
-                                avail_subvol->name, subvol->name);
-                        dht_linkfile_create (frame, 
-                                             dht_create_linkfile_create_cbk,
-                                             avail_subvol, subvol, loc);
-                } else {
-                        gf_log (this->name, GF_LOG_TRACE,
-                                "creating %s on %s", loc->path, subvol->name);
-                        STACK_WIND (frame, dht_create_cbk,
-                                    subvol, subvol->fops->create,
-                                    loc, flags, mode, fd);
-                        
-                }
+                goto done;
         }
-
+        /* Choose the minimum filled volume, and create the 
+           files there */
+        /* TODO */
+        avail_subvol = dht_free_disk_available_subvol (this, subvol);
+        if (avail_subvol != subvol) {
+                local->fd = fd_ref (fd);
+                local->flags = flags;
+                local->mode = mode;
+
+                local->cached_subvol = avail_subvol;
+                local->hashed_subvol = subvol;
+                gf_log (this->name, GF_LOG_TRACE,
+                        "creating %s on %s (link at %s)", loc->path,
+                        avail_subvol->name, subvol->name);
+                dht_linkfile_create (frame,
+                                     dht_create_linkfile_create_cbk,
+                                     avail_subvol, subvol, loc);
+                goto done;
+        }
+        gf_log (this->name, GF_LOG_TRACE,
+                "creating %s on %s", loc->path, subvol->name);
+        STACK_WIND (frame, dht_create_cbk,
+                    subvol, subvol->fops->create,
+                    loc, flags, mode, fd);
+done:
 	return 0;
 
 err:
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index d5a5c7b2c59..dfdaf7e861c 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -31,6 +31,8 @@
 #define GF_DHT_LOOKUP_UNHASHED_ON   1
 #define GF_DHT_LOOKUP_UNHASHED_AUTO 2
 
+#include <fnmatch.h>
+
 typedef int (*dht_selfheal_dir_cbk_t) (call_frame_t *frame, void *cookie,
 				       xlator_t *this,
 				       int32_t op_ret, int32_t op_errno);
@@ -121,6 +123,8 @@ struct dht_local {
         /* need for file-info */
         char   *pathinfo;
         char   *key;
+
+        char   *newpath;
 };
 typedef struct dht_local dht_local_t;
 
@@ -289,4 +293,8 @@ int dht_frame_su_undo (call_frame_t *frame);
 
 int dht_build_child_loc (xlator_t *this, loc_t *child, loc_t *parent, char *name);
 
+int dht_filter_loc_subvol_key (xlator_t *this, loc_t *loc, loc_t *new_loc,
+                               xlator_t **subvol);
+
+
 #endif /* _DHT_H */
diff --git a/xlators/cluster/dht/src/dht-helper.c b/xlators/cluster/dht/src/dht-helper.c
index 767be38b41c..6580f788db9 100644
--- a/xlators/cluster/dht/src/dht-helper.c
+++ b/xlators/cluster/dht/src/dht-helper.c
@@ -77,6 +77,65 @@ out:
 	return 0;
 }
 
+int
+dht_filter_loc_subvol_key (xlator_t *this, loc_t *loc, loc_t *new_loc,
+                           xlator_t **subvol)
+{
+        char          *new_name  = NULL;
+        char          *new_path  = NULL;
+        xlator_list_t *trav      = NULL;
+        char           key[1024] = {0,};
+        int            ret       = 0; /* not found */
+
+        /* Why do other tasks if first required 'char' itself is not there */
+        if (loc->name && !strchr (loc->name, '@'))
+                goto out;
+
+        trav = this->children;
+        while (trav) {
+                snprintf (key, 1024, "*@%s:%s", this->name, trav->xlator->name);
+                if (fnmatch (key, loc->name, FNM_NOESCAPE) == 0) {
+                        new_name = GF_CALLOC(strlen (loc->name),
+                                             sizeof (char),
+                                             gf_common_mt_char);
+                        if (!new_name)
+                                goto out;
+                        if (fnmatch (key, loc->path, FNM_NOESCAPE) == 0) {
+                                new_path = GF_CALLOC(strlen (loc->path),
+                                                     sizeof (char),
+                                                     gf_common_mt_char);
+                                if (!new_path)
+                                        goto out;
+                                strncpy (new_path, loc->path, (strlen (loc->path) -
+                                                               strlen (key) + 1));
+                        }
+                        strncpy (new_name, loc->name, (strlen (loc->name) -
+                                                       strlen (key) + 1));
+
+                        if (new_loc) {
+                                new_loc->path   = ((new_path) ? new_path:
+                                                   gf_strdup (loc->path));
+                                new_loc->name   = new_name;
+                                new_loc->ino    = loc->ino;
+                                new_loc->inode  = inode_ref (loc->inode);
+                                new_loc->parent = inode_ref (loc->parent);
+                        }
+                        *subvol         = trav->xlator;
+                        ret = 1;  /* success */
+                        goto out;
+                }
+                trav = trav->next;
+        }
+out:
+        if (!ret) {
+                /* !success */
+                if (new_path)
+                        GF_FREE (new_path);
+                if (new_name)
+                        GF_FREE (new_name);
+        }
+        return ret;
+}
 
 int
 dht_deitransform (xlator_t *this, uint64_t y, xlator_t **subvol_p,
@@ -148,6 +207,10 @@ dht_local_wipe (xlator_t *this, dht_local_t *local)
                 local->selfheal.layout = NULL;
         }
 
+        if (local->newpath) {
+                GF_FREE (local->newpath);
+        }
+
 	GF_FREE (local);
 }