cluster/dht: synchronize rename and file-migration

Change-Id: I4f243c946f76d440680b651235f925e3d0ebf0fd
BUG: 1130888
Signed-off-by: Raghavendra G <rgowdapp@redhat.com>
Reviewed-on: http://review.gluster.org/8523
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

3 files changed, 293 insertions, 34 deletions

diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index c310b6c543b..62fef87a51c 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -28,6 +28,7 @@
 #define GF_DHT_LOOKUP_UNHASHED_ON   1
 #define GF_DHT_LOOKUP_UNHASHED_AUTO 2
 #define DHT_PATHINFO_HEADER         "DISTRIBUTE:"
+#define DHT_FILE_MIGRATE_DOMAIN     "dht.file.migrate"
 
 #include <fnmatch.h>
 
@@ -220,6 +221,7 @@ struct dht_local {
         gf_boolean_t     quota_deem_statfs;
 
         gf_boolean_t     added_link;
+        gf_boolean_t     is_linkfile;
 
         struct dht_skip_linkto_unlink  skip_unlink;
 
@@ -889,5 +891,7 @@ dht_unlock_inodelk (call_frame_t *frame, dht_lock_t **lk_array, int lk_count,
 dht_lock_t *
 dht_lock_new (xlator_t *this, xlator_t *xl, loc_t *loc, short type,
               const char *domain);
+void
+dht_lock_array_free (dht_lock_t **lk_array, int count);
 
 #endif/* _DHT_H */
diff --git a/xlators/cluster/dht/src/dht-rebalance.c b/xlators/cluster/dht/src/dht-rebalance.c
index 365099b9e0b..358d221c81c 100644
--- a/xlators/cluster/dht/src/dht-rebalance.c
+++ b/xlators/cluster/dht/src/dht-rebalance.c
@@ -858,19 +858,22 @@ int
 dht_migrate_file (xlator_t *this, loc_t *loc, xlator_t *from, xlator_t *to,
                   int flag)
 {
-        int             ret            = -1;
-        struct iatt     new_stbuf      = {0,};
-        struct iatt     stbuf          = {0,};
-        struct iatt     empty_iatt     = {0,};
-        ia_prot_t       src_ia_prot    = {0,};
-        fd_t           *src_fd         = NULL;
-        fd_t           *dst_fd         = NULL;
-        dict_t         *dict           = NULL;
-        dict_t         *xattr          = NULL;
-        dict_t         *xattr_rsp      = NULL;
-        int             file_has_holes = 0;
-        dht_conf_t     *conf           = this->private;
-        int            rcvd_enoent_from_src = 0;
+        int             ret                  = -1;
+        struct iatt     new_stbuf            = {0,};
+        struct iatt     stbuf                = {0,};
+        struct iatt     empty_iatt           = {0,};
+        ia_prot_t       src_ia_prot          = {0,};
+        fd_t           *src_fd               = NULL;
+        fd_t           *dst_fd               = NULL;
+        dict_t         *dict                 = NULL;
+        dict_t         *xattr                = NULL;
+        dict_t         *xattr_rsp            = NULL;
+        int             file_has_holes       = 0;
+        dht_conf_t     *conf                 = this->private;
+        int             rcvd_enoent_from_src = 0;
+        struct gf_flock flock                = {0, };
+        loc_t           tmp_loc              = {0, };
+        gf_boolean_t    locked               = _gf_false;
 
         gf_log (this->name, GF_LOG_INFO, "%s: attempting to move from %s to %s",
                 loc->path, from->name, to->name);
@@ -888,6 +891,25 @@ dht_migrate_file (xlator_t *this, loc_t *loc, xlator_t *from, xlator_t *to,
                 goto out;
         }
 
+        flock.l_type = F_WRLCK;
+
+        tmp_loc.inode = inode_ref (loc->inode);
+        uuid_copy (tmp_loc.gfid, loc->gfid);
+
+        ret = syncop_inodelk (from, DHT_FILE_MIGRATE_DOMAIN, &tmp_loc, F_SETLKW,
+                              &flock, NULL, NULL);
+        if (ret < 0) {
+                gf_msg (this->name, GF_LOG_WARNING, 0,
+                        DHT_MSG_MIGRATE_FILE_FAILED,
+                        "migrate file failed: "
+                        "%s: failed to lock file on %s (%s)",
+                        loc->path, from->name, strerror (-ret));
+                ret = -1;
+                goto out;
+        }
+
+        locked = _gf_true;
+
         /* Phase 1 - Data migration is in progress from now on */
         ret = syncop_lookup (from, loc, dict, &stbuf, &xattr_rsp, NULL);
         if (ret) {
@@ -1136,6 +1158,19 @@ dht_migrate_file (xlator_t *this, loc_t *loc, xlator_t *from, xlator_t *to,
 
         ret = 0;
 out:
+        if (locked) {
+                flock.l_type = F_UNLCK;
+
+                ret = syncop_inodelk (from, DHT_FILE_MIGRATE_DOMAIN, &tmp_loc,
+                                      F_SETLK, &flock, NULL, NULL);
+                if (ret < 0) {
+                        gf_msg (this->name, GF_LOG_WARNING, 0,
+                                DHT_MSG_MIGRATE_FILE_FAILED,
+                                "%s: failed to unlock file on %s (%s)",
+                                loc->path, from->name, strerror (-ret));
+                }
+        }
+
         if (dict)
                 dict_unref (dict);
 
@@ -1149,6 +1184,8 @@ out:
         if (src_fd)
                 syncop_close (src_fd);
 
+        loc_wipe (&tmp_loc);
+
         return ret;
 }
 
diff --git a/xlators/cluster/dht/src/dht-rename.c b/xlators/cluster/dht/src/dht-rename.c
index ce7df8442bf..10040a33958 100644
--- a/xlators/cluster/dht/src/dht-rename.c
+++ b/xlators/cluster/dht/src/dht-rename.c
@@ -355,26 +355,72 @@ err:
         }while (0)
 
 int
-dht_rename_done (call_frame_t *frame, xlator_t *this)
+dht_rename_unlock_cbk (call_frame_t *frame, void *cookie,
+                       xlator_t *this, int32_t op_ret, int32_t op_errno,
+                       dict_t *xdata)
 {
-        dht_local_t     *local = NULL;
+        dht_local_t *local = NULL;
 
         local = frame->local;
 
-        if (local->linked == _gf_true) {
-                local->linked = _gf_false;
-                dht_linkfile_attr_heal (frame, this);
-        }
         DHT_STRIP_PHASE1_FLAGS (&local->stbuf);
         DHT_STACK_UNWIND (rename, frame, local->op_ret, local->op_errno,
                           &local->stbuf, &local->preoldparent,
                           &local->postoldparent, &local->preparent,
                           &local->postparent, NULL);
+        return 0;
+}
+
+int
+dht_rename_unlock (call_frame_t *frame, xlator_t *this)
+{
+        dht_local_t *local                      = NULL;
+        int          op_ret                     = -1;
+        char         src_gfid[GF_UUID_BUF_SIZE] = {0};
+        char         dst_gfid[GF_UUID_BUF_SIZE] = {0};
+
+        local = frame->local;
+
+        op_ret = dht_unlock_inodelk (frame, local->lock.locks,
+                                     local->lock.lk_count,
+                                     dht_rename_unlock_cbk);
+        if (op_ret < 0) {
+                uuid_utoa_r (local->loc.inode->gfid, src_gfid);
+
+                if (local->loc2.inode)
+                        uuid_utoa_r (local->loc2.inode->gfid, dst_gfid);
+
+                gf_log (this->name, GF_LOG_WARNING,
+                        "winding unlock inodelk failed "
+                        "rename (%s:%s:%s %s:%s:%s), "
+                        "stale locks left on bricks",
+                        local->loc.path, src_gfid, local->src_cached->name,
+                        local->loc2.path, dst_gfid,
+                        local->dst_cached ? local->dst_cached->name : NULL);
+
+                dht_rename_unlock_cbk (frame, NULL, this, 0, 0, NULL);
+        }
 
         return 0;
 }
 
 int
+dht_rename_done (call_frame_t *frame, xlator_t *this)
+{
+        dht_local_t *local                      = NULL;
+
+        local = frame->local;
+
+        if (local->linked == _gf_true) {
+                local->linked = _gf_false;
+                dht_linkfile_attr_heal (frame, this);
+        }
+
+        dht_rename_unlock (frame, this);
+        return 0;
+}
+
+int
 dht_rename_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                        int32_t op_ret, int32_t op_errno, struct iatt *preparent,
                        struct iatt *postparent, dict_t *xdata)
@@ -418,14 +464,14 @@ out:
 int
 dht_rename_cleanup (call_frame_t *frame)
 {
-        dht_local_t *local = NULL;
-        xlator_t    *this = NULL;
-        xlator_t    *src_hashed = NULL;
-        xlator_t    *src_cached = NULL;
-        xlator_t    *dst_hashed = NULL;
-        xlator_t    *dst_cached = NULL;
-        int          call_cnt = 0;
-        dict_t      *xattr      = NULL;
+        dht_local_t *local                  = NULL;
+        xlator_t    *this                   = NULL;
+        xlator_t    *src_hashed             = NULL;
+        xlator_t    *src_cached             = NULL;
+        xlator_t    *dst_hashed             = NULL;
+        xlator_t    *dst_cached             = NULL;
+        int          call_cnt               = 0;
+        dict_t      *xattr                  = NULL;
         char         gfid[GF_UUID_BUF_SIZE] = {0};
 
         local = frame->local;
@@ -513,12 +559,7 @@ nolinks:
         WIPE (&local->preparent);
         WIPE (&local->postparent);
 
-        DHT_STRIP_PHASE1_FLAGS (&local->stbuf);
-        DHT_STACK_UNWIND (rename, frame, local->op_ret, local->op_errno,
-                          &local->stbuf, &local->preoldparent,
-                          &local->postoldparent, &local->preparent,
-                          &local->postparent, NULL);
-
+        dht_rename_unlock (frame, this);
         return 0;
 }
 
@@ -954,6 +995,181 @@ nolinks:
         return 0;
 }
 
+int
+dht_rename_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                       int op_ret, int op_errno,
+                       inode_t *inode, struct iatt *stbuf, dict_t *xattr,
+                       struct iatt *postparent)
+{
+        dht_local_t *local                      = NULL;
+        int          call_cnt                   = 0;
+        dht_conf_t  *conf                       = NULL;
+
+        local = frame->local;
+        conf = this->private;
+
+        if (op_ret < 0) {
+                /* The meaning of is_linkfile is overloaded here. For locking
+                 * to work properly both rebalance and rename should acquire
+                 * lock on datafile. The reason for sending this lookup is to
+                 * find out whether we've acquired a lock on data file.
+                 * Between the lookup before rename and this rename, the
+                 * file could be migrated by a rebalance process and now this
+                 * file this might be a linkto file. We verify that by sending
+                 * this lookup. However, if this lookup fails we cannot really
+                 * say whether we've acquired lock on a datafile or linkto file.
+                 * So, we act conservatively and _assume_
+                 * that this is a linkfile and fail the rename operation.
+                 */
+                local->is_linkfile = _gf_true;
+        } else if (xattr && check_is_linkfile (inode, stbuf, xattr,
+                                               conf->link_xattr_name)) {
+                local->is_linkfile = _gf_true;
+        }
+
+        call_cnt = dht_frame_return (frame);
+        if (is_last_call (call_cnt)) {
+                if (local->is_linkfile) {
+                        local->op_ret = -1;
+                        local->op_errno = EBUSY;
+                        goto fail;
+                }
+
+                dht_rename_create_links (frame);
+        }
+
+        return 0;
+fail:
+        dht_rename_unlock (frame, this);
+        return 0;
+}
+
+int32_t
+dht_rename_lock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                     int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t *local                      = NULL;
+        char         src_gfid[GF_UUID_BUF_SIZE] = {0};
+        char         dst_gfid[GF_UUID_BUF_SIZE] = {0};
+        dict_t      *xattr_req                  = NULL;
+        dht_conf_t  *conf                       = NULL;
+        int          i                          = 0;
+
+        local = frame->local;
+        conf = this->private;
+
+        if (op_ret < 0) {
+                uuid_utoa_r (local->loc.inode->gfid, src_gfid);
+
+                if (local->loc2.inode)
+                        uuid_utoa_r (local->loc2.inode->gfid, dst_gfid);
+
+                gf_log (this->name, GF_LOG_WARNING,
+                        "acquiring inodelk failed (%s) "
+                        "rename (%s:%s:%s %s:%s:%s), returning EBUSY",
+                        strerror (op_errno),
+                        local->loc.path, src_gfid, local->src_cached->name,
+                        local->loc2.path, dst_gfid,
+                        local->dst_cached ? local->dst_cached->name : NULL);
+
+                local->op_ret = -1;
+                local->op_errno = (op_errno == EAGAIN) ? EBUSY : op_errno;
+
+                goto done;
+        }
+
+        xattr_req = dict_new ();
+        if (xattr_req == NULL) {
+                local->op_ret = -1;
+                local->op_errno = ENOMEM;
+                goto done;
+        }
+
+        op_ret = dict_set_uint32 (xattr_req,
+                                  conf->link_xattr_name, 256);
+        if (op_ret < 0) {
+                local->op_ret = -1;
+                local->op_errno = -op_ret;
+                goto done;
+        }
+
+        local->call_cnt = local->lock.lk_count;
+
+        for (i = 0; i < local->lock.lk_count; i++) {
+                STACK_WIND (frame, dht_rename_lookup_cbk,
+                            local->lock.locks[i]->xl,
+                            local->lock.locks[i]->xl->fops->lookup,
+                            &local->lock.locks[i]->loc, xattr_req);
+        }
+
+        dict_unref (xattr_req);
+        return 0;
+
+done:
+        /* Its fine to call unlock even when no locks are acquired, as we check
+         * for lock->locked before winding a unlock call.
+         */
+        dht_rename_unlock (frame, this);
+
+        if (xattr_req)
+                dict_unref (xattr_req);
+
+        return 0;
+}
+
+int
+dht_rename_lock (call_frame_t *frame)
+{
+        dht_local_t  *local    = NULL;
+        int           count    = 1, ret = -1;
+        dht_lock_t  **lk_array = NULL;
+
+        local = frame->local;
+
+        if (local->dst_cached)
+                count++;
+
+        lk_array = GF_CALLOC (count, sizeof (*lk_array), gf_common_mt_char);
+        if (lk_array == NULL)
+                goto err;
+
+        lk_array[0] = dht_lock_new (frame->this, local->src_cached, &local->loc,
+                                    F_WRLCK, DHT_FILE_MIGRATE_DOMAIN);
+        if (lk_array[0] == NULL)
+                goto err;
+
+        if (local->dst_cached) {
+                lk_array[1] = dht_lock_new (frame->this, local->dst_cached,
+                                            &local->loc2, F_WRLCK,
+                                            DHT_FILE_MIGRATE_DOMAIN);
+                if (lk_array[1] == NULL)
+                        goto err;
+        }
+
+        local->lock.locks = lk_array;
+        local->lock.lk_count = count;
+
+        ret = dht_nonblocking_inodelk (frame, lk_array, count,
+                                       dht_rename_lock_cbk);
+        if (ret < 0) {
+                local->lock.locks = NULL;
+                local->lock.lk_count = 0;
+                goto err;
+        }
+
+        return 0;
+err:
+        if (lk_array != NULL) {
+                int tmp_count = 0, i = 0;
+
+                for (i = 0; (i < count) && (lk_array[i]); i++, tmp_count++);
+
+                dht_lock_array_free (lk_array, tmp_count);
+                GF_FREE (lk_array);
+        }
+
+        return -1;
+}
 
 int
 dht_rename (call_frame_t *frame, xlator_t *this,
@@ -1040,7 +1256,9 @@ dht_rename (call_frame_t *frame, xlator_t *this,
                 dht_rename_dir (frame, this);
         } else {
                 local->op_ret = 0;
-                dht_rename_create_links (frame);
+                ret = dht_rename_lock (frame);
+                if (ret < 0)
+                        goto err;
         }
 
         return 0;