cluster/dht: rmdir should succeed last on the hashed subvol

mkdir and rename both first happen on the hashed subvol.
Rmdir should succeed, only if all non hashed subvol succeed,
and finally removal from hashed_subvol succeeds.

if hashed_subvol is unavailable, then rmdir should be sent
on all subvols

Change-Id: I1bf5d5fc25393d0e2fde13ae403cf867a01f5ec7
BUG: 773187
Signed-off-by: shishir gowda <shishirng@gluster.com>
Reviewed-on: http://review.gluster.com/3202
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Amar Tumballi <amarts@redhat.com>
Reviewed-by: Anand Avati <avati@redhat.com>

1 files changed, 112 insertions, 2 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 500c9735f..c7480434e 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -3923,6 +3923,72 @@ dht_rmdir_selfheal_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
 
 int
+dht_rmdir_hashed_subvol_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+               int op_ret, int op_errno, struct iatt *preparent,
+               struct iatt *postparent, dict_t *xdata)
+{
+        dht_local_t  *local = NULL;
+        int           this_call_cnt = 0;
+        call_frame_t *prev = NULL;
+
+        local = frame->local;
+        prev  = cookie;
+
+        LOCK (&frame->lock);
+        {
+                if (op_ret == -1) {
+                        local->op_errno = op_errno;
+                        local->op_ret   = -1;
+                        if (op_errno != ENOENT && op_errno != EACCES) {
+                                local->need_selfheal = 1;
+                        }
+
+
+                        gf_log (this->name, GF_LOG_DEBUG,
+                                "rmdir on %s for %s failed (%s)",
+                                prev->this->name, local->loc.path,
+                                strerror (op_errno));
+                        goto unlock;
+                }
+
+                dht_iatt_merge (this, &local->preparent, preparent, prev->this);
+                dht_iatt_merge (this, &local->postparent, postparent,
+                                prev->this);
+
+        }
+unlock:
+        UNLOCK (&frame->lock);
+
+        this_call_cnt = dht_frame_return (frame);
+        if (is_last_call (this_call_cnt)) {
+               if (local->need_selfheal) {
+                        local->layout =
+                                dht_layout_get (this, local->loc.inode);
+
+                        /* TODO: neater interface needed below */
+                        local->stbuf.ia_type = local->loc.inode->ia_type;
+
+                        uuid_copy (local->gfid, local->loc.inode->gfid);
+                        dht_selfheal_restore (frame, dht_rmdir_selfheal_cbk,
+                                              &local->loc, local->layout);
+               } else {
+
+                        if (local->loc.parent) {
+                                WIPE (&local->preparent);
+                                WIPE (&local->postparent);
+                        }
+
+                        DHT_STACK_UNWIND (rmdir, frame, local->op_ret,
+                                          local->op_errno, &local->preparent,
+                                          &local->postparent, NULL);
+               }
+        }
+
+        return 0;
+}
+
+
+int
 dht_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int op_ret, int op_errno, struct iatt *preparent,
                struct iatt *postparent, dict_t *xdata)
@@ -3930,6 +3996,7 @@ dht_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         dht_local_t  *local = NULL;
         int           this_call_cnt = 0;
         call_frame_t *prev = NULL;
+        int           done = 0;
 
         local = frame->local;
         prev  = cookie;
@@ -3962,7 +4029,16 @@ unlock:
 
 
         this_call_cnt = dht_frame_return (frame);
-        if (is_last_call (this_call_cnt)) {
+
+        /* if local->hashed_subvol, we are yet to wind to hashed_subvol. */
+        if (local->hashed_subvol && (this_call_cnt == 1)) {
+                done = 1;
+        } else if (!local->hashed_subvol && !this_call_cnt) {
+                done = 1;
+        }
+
+
+        if (done) {
                 if (local->need_selfheal && local->fop_succeeded) {
                         local->layout =
                                 dht_layout_get (this, local->loc.inode);
@@ -3973,7 +4049,17 @@ unlock:
                         uuid_copy (local->gfid, local->loc.inode->gfid);
                         dht_selfheal_restore (frame, dht_rmdir_selfheal_cbk,
                                               &local->loc, local->layout);
-                } else {
+                } else if (this_call_cnt) {
+                        /* If non-hashed subvol's have responded, proceed */
+
+                        local->need_selfheal = 0;
+                        STACK_WIND (frame, dht_rmdir_hashed_subvol_cbk,
+                                    local->hashed_subvol,
+                                    local->hashed_subvol->fops->rmdir,
+                                    &local->loc, local->flags, NULL);
+                } else if (!this_call_cnt) {
+                        /* All subvol's have responded, proceed */
+
                         if (local->loc.parent) {
                                 WIPE (&local->preparent);
                                 WIPE (&local->postparent);
@@ -3995,6 +4081,7 @@ dht_rmdir_do (call_frame_t *frame, xlator_t *this)
         dht_local_t  *local = NULL;
         dht_conf_t   *conf = NULL;
         int           i = 0;
+        xlator_t     *hashed_subvol = NULL;
 
         VALIDATE_OR_GOTO (this->private, err);
 
@@ -4006,7 +4093,30 @@ dht_rmdir_do (call_frame_t *frame, xlator_t *this)
 
         local->call_cnt = conf->subvolume_cnt;
 
+        /* first remove from non-hashed_subvol */
+        hashed_subvol = dht_subvol_get_hashed (this, &local->loc);
+
+        if (!hashed_subvol) {
+                gf_log (this->name, GF_LOG_WARNING, "failed to get hashed "
+                        "subvol for %s",local->loc.path);
+        } else {
+                local->hashed_subvol = hashed_subvol;
+        }
+
+        /* When DHT has only 1 child */
+        if (conf->subvolume_cnt == 1) {
+                STACK_WIND (frame, dht_rmdir_hashed_subvol_cbk,
+                            conf->subvolumes[0],
+                            conf->subvolumes[0]->fops->rmdir,
+                            &local->loc, local->flags, NULL);
+                return 0;
+        }
+
         for (i = 0; i < conf->subvolume_cnt; i++) {
+                if (hashed_subvol &&
+                    (hashed_subvol == conf->subvolumes[i]))
+                        continue;
+
                 STACK_WIND (frame, dht_rmdir_cbk,
                             conf->subvolumes[i],
                             conf->subvolumes[i]->fops->rmdir,