cluster/dht: synchronize with other concurrent healers while healing layout.

Current layout heal code assumes layout setting is idempotent. This
allowed multiple concurrent healers to set the layout without any
synchronization. However, this is not the case as different healers
can come up with different layout for same directory and making layout
setting non-idempotent. So, we bring in synchronization among healers
to
   1. Not to overwrite an ondisk well-formed layout.
   2. Refresh the in-memory layout with the ondisk layout if in-memory
   layout needs healing and ondisk layout is well formed.

This patch can synchronize
   1. among multiple healers.
   2. among multiple fix-layouts (which extends layout to consider
   added or removed brick)
   3. (but) not between healers and fix-layouts. So, the problem of
   in-memory stale layouts (not matching with layout ondisk), is not
   _completely_ fixed by this patch.

Signed-off-by: Raghavendra G <rgowdapp@redhat.com>
Change-Id: Ia285f25e8d043bb3175c61468d0d11090acee539
BUG: 1176008
Reviewed-on: http://review.gluster.org/9302
Reviewed-by: N Balachandran <nbalacha@redhat.com>

1 files changed, 22 insertions, 1 deletions

diff --git a/xlators/cluster/dht/src/dht-helper.c b/xlators/cluster/dht/src/dht-helper.c
index c25646c3c5d..f6b3362da4f 100644
--- a/xlators/cluster/dht/src/dht-helper.c
+++ b/xlators/cluster/dht/src/dht-helper.c
@@ -1370,6 +1370,22 @@ out:
         return ret;
 }
 
+void
+dht_set_lkowner (dht_lock_t **lk_array, int count, gf_lkowner_t *lkowner)
+{
+        int i = 0;
+
+        if (!lk_array || !lkowner)
+                goto out;
+
+        for (i = 0; i < count; i++) {
+                lk_array[i]->lk_owner = *lkowner;
+        }
+
+out:
+        return;
+}
+
 int
 dht_subvol_status (dht_conf_t *conf, xlator_t *subvol)
 {
@@ -1415,7 +1431,7 @@ dht_inodelk_cleanup_cbk (call_frame_t *frame, void *cookie,
         return 0;
 }
 
-static inline int32_t
+inline int32_t
 dht_lock_count (dht_lock_t **lk_array, int lk_count)
 {
         int i = 0, locked = 0;
@@ -1553,6 +1569,7 @@ dht_unlock_inodelk (call_frame_t *frame, dht_lock_t **lk_array, int lk_count,
                 if (!local->lock.locks[i]->locked)
                         continue;
 
+                lock_frame->root->lk_owner = local->lock.locks[i]->lk_owner;
                 STACK_WIND_COOKIE (lock_frame, dht_unlock_inodelk_cbk,
                                    (void *)(long)i,
                                    local->lock.locks[i]->xl,
@@ -1642,6 +1659,8 @@ dht_nonblocking_inodelk (call_frame_t *frame, dht_lock_t **lk_array,
                 goto out;
         }
 
+        dht_set_lkowner (lk_array, lk_count, &lock_frame->root->lk_owner);
+
         local = lock_frame->local;
         local->main_frame = frame;
 
@@ -1780,6 +1799,8 @@ dht_blocking_inodelk (call_frame_t *frame, dht_lock_t **lk_array,
                 goto out;
         }
 
+        dht_set_lkowner (lk_array, lk_count, &lock_frame->root->lk_owner);
+
         local = lock_frame->local;
         local->main_frame = frame;