dht/tier :rename fails with EBUSY

When the files was in hot tier and the look up was done already, then
hashed and cached subvolume will be hot-tier. Once the file is moved
from hot-tier to cold-tier, then subsequent lookup will send a
revalidate lookup to hot-tier and it will find out that the file was
actually moved and there is only link in the cached subvolume. So dht
will return an ESTALE to fuse. Upon receiving ESTALE for a lookup, fuse
will create a new inode and sent a fresh lookup. This lookup will be
successful, and it will locate the file properly. Then fuse try to link
the inode, but the older inode was already there in inmemory inode cache
with same gfid and that is also shared with fuse kernal. So inode_link
will return the older ionode itself. So the subsequent rename fop will
come to gluster with the older inode. From dht_rename, we will take a
lock on the inode and after successful inodelk on inode dht will send
lookup before creating a link. this lookup will again find out that the
file is a link file, and then dht will think that file is
migrating/migrated in the mean time, and will send EBUSY.

Back port of :
>Change-Id: Ib3a01e5b1d7f64514b04bb6234026d049f082679
>BUG: 1248306
>Signed-off-by: Mohammed Rafi KC <rkavunga@redhat.com>
>Reviewed-on: http://review.gluster.org/11768
>Tested-by: Gluster Build System <jenkins@build.gluster.com>
>Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
>Reviewed-by: Dan Lambright <dlambrig@redhat.com>
>Tested-by: NetBSD Build System <jenkins@build.gluster.org>
>Tested-by: Dan Lambright <dlambrig@redhat.com>

(cherry picked from commit 0ad26041fbf65ab36856a0ad178c32e51bf87319)

Change-Id: I1278a2c2ccc2cadcbe147db836f0526f079f6038
BUG: 1254437
Signed-off-by: Mohammed Rafi KC <rkavunga@redhat.com>
Reviewed-on: http://review.gluster.org/11944
Tested-by: NetBSD Build System <jenkins@build.gluster.org>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Dan Lambright <dlambrig@redhat.com>
Tested-by: Dan Lambright <dlambrig@redhat.com>

1 files changed, 20 insertions, 8 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 2a857f27021..da36a3b33a5 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -678,10 +678,12 @@ dht_revalidate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         int           ret  = -1;
         int           is_dir = 0;
         int           is_linkfile = 0;
+        int           follow_link = 0;
         call_frame_t *copy          = NULL;
         dht_local_t  *copy_local    = NULL;
         char gfid[GF_UUID_BUF_SIZE] = {0};
         uint32_t      vol_commit_hash = 0;
+        xlator_t      *subvol = NULL;
 
         GF_VALIDATE_OR_GOTO ("dht", frame, err);
         GF_VALIDATE_OR_GOTO ("dht", this, err);
@@ -772,17 +774,10 @@ dht_revalidate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 is_dir = check_is_dir (inode, stbuf, xattr);
                 is_linkfile = check_is_linkfile (inode, stbuf, xattr,
                                                  conf->link_xattr_name);
-
                 if (is_linkfile) {
-                        gf_msg (this->name, GF_LOG_INFO, 0,
-                                DHT_MSG_REVALIDATE_CBK_INFO,
-                                "Revalidate: linkfile found %s, (gfid = %s)",
-                                local->loc.path, gfid);
-                        local->return_estale = 1;
-
+                        follow_link = 1;
                         goto unlock;
                 }
-
                 if (is_dir) {
                         ret = dht_dir_has_layout (xattr, conf->xattr_name);
                         if (ret >= 0) {
@@ -830,6 +825,23 @@ dht_revalidate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         }
 unlock:
         UNLOCK (&frame->lock);
+
+        if (follow_link) {
+                gf_uuid_copy (local->gfid, stbuf->ia_gfid);
+
+                subvol = dht_linkfile_subvol (this, inode, stbuf, xattr);
+                if (!subvol) {
+                        op_errno = ESTALE;
+                        local->op_ret = -1;
+                } else {
+
+                        STACK_WIND (frame, dht_lookup_linkfile_cbk,
+                        subvol, subvol->fops->lookup,
+                        &local->loc, local->xattr_req);
+                        return 0;
+                }
+        }
+
 out:
         this_call_cnt = dht_frame_return (frame);