cluster/afr: Fixing the flaws in arbiter becoming source patch

Problem:
Setting the write_subvol value to read_subvol in case of metadata
transaction during pre-op (commit 19f9bcff4aada589d4321356c2670ed283f02c03)
might lead to the original problem of arbiter becoming source.

Scenario:
1) All bricks are up and good
2) 2 writes w1 and w2 are in progress in parallel
3) ctx->read_subvol is good for all the subvolumes
4) w1 succeeds on brick0 and fails on brick1, yet to do post-op on
   the disk
5) read/lookup comes on the same file and refreshes read_subvols back
   to all good
6) metadata transaction happens which makes ctx->write_subvol to be
   assigned with ctx->read_subvol which is all good
7) w2 succeeds on brick1 and fails on brick0 and this will update the
   brick in reverse order leading to arbiter becoming source

Fix:
Instead of setting the ctx->write_subvol to ctx->read_subvol in the
pre-op statge, if there is a metadata transaction, check in the
function __afr_set_in_flight_sb_status() if it is a data/metadata
transaction. Use the value of ctx->write_subvol if it is a data
transactions and ctx->read_subvol value for other transactions.

With this patch we assign the value of ctx->write_subvol in the
afr_transaction_perform_fop() with the on disk value, instead of
assigning it in the afr_changelog_pre_op() with the in memory value.

Change-Id: Id2025a7e965f0578af35b1abaac793b019c43cc4
BUG: 1482064
Signed-off-by: karthik-us <ksubrahm@redhat.com>

1 files changed, 27 insertions, 15 deletions

diff --git a/xlators/cluster/afr/src/afr-lk-common.c b/xlators/cluster/afr/src/afr-lk-common.c
index c17f60f62c4..f50c7b6464a 100644
--- a/xlators/cluster/afr/src/afr-lk-common.c
+++ b/xlators/cluster/afr/src/afr-lk-common.c
@@ -615,14 +615,14 @@ afr_unlock_common_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 {
         afr_local_t             *local          = NULL;
         afr_internal_lock_t     *int_lock       = NULL;
-        afr_fd_ctx_t            *fd_ctx         = NULL;
-        afr_private_t           *priv           = NULL;
         int                      call_count     = 0;
         int                      ret            = 0;
 
         local    = frame->local;
         int_lock = &local->internal_lock;
-        priv = this->private;
+
+        if (local->transaction.type == AFR_DATA_TRANSACTION && op_ret != 1)
+                ret = afr_write_subvol_reset (frame, this);
 
         LOCK (&frame->lock);
         {
@@ -633,11 +633,6 @@ afr_unlock_common_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (call_count == 0) {
                 gf_msg_trace (this->name, 0,
                               "All internal locks unlocked");
-                if (local->fd) {
-                        fd_ctx = afr_fd_ctx_get (local->fd, this);
-                        if (0 == AFR_COUNT (fd_ctx->lock_acquired, priv->child_count))
-                                ret = afr_write_subvol_reset (frame, this);
-                }
                 int_lock->lock_cbk (frame, this);
         }
 
@@ -947,6 +942,15 @@ afr_lock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         } else {
                                 int_lock->locked_nodes[child_index] |= LOCKED_YES;
                                 int_lock->lock_count++;
+
+                                if (local->transaction.type ==
+                                    AFR_DATA_TRANSACTION) {
+                                        LOCK(&local->inode->lock);
+                                        {
+                                                local->inode_ctx->lock_count++;
+                                        }
+                                        UNLOCK (&local->inode->lock);
+                                }
                         }
                 }
                 afr_lock_blocking (frame, this, cky + 1);
@@ -1502,13 +1506,12 @@ int32_t
 afr_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                              int32_t op_ret, int32_t op_errno, dict_t *xdata)
 {
-        afr_internal_lock_t *int_lock = NULL;
-        afr_inodelk_t       *inodelk  = NULL;
-        afr_local_t         *local    = NULL;
-        int call_count  = 0;
-        int child_index = (long) cookie;
-        afr_fd_ctx_t        *fd_ctx = NULL;
-
+        afr_internal_lock_t *int_lock    = NULL;
+        afr_inodelk_t       *inodelk     = NULL;
+        afr_local_t         *local       = NULL;
+        afr_fd_ctx_t        *fd_ctx      = NULL;
+        int                  call_count  = 0;
+        int                  child_index = (long) cookie;
 
         local    = frame->local;
         int_lock = &local->internal_lock;
@@ -1553,6 +1556,15 @@ afr_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                                         fd_ctx->lock_acquired[child_index]++;
 				}
 			}
+
+                        if (local->transaction.type == AFR_DATA_TRANSACTION &&
+                            op_ret == 0) {
+                                LOCK(&local->inode->lock);
+                                {
+                                        local->inode_ctx->lock_count++;
+                                }
+                                UNLOCK (&local->inode->lock);
+                        }
 		}
 
                 call_count = --int_lock->lk_call_count;