marker-quota/rename: use contribution values from backend instead of in-memory while reducing parent sizes during rename

Signed-off-by: Raghavendra G <raghavendra@gluster.com>
Signed-off-by: Anand Avati <avati@gluster.com>

BUG: 2697 (Quota: add-brick creates the size go awkward, though it was perfect earlier)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=2697

2 files changed, 194 insertions, 97 deletions

diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index 3d42df04fad..84d0da9184a 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -180,11 +180,11 @@ mark_inode_undirty (call_frame_t *frame, void *cookie, xlator_t *this,
         if (ret)
                 goto wind;
 
-	LOCK (&local->ctx->lock);
-	{
-	  local->ctx->size = ntoh64 (*size);
-	}
-	UNLOCK (&local->ctx->lock);
+        LOCK (&local->ctx->lock);
+        {
+                local->ctx->size = ntoh64 (*size);
+        }
+        UNLOCK (&local->ctx->lock);
 
 wind:
         newdict = dict_new ();
@@ -1110,11 +1110,11 @@ quota_update_parent_size (call_frame_t *frame,
                 goto err;
         }
 
-	LOCK (&local->contri->lock);
-	{
-	  local->contri->contribution += local->delta;
-	}
-	UNLOCK (&local->contri->lock);
+        LOCK (&local->contri->lock);
+        {
+                local->contri->contribution += local->delta;
+        }
+        UNLOCK (&local->contri->lock);
 
         gf_log (this->name, GF_LOG_DEBUG, "%s %"PRId64 "%"PRId64,
                 local->loc.path, local->ctx->size,
@@ -1213,27 +1213,27 @@ quota_update_inode_contribution (call_frame_t *frame, void *cookie,
                 } else
                         ctx->size = buf->ia_blocks * 512;
 
-		size_int = ctx->size;
-	}
- unlock:
-	UNLOCK  (&ctx->lock);
+                size_int = ctx->size;
+        }
+unlock:
+        UNLOCK  (&ctx->lock);
 
-	if (ret < 0) {
-	  goto err;
-	}
+        if (ret < 0) {
+                goto err;
+        }
 
-	ret = dict_get_bin (dict, contri_key, (void **) &contri);
+        ret = dict_get_bin (dict, contri_key, (void **) &contri);
 
