features/shard: Fill inode ctx in readdir(p) callback too

The only place where shard translator was initialising inode ctx
was lookup callback. But if the inodes are created and linked through
readdirp, shard_lookup() path _may_ not be exercised before FUSE
winds other fops on them. Since shard translator does an
inode_ctx_get() first thing in most fops, an uninitialised ctx could
cause it to fail the operation with ENOMEM.

The solution would be to also initialise inode ctx if it has not been
done already in readdir(p) callback.

Change-Id: I3e058cd2a29bc6a69a96aaac89165c3251315625
BUG: 1250855
Signed-off-by: Krutika Dhananjay <kdhananj@redhat.com>
Reviewed-on: http://review.gluster.org/11854
Tested-by: NetBSD Build System <jenkins@build.gluster.org>
Reviewed-by: Pranith Kumar Karampuri <pkarampu@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>

2 files changed, 79 insertions, 26 deletions

diff --git a/tests/bugs/shard/bug-1250855.t b/tests/bugs/shard/bug-1250855.t
new file mode 100644
index 00000000000..b8bc3b42513
--- /dev/null
+++ b/tests/bugs/shard/bug-1250855.t
@@ -0,0 +1,34 @@
+#!/bin/bash
+
+. $(dirname $0)/../../include.rc
+
+cleanup;
+
+TESTS_EXPECTED_IN_LOOP=40
+
+TEST glusterd
+TEST pidof glusterd
+TEST $CLI volume create $V0 $H0:$B0/${V0}{0,1}
+TEST $CLI volume start $V0
+
+TEST $GFS --volfile-id=$V0 --volfile-server=$H0 $M0
+TEST mkdir $M0/dir
+
+for i in {1..20}; do
+        TEST_IN_LOOP touch $M0/dir/$i;
+done
+
+TEST $CLI volume set $V0 features.shard on
+
+TEST ls $M0
+TEST ls $M0/dir
+
+for i in {1..10}; do
+        TEST_IN_LOOP mv $M0/dir/$i $M0/dir/$i-sharded;
+done
+
+for i in {11..20}; do
+        TEST_IN_LOOP unlink $M0/dir/$i;
+done
+
+cleanup;
diff --git a/xlators/features/shard/src/shard.c b/xlators/features/shard/src/shard.c
index 8c9f9f8978e..52b95472ba7 100644
--- a/xlators/features/shard/src/shard.c
+++ b/xlators/features/shard/src/shard.c
@@ -550,18 +550,35 @@ err:
         return 0;
 }
 
-int
-shard_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-                  int32_t op_ret, int32_t op_errno, inode_t *inode,
-                  struct iatt *buf, dict_t *xdata, struct iatt *postparent)
+static void
+shard_inode_ctx_set_if_absent (inode_t *inode, xlator_t *this, dict_t *xdata,
+                               struct iatt *buf)
 {
         int                ret           = 0;
         uint64_t           size          = 0;
         void              *bsize         = NULL;
-        void              *size_attr     = NULL;
         shard_inode_ctx_t  ctx_tmp       = {0,};
-        uint64_t           size_array[4];
 
+        if (shard_inode_ctx_get_block_size (inode, this, &size)) {
+                ret = dict_get_ptr (xdata, GF_XATTR_SHARD_BLOCK_SIZE, &bsize);
+                if (!ret) {
+                        ctx_tmp.block_size = ntoh64 (*((uint64_t *)bsize));
+                        ctx_tmp.mode = st_mode_from_ia (buf->ia_prot,
+                                                        buf->ia_type);
+                        ctx_tmp.rdev = buf->ia_rdev;
+                }
+                ret = shard_inode_ctx_set_all (inode, this, &ctx_tmp);
+                if (ret)
+                        gf_log (this->name, GF_LOG_WARNING, "Failed to set "
+                                "inode ctx for %s", uuid_utoa (buf->ia_gfid));
+        }
+}
+
+int
+shard_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                  int32_t op_ret, int32_t op_errno, inode_t *inode,
+                  struct iatt *buf, dict_t *xdata, struct iatt *postparent)
+{
         if (op_ret < 0)
                 goto unwind;
 
@@ -576,31 +593,15 @@ shard_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
          *    (which are anyway don't cares) in inode ctx. Since @ctx_tmp is
          *    already initialised to all zeroes, nothing more needs to be done.
          */
-        if (shard_inode_ctx_get_block_size (inode, this, &size)) {
-                ret = dict_get_ptr (xdata, GF_XATTR_SHARD_BLOCK_SIZE, &bsize);
-                if (!ret) {
-                        ctx_tmp.block_size = ntoh64 (*((uint64_t *)bsize));
-                        ctx_tmp.mode = st_mode_from_ia (buf->ia_prot,
-                                                        buf->ia_type);
-                        ctx_tmp.rdev = buf->ia_rdev;
-                }
-                ret = shard_inode_ctx_set_all (inode, this, &ctx_tmp);
-                if (ret)
-                        gf_log (this->name, GF_LOG_WARNING, "Failed to set "
-                                "inode ctx for %s", uuid_utoa (buf->ia_gfid));
-        }
+
+        (void) shard_inode_ctx_set_if_absent (inode, this, xdata, buf);
 
         /* Also, if the file is sharded, get the file size and block cnt xattr,
          * and store them in the stbuf appropriately.
          */
 
-        ret = dict_get_ptr (xdata, GF_XATTR_SHARD_FILE_SIZE, &size_attr);
-        if (!ret) {
-                memcpy (size_array, size_attr, sizeof (size_array));
-
-                buf->ia_size = ntoh64 (size_array[0]);
-                buf->ia_blocks = ntoh64 (size_array[2]);
-        }
+        if (dict_get (xdata, GF_XATTR_SHARD_FILE_SIZE))
+                shard_modify_size_and_block_count (buf, xdata);
 
 unwind:
         SHARD_STACK_UNWIND (lookup, frame, op_ret, op_errno, inode, buf,
@@ -3403,6 +3404,12 @@ shard_readdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 if (dict_get (entry->dict, GF_XATTR_SHARD_FILE_SIZE))
                         shard_modify_size_and_block_count (&entry->d_stat,
                                                            entry->dict);
+
+                if (!entry->inode)
+                        continue;
+
+                shard_inode_ctx_set_if_absent (entry->inode, this, entry->dict,
+                                               &entry->d_stat);
         }
 
         local->op_ret = op_ret;
@@ -3438,6 +3445,7 @@ int
 shard_readdir_do (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                off_t offset, int whichop, dict_t *xdata)
 {
+        int             ret      = 0;
         shard_local_t  *local    = NULL;
 
         local = mem_get0 (this->local_pool);
@@ -3463,6 +3471,17 @@ shard_readdir_do (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                 local->xattr_req = (xdata) ? dict_ref (xdata) : dict_new ();
                 SHARD_MD_READ_FOP_INIT_REQ_DICT (this, local->xattr_req,
                                                  fd->inode->gfid, local, err);
+                ret = dict_set_uint64 (local->xattr_req,
+                                       GF_XATTR_SHARD_BLOCK_SIZE, 0);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_WARNING, "Failed to set "
+                                "dict value: key:%s, directory gfid=%s",
+                                GF_XATTR_SHARD_BLOCK_SIZE,
+                                uuid_utoa (fd->inode->gfid));
+                        local->op_ret = -1;
+                        local->op_errno = ENOMEM;
+                        goto err;
+                }
 
                 STACK_WIND (frame, shard_readdir_cbk, FIRST_CHILD(this),
                             FIRST_CHILD(this)->fops->readdirp, fd, size, offset,