storage/posix: send proper iatt attributes for the root inode

* changes in posix to send proper iatt attributes for the root directory
  when ancestry is built. Before posix was filling only the gfid and the
  inode type in the iatt structure keeping rest of the fields zeros. This
  was cached by posix-acl and used to send EACCES when some fops came on
  that object if the uid of the caller is same as the uid of the object on
  the disk.

* getting and setting inode_ctx in function 'posix_acl_ctx_get' is not atomic
  and can lead to memory leak when there are multiple looups for an
  inode at same time. This patch fix this problem

* Linking an inode in posix_build_ancestry, can cause a race in
  posix_acl.
  When parent inode is linked in posix_build_ancestry, and before
  it reaches posix_acl_readdirp_cbkc, reate/lookup can
  come on a leaf-inode, as parent-inode-ctx not yet updated
  in posix_acl_readdirp_cbk, create/lookup can fail
  with EACCESS. So do the inode linking in the quota xlator

Change-Id: I3101eefb65551cc4162c4ff2963be1b73deacd6d
BUG: 1320818
Signed-off-by: Raghavendra Bhat <raghavendra@redhat.com>
Reviewed-on: http://review.gluster.org/13730
Tested-by: Vijaikumar Mallikarjuna <vmallika@redhat.com>
Smoke: Gluster Build System <jenkins@build.gluster.com>
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
CentOS-regression: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>

6 files changed, 149 insertions, 70 deletions

diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index 48a275e2783..f4fdf2e415b 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -2982,38 +2982,17 @@ marker_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                            int op_ret, int op_errno, gf_dirent_t *entries,
                            dict_t *xdata)
 {
-        gf_dirent_t *entry = NULL;
-        loc_t        loc    = {0, };
-        inode_t     *parent = NULL;
-        int          ret    = -1;
+        gf_dirent_t        *entry  = NULL;
+        quota_inode_ctx_t  *ctx    = NULL;
+        int                 ret    = -1;
 
         if ((op_ret <= 0) || (entries == NULL)) {
                 goto out;
         }
 
         list_for_each_entry (entry, &entries->list, list) {
-                if (entry->inode == entry->inode->table->root) {
-                        inode_unref (parent);
-                        parent = NULL;
-                }
-
-                if (parent)
-                        _marker_inode_loc_fill (entry->inode, parent,
-                                                entry->d_name, &loc);
-                else
-                        ret = marker_inode_loc_fill (entry->inode, &loc);
-
-                if (ret) {
-                        gf_log (this->name, GF_LOG_WARNING, "Couldn't build "
-                                "loc for %s/%s",
-                                parent? uuid_utoa (parent->gfid): NULL,
-                                entry->d_name);
+                if (entry->inode == NULL)
                         continue;
-                }
-
-                inode_unref (parent);
-                parent = inode_ref (entry->inode);
-                loc_wipe (&loc);
 
                 ret = marker_key_set_ver (this, entry->dict);
                 if (ret < 0) {
@@ -3021,10 +3000,13 @@ marker_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         op_errno = ENOMEM;
                         break;
                 }
-        }
 
-        if (parent)
-                inode_unref (parent);
+                ctx = mq_inode_ctx_new (entry->inode, this);
+                if (ctx == NULL)
+                        gf_log (this->name, GF_LOG_WARNING, "mq_inode_ctx_new "
+                                "failed for %s",
+                                uuid_utoa (entry->inode->gfid));
+        }
 
 out:
         STACK_UNWIND_STRICT (readdirp, frame, op_ret, op_errno, entries, xdata);
diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index 105d70cbea6..847c93f37c0 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -737,13 +737,17 @@ quota_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                           int32_t op_ret, int32_t op_errno,
                           gf_dirent_t *entries, dict_t *xdata)
 {
-        inode_t           *parent  = NULL, *tmp_parent = NULL;
-        gf_dirent_t       *entry   = NULL;
-        loc_t              loc     = {0, };
-        quota_dentry_t    *dentry  = NULL, *tmp = NULL;
-        quota_inode_ctx_t *ctx     = NULL;
-        struct list_head   parents = {0, };
-        quota_local_t     *local   = NULL;
+        inode_t           *parent       = NULL;
+        inode_t           *tmp_parent   = NULL;
+        inode_t           *linked_inode = NULL;
+        inode_t           *tmp_inode    = NULL;
+        gf_dirent_t       *entry        = NULL;
+        loc_t              loc          = {0, };
+        quota_dentry_t    *dentry       = NULL;
+        quota_dentry_t    *tmp          = NULL;
+        quota_inode_ctx_t *ctx          = NULL;
+        struct list_head   parents      = {0, };
+        quota_local_t     *local        = NULL;
 
         INIT_LIST_HEAD (&parents);
 
@@ -753,14 +757,6 @@ quota_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (op_ret < 0)
                 goto err;
 
-        parent = inode_parent (local->loc.inode, 0, NULL);
-        if (parent == NULL) {
-                gf_msg (this->name, GF_LOG_WARNING, EINVAL,
-                        Q_MSG_PARENT_NULL, "parent is NULL");
-                op_errno = EINVAL;
-                goto err;
-        }
-
         if ((op_ret > 0) && (entries != NULL)) {
                 list_for_each_entry (entry, &entries->list, list) {
                         if (__is_root_gfid (entry->inode->gfid)) {
@@ -776,6 +772,23 @@ quota_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                                  */
 
                                 tmp_parent = NULL;
+                        } else {
+                                /* For a non-root entry, link this inode */
+                                linked_inode = inode_link (entry->inode,
+                                                           tmp_parent,
+                                                           entry->d_name,
+                                                           &entry->d_stat);
+                                if (linked_inode) {
+                                        tmp_inode = entry->inode;
+                                        entry->inode = linked_inode;
+                                        inode_unref (tmp_inode);
+                                } else {
+                                        gf_msg (this->name, GF_LOG_WARNING,
+                                                EINVAL, Q_MSG_PARENT_NULL,
+                                                "inode link failed");
+                                                op_errno = EINVAL;
+                                                goto err;
+                                }
                         }
 
                         gf_uuid_copy (loc.gfid, entry->d_stat.ia_gfid);
@@ -793,6 +806,14 @@ quota_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 }
         }
 
+        parent = inode_parent (local->loc.inode, 0, NULL);
+        if (parent == NULL) {
+                gf_msg (this->name, GF_LOG_WARNING, EINVAL,
+                        Q_MSG_PARENT_NULL, "parent is NULL");
+                op_errno = EINVAL;
+                goto err;
+        }
+
         quota_inode_ctx_get (local->loc.inode, this, &ctx, 0);
 
         quota_add_parents_from_ctx (ctx, &parents);
diff --git a/xlators/storage/posix/src/posix-handle.c b/xlators/storage/posix/src/posix-handle.c
index 7f5cd5237ae..8e561e4a76f 100644
--- a/xlators/storage/posix/src/posix-handle.c
+++ b/xlators/storage/posix/src/posix-handle.c
@@ -30,24 +30,42 @@ inode_t *
 posix_resolve (xlator_t *this, inode_table_t *itable, inode_t *parent,
                char *bname, struct iatt *iabuf)
 {
-        inode_t     *inode = NULL, *linked_inode = NULL;
+        inode_t     *inode = NULL;
         int          ret   = -1;
 
         ret = posix_istat (this, parent->gfid, bname, iabuf);
-        if (ret < 0)
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING, "gfid: %s, bname: %s "
+                        "failed", uuid_utoa (parent->gfid), bname);
                 goto out;
+        }
 
-        inode = inode_find (itable, iabuf->ia_gfid);
-        if (inode == NULL) {
-                inode = inode_new (itable);
+        if (__is_root_gfid (iabuf->ia_gfid) && !strcmp (bname, "/")) {
+                inode = itable->root;
+        } else {
+                inode = inode_find (itable, iabuf->ia_gfid);
+                if (inode == NULL) {
+                        inode = inode_new (itable);
+                        gf_uuid_copy (inode->gfid, iabuf->ia_gfid);
+                }
         }
 
-        linked_inode = inode_link (inode, parent, bname, iabuf);
+        /* Linking an inode here, can cause a race in posix_acl.
+           Parent inode gets linked here, but before
+           it reaches posix_acl_readdirp_cbk, create/lookup can
+           come on a leaf-inode, as parent-inode-ctx not yet updated
+           in posix_acl_readdirp_cbk, create and lookup can fail
+           with EACCESS. So do the inode linking in the quota xlator
+
+        if (__is_root_gfid (iabuf->ia_gfid) && !strcmp (bname, "/"))
+                linked_inode = itable->root;
+        else
+                linked_inode = inode_link (inode, parent, bname, iabuf);
 
-        inode_unref (inode);
+        inode_unref (inode);*/
 
 out:
-        return linked_inode;
+        return inode;
 }
 
 int
