afr: [Un]Set the 'right' lkowner for [f]{inode|entry}_lk and the 'enclosed' fop.

afr 'mangles' the lkowner inorder to ensure [f]inodelk/[f]entrylk fops from the
same application contend. But other fops that are 'visible' to the application
should operate with the lkowner provided by fuse for correct functioning of
posix-locks xlator.

Change-Id: I7e71f35ae7df2a070f1f46d4fc77eed26a717673
BUG: 790743
Signed-off-by: Krishnan Parthasarathi <kp@gluster.com>
Reviewed-on: http://review.gluster.com/2752
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vijay@gluster.com>

6 files changed, 94 insertions, 22 deletions

diff --git a/xlators/cluster/afr/src/afr-lk-common.c b/xlators/cluster/afr/src/afr-lk-common.c
index e9a3a430dd0..ebeb588e99c 100644
--- a/xlators/cluster/afr/src/afr-lk-common.c
+++ b/xlators/cluster/afr/src/afr-lk-common.c
@@ -31,6 +31,14 @@
 #define LOCKED_YES      0x1        /* for DATA, METADATA, ENTRY and higher_path */
 #define LOCKED_LOWER    0x2        /* for lower path */
 
+static inline void
+afr_unset_lk_owner_and_call_cbk (call_frame_t *frame, xlator_t *this,
+                                afr_internal_lock_t *int_lock)
+{
+         afr_unset_lk_owner (frame, this, int_lock);
+         int_lock->lock_cbk (frame, this);
+}
+
 int
 afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index);
 
@@ -56,14 +64,58 @@ afr_set_lock_number (call_frame_t *frame, xlator_t *this)
         return 0;
 }
 
+static void
+afr_lkowner_trace_log (xlator_t *this, gf_lkowner_t *from, gf_lkowner_t *to)
+{
+        char            from_lkowner[1024]      = {0, };
+        char            to_lkowner[1024]        = {0, };
+        afr_private_t   *priv                   = NULL;
+
+        priv = this->private;
+
+        if (!priv->entrylk_trace && !priv->inodelk_trace)
+                return;
+
+        lkowner_utoa_r (from, from_lkowner, sizeof (from_lkowner));
+        lkowner_utoa_r (to, to_lkowner, sizeof (to_lkowner));
+
+        gf_log (this->name, GF_LOG_INFO, "Changing lk-owner from %s to %s",
+                from_lkowner, to_lkowner);
+}
+
 void
-afr_set_lk_owner (call_frame_t *frame, xlator_t *this)
+afr_set_lk_owner (call_frame_t *frame, xlator_t *this,
+                  afr_internal_lock_t *int_lock)
 {
-        gf_log (this->name, GF_LOG_TRACE,
-                "Setting lk-owner=%llu",
-                (unsigned long long) (unsigned long)frame->root);
+
+        if (int_lock)
+                memcpy (&int_lock->fop_lkowner, &frame->root->lk_owner,
+                        sizeof (int_lock->fop_lkowner));
 
         set_lk_owner_from_ptr (&frame->root->lk_owner, frame->root);
+
+        afr_lkowner_trace_log (this, &int_lock->fop_lkowner,
+                               &frame->root->lk_owner);
+}
+
+void
+afr_unset_lk_owner (call_frame_t *frame, xlator_t *this,
+                    afr_internal_lock_t *int_lock)
+{
+        if (!int_lock) {
+                gf_log (this->name, GF_LOG_CRITICAL, "Attempting to unset "
+                        "lk_owner without saved fop lk_owner");
+                goto out;
+        }
+
+        memcpy (&frame->root->lk_owner, &int_lock->fop_lkowner,
+                sizeof (int_lock->fop_lkowner));
+
+        afr_lkowner_trace_log (this, &int_lock->fop_lkowner,
+                               &frame->root->lk_owner);
+
+out:
+        return;
 }
 
 static int
@@ -463,6 +515,8 @@ initialize_entrylk_variables (call_frame_t *frame, xlator_t *this)
                 int_lock->entry_locked_nodes[i] = 0;
         }
 
+        afr_set_lk_owner (frame, this, int_lock);
+
         return 0;
 }
 
@@ -486,6 +540,8 @@ initialize_inodelk_variables (call_frame_t *frame, xlator_t *this)
                 int_lock->inode_locked_nodes[i] = 0;
         }
 
+        afr_set_lk_owner (frame, this, int_lock);
+
         return 0;
 }
 
@@ -541,7 +597,7 @@ afr_unlock_common_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         if (call_count == 0) {
                 gf_log (this->name, GF_LOG_TRACE,
                         "All internal locks unlocked");
-                int_lock->lock_cbk (frame, this);
+                afr_unset_lk_owner_and_call_cbk (frame, this, int_lock);
         }
 
         return 0;
