cluster/afr: post-op should complete before starting flush

Problem:
At the moment afr-flush makes sure that a delayed post-op
is woken up but it does not wait for it to complete the
post-op before flush unwinds.
These are the steps that are happening:
1) flush fop comes on an fd which wakes up a delayed post-op
and continues with the flush fop.
2) post-op sends fsync on the wire.
3) flush completes and unwinds to fuse.
4) graph switch happens on the fuse mount disconnecting the
old graph's client connections to bricks.
5) xattrop after fsync fails with ENOTCONN because the
connections from old graph are taken down now.

Fix:
Wait for post-op to complete before starting to flush.
We could make flush act similar to fsync (i.e.) wind
flush as is but wait for post-op to complete before unwinding
flush, but it is better to send flush as the final fop. So
wind of flush will start after post-op is complete. Had to
change fsync to accommodate this change.

Change-Id: I93aa642647751969511718b0e137afbd067b388a
BUG: 980548
Signed-off-by: Pranith Kumar K <pkarampu@redhat.com>
Reviewed-on: http://review.gluster.org/5274
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

3 files changed, 43 insertions, 36 deletions

diff --git a/xlators/cluster/afr/src/afr-common.c b/xlators/cluster/afr/src/afr-common.c
index 3fcf53d8153..6c6964deceb 100644
--- a/xlators/cluster/afr/src/afr-common.c
+++ b/xlators/cluster/afr/src/afr-common.c
@@ -2573,15 +2573,37 @@ afr_flush_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
         return 0;
 }
 
+static int
+afr_flush_wrapper (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
+{
+        int           i      = 0;
+        afr_local_t   *local = NULL;
+        afr_private_t *priv  = NULL;
+
+        priv = this->private;
+        local = frame->local;
+
+        for (i = 0; i < priv->child_count; i++) {
+                if (local->child_up[i]) {
+                        STACK_WIND_COOKIE (frame, afr_flush_cbk,
+                                           (void *) (long) i,
+                                           priv->children[i],
+                                           priv->children[i]->fops->flush,
+                                           local->fd, NULL);
+                }
+        }
+
+        return 0;
+}
+
 int
 afr_flush (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
 {
         afr_private_t *priv  = NULL;
         afr_local_t   *local = NULL;
+        call_stub_t   *stub = NULL;
         int            ret        = -1;
         int            op_errno   = 0;
-	int	       call_count = -1;
-	int	       i = 0;
 
         VALIDATE_OR_GOTO (frame, out);
         VALIDATE_OR_GOTO (this, out);
@@ -2597,23 +2619,14 @@ afr_flush (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
 		goto out;
 
 	local->fd = fd_ref(fd);
-        call_count = local->call_count;
-
-	afr_delayed_changelog_wake_up (this, fd);
-
-        for (i = 0; i < priv->child_count; i++) {
-                if (local->child_up[i]) {
-                        STACK_WIND_COOKIE (frame, afr_flush_cbk,
-                                           (void *) (long) i,
-                                           priv->children[i],
-                                           priv->children[i]->fops->flush,
-                                           local->fd, NULL);
-
-                        if (!--call_count)
-                                break;
-                }
+        stub = fop_flush_stub (frame, afr_flush_wrapper, fd, xdata);
+        if (!stub) {
+                ret = -1;
+                op_errno = ENOMEM;
+                goto out;
         }
 
+        afr_delayed_changelog_wake_resume (this, fd, stub);
 	ret = 0;
 
 out:
@@ -2689,6 +2702,16 @@ afr_release (xlator_t *this, fd_t *fd)
 /* {{{ fsync */
 
 int
+afr_fsync_unwind_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                      int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                      struct iatt *postbuf, dict_t *xdata)
+{
+        AFR_STACK_UNWIND (fsync, frame, op_ret, op_errno, prebuf, postbuf,
+                          xdata);
+        return 0;
+}
+
+int
 afr_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                struct iatt *postbuf, dict_t *xdata)
@@ -2741,7 +2764,7 @@ afr_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 		   post-op. This guarantee is expected by FUSE graph switching
 		   for example.
 		*/
-		stub = fop_fsync_cbk_stub (frame, default_fsync_cbk,
+		stub = fop_fsync_cbk_stub (frame, afr_fsync_unwind_cbk,
                                            local->op_ret, local->op_errno,
                                            &local->cont.fsync.prebuf,
                                            &local->cont.fsync.postbuf, xdata);
diff --git a/xlators/cluster/afr/src/afr-transaction.c b/xlators/cluster/afr/src/afr-transaction.c
index 0ae4fe47761..dbc74ceba53 100644
--- a/xlators/cluster/afr/src/afr-transaction.c
+++ b/xlators/cluster/afr/src/afr-transaction.c
@@ -369,7 +369,7 @@ afr_changelog_post_op_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         if (call_count == 0) {
                 if (local->transaction.resume_stub) {
-			AFR_CALL_RESUME (local->transaction.resume_stub);
+			call_resume (local->transaction.resume_stub);
                         local->transaction.resume_stub = NULL;
                 }
 
@@ -1598,7 +1598,7 @@ out:
 		local->transaction.resume_stub = stub;
 		afr_changelog_post_op_safe (prev_frame, this);
 	} else if (stub) {
-		AFR_CALL_RESUME (stub);
+		call_resume (stub);
 	}
 }
 
diff --git a/xlators/cluster/afr/src/afr.h b/xlators/cluster/afr/src/afr.h
index 9594a828773..836a64d0a1f 100644
--- a/xlators/cluster/afr/src/afr.h
+++ b/xlators/cluster/afr/src/afr.h
@@ -971,22 +971,6 @@ afr_launch_openfd_self_heal (call_frame_t *frame, xlator_t *this, fd_t *fd);
                 }                                               \
         } while (0);
 
-#define AFR_CALL_RESUME(stub)					\
-        do {                                                    \
-                afr_local_t *__local = NULL;                    \
-                xlator_t    *__this = NULL;                     \
-								\
-		__local = stub->frame->local;			\
-		__this = stub->frame->this;			\
-		stub->frame->local = NULL;			\
-								\
-		call_resume (stub);				\
-                if (__local) {                                  \
-                        afr_local_cleanup (__local, __this);    \
-                        mem_put (__local);                      \
-                }                                               \
-        } while (0)
-
 #define AFR_NUM_CHANGE_LOGS            3 /*data + metadata + entry*/
 /* allocate and return a string that is the basename of argument */
 static inline char *