features/quota: Check quota on all parents after building ancestry.

This patch also fixes a potential race-condition while accessing
dentry list stored in inode context while updating size and enforcing
limits.

Change-Id: I76c85fe782f6a53d74b453dd7dcb893e33003620
BUG: 969461
Signed-off-by: Raghavendra G <rgowdapp@redhat.com>

1 files changed, 94 insertions, 42 deletions

diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index 114febff..b398d0d8 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -24,7 +24,6 @@ quota_fill_inodectx (xlator_t *this, inode_t *inode, dict_t *dict,
 
 struct volume_options options[];
 
-
 static int32_t
 __quota_init_inode_ctx (inode_t *inode, xlator_t *this,
                         quota_inode_ctx_t **context)
@@ -56,6 +55,29 @@ out:
         return ret;
 }
 
+
+static int32_t
+quota_inode_ctx_get (inode_t *inode, xlator_t *this,
+                     quota_inode_ctx_t **ctx, char create_if_absent)
+{
+        int32_t  ret = 0;
+        uint64_t ctx_int;
+
+        LOCK (&inode->lock);
+        {
+                ret = __inode_ctx_get (inode, this, &ctx_int);
+
+                if ((ret == 0) && (ctx != NULL)) {
+                        *ctx = (quota_inode_ctx_t *) (unsigned long)ctx_int;
+                } else if (create_if_absent) {
+                        ret = __quota_init_inode_ctx (inode, this, ctx);
+                }
+        }
+        UNLOCK (&inode->lock);
+
+        return ret;
+}
+
 int
 quota_loc_fill (loc_t *loc, inode_t *inode, inode_t *parent, char *path)
 {
@@ -193,7 +215,9 @@ __quota_dentry_new (quota_inode_ctx_t *ctx, char *name, uuid_t par)
 
         uuid_copy (dentry->par, par);
 
-        list_add_tail (&dentry->next, &ctx->parents);
+        if (ctx != NULL)
+                list_add_tail (&dentry->next, &ctx->parents);
+
 err:
         return dentry;
 }
@@ -354,12 +378,16 @@ quota_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                           int32_t op_ret, int32_t op_errno, dict_t *dict,
                           dict_t *xdata)
 {
-        quota_local_t *local  = NULL;
-        inode_t       *parent = NULL;
-        gf_dirent_t   *head   = NULL, *entry = NULL;
-        loc_t          loc    = {0, };
-        int            ret    = 0;
-        dict_t        *tmp    = NULL;
+        inode_t           *parent  = NULL;
+        gf_dirent_t       *head    = NULL, *entry = NULL;
+        loc_t              loc     = {0, };
+        int                ret     = 0;
+        quota_dentry_t    *dentry  = NULL, *tmp = NULL;
+        quota_inode_ctx_t *ctx     = NULL;
+        struct list_head   parents = {0, };
+        quota_local_t     *local   = NULL;
+
+        INIT_LIST_HEAD (&parents);
 
         if (op_ret < 0)
                 goto err;
@@ -373,8 +401,6 @@ quota_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto err;
         }
 
-        tmp = dict_new ();
-
         ret = dict_get_bin (dict, GET_ANCESTRY_DENTRY_KEY, (void **)&head);
 
         if (ret == 0 && head) {
@@ -403,7 +429,37 @@ quota_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 }
         }
 
-        quota_check_limit (frame, parent, this, NULL, NULL);
+        quota_inode_ctx_get (local->validate_loc.inode, this, &ctx, 0);
+
+        if (ctx == NULL) {
+                op_errno = EINVAL;
+                goto err;
+        }
+
+        local->link_count = 0;
+
+        LOCK (&ctx->lock);
+        {
+                list_for_each_entry (dentry, &ctx->parents, next) {
+                        tmp = __quota_dentry_new (NULL, dentry->name,
+                                                  dentry->par);
+                        list_add_tail (&tmp->next, &parents);
+                        local->link_count++;
+                }
+        }
+        UNLOCK (&ctx->lock);
+
+        if (local->link_count != 0) {
+                list_for_each_entry_safe (dentry, tmp, &parents, next) {
+                        quota_check_limit (frame, local->validate_loc.inode,
+                                           this, dentry->name, dentry->par);
+                        __quota_dentry_free (dentry);
+                }
+        } else {
+                local->link_count = 1;
+                quota_check_limit (frame, parent, this, NULL, NULL);
+        }
+
         return 0;
 
 err:
@@ -461,11 +517,11 @@ err:
 int
 quota_validate (call_frame_t *frame, inode_t *inode, xlator_t *this)
 {
-        quota_local_t *local = NULL;
-        int            ret   = 0;
-        dict_t        *xdata = NULL;
-        quota_priv_t  *priv  = NULL;
-
+        quota_local_t     *local = NULL;
+        int                ret   = 0;
+        dict_t            *xdata = NULL;
+        quota_priv_t      *priv  = NULL;
+ 
         local = frame->local;
         priv = this->private;
 
@@ -683,28 +739,6 @@ err:
         return 0;
 }
 
-static int32_t
-quota_inode_ctx_get (inode_t *inode, xlator_t *this,
-                     quota_inode_ctx_t **ctx, char create_if_absent)
-{
-        int32_t  ret = 0;
-        uint64_t ctx_int;
-
-        LOCK (&inode->lock);
-        {
-                ret = __inode_ctx_get (inode, this, &ctx_int);
-
-                if ((ret == 0) && (ctx != NULL)) {
-                        *ctx = (quota_inode_ctx_t *) (unsigned long)ctx_int;
-                } else if (create_if_absent) {
-                        ret = __quota_init_inode_ctx (inode, this, ctx);
-                }
-        }
-        UNLOCK (&inode->lock);
-
-        return ret;
-}
-
 inline int
 quota_get_limits (xlator_t *this, dict_t *dict, int64_t *hard_lim,
                   int64_t *soft_lim)
@@ -977,8 +1011,9 @@ quota_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         uint64_t                 ctx_int        = 0;
         quota_inode_ctx_t       *ctx            = NULL;
         quota_local_t           *local          = NULL;
-        quota_dentry_t          *dentry         = NULL;
+        quota_dentry_t          *dentry         = NULL, *tmp = NULL;
         int64_t                  delta          = 0;
+        struct list_head         head           = {0, };
 
         local = frame->local;
 
@@ -986,6 +1021,8 @@ quota_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
+        INIT_LIST_HEAD (&head);
+
         ret = inode_ctx_get (local->loc.inode, this, &ctx_int);
         if (ret) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -1005,15 +1042,23 @@ quota_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         LOCK (&ctx->lock);
         {
                 ctx->buf = *postbuf;
+
+                list_for_each_entry (dentry, &ctx->parents, next) {
+                        tmp = __quota_dentry_new (NULL, dentry->name,
+                                                  dentry->par);
+                        list_add_tail (&tmp->next, &head);
+                }
+
         }
         UNLOCK (&ctx->lock);
 
         if (postbuf->ia_blocks != prebuf->ia_blocks)
                 delta = local->delta;
 
-        list_for_each_entry (dentry, &ctx->parents, next) {
+        list_for_each_entry_safe (dentry, tmp, &head, next) {
                 quota_update_size (this, local->loc.inode, dentry->name,
                                    dentry->par, delta);
+                __quota_dentry_free (dentry);
         }
 
 out:
@@ -1069,13 +1114,16 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         uint64_t           size    = 0;
         quota_local_t     *local   = NULL;
         quota_inode_ctx_t *ctx     = NULL;
-        quota_dentry_t    *dentry  = NULL;
+        quota_dentry_t    *dentry  = NULL, *tmp = NULL;
         call_stub_t       *stub    = NULL;
+        struct list_head   head    = {0, };
 
         priv = this->private;
 
         WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
+        INIT_LIST_HEAD (&head);
+
         GF_ASSERT (frame);
         GF_VALIDATE_OR_GOTO ("quota", this, unwind);
         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);
@@ -1110,6 +1158,9 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         LOCK (&ctx->lock);
         {
                 list_for_each_entry (dentry, &ctx->parents, next) {
+                        tmp = __quota_dentry_new (NULL, dentry->name,
+                                                  dentry->par);
+                        list_add_tail (&tmp->next, &head);
                         parents++;
                 }
         }
@@ -1127,9 +1178,10 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 local->link_count = 1;
                 quota_check_limit (frame, fd->inode, this, NULL, NULL);
         } else {
-                list_for_each_entry (dentry, &ctx->parents, next) {
+                list_for_each_entry_safe (dentry, tmp, &head, next) {
                         quota_check_limit (frame, fd->inode, this, dentry->name,
                                            dentry->par);
+                        __quota_dentry_free (dentry);
                 }
         }