@@ -614,7 +670,7 @@ afr_unlock_inodelk (call_frame_t *frame, xlator_t *this)
         if (!call_count) {
                 gf_log (this->name, GF_LOG_TRACE,
                         "No internal locks unlocked");
-                int_lock->lock_cbk (frame, this);
+                afr_unset_lk_owner_and_call_cbk (frame, this, int_lock);
                 goto out;
         }
 
@@ -734,7 +790,7 @@ afr_unlock_entrylk (call_frame_t *frame, xlator_t *this)
         if (!call_count){
                 gf_log (this->name, GF_LOG_TRACE,
                         "No internal locks unlocked");
-                int_lock->lock_cbk (frame, this);
+                afr_unset_lk_owner_and_call_cbk (frame, this, int_lock);
                 goto out;
         }
 
@@ -792,6 +848,7 @@ afr_lock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         if ((op_ret == -1) &&
             (op_errno == ENOSYS)) {
+                afr_unset_lk_owner (frame, this, int_lock);
                 afr_unlock (frame, this);
         } else {
                 if (op_ret == 0) {
@@ -852,6 +909,7 @@ afr_lock_lower_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         UNLOCK (&frame->lock);
 
         if (op_ret != 0) {
+                afr_unset_lk_owner (frame, this, int_lock);
                 afr_unlock (frame, this);
                 goto out;
         } else {
@@ -967,6 +1025,7 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
 
                         afr_copy_locked_nodes (frame, this);
 
+                        afr_unset_lk_owner (frame, this, int_lock);
                         afr_unlock (frame, this);
 
                         return 0;
@@ -998,6 +1057,7 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
 
                 afr_copy_locked_nodes (frame, this);
 
+                afr_unset_lk_owner (frame, this, int_lock);
                 afr_unlock(frame, this);
 
                 return 0;
@@ -1015,7 +1075,7 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
                 afr_copy_locked_nodes (frame, this);
 
                 int_lock->lock_op_ret = 0;
-                int_lock->lock_cbk (frame, this);
+                afr_unset_lk_owner_and_call_cbk (frame, this, int_lock);
                 return 0;
         }
 
@@ -1190,7 +1250,7 @@ afr_nonblocking_entrylk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         gf_log (this->name, GF_LOG_TRACE,
                                 "All servers locked. Calling the cbk");
                         int_lock->lock_op_ret = 0;
-                        int_lock->lock_cbk (frame, this);
+                        afr_unset_lk_owner_and_call_cbk (frame, this, int_lock);
                 }
                 /* Not all locks were successful. Unlock and try locking
                    again, this time with serially blocking locks */
@@ -1199,6 +1259,7 @@ afr_nonblocking_entrylk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                                 "%d servers locked. Trying again with blocking calls",
                                 int_lock->lock_count);
 
+                        afr_unset_lk_owner (frame, this, int_lock);
                         afr_unlock(frame, this);
                 }
         }
@@ -1265,6 +1326,7 @@ afr_nonblocking_entrylk (call_frame_t *frame, xlator_t *this)
                 if (!call_count) {
                         gf_log (this->name, GF_LOG_INFO,
                                 "fd not open on any subvolumes. aborting.");
+                        afr_unset_lk_owner (frame, this, int_lock);
                         afr_unlock (frame, this);
                         goto out;
                 }
@@ -1383,7 +1445,7 @@ afr_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         gf_log (this->name, GF_LOG_TRACE,
                                 "All servers locked. Calling the cbk");
                         int_lock->lock_op_ret = 0;
-                        int_lock->lock_cbk (frame, this);
+                        afr_unset_lk_owner_and_call_cbk (frame, this, int_lock);
                 }
                 /* Not all locks were successful. Unlock and try locking
                    again, this time with serially blocking locks */
@@ -1392,6 +1454,7 @@ afr_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                                 "%d servers locked. Trying again with blocking calls",
                                 int_lock->lock_count);
 
+                        afr_unset_lk_owner (frame, this, int_lock);
                         afr_unlock(frame, this);
                 }
         }
@@ -1454,6 +1517,7 @@ afr_nonblocking_inodelk (call_frame_t *frame, xlator_t *this)
                 if (!call_count) {
                         gf_log (this->name, GF_LOG_INFO,
                                 "fd not open on any subvolumes. aborting.");
+                        afr_unset_lk_owner (frame, this, int_lock);
                         afr_unlock (frame, this);
                         goto out;
                 }
@@ -1647,6 +1711,7 @@ afr_post_unlock_lower_cbk (call_frame_t *frame, xlator_t *this)
 
         local    = frame->local;
         int_lock = &local->internal_lock;
