marker: do mq_reduce_parent_size_txn in FG for unlink & rmdir

* If a "rm -rf" is performed by a client, we initiate a
  marker background operation mq_reduce_parent_size_txn
  for rmdir and unlink.

  mq_reduce_parent_size_txn can fail when updating
  size on the ancestor directories, if these directories
  are removed during the txn as the child-parent association
  removed in the dentry list.

  So execute mq_reduce_parent_size_txn in foreground
  and then do the UNWIND for rmdir and unlink FOP

Change-Id: Iefcdced4c6ae0dbd43f92814d0ddcd1e33825864
BUG: 1322489
Signed-off-by: vmallika <vmallika@redhat.com>
Reviewed-on: http://review.gluster.org/13874
Smoke: Gluster Build System <jenkins@build.gluster.com>
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
CentOS-regression: Gluster Build System <jenkins@build.gluster.com>

3 files changed, 90 insertions, 32 deletions

diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index 9d2f3d375e5..c7d050a42a2 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -1046,6 +1046,9 @@ mq_synctask_cleanup (int ret, call_frame_t *frame, void *opaque)
         args = (quota_synctask_t *) opaque;
         loc_wipe (&args->loc);
 
+        if (args->stub)
+                call_resume (args->stub);
+
         if (!args->is_static)
                 GF_FREE (args);
 
@@ -1054,7 +1057,8 @@ mq_synctask_cleanup (int ret, call_frame_t *frame, void *opaque)
 
 int
 mq_synctask1 (xlator_t *this, synctask_fn_t task, gf_boolean_t spawn,
-              loc_t *loc, quota_meta_t *contri, uint32_t nlink)
+              loc_t *loc, quota_meta_t *contri, uint32_t nlink,
+              call_stub_t *stub)
 {
         int32_t              ret         = -1;
         quota_synctask_t    *args        = NULL;
@@ -1069,6 +1073,7 @@ mq_synctask1 (xlator_t *this, synctask_fn_t task, gf_boolean_t spawn,
         }
 
         args->this = this;
+        args->stub = stub;
         loc_copy (&args->loc, loc);
         args->ia_nlink = nlink;
 
@@ -1100,7 +1105,7 @@ out:
 int
 mq_synctask (xlator_t *this, synctask_fn_t task, gf_boolean_t spawn, loc_t *loc)
 {
-        return mq_synctask1 (this, task, spawn, loc, NULL, -1);
+        return mq_synctask1 (this, task, spawn, loc, NULL, -1, NULL);
 }
 
 int32_t
@@ -1407,10 +1412,12 @@ out:
 
 int32_t
 mq_reduce_parent_size_txn (xlator_t *this, loc_t *origin_loc,
-                           quota_meta_t *contri, uint32_t nlink)
+                           quota_meta_t *contri, uint32_t nlink,
+                           call_stub_t *stub)
 {
         int32_t                  ret           = -1;
         loc_t                    loc           = {0, };
+        gf_boolean_t             resume_stub   = _gf_true;
 
         GF_VALIDATE_OR_GOTO ("marker", this, out);
         GF_VALIDATE_OR_GOTO ("marker", origin_loc, out);
@@ -1424,11 +1431,19 @@ mq_reduce_parent_size_txn (xlator_t *this, loc_t *origin_loc,
                 goto out;
         }
 
+        resume_stub = _gf_false;
         ret = mq_synctask1 (this, mq_reduce_parent_size_task, _gf_true, &loc,
-                            contri, nlink);
+                            contri, nlink, stub);
 out:
         loc_wipe (&loc);
 
+        if (resume_stub && stub)
+                call_resume (stub);
+
+        if (ret)
+                gf_log_callingfn (this->name, GF_LOG_ERROR,
+                                  "mq_reduce_parent_size_txn failed");
+
         return ret;
 }
 
diff --git a/xlators/features/marker/src/marker-quota.h b/xlators/features/marker/src/marker-quota.h
index 319fafd9983..51e062537b8 100644
--- a/xlators/features/marker/src/marker-quota.h
+++ b/xlators/features/marker/src/marker-quota.h
@@ -14,6 +14,7 @@
 #include "marker-mem-types.h"
 #include "refcount.h"
 #include "quota-common-utils.h"
+#include "call-stub.h"
 
 #define QUOTA_XATTR_PREFIX "trusted.glusterfs"
 #define QUOTA_DIRTY_KEY "trusted.glusterfs.quota.dirty"
@@ -116,6 +117,7 @@ struct quota_synctask {
         quota_meta_t   contri;
         gf_boolean_t   is_static;
         uint32_t       ia_nlink;
+        call_stub_t   *stub;
 };
 typedef struct quota_synctask quota_synctask_t;
 
