dht-rename: Unlink older link files before creating new one

If a older link file exists, unlink it and then create the linkfile.
This will prevent mis-match of gfid's.

Signed-off-by: shishir gowda <shishirng@gluster.com>
Signed-off-by: Anand Avati <avati@gluster.com>

BUG: 2464 ([7b07d444a77526f27f860210930bf1d4c7fbea9b]: rm -rf gives Invalid argumenrt error)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=2464

3 files changed, 150 insertions, 2 deletions

diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index bc84098b7ae..99b626c1695 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -309,4 +309,12 @@ 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);
 
+int dht_rename_cleanup (call_frame_t *frame);
+int dht_rename_links_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                      int32_t op_ret, int32_t op_errno,
+                      inode_t *inode, struct iatt *stbuf,
+                      struct iatt *preparent, struct iatt *postparent);
+
+int dht_linkfile_recreate(call_frame_t *frame, fop_mknod_cbk_t linkfile_cbk,
+                         xlator_t *tovol, xlator_t *fromvol, loc_t *loc);
 #endif /* _DHT_H */
diff --git a/xlators/cluster/dht/src/dht-linkfile.c b/xlators/cluster/dht/src/dht-linkfile.c
index 7acebd290e3..82419a68bc1 100644
--- a/xlators/cluster/dht/src/dht-linkfile.c
+++ b/xlators/cluster/dht/src/dht-linkfile.c
@@ -242,3 +242,143 @@ dht_linkfile_subvol (xlator_t *this, inode_t *inode, struct iatt *stbuf,
 out:
         return subvol;
 }
+
+int
+dht_recreate_linkfile_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)
+{
+        dht_local_t   *local            = NULL;
+        xlator_t      *subvol           = NULL;
+        xlator_t      *src_cached       = NULL;
+        xlator_t      *dst_hashed       = NULL;
+        call_frame_t  *prev             = NULL;
+        loc_t         *loc              = NULL;
+
+        GF_VALIDATE_OR_GOTO ("dht", frame, err);
+        GF_VALIDATE_OR_GOTO ("dht", this, err);
+        GF_VALIDATE_OR_GOTO ("dht", frame->local, err);
+
+        local = frame->local;
+        prev = cookie;
+        subvol = prev->this;
+        src_cached = local->src_cached;
+        dst_hashed = local->dst_hashed;
+        loc = &local->loc;
+
+        if (!src_cached || !dst_hashed) {
+                gf_log (this->name, GF_LOG_ERROR, "src_cached or dst_hashed"
+                        "subvol is null");
+                local->op_ret = -1;
+                local->op_errno = EINVAL;
+                goto out;
+        }
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "unlinking linkfile %s on %s failed (%s)",
+                        loc->path, subvol->name, strerror (op_errno));
+                local->op_ret = op_ret;
+                local->op_errno = op_errno;
+                goto out;
+        }
+        gf_log (this->name, GF_LOG_DEBUG, "unlink successfull. Proceeding with"
+                " creation of link file %s", loc->path);
+        dht_linkfile_create (frame, local->linkfile.linkfile_cbk,
+                             src_cached, dst_hashed, loc);
+
+        return 0;
+out:
+        local->linkfile.linkfile_cbk (frame, NULL, frame->this, -1, EINVAL,
+                                      local->loc.inode, NULL, NULL, NULL);
+err:
+        return -1;
+}
+
+int
+dht_linkfile_recreate_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;
+        xlator_t     *src_cached    = NULL;
+        xlator_t     *dst_hashed    = NULL;
+        loc_t        *loc           = NULL;
+
+        GF_VALIDATE_OR_GOTO ("dht", frame, err);
+        GF_VALIDATE_OR_GOTO ("dht", this, err);
+        GF_VALIDATE_OR_GOTO ("dht", frame->local, err);
+
+        local = frame->local;
+        loc = &local->loc;
+
+        src_cached = local->src_cached;
+        dst_hashed = local->dst_hashed;
+
+        if (!src_cached || !dst_hashed) {
+                gf_log (this->name, GF_LOG_ERROR, "src_cached or dst_hashed"
+                        " or hashed_subvol is null");
+                local->op_ret = -1;
+                local->op_errno = EINVAL;
+                goto out;
+        }
+
+        /* if link_file exists, remove it, else create it*/
+        if (!op_ret) {
+               if (!check_is_linkfile (inode, stbuf, xattr))
+                       goto out;
+               gf_log (this->name, GF_LOG_DEBUG, "link file exists."
+                        " Calling unlink on %s", loc->path);
+                STACK_WIND (frame, dht_recreate_linkfile_unlink_cbk,
+                            dst_hashed, dst_hashed->fops->unlink,
+                            loc);
+
+        } else if (op_errno == ENOENT) {
+                gf_log (this->name, GF_LOG_DEBUG, "link file does not exist."
+                        " Proceeding to creation of linkfile %s", loc->path);
+                dht_linkfile_create (frame, local->linkfile.linkfile_cbk,
+                                     src_cached, dst_hashed, loc);
+        } else {
+                gf_log (this->name, GF_LOG_ERROR, "returned error %s",
+                        strerror(op_errno));
+                local->op_ret = op_ret;
+                local->op_errno = op_errno;
+                goto out;
+        }
+
+        return 0;
+out:
+        local->linkfile.linkfile_cbk (frame, NULL, frame->this, -1, EINVAL,
+                                      loc->inode, NULL, NULL, NULL);
+err:
+        return -1;
+}
+
+int
+dht_linkfile_recreate (call_frame_t *frame, fop_mknod_cbk_t linkfile_cbk,
+                       xlator_t *tovol, xlator_t *fromvol, loc_t *loc)
+{
+        dht_local_t     *local          = NULL;
+
+        GF_VALIDATE_OR_GOTO ("dht", frame, out);
+        GF_VALIDATE_OR_GOTO ("dht", loc, out);
+        GF_VALIDATE_OR_GOTO ("dht", tovol, out);
+        GF_VALIDATE_OR_GOTO ("dht", fromvol, out);
+
+        local = frame->local;
+        local->linkfile.linkfile_cbk = linkfile_cbk;
+        local->linkfile.srcvol = tovol;
+        loc_copy (&local->linkfile.loc, loc);
+
+        STACK_WIND (frame, dht_linkfile_recreate_cbk,
+                    fromvol, fromvol->fops->lookup, loc, NULL);
+
+        return 0;
+
+out:
+        linkfile_cbk (frame, NULL, frame->this, -1, EINVAL, loc->inode, NULL,
+                      NULL, NULL);
+
+        return -1;
+}
diff --git a/xlators/cluster/dht/src/dht-rename.c b/xlators/cluster/dht/src/dht-rename.c
index f6ed8769d8d..cef4a7b6fbc 100644
--- a/xlators/cluster/dht/src/dht-rename.c
+++ b/xlators/cluster/dht/src/dht-rename.c
@@ -593,8 +593,8 @@ dht_rename_create_links (call_frame_t *frame)
                         "linkfile %s @ %s => %s",
                         local->loc.path, dst_hashed->name, src_cached->name);
                 memcpy (local->gfid, local->loc.inode->gfid, 16);
-                dht_linkfile_create (frame, dht_rename_links_cbk,
-                                     src_cached, dst_hashed, &local->loc);
+                dht_linkfile_recreate (frame, dht_rename_links_cbk,
+                                       src_cached, dst_hashed, &local->loc);
         }
 
         if (src_cached != dst_hashed) {