+        afr_set_lk_owner (frame, this, int_lock);
 
         lower = lower_path (&local->transaction.parent_loc,
                             local->transaction.basename,
@@ -1724,9 +1789,13 @@ afr_rename_transaction (call_frame_t *frame, xlator_t *this)
 int32_t
 afr_unlock (call_frame_t *frame, xlator_t *this)
 {
-        afr_local_t *local = NULL;
+        afr_local_t             *local = NULL;
+        afr_internal_lock_t     *int_lock = NULL;
 
         local = frame->local;
+        int_lock = &local->internal_lock;
+
+        afr_set_lk_owner (frame, this, int_lock);
 
         if (transaction_lk_op (local)) {
                 if (is_afr_lock_transaction (local))
diff --git a/xlators/cluster/afr/src/afr-self-heal-common.c b/xlators/cluster/afr/src/afr-self-heal-common.c
index 0ff959240bc..b7649b1271d 100644
--- a/xlators/cluster/afr/src/afr-self-heal-common.c
+++ b/xlators/cluster/afr/src/afr-self-heal-common.c
@@ -2070,7 +2070,6 @@ afr_self_heal (call_frame_t *frame, xlator_t *this, inode_t *inode)
         sh_frame        = copy_frame (frame);
         if (!sh_frame)
                 goto out;
-        afr_set_lk_owner (sh_frame, this);
         afr_set_low_priority (sh_frame);
 
         sh_local        = afr_local_copy (local, this);
diff --git a/xlators/cluster/afr/src/afr-self-heald.c b/xlators/cluster/afr/src/afr-self-heald.c
index fa7e61e49e8..c8fb4d38ee6 100644
--- a/xlators/cluster/afr/src/afr-self-heald.c
+++ b/xlators/cluster/afr/src/afr-self-heald.c
@@ -1031,7 +1031,6 @@ afr_start_crawl (xlator_t *this, int idx, afr_crawl_type_t crawl,
         if (!frame)
                 goto out;
 
-        afr_set_lk_owner (frame, this);
         afr_set_low_priority (frame);
         crawl_data = GF_CALLOC (1, sizeof (*crawl_data),
                                 gf_afr_mt_crawl_data_t);
diff --git a/xlators/cluster/afr/src/afr-transaction.c b/xlators/cluster/afr/src/afr-transaction.c
index 36e2812f968..eb41c51841c 100644
--- a/xlators/cluster/afr/src/afr-transaction.c
+++ b/xlators/cluster/afr/src/afr-transaction.c
@@ -1168,8 +1168,6 @@ afr_lock (call_frame_t *frame, xlator_t *this)
 
         frame->root->pid = (long) frame->root;
 
-        afr_set_lk_owner (frame, this);
-
         afr_set_lock_number (frame, this);
 
         return afr_lock_rec (frame, this);
diff --git a/xlators/cluster/afr/src/afr.h b/xlators/cluster/afr/src/afr.h
index f0cb32c12b1..87b557bcb1e 100644
--- a/xlators/cluster/afr/src/afr.h
+++ b/xlators/cluster/afr/src/afr.h
@@ -335,6 +335,9 @@ afr_index_for_transaction_type (afr_transaction_type type)
 typedef struct {
         loc_t *lk_loc;
         struct gf_flock lk_flock;
+        gf_lkowner_t fop_lkowner; /* 'copy' of fop's lk_owner to protect
+                                      against afr internal locks 'irreversibly'
+                                      overloading lk_owner.*/
 
         const char *lk_basename;
         const char *lower_basename;
@@ -769,7 +772,12 @@ afr_mark_locked_nodes (xlator_t *this, fd_t *fd,
                        unsigned char *locked_nodes);
 
 void
-afr_set_lk_owner (call_frame_t *frame, xlator_t *this);
+afr_set_lk_owner (call_frame_t *frame, xlator_t *this,
+                  afr_internal_lock_t *int_lock);
+
+void
+afr_unset_lk_owner (call_frame_t *frame, xlator_t *this,
+                    afr_internal_lock_t *int_lock);
 
 int
 afr_set_lock_number (call_frame_t *frame, xlator_t *this);
diff --git a/xlators/cluster/afr/src/pump.c b/xlators/cluster/afr/src/pump.c
index e795c38e096..bc95791004b 100644
--- a/xlators/cluster/afr/src/pump.c
+++ b/xlators/cluster/afr/src/pump.c
@@ -696,15 +696,14 @@ pump_task_completion (int ret, call_frame_t *sync_frame, void *data)
 int
 pump_start (call_frame_t *pump_frame, xlator_t *this)
 {
-	afr_private_t *priv = NULL;
-	pump_private_t *pump_priv = NULL;
-
-	int ret = -1;
+        afr_private_t           *priv           = NULL;
+        pump_private_t          *pump_priv      = NULL;
+        int ret = -1;
 
-	priv = this->private;
+        priv = this->private;
         pump_priv = priv->pump_private;
 
-        afr_set_lk_owner (pump_frame, this);
+        afr_set_lk_owner (pump_frame, this, NULL);
 	pump_pid = (uint64_t) (unsigned long)pump_frame->root;
 
 	ret = synctask_new (pump_priv->env, pump_task,