-	LOCK (&contribution->lock);
-	{
-	  if (ret < 0)
-	    contribution->contribution = 0;
-	  else
-	    contribution->contribution = ntoh64 (*contri);
+        LOCK (&contribution->lock);
+        {
+                if (ret < 0)
+                        contribution->contribution = 0;
+                else
+                        contribution->contribution = ntoh64 (*contri);
 
-	  contri_int = contribution->contribution;
+                contri_int = contribution->contribution;
         }
-	UNLOCK (&contribution->lock);
+        UNLOCK (&contribution->lock);
 
         gf_log (this->name, GF_LOG_DEBUG, "%s %"PRId64 "%"PRId64,
                 local->loc.path, size_int, contri_int);
@@ -1517,7 +1517,7 @@ out:
 
 /* int32_t */
 /* validate_inode_size_contribution (xlator_t *this, loc_t *loc, int64_t size, */
-/* 				  int64_t contribution) */
+/*                                int64_t contribution) */
 /* { */
 /*   if (size != contribution) { */
 /*     initiate_quota_txn (this, loc); */
@@ -1582,29 +1582,29 @@ inspect_directory_xattr (xlator_t *this,
                 if (ret < 0)
                         goto out;
 
-		LOCK (&contribution->lock);
-		{
-		  contribution->contribution = ntoh64 (*contri);
-		  contri_int = contribution->contribution;
-		}
-		UNLOCK (&contribution->lock);
+                LOCK (&contribution->lock);
+                {
+                        contribution->contribution = ntoh64 (*contri);
+                        contri_int = contribution->contribution;
+                }
+                UNLOCK (&contribution->lock);
         }
 
-	LOCK (&ctx->lock);
-	{
-	  ctx->size = ntoh64 (*size);
-	  ctx->dirty = dirty;
-	  size_int = ctx->size;
-	}
-	UNLOCK (&ctx->lock);
+        LOCK (&ctx->lock);
+        {
+                ctx->size = ntoh64 (*size);
+                ctx->dirty = dirty;
+                size_int = ctx->size;
+        }
+        UNLOCK (&ctx->lock);
 
         gf_log (this->name, GF_LOG_DEBUG, "size=%"PRId64
                 " contri=%"PRId64, size_int, contri_int);
 
         if (dirty) {
-	  update_dirty_inode (this, loc, ctx, contribution);
-	} else if ((not_root == _gf_true) && (size_int != contri_int)) {
-	  initiate_quota_txn (this, loc);
+                update_dirty_inode (this, loc, ctx, contribution);
+        } else if ((not_root == _gf_true) && (size_int != contri_int)) {
+                initiate_quota_txn (this, loc);
         }
 
         ret = 0;
@@ -1649,7 +1649,7 @@ inspect_file_xattr (xlator_t *this,
         LOCK (&ctx->lock);
         {
                 ctx->size = 512 * buf.ia_blocks;
-		size = ctx->size;
+                size = ctx->size;
         }
         UNLOCK (&ctx->lock);
 
@@ -1663,19 +1663,19 @@ inspect_file_xattr (xlator_t *this,
                 if (ret == 0) {
                         contri_ptr = (int64_t *)(unsigned long)contri_int;
 
-			LOCK (&contribution->lock);
-			{
-			  contribution->contribution = ntoh64 (*contri_ptr);
-			  contri_int = contribution->contribution;
-			}
-			UNLOCK (&contribution->lock);
+                        LOCK (&contribution->lock);
+                        {
+                                contribution->contribution = ntoh64 (*contri_ptr);
+                                contri_int = contribution->contribution;
+                        }
+                        UNLOCK (&contribution->lock);
 
                         gf_log (this->name, GF_LOG_DEBUG,
                                 "size=%"PRId64 " contri=%"PRId64, size, contri_int);
 
-			if (size != contri_int) {
-			  initiate_quota_txn (this, loc);
-			}
+                        if (size != contri_int) {
+                                initiate_quota_txn (this, loc);
+                        }
                 } else
                         initiate_quota_txn (this, loc);
         }
@@ -1793,7 +1793,7 @@ mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
         struct gf_flock    lock  = {0, };
         quota_inode_ctx_t *ctx   = NULL;
         quota_local_t     *local = NULL;
-	int64_t contribution = 0;
+        int64_t contribution = 0;
 
         local = frame->local;
         if (op_ret == -1)
@@ -1801,26 +1801,26 @@ mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
 
         ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx);
 
-	LOCK (&local->contri->lock);
-	{
-	  contribution = local->contri->contribution;
-	}
-	UNLOCK (&local->contri->lock);
+        LOCK (&local->contri->lock);
+        {
+                contribution = local->contri->contribution;
+        }
+        UNLOCK (&local->contri->lock);
 
         if (contribution == local->size) {
-	  if (ret == 0) {
-	    LOCK (&ctx->lock);
-	    {
-	      ctx->size -= contribution;
-	    }
-	    UNLOCK (&ctx->lock);
-
-	    LOCK (&local->contri->lock);
-	    {
-	      local->contri->contribution = 0;
-	    }
-	    UNLOCK (&local->contri->lock);
-	  }
+                if (ret == 0) {
+                        LOCK (&ctx->lock);
+                        {
+                                ctx->size -= contribution;
+                        }
+                        UNLOCK (&ctx->lock);
+
+                        LOCK (&local->contri->lock);
+                        {
+                                local->contri->contribution = 0;
+                        }
+                        UNLOCK (&local->contri->lock);
+                }
         }
 
         lock.l_type   = F_UNLCK;
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index b00d0a3d2ff..e6fd66a6708 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -29,6 +29,7 @@
 #include "marker-quota.h"
 #include "marker-quota-helper.h"
 #include "marker-common.h"
+#include "byte-order.h"
 
 void
 fini (xlator_t *this);
@@ -1119,49 +1120,145 @@ quota_err:
 
 int32_t
 marker_do_rename (call_frame_t *frame, void *cookie, xlator_t *this,
-                  int32_t op_ret, int32_t op_errno)
+                  int32_t op_ret, int32_t op_errno, dict_t *dict)
+
 {
-        marker_local_t       *local        = NULL, *oplocal = NULL;
-        inode_contribution_t *contribution = NULL;
+        marker_local_t       *local           = NULL, *oplocal = NULL;
+        char                  contri_key[512] = {0, };
+        int32_t               ret             = 0;
+        int64_t              *contribution    = 0;
 
         local = frame->local;
         oplocal = local->oplocal;
 
-        if (op_ret < 0) {
+        if ((op_ret < 0) && (op_errno != ENOATTR)) {
                 local->err = op_errno;
                 gf_log (this->name, GF_LOG_WARNING,
-                        "cannot hold inodelk on %s (ino:%"PRId64", gfid:%s)"
-                        "(%s)",
-                        local->next_lock_on->path,
-                        local->next_lock_on->inode->ino,
-                        uuid_utoa (local->next_lock_on->inode->gfid),
+                        "fetching contribution values from %s (ino:%"PRId64", "
+                        "gfid:%s) failed (%s)", local->loc.path,
+                        local->loc.inode->ino,
+                        uuid_utoa (local->loc.inode->gfid),
                         strerror (op_errno));
-                goto lock_err;
+                goto err;
+        }
+
+        if (local->loc.inode != NULL) {
+                GET_CONTRI_KEY (contri_key, local->loc.parent->gfid, ret);
+                if (ret < 0) {
+                        local->err = errno;
+                        goto err;
+                }
+
+                if (dict_get_bin (dict, contri_key,
+                                  (void **) &contribution) == 0) {
+                        local->contribution = ntoh64 (*contribution);
+                }
         }
 
+        STACK_WIND (frame, marker_rename_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->rename, &oplocal->loc,
+                    &local->loc);
+
+        return 0;
+
+err:
+        marker_rename_release_oldp_lock (frame, NULL, this, 0, 0);
+        return 0;
+}
+
+
+int32_t
+marker_get_newpath_contribution (call_frame_t *frame, void *cookie,
+                                 xlator_t *this, int32_t op_ret,
+                                 int32_t op_errno, dict_t *dict)
+{
+        marker_local_t *local           = NULL, *oplocal = NULL;
+        char            contri_key[512] = {0, };
+        int32_t         ret             = 0;
+        int64_t        *contribution    = 0;
+
         local = frame->local;
         oplocal = local->oplocal;
 
+        if ((op_ret < 0) && (op_errno != ENOATTR)) {
+                local->err = op_errno;
+                gf_log (this->name, GF_LOG_WARNING,
+                        "fetching contribution values from %s (ino:%"PRId64", "
+                        "gfid:%s) failed (%s)", oplocal->loc.path,
+                        oplocal->loc.inode->ino,
+                        uuid_utoa (oplocal->loc.inode->gfid),
+                        strerror (op_errno));
+                goto err;
+        }
+
+        GET_CONTRI_KEY (contri_key, oplocal->loc.parent->gfid, ret);
+        if (ret < 0) {
+                local->err = errno;
+                goto err;
+        }
+
+        if (dict_get_bin (dict, contri_key, (void **) &contribution) == 0)
+                oplocal->contribution = ntoh64 (*contribution);
+
         if (local->loc.inode != NULL) {
-                contribution = get_contribution_from_loc (this, &local->loc);
-                if (contribution == NULL) {
-                        local->contribution = 0;
-                } else {
-                        local->contribution = contribution->contribution;
+                GET_CONTRI_KEY (contri_key, local->loc.parent->gfid, ret);
+                if (ret < 0) {
+                        local->err = errno;
+                        goto err;
                 }
-        }
 
-        contribution = get_contribution_from_loc (this, &oplocal->loc);
-        if (contribution == NULL) {
-                oplocal->contribution = 0;
+                STACK_WIND (frame, marker_do_rename,
+                            FIRST_CHILD(this),
+                            FIRST_CHILD(this)->fops->getxattr, &local->loc,
+                            contri_key);
         } else {
-                oplocal->contribution = contribution->contribution;
+                marker_do_rename (frame, NULL, this, 0, 0, NULL);
         }
 
-        STACK_WIND (frame, marker_rename_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->rename, &oplocal->loc,
-                    &local->loc);
+        return 0;
+err:
+        marker_rename_release_oldp_lock (frame, NULL, this, 0, 0);
+        return 0;
+}
+
 
+int32_t
+marker_get_oldpath_contribution (call_frame_t *frame, void *cookie,
+                                 xlator_t *this, int32_t op_ret,
+                                 int32_t op_errno)
+{
+        marker_local_t *local           = NULL, *oplocal = NULL;
+        char            contri_key[512] = {0, };
+        int32_t         ret             = 0;
+
+        local = frame->local;
+        oplocal = local->oplocal;
+
+        if (op_ret < 0) {
+                local->err = op_errno;
+                gf_log (this->name, GF_LOG_WARNING,
+                        "cannot hold inodelk on %s (ino:%"PRId64", gfid:%s)"
+                        "(%s)",
+                        local->next_lock_on->path,
+                        local->next_lock_on->inode->ino,
+                        uuid_utoa (local->next_lock_on->inode->gfid),
+                        strerror (op_errno));
+                goto lock_err;
+        }
+
+        GET_CONTRI_KEY (contri_key, oplocal->loc.parent->gfid, ret);
+        if (ret < 0) {
+                local->err = errno;
+                goto quota_err;
+        }
+
+        STACK_WIND (frame, marker_get_newpath_contribution, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->getxattr, &oplocal->loc,
+                    contri_key);
+        return 0;
+
+quota_err:
+        marker_rename_release_oldp_lock (frame, NULL, this, 0, 0);
         return 0;
 
 lock_err:
@@ -1211,13 +1308,13 @@ marker_rename_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 lock.l_whence = SEEK_SET;
 
                 STACK_WIND (frame,
-                            marker_do_rename,
+                            marker_get_oldpath_contribution,
                             FIRST_CHILD(this),
                             FIRST_CHILD(this)->fops->inodelk,
                             this->name, local->next_lock_on,
                             F_SETLKW, &lock);
         } else {
-                marker_do_rename (frame, 0, this, 0, 0);
+                marker_get_oldpath_contribution (frame, 0, this, 0, 0);
         }
 
         return 0;