quota: For a rename operation, do quota_check_limit only till the

common ancestor of src and dst file

Example:
set quota limit set to 1GB on /
create a file /a1/b1/file1 of 600MB
mv /a1/b1/file1 /a1/b1/file2
This rename fails as it takes delta into account which sums up to 1.2BG.
Though we are not creating new file, we still get quota exceeded error.
So quota enforce should happen only till b1.

Similarly:
mv /a/b/c/file /a/b/x/y/file
quota enforce should happen only till dir 'b'

Change-Id: Ia1e5363da876c3d71bd424e67a8bb28b7ac1c7c1
BUG: 1153964
Signed-off-by: vmallika <vmallika@redhat.com>
Reviewed-on: http://review.gluster.org/8940
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
Tested-by: Raghavendra G <rgowdapp@redhat.com>

2 files changed, 291 insertions, 54 deletions

diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index 3c1b8e09c5c..f903b4e57b7 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -14,17 +14,6 @@
 #include "defaults.h"
 #include "statedump.h"
 
-void
-quota_get_limit_dir (call_frame_t *frame, inode_t *cur_inode, xlator_t *this);
-
-int32_t
-quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
-                   char *name, uuid_t par);
-
-int
-quota_fill_inodectx (xlator_t *this, inode_t *inode, dict_t *dict,
-                     loc_t *loc, struct iatt *buf, int32_t *op_errno);
-
 struct volume_options options[];
 
 static int32_t
@@ -251,6 +240,164 @@ out:
         return;
 }
 