@@ -67,6 +85,8 @@ posix_make_ancestral_node (const char *priv_base_path, char *path, int pathsize,
         }
 
         strcat (path, dir_name);
+        if (*dir_name != '/')
+                strcat (path, "/");
 
         if (type & POSIX_ANCESTRY_DENTRY) {
                 entry = gf_dirent_for_name (dir_name);
@@ -127,11 +147,16 @@ posix_make_ancestryfromgfid (xlator_t *this, char *path, int pathsize,
                         *parent = inode_ref (itable->root);
                 }
 
-                inode = itable->root;
 
-                memset (&iabuf, 0, sizeof (iabuf));
-                gf_uuid_copy (iabuf.ia_gfid, inode->gfid);
-                iabuf.ia_type = inode->ia_type;
+                inode = posix_resolve (this, itable, *parent, "/", &iabuf);
+                if (!inode) {
+                        gf_msg (this->name, GF_LOG_ERROR,
+                                P_MSG_INODE_RESOLVE_FAILED, 0,
+                                "posix resolve on the root inode %s failed",
+                                uuid_utoa (gfid));
+                        *op_errno = ESTALE;
+                        goto out;
+                }
 
                 ret = posix_make_ancestral_node (priv_base_path, path, pathsize,
                                                  head, "/", &iabuf, inode, type,
@@ -177,12 +202,14 @@ posix_make_ancestryfromgfid (xlator_t *this, char *path, int pathsize,
 
         inode = posix_resolve (this, itable, *parent, dir_name, &iabuf);
         if (inode == NULL) {
+                gf_msg (this->name, GF_LOG_ERROR, P_MSG_INODE_RESOLVE_FAILED,
+                        0, "posix resolve on the root inode %s failed",
+                        uuid_utoa (gfid));
                 *op_errno = ESTALE;
                 ret = -1;
                 goto out;
         }
 
-        strcat (dir_name, "/");
         ret = posix_make_ancestral_node (priv_base_path, path, pathsize, head,
                                          dir_name, &iabuf, inode, type, xdata);
         if (*parent != NULL) {
diff --git a/xlators/storage/posix/src/posix-messages.h b/xlators/storage/posix/src/posix-messages.h
index ad16c1dc00c..4efdef0a6b9 100644
--- a/xlators/storage/posix/src/posix-messages.h
+++ b/xlators/storage/posix/src/posix-messages.h
@@ -45,7 +45,7 @@
  */
 
 #define POSIX_COMP_BASE         GLFS_MSGID_COMP_POSIX
-#define GLFS_NUM_MESSAGES       107
+#define GLFS_NUM_MESSAGES       108
 #define GLFS_MSGID_END          (POSIX_COMP_BASE + GLFS_NUM_MESSAGES + 1)
 /* Messaged with message IDs */
 #define glfs_msg_start_x POSIX_COMP_BASE, "Invalid: Start of messages"
@@ -917,6 +917,14 @@
  *
  */
 
+#define P_MSG_INODE_RESOLVE_FAILED              (POSIX_COMP_BASE + 108)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+
 /*------------*/
 #define glfs_msg_end_x GLFS_MSGID_END, "Invalid: End of messages"
 
diff --git a/xlators/storage/posix/src/posix.c b/xlators/storage/posix/src/posix.c
index dcd54bd35d6..b5df1d082ee 100644
--- a/xlators/storage/posix/src/posix.c
+++ b/xlators/storage/posix/src/posix.c
@@ -3803,11 +3803,18 @@ posix_links_in_same_directory (char *dirpath, int count, inode_t *leaf_inode,
                 if (entry->d_ino != stbuf->st_ino)
                         continue;
 
+                /* Linking an inode here, can cause a race in posix_acl.
+                   Parent inode gets linked here, but before
+                   it reaches posix_acl_readdirp_cbk, create/lookup can
+                   come on a leaf-inode, as parent-inode-ctx not yet updated
+                   in posix_acl_readdirp_cbk, create and lookup can fail
+                   with EACCESS. So do the inode linking in the quota xlator
+
                 linked_inode = inode_link (leaf_inode, parent,
                                            entry->d_name, NULL);
 
                 GF_ASSERT (linked_inode == leaf_inode);
-                inode_unref (linked_inode);
+                inode_unref (linked_inode);*/
 
                 if (type & POSIX_ANCESTRY_DENTRY) {
                         loc_t loc = {0, };
diff --git a/xlators/system/posix-acl/src/posix-acl.c b/xlators/system/posix-acl/src/posix-acl.c
index 2edaa408135..61d2d6647f6 100644
--- a/xlators/system/posix-acl/src/posix-acl.c
+++ b/xlators/system/posix-acl/src/posix-acl.c
@@ -193,8 +193,12 @@ acl_permits (call_frame_t *frame, inode_t *inode, int want)
         conf = frame->this->private;
 
         ctx = posix_acl_ctx_get (inode, frame->this);
-        if (!ctx)
+        if (!ctx) {
+                gf_log_callingfn (frame->this->name, GF_LOG_ERROR,
+                                  "inode ctx is NULL for %s",
+                                  uuid_utoa (inode->gfid));
                 goto red;
+        }
 
         if (frame_is_super_user (frame))
                 goto green;
@@ -287,21 +291,52 @@ out:
 
 
 struct posix_acl_ctx *
-posix_acl_ctx_get (inode_t *inode, xlator_t *this)
+__posix_acl_ctx_get (inode_t *inode, xlator_t *this, gf_boolean_t create)
 {
         struct posix_acl_ctx *ctx = NULL;
         uint64_t              int_ctx = 0;
         int                   ret = 0;
 
-        ret = inode_ctx_get (inode, this, &int_ctx);
+        ret = __inode_ctx_get (inode, this, &int_ctx);
         if ((ret == 0) && (int_ctx))
                 return PTR(int_ctx);
 
+        if (create == _gf_false)
+                return NULL;
+
         ctx = GF_CALLOC (1, sizeof (*ctx), gf_posix_acl_mt_ctx_t);
         if (!ctx)
                 return NULL;
 
-        ret = inode_ctx_put (inode, this, UINT64 (ctx));
+        ret = __inode_ctx_put (inode, this, UINT64 (ctx));
+
+        return ctx;
+}
+
+struct posix_acl_ctx *
+posix_acl_ctx_new (inode_t *inode, xlator_t *this)
+{
+        struct posix_acl_ctx *ctx = NULL;
+
+        LOCK (&inode->lock);
+        {
+                ctx = __posix_acl_ctx_get (inode, this, _gf_true);
+        }
+        UNLOCK (&inode->lock);
+
+        return ctx;
+}
+
+struct posix_acl_ctx *
+posix_acl_ctx_get (inode_t *inode, xlator_t *this)
+{
+        struct posix_acl_ctx *ctx = NULL;
+
+        LOCK (&inode->lock);
+        {
+                ctx = __posix_acl_ctx_get (inode, this, _gf_false);
+        }
+        UNLOCK (&inode->lock);
 
         return ctx;
 }
@@ -636,7 +671,7 @@ posix_acl_inherit (xlator_t *this, loc_t *loc, dict_t *params, mode_t mode,
         if (!par_default)
                 goto out;
 
-        ctx = posix_acl_ctx_get (loc->inode, this);
+        ctx = posix_acl_ctx_new (loc->inode, this);
 
         acl_access = posix_acl_dup (this, par_default);
         if (!acl_access)
@@ -742,14 +777,14 @@ posix_acl_ctx_update (inode_t *inode, xlator_t *this, struct iatt *buf)
         int                   ret = 0;
 	int                   i = 0;
 
-        ctx = posix_acl_ctx_get (inode, this);
-        if (!ctx) {
-                ret = -1;
-                goto out;
-        }
-
         LOCK(&inode->lock);
         {
+                ctx = __posix_acl_ctx_get (inode, this, _gf_true);
+                if (!ctx) {
+                        ret = -1;
+                        goto unlock;
+                }
+
                 ctx->uid   = buf->ia_uid;
                 ctx->gid   = buf->ia_gid;
                 ctx->perm  = st_mode_from_ia (buf->ia_prot, buf->ia_type);
@@ -794,7 +829,6 @@ posix_acl_ctx_update (inode_t *inode, xlator_t *this, struct iatt *buf)
         }
 unlock:
         UNLOCK(&inode->lock);
-out:
         return ret;
 }