@@ -147,7 +149,7 @@ mq_create_xattrs_txn (xlator_t *this, loc_t *loc, struct iatt *buf);
 
 int32_t
 mq_reduce_parent_size_txn (xlator_t *, loc_t *, quota_meta_t *,
-                           uint32_t nlink);
+                           uint32_t nlink, call_stub_t *stub);
 
 int32_t
 mq_forget (xlator_t *, quota_inode_ctx_t *);
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index f4fdf2e415b..a1da66bf4e3 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -959,6 +959,7 @@ marker_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 {
         marker_conf_t      *priv    = NULL;
         marker_local_t     *local   = NULL;
+        call_stub_t        *stub    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_TRACE, "error occurred "
@@ -968,21 +969,40 @@ marker_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         local = (marker_local_t *) frame->local;
 
         frame->local = NULL;
-
-        STACK_UNWIND_STRICT (rmdir, frame, op_ret, op_errno, preparent,
-                             postparent, xdata);
+        priv = this->private;
 
         if (op_ret == -1 || local == NULL)
                 goto out;
 
-        priv = this->private;
-
-        if (priv->feature_enabled & GF_QUOTA)
-                mq_reduce_parent_size_txn (this, &local->loc, NULL, 1);
-
         if (priv->feature_enabled & GF_XTIME)
                 marker_xtime_update_marks (this, local);
+
+        if (priv->feature_enabled & GF_QUOTA) {
+                /* If a 'rm -rf' is performed by a client, rmdir can be faster
+                   than marker background mq_reduce_parent_size_txn.
+                   In this case, as part of rmdir parent child association
+                   will be removed in the server protocol.
+                   This can lead to mq_reduce_parent_size_txn failures.
+
+                   So perform mq_reduce_parent_size_txn in foreground
+                   and unwind to server once txn is complete
+                 */
+
+                stub = fop_rmdir_cbk_stub (frame, default_rmdir_cbk, op_ret,
+                                           op_errno, preparent, postparent,
+                                           xdata);
+                mq_reduce_parent_size_txn (this, &local->loc, NULL, 1, stub);
+
+                if (stub) {
+                        marker_local_unref (local);
+                        return 0;
+                }
+        }
+
 out:
+        STACK_UNWIND_STRICT (rmdir, frame, op_ret, op_errno, preparent,
+                             postparent, xdata);
+
         marker_local_unref (local);
 
         return 0;
@@ -1029,6 +1049,7 @@ marker_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         marker_local_t     *local   = NULL;
         uint32_t            nlink   = -1;
         GF_UNUSED int32_t   ret     = 0;
+        call_stub_t        *stub    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_TRACE,
@@ -1038,34 +1059,54 @@ marker_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         local = (marker_local_t *) frame->local;
 
         frame->local = NULL;
-
-        STACK_UNWIND_STRICT (unlink, frame, op_ret, op_errno, preparent,
-                             postparent, xdata);
+        priv = this->private;
 
         if (op_ret == -1 || local == NULL)
                 goto out;
 
-        priv = this->private;
+        if (priv->feature_enabled & GF_XTIME)
+                marker_xtime_update_marks (this, local);
 
         if (priv->feature_enabled & GF_QUOTA) {
-                if (!local->skip_txn) {
-                        if (xdata) {
-                                ret = dict_get_uint32 (xdata,
-                                        GF_RESPONSE_LINK_COUNT_XDATA, &nlink);
-                                if (ret) {
-                                        gf_log (this->name, GF_LOG_TRACE,
-                                                "dict get failed %s ",
-                                                strerror (-ret));
-                                }
+                if (local->skip_txn)
+                        goto out;
+
+                if (xdata) {
+                        ret = dict_get_uint32 (xdata,
+                                GF_RESPONSE_LINK_COUNT_XDATA, &nlink);
+                        if (ret) {
+                                gf_log (this->name, GF_LOG_TRACE,
+                                        "dict get failed %s ",
+                                        strerror (-ret));
                         }
-                        mq_reduce_parent_size_txn (this, &local->loc, NULL,
-                                                   nlink);
+                }
+
+                /* If a 'rm -rf' is performed by a client, unlink can be faster
+                   than marker background mq_reduce_parent_size_txn.
+                   In this case, as part of unlink parent child association
+                   will be removed in the server protocol.
+                   This can lead to mq_reduce_parent_size_txn failures.
+
+                  So perform mq_reduce_parent_size_txn in foreground
+                  and unwind to server once txn is complete
+                */
+
+                stub = fop_unlink_cbk_stub (frame, default_unlink_cbk, op_ret,
+                                            op_errno, preparent, postparent,
+                                            xdata);
+                mq_reduce_parent_size_txn (this, &local->loc, NULL, nlink,
+                                           stub);
+
+                if (stub) {
+                        marker_local_unref (local);
+                        return 0;
                 }
         }
 
-        if (priv->feature_enabled & GF_XTIME)
-                marker_xtime_update_marks (this, local);
 out:
+        STACK_UNWIND_STRICT (unlink, frame, op_ret, op_errno, preparent,
+                             postparent, xdata);
+
         marker_local_unref (local);
 
         return 0;
@@ -1229,14 +1270,14 @@ marker_rename_done (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto err;
 
         mq_reduce_parent_size_txn (this, &oplocal->loc, &oplocal->contribution,
-                                   -1);
+                                   -1, NULL);
 
         if (local->loc.inode != NULL) {
                 /* If destination file exits before rename, it would have
                  * been unlinked while renaming a file
                  */
                 mq_reduce_parent_size_txn (this, &local->loc, NULL,
-                                           local->ia_nlink);
+                                           local->ia_nlink, NULL);
         }
 
         newloc.inode = inode_ref (oplocal->loc.inode);