+static inline inode_t*
+__quota_inode_parent (inode_t *inode, uuid_t pargfid, const char *name)
+{
+        inode_t *parent            = NULL;
+
+        parent = inode_parent (inode, pargfid, name);
+        inode_unref (inode);
+        return parent;
+}
+
+static inline inode_t*
+quota_inode_parent (inode_t *inode, uuid_t pargfid, const char *name)
+{
+        inode_t *parent            = NULL;
+
+        parent = __quota_inode_parent (inode, pargfid, name);
+        if (!parent)
+                gf_log_callingfn (THIS->name, GF_LOG_ERROR, "Failed to find "
+                                  "ancestor for inode (%s)",
+                                  uuid_utoa(inode->gfid));
+
+        return parent;
+}
+
+int32_t
+quota_inode_depth (inode_t *inode)
+{
+        int      depth             = 0;
+        inode_t *cur_inode         = NULL;
+
+        cur_inode = inode_ref (inode);
+        while (cur_inode && !__is_root_gfid (cur_inode->gfid)) {
+                depth++;
+                cur_inode = quota_inode_parent (cur_inode, 0 , NULL);
+                if (!cur_inode)
+                        depth = -1;
+        }
+
+        if (cur_inode)
+                inode_unref (cur_inode);
+
+        return depth;
+}
+
+int32_t quota_find_common_ancestor (inode_t *inode1, inode_t *inode2,
+                                    uuid_t *common_ancestor)
+{
+        int32_t         depth1         = 0;
+        int32_t         depth2         = 0;
+        int32_t         ret            = -1;
+        inode_t        *cur_inode1     = NULL;
+        inode_t        *cur_inode2     = NULL;
+
+        depth1 = quota_inode_depth (inode1);
+        if (depth1 < 0)
+                goto out;
+
+        depth2 = quota_inode_depth (inode2);
+        if (depth2 < 0)
+                goto out;
+
+        cur_inode1 = inode_ref (inode1);
+        cur_inode2 = inode_ref (inode2);
+
+        while (cur_inode1 && depth1 > depth2) {
+                cur_inode1 = quota_inode_parent (cur_inode1, 0 , NULL);
+                depth1--;
+        }
+
+        while (cur_inode2 && depth2 > depth1) {
+                cur_inode2 = quota_inode_parent (cur_inode2, 0 , NULL);
+                depth2--;
+        }
+
+        while (depth1 && cur_inode1 && cur_inode2 && cur_inode1 != cur_inode2) {
+                cur_inode1 = quota_inode_parent (cur_inode1, 0 , NULL);
+                cur_inode2 = quota_inode_parent (cur_inode2, 0 , NULL);
+                depth1--;
+        }
+
+        if (cur_inode1 && cur_inode2) {
+                uuid_copy (*common_ancestor, cur_inode1->gfid);
+                ret = 0;
+        }
+out:
+        if (cur_inode1)
+                inode_unref (cur_inode1);
+
+        if (cur_inode2)
+                inode_unref (cur_inode2);
+
+        return ret;
+ }
+
+void
+check_ancestory_continue (struct list_head *parents, inode_t *inode,
+                          int32_t op_ret, int32_t op_errno, void *data)
+{
+        call_frame_t   *frame        = NULL;
+        quota_local_t  *local        = NULL;
+        uint32_t        link_count   = 0;
+
+        frame = data;
+        local = frame->local;
+
+        if (op_ret < 0 || (parents && list_empty (parents))) {
+                if (op_ret >= 0) {
+                        gf_log (THIS->name, GF_LOG_WARNING,
+                                "Couldn't build ancestry for inode (gfid:%s). "
+                                "Without knowing ancestors till root, quota "
+                                "cannot be enforced. "
+                                "Hence, failing fop with EIO",
+                                uuid_utoa (inode->gfid));
+                        op_errno = EIO;
+                        op_ret = -1;
+                }
+        }
+
+        LOCK (&local->lock);
+        {
+                link_count = --local->link_count;
+                if (op_ret < 0) {
+                        local->op_ret = op_ret;
+                        local->op_errno = op_errno;
+                }
+        }
+        UNLOCK (&local->lock);
+
+        if (link_count == 0)
+                local->fop_continue_cbk (frame);
+}
+
+void
+check_ancestory (call_frame_t *frame, inode_t *inode)
+{
+        inode_t *cur_inode = NULL;
+        inode_t *parent    = NULL;
+
+        cur_inode = inode_ref (inode);
+        while (cur_inode && !__is_root_gfid (cur_inode->gfid)) {
+                parent = inode_parent (cur_inode, 0, NULL);
+                if (!parent) {
+                        quota_build_ancestry (cur_inode,
+                                              check_ancestory_continue, frame);
+                        return;
+                }
+                inode_unref (cur_inode);
+                cur_inode = parent;
+        }
+
+        if (cur_inode) {
+                inode_unref (cur_inode);
+                check_ancestory_continue (NULL, NULL, 0, 0, frame);
+        } else {
+                check_ancestory_continue (NULL, NULL, -1, ESTALE, frame);
+        }
+}
+
 static inline void
 quota_link_count_decrement (quota_local_t *local)
 {
@@ -827,6 +974,14 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
         }
 
         do {
+                /* In a rename operation, enforce should be stopped at common
+                   ancestor */
+                if (!uuid_is_null (local->common_ancestor) &&
+                    !uuid_compare (_inode->gfid, local->common_ancestor)) {
+                        quota_link_count_decrement (local);
+                        break;
+                }
+
                 if (ctx != NULL && (ctx->hard_lim > 0 || ctx->soft_lim > 0)) {
                         wouldbe_size = ctx->size + delta;
 
@@ -2046,63 +2201,51 @@ out:
         return 0;
 }
 
-int32_t
-quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
-              loc_t *newloc, dict_t *xdata)
+void
+quota_rename_continue (call_frame_t *frame)
 {
-        quota_priv_t      *priv  = NULL;
-        int32_t            ret   = -1, op_errno = ENOMEM;
-        quota_local_t     *local = NULL;
-        quota_inode_ctx_t *ctx   = NULL;
-        call_stub_t       *stub  = NULL;
-
-        priv = this->private;
-
-        WIND_IF_QUOTAOFF (priv->is_quota_on, off);
-
-        local = quota_local_new ();
-        if (local == NULL) {
-                goto err;
-        }
-
-        frame->local = local;
+        int32_t            ret               = -1;
+        int32_t            op_errno          = EIO;
+        quota_local_t     *local             = NULL;
+        uuid_t             common_ancestor   = {0};
+        xlator_t          *this              = NULL;
+        quota_inode_ctx_t *ctx               = NULL;
 
-        ret = loc_copy (&local->oldloc, oldloc);
-        if (ret < 0) {
-                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
-                goto err;
-        }
+        local = frame->local;
+        this = THIS;
 
-        ret = loc_copy (&local->newloc, newloc);
-        if (ret < 0) {
-                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
+        if (local->op_ret < 0) {
+                op_errno = local->op_errno;
                 goto err;
         }
 
-        stub = fop_rename_stub (frame, quota_rename_helper, oldloc, newloc,
-                                xdata);
-        if (stub == NULL) {
+        ret = quota_find_common_ancestor (local->oldloc.parent,
+                                          local->newloc.parent,
+                                          &common_ancestor);
+        if (ret < 0 || uuid_is_null(common_ancestor)) {
+                gf_log (this->name, GF_LOG_ERROR, "failed to get "
+                        "common_ancestor for %s and %s",
+                        local->oldloc.path, local->newloc.path);
+                op_errno = ESTALE;
                 goto err;
         }
 
         LOCK (&local->lock);
         {
                 local->link_count = 1;
-                local->stub = stub;
+                uuid_copy (local->common_ancestor, common_ancestor);
         }
         UNLOCK (&local->lock);
 
-        if (QUOTA_REG_OR_LNK_FILE (oldloc->inode->ia_type)) {
-                ret = quota_inode_ctx_get (oldloc->inode, this, &ctx, 0);
+        if (QUOTA_REG_OR_LNK_FILE (local->oldloc.inode->ia_type)) {
+                ret = quota_inode_ctx_get (local->oldloc.inode, this, &ctx, 0);
                 if (ctx == NULL) {
                         gf_log (this->name, GF_LOG_WARNING,
                                 "quota context not set in inode (gfid:%s), "
                                 "considering file size as zero while enforcing "
                                 "quota on new ancestry",
-                                oldloc->inode ? uuid_utoa (oldloc->inode->gfid)
-                                : "0");
+                                uuid_utoa (local->oldloc.inode->gfid));
                         local->delta = 0;
-
                 } else {
 
                     /* FIXME: We need to account for the size occupied by this
@@ -2112,25 +2255,99 @@ quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                      * directory inode*/
 
                     /* FIXME: The following code assumes that regular files and
-                     *linkfiles are present, in their entirety, in a single
-                     brick. This *assumption is invalid in the case of
-                     stripe.*/
+                     * linkfiles are present, in their entirety, in a single
+                     * brick. This *assumption is invalid in the case of
+                     * stripe.*/
 
                     local->delta = ctx->buf.ia_blocks * 512;
                 }
 
-        } else if (IA_ISDIR (oldloc->inode->ia_type)) {
-                        ret = quota_validate (frame, oldloc->inode, this,
+        } else if (IA_ISDIR (local->oldloc.inode->ia_type)) {
+                        ret = quota_validate (frame, local->oldloc.inode, this,
                                               quota_rename_get_size_cbk);
                         if (ret){
                                 op_errno = -ret;
                                 goto err;
                         }
 
-                        return 0;
+                        return;
         }
 
-        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
+        quota_check_limit (frame, local->newloc.parent, this, NULL, NULL);
+        return;
+
+err:
+        if (local && local->stub)
+                call_stub_destroy (local->stub);
+
+        QUOTA_STACK_UNWIND (rename, frame, -1, op_errno, NULL,
+                            NULL, NULL, NULL, NULL, NULL);
+        return;
+
+}
+
+int32_t
+quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
+              loc_t *newloc, dict_t *xdata)
+{
+        quota_priv_t      *priv              = NULL;
+        int32_t            ret               = -1;
+        int32_t            op_errno          = ENOMEM;
+        quota_local_t     *local             = NULL;
+        call_stub_t       *stub              = NULL;
+        uuid_t             common_ancestor   = {0};
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, off);
+
+        /* No need to check quota limit if src and dst parents are same */
+        if (oldloc->parent && newloc->parent &&
+            !uuid_compare(oldloc->parent->gfid, newloc->parent->gfid)) {
+                gf_log (this->name, GF_LOG_DEBUG, "rename %s -> %s are "
+                        "in the same directory, so skip check limit",
+                        oldloc->path, newloc->path);
+                goto off;
+        }
+
+        local = quota_local_new ();
+        if (local == NULL) {
+                goto err;
+        }
+
+        frame->local = local;
+
+        ret = loc_copy (&local->oldloc, oldloc);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
+                goto err;
+        }
+
+        ret = loc_copy (&local->newloc, newloc);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
+                goto err;
+        }
+
+        stub = fop_rename_stub (frame, quota_rename_helper, oldloc, newloc,
+                                xdata);
+        if (stub == NULL) {
+                goto err;
+        }
+
+        LOCK (&local->lock);
+        {
+                /* link_count here tell how many check_ancestory should be done
+                 * before continuing the FOP
+                 */
+                local->link_count = 2;
+                local->stub = stub;
+                local->fop_continue_cbk = quota_rename_continue;
+        }
+        UNLOCK (&local->lock);
+
+        check_ancestory (frame, newloc->parent);
+        check_ancestory (frame, oldloc->parent);
         return 0;
 
 err:
diff --git a/xlators/features/quota/src/quota.h b/xlators/features/quota/src/quota.h
index 5a4bcb2b1e0..3d6c65f8fb6 100644
--- a/xlators/features/quota/src/quota.h
+++ b/xlators/features/quota/src/quota.h
@@ -181,6 +181,9 @@ typedef void
 (*quota_ancestry_built_t) (struct list_head *parents, inode_t *inode,
                            int32_t op_ret, int32_t op_errno, void *data);
 
+typedef void
+(*quota_fop_continue_t) (call_frame_t *frame);
+
 struct quota_local {
         gf_lock_t               lock;
         uint32_t                validate_count;
@@ -196,7 +199,9 @@ struct quota_local {
         gf_boolean_t            skip_check;
         char                    just_validated;
         fop_lookup_cbk_t        validate_cbk;
+        quota_fop_continue_t    fop_continue_cbk;
         inode_t                *inode;
+        uuid_t                  common_ancestor; /* Used by quota_rename */
         call_stub_t            *stub;
         struct iobref          *iobref;
         quota_limit_t           limit;
@@ -235,4 +240,19 @@ void
 quota_log_usage (xlator_t *this, quota_inode_ctx_t *ctx, inode_t *inode,
                  int64_t delta);
 
+int
+quota_build_ancestry (inode_t *inode, quota_ancestry_built_t ancestry_cbk,
+                      void *data);
+
+void
+quota_get_limit_dir (call_frame_t *frame, inode_t *cur_inode, xlator_t *this);
+
+int32_t
+quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
+                   char *name, uuid_t par);
+
+int
+quota_fill_inodectx (xlator_t *this, inode_t *inode, dict_t *dict,
+                     loc_t *loc, struct iatt *buf, int32_t *op_errno);
+
 #endif