cluster/afr: Enable eager-lock

Eager-lock is disabled by default.
Use cluster.eager-lock on/off to change the config.
write-behind on and eager-lock off is not supported configuration.

In afr, when eager-lock is enabled the inode lock on fd is taken
using the fd address as the lk-owner. So the lock is
interchangableale between the inode-locks on the same fd.

Change-Id: I7eef1ecd510f8028f5395dee882782da53c0de3f
BUG: 802515
Signed-off-by: Pranith Kumar K <pranithk@gluster.com>
Reviewed-on: http://review.gluster.com/2925
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Jeff Darcy <jdarcy@redhat.com>
Reviewed-by: Anand Avati <avati@redhat.com>

9 files changed, 219 insertions, 122 deletions

diff --git a/cli/src/cli-rpc-ops.c b/cli/src/cli-rpc-ops.c
index 4be25c9e4..1477ef188 100644
--- a/cli/src/cli-rpc-ops.c
+++ b/cli/src/cli-rpc-ops.c
@@ -1318,7 +1318,7 @@ gf_cli3_1_set_volume_cbk (struct rpc_req *req, struct iovec *iov,
         if (ret)
                 goto out;
 
-        if (dict_get_str (dict, "help-str", &help_str))
+        if (dict_get_str (dict, "help-str", &help_str) && !msg[0])
                 snprintf (msg, sizeof (msg), "Set volume %s",
                           (rsp.op_ret) ? "unsuccessful": "successful");
 
diff --git a/xlators/cluster/afr/src/afr-lk-common.c b/xlators/cluster/afr/src/afr-lk-common.c
index 39646ff85..aeac72d16 100644
--- a/xlators/cluster/afr/src/afr-lk-common.c
+++ b/xlators/cluster/afr/src/afr-lk-common.c
@@ -31,6 +31,38 @@
 #define LOCKED_YES      0x1        /* for DATA, METADATA, ENTRY and higher_path */
 #define LOCKED_LOWER    0x2        /* for lower path */
 
+#define AFR_TRACE_INODELK_IN(frame, this, params ...)           \
+        do {                                                    \
+                afr_private_t *_priv = this->private;           \
+                if (!_priv->inodelk_trace)                      \
+                        break;                                  \
+                afr_trace_inodelk_in (frame, this, params);     \
+        } while (0);
+
+#define AFR_TRACE_INODELK_OUT(frame, this, params ...)          \
+        do {                                                    \
+                afr_private_t *_priv = this->private;           \
+                if (!_priv->inodelk_trace)                      \
+                        break;                                  \
+                afr_trace_inodelk_out (frame, this, params);    \
+        } while (0);
+
+#define AFR_TRACE_ENTRYLK_IN(frame, this, params ...)           \
+        do {                                                    \
+                afr_private_t *_priv = this->private;           \
+                if (!_priv->entrylk_trace)                      \
+                        break;                                  \
+                afr_trace_entrylk_in (frame, this, params);     \
+        } while (0);
+
+#define AFR_TRACE_ENTRYLK_OUT(frame, this, params ...)          \
+        do {                                                    \
+                afr_private_t *_priv = this->private;           \
+                if (!_priv->entrylk_trace)                      \
+                        break;                                  \
+                afr_trace_entrylk_out (frame, this, params);    \
+        } while (0);
+
 int
 afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index);
 
@@ -57,13 +89,13 @@ afr_set_lock_number (call_frame_t *frame, xlator_t *this)
 }
 
 void
-afr_set_lk_owner (call_frame_t *frame, xlator_t *this)
+afr_set_lk_owner (call_frame_t *frame, xlator_t *this, void *lk_owner)
 {
         gf_log (this->name, GF_LOG_TRACE,
                 "Setting lk-owner=%llu",
-                (unsigned long long) (unsigned long)frame->root);
+                (unsigned long long) (unsigned long)lk_owner);
 
-        set_lk_owner_from_ptr (&frame->root->lk_owner, frame->root);
+        set_lk_owner_from_ptr (&frame->root->lk_owner, lk_owner);
 }
 
 static int
@@ -242,27 +274,20 @@ afr_set_lock_call_type (afr_lock_call_type_t lock_call_type,
 }
 
 static void
-afr_trace_inodelk_out (call_frame_t *frame, afr_lock_call_type_t lock_call_type,
+afr_trace_inodelk_out (call_frame_t *frame, xlator_t *this,
+                       afr_lock_call_type_t lock_call_type,
                        afr_lock_op_type_t lk_op_type, struct gf_flock *flock,
                        int op_ret, int op_errno, int32_t child_index)
 {
-        xlator_t            *this     = NULL;
         afr_internal_lock_t *int_lock = NULL;
         afr_local_t         *local    = NULL;
-        afr_private_t       *priv     = NULL;
 
         char lockee[256];
         char lock_call_type_str[256];
         char verdict[16];
 
-        this     = THIS;
         local    = frame->local;
         int_lock = &local->internal_lock;
-        priv     = this->private;
-
-        if (!priv->inodelk_trace) {
-                return;
-        }
 
         afr_print_lockee (lockee, 256, &local->loc, local->fd, child_index);
 
@@ -281,27 +306,20 @@ afr_trace_inodelk_out (call_frame_t *frame, afr_lock_call_type_t lock_call_type,
 }
 
 static void
-afr_trace_inodelk_in (call_frame_t *frame, afr_lock_call_type_t lock_call_type,
+afr_trace_inodelk_in (call_frame_t *frame, xlator_t *this,
+                      afr_lock_call_type_t lock_call_type,
                       afr_lock_op_type_t lk_op_type, struct gf_flock *flock,
                       int32_t cmd, int32_t child_index)
 {
-        xlator_t            *this     = NULL;
         afr_local_t         *local    = NULL;
         afr_internal_lock_t *int_lock = NULL;
-        afr_private_t       *priv     = NULL;
 
         char lock[256];
         char lockee[256];
         char lock_call_type_str[256];
 
-        this     = THIS;
         local    = frame->local;
         int_lock = &local->internal_lock;
-        priv     = this->private;
-
-        if (!priv->inodelk_trace) {
-                return;
-        }
 
         afr_print_inodelk (lock, 256, cmd, flock, &frame->root->lk_owner);
         afr_print_lockee (lockee, 256, &local->loc, local->fd, child_index);
@@ -318,27 +336,20 @@ afr_trace_inodelk_in (call_frame_t *frame, afr_lock_call_type_t lock_call_type,
 }
 
 static void
-afr_trace_entrylk_in (call_frame_t *frame, afr_lock_call_type_t lock_call_type,
+afr_trace_entrylk_in (call_frame_t *frame, xlator_t *this,
+                      afr_lock_call_type_t lock_call_type,
                       afr_lock_op_type_t lk_op_type, const char *basename,
                       int32_t child_index)
 {
-        xlator_t            *this     = NULL;
         afr_local_t         *local    = NULL;
         afr_internal_lock_t *int_lock = NULL;
-        afr_private_t       *priv     = NULL;
 
         char lock[256];
         char lockee[256];
         char lock_call_type_str[256];
 
-        this     = THIS;
         local    = frame->local;
         int_lock = &local->internal_lock;
-        priv     = this->private;
-
-        if (!priv->entrylk_trace) {
-                return;
-        }
 
         afr_print_entrylk (lock, 256, basename, &frame->root->lk_owner);
         afr_print_lockee (lockee, 256, &local->loc, local->fd, child_index);
@@ -354,28 +365,21 @@ afr_trace_entrylk_in (call_frame_t *frame, afr_lock_call_type_t lock_call_type,
 }
 
 static void
-afr_trace_entrylk_out (call_frame_t *frame, afr_lock_call_type_t lock_call_type,
-                       afr_lock_op_type_t lk_op_type, const char *basename, int op_ret,
-                       int op_errno, int32_t child_index)
+afr_trace_entrylk_out (call_frame_t *frame, xlator_t *this,
+                       afr_lock_call_type_t lock_call_type,
+                       afr_lock_op_type_t lk_op_type, const char *basename,
+                       int op_ret, int op_errno, int32_t child_index)
 {
-        xlator_t            *this     = NULL;
         afr_internal_lock_t *int_lock = NULL;
         afr_local_t         *local    = NULL;
-        afr_private_t       *priv     = NULL;
 
         char lock[256];
         char lockee[256];
         char lock_call_type_str[256];
         char verdict[16];
 
-        this     = THIS;
         local    = frame->local;
         int_lock = &local->internal_lock;
-        priv     = this->private;
-
-        if (!priv->entrylk_trace) {
-                return;
-        }
 
         afr_print_lockee (lockee, 256, &local->loc, local->fd, child_index);
 
@@ -558,22 +562,20 @@ afr_unlock_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         local = frame->local;
         int_lock = &local->internal_lock;
 
-        afr_trace_inodelk_out (frame, AFR_INODELK_TRANSACTION,
+        AFR_TRACE_INODELK_OUT (frame, this, AFR_INODELK_TRANSACTION,
                                AFR_UNLOCK_OP, NULL, op_ret,
                                op_errno, child_index);
 
         if (op_ret < 0 && op_errno != ENOTCONN && op_errno != EBADFD) {
-                gf_log (this->name, GF_LOG_ERROR,
-                        "%s: unlock failed on %d, reason: %s",
-                        local->loc.path, child_index, strerror (op_errno));
+                gf_log (this->name, GF_LOG_INFO, "%s: unlock failed on %d "
+                        "unlock by %s", local->loc.path, child_index,
+                        lkowner_utoa (&frame->root->lk_owner));
         }
 
 
         int_lock->inode_locked_nodes[child_index] &= LOCKED_NO;
-
-        if (op_ret == 1) {
+        if (local->transaction.eager_lock)
                 local->transaction.eager_lock[child_index] = 0;
-        }
 
         afr_unlock_common_cbk (frame, cookie, this, op_ret, op_errno);
 
@@ -588,10 +590,13 @@ afr_unlock_inodelk (call_frame_t *frame, xlator_t *this)
         afr_local_t         *local    = NULL;
         afr_private_t       *priv     = NULL;
         struct gf_flock flock = {0,};
+        struct gf_flock full_flock = {0,};
+        struct gf_flock *flock_use = NULL;
         int call_count = 0;
         int i = 0;
         int piggyback = 0;
         afr_fd_ctx_t        *fd_ctx      = NULL;
+        gf_boolean_t    fd_lock_owner = _gf_false;
 
 
         local    = frame->local;
@@ -602,10 +607,7 @@ afr_unlock_inodelk (call_frame_t *frame, xlator_t *this)
         flock.l_len   = int_lock->lk_flock.l_len;
         flock.l_type  = F_UNLCK;
 
-        gf_log (this->name, GF_LOG_DEBUG, "attempting data unlock range %"PRIu64
-                " %"PRIu64" by %s", flock.l_start, flock.l_len,
-                lkowner_utoa (&frame->root->lk_owner));
-
+        full_flock.l_type = F_UNLCK;
         call_count = afr_locked_nodes_count (int_lock->inode_locked_nodes,
                                              priv->child_count);
 
@@ -627,7 +629,13 @@ afr_unlock_inodelk (call_frame_t *frame, xlator_t *this)
                         continue;
 
                 if (local->fd) {
+                        flock_use = &flock;
                         if (!local->transaction.eager_lock[i]) {
+                                if (fd_lock_owner) {
+                                        afr_set_lk_owner (frame, this,
+                                                          frame->root);
+                                        fd_lock_owner = _gf_false;
+                                }
                                 goto wind;
                         }
 
@@ -638,6 +646,8 @@ afr_unlock_inodelk (call_frame_t *frame, xlator_t *this)
                                 if (fd_ctx->lock_piggyback[i]) {
                                         fd_ctx->lock_piggyback[i]--;
                                         piggyback = 1;
+                                } else {
+                                        fd_ctx->lock_acquired[i]--;
                                 }
                         }
                         UNLOCK (&local->fd->lock);
@@ -650,23 +660,30 @@ afr_unlock_inodelk (call_frame_t *frame, xlator_t *this)
                                 continue;
                         }
 
-                        fd_ctx->lock_acquired[i]--;
+                        if (!fd_lock_owner) {
+                                afr_set_lk_owner (frame, this, local->fd);
+                                fd_lock_owner = _gf_true;
+                        }
+                        flock_use = &full_flock;
                 wind:
-                        afr_trace_inodelk_in (frame, AFR_INODELK_TRANSACTION,
-                                              AFR_UNLOCK_OP, &flock, F_SETLK, i);
+                        AFR_TRACE_INODELK_IN (frame, this,
+                                              AFR_INODELK_TRANSACTION,
+                                              AFR_UNLOCK_OP, flock_use, F_SETLK,
+                                              i);
 
                         STACK_WIND_COOKIE (frame, afr_unlock_inodelk_cbk,
                                            (void *) (long)i,
                                            priv->children[i],
                                            priv->children[i]->fops->finodelk,
                                            this->name, local->fd,
-                                           F_SETLK, &flock);
+                                           F_SETLK, flock_use);
 
                         if (!--call_count)
                                 break;
 
                 } else {
-                        afr_trace_inodelk_in (frame, AFR_INODELK_TRANSACTION,
+                        AFR_TRACE_INODELK_IN (frame, this,
+                                              AFR_INODELK_TRANSACTION,
                                               AFR_UNLOCK_OP, &flock, F_SETLK, i);
 
                         STACK_WIND_COOKIE (frame, afr_unlock_inodelk_cbk,
@@ -693,7 +710,7 @@ afr_unlock_entrylk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         local = frame->local;
 
-        afr_trace_entrylk_out (frame, AFR_ENTRYLK_TRANSACTION,
+        AFR_TRACE_ENTRYLK_OUT (frame, this, AFR_ENTRYLK_TRANSACTION,
                                AFR_UNLOCK_OP, NULL, op_ret,
                                op_errno, child_index);
 
@@ -740,7 +757,8 @@ afr_unlock_entrylk (call_frame_t *frame, xlator_t *this)
 
         for (i = 0; i < priv->child_count; i++) {
                 if (int_lock->entry_locked_nodes[i] & LOCKED_YES) {
-                        afr_trace_entrylk_in (frame, AFR_ENTRYLK_NB_TRANSACTION,
+                        AFR_TRACE_ENTRYLK_IN (frame, this,
+                                              AFR_ENTRYLK_NB_TRANSACTION,
                                               AFR_UNLOCK_OP, basename, i);
 
                         STACK_WIND_COOKIE (frame, afr_unlock_entrylk_cbk,
@@ -808,7 +826,7 @@ static int32_t
 afr_blocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                           int32_t op_ret, int32_t op_errno)
 {
-        afr_trace_inodelk_out (frame, AFR_INODELK_TRANSACTION,
+        AFR_TRACE_INODELK_OUT (frame, this, AFR_INODELK_TRANSACTION,
                                AFR_LOCK_OP, NULL, op_ret,
                                op_errno, (long) cookie);
 
@@ -874,7 +892,7 @@ afr_lock_lower_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                        local->transaction.basename :
                        local->transaction.new_basename);
 
-        afr_trace_entrylk_in (frame, AFR_ENTRYLK_TRANSACTION,
+        AFR_TRACE_ENTRYLK_IN (frame, this, AFR_ENTRYLK_TRANSACTION,
                               AFR_LOCK_OP, higher_name, child_index);
 
 
@@ -893,7 +911,7 @@ static int32_t
 afr_blocking_entrylk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                           int32_t op_ret, int32_t op_errno)
 {
-        afr_trace_entrylk_out (frame, AFR_ENTRYLK_TRANSACTION,
+        AFR_TRACE_ENTRYLK_OUT (frame, this, AFR_ENTRYLK_TRANSACTION,
                                AFR_LOCK_OP, NULL, op_ret,
                                op_errno, (long)cookie);
 
@@ -1024,7 +1042,8 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
         case AFR_METADATA_TRANSACTION:
 
                 if (local->fd) {
-                        afr_trace_inodelk_in (frame, AFR_INODELK_TRANSACTION,
+                        AFR_TRACE_INODELK_IN (frame, this,
+                                              AFR_INODELK_TRANSACTION,
                                               AFR_LOCK_OP, &flock, F_SETLKW,
                                               child_index);
 
@@ -1036,7 +1055,8 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
                                            F_SETLKW, &flock);
 
                 } else {
-                        afr_trace_inodelk_in (frame, AFR_INODELK_TRANSACTION,
+                        AFR_TRACE_INODELK_IN (frame, this,
+                                              AFR_INODELK_TRANSACTION,
                                               AFR_LOCK_OP, &flock, F_SETLKW,
                                               child_index);
 
@@ -1061,7 +1081,7 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
                               local->transaction.basename :
                               local->transaction.new_basename);
 
-                afr_trace_entrylk_in (frame, AFR_ENTRYLK_TRANSACTION,
+                AFR_TRACE_ENTRYLK_IN (frame, this, AFR_ENTRYLK_TRANSACTION,
                                       AFR_LOCK_OP, lower_name, child_index);
 
 
@@ -1077,7 +1097,8 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
 
         case AFR_ENTRY_TRANSACTION:
                 if (local->fd) {
-                        afr_trace_entrylk_in (frame, AFR_ENTRYLK_TRANSACTION,
+                        AFR_TRACE_ENTRYLK_IN (frame, this,
+                                              AFR_ENTRYLK_TRANSACTION,
                                               AFR_LOCK_OP, local->transaction.basename,
                                               child_index);
 
@@ -1089,7 +1110,8 @@ afr_lock_blocking (call_frame_t *frame, xlator_t *this, int child_index)
                                            local->transaction.basename,
                                            ENTRYLK_LOCK, ENTRYLK_WRLCK);
                 } else {
-                        afr_trace_entrylk_in (frame, AFR_ENTRYLK_TRANSACTION,
+                        AFR_TRACE_ENTRYLK_IN (frame, this,
+                                              AFR_ENTRYLK_TRANSACTION,
                                               AFR_LOCK_OP, local->transaction.basename,
                                               child_index);
 
@@ -1148,13 +1170,13 @@ afr_nonblocking_entrylk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 {
         afr_internal_lock_t *int_lock = NULL;
         afr_local_t         *local    = NULL;
-        int call_count          = 0;
-        int child_index         = (long) cookie;
+        int                 call_count  = 0;
+        int                 child_index = (long) cookie;
 
         local    = frame->local;
         int_lock = &local->internal_lock;
 
-        afr_trace_entrylk_out (frame, AFR_ENTRYLK_TRANSACTION,
+        AFR_TRACE_ENTRYLK_OUT (frame, this, AFR_ENTRYLK_TRANSACTION,
                                AFR_LOCK_OP, NULL, op_ret,
                                op_errno, (long) cookie);
 
@@ -1273,7 +1295,8 @@ afr_nonblocking_entrylk (call_frame_t *frame, xlator_t *this)
                    and where the fd has been opened */
                 for (i = 0; i < priv->child_count; i++) {
                         if (local->child_up[i] && local->fd_open_on[i]) {
-                                afr_trace_entrylk_in (frame, AFR_ENTRYLK_NB_TRANSACTION,
+                                AFR_TRACE_ENTRYLK_IN (frame, this,
+                                                      AFR_ENTRYLK_NB_TRANSACTION,
                                                       AFR_LOCK_OP, basename, i);
 
                                 STACK_WIND_COOKIE (frame, afr_nonblocking_entrylk_cbk,
@@ -1294,7 +1317,8 @@ afr_nonblocking_entrylk (call_frame_t *frame, xlator_t *this)
 
                 for (i = 0; i < priv->child_count; i++) {
                         if (local->child_up[i]) {
-                                afr_trace_entrylk_in (frame, AFR_ENTRYLK_NB_TRANSACTION,
+                                AFR_TRACE_ENTRYLK_IN (frame, this,
+                                                      AFR_ENTRYLK_NB_TRANSACTION,
                                                       AFR_LOCK_OP, basename, i);
 
                                 STACK_WIND_COOKIE (frame, afr_nonblocking_entrylk_cbk,
@@ -1323,23 +1347,15 @@ afr_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         int call_count  = 0;
         int child_index = (long) cookie;
         afr_fd_ctx_t        *fd_ctx = NULL;
-        afr_private_t       *priv     = NULL;
 
 
-        priv = this->private;
         local    = frame->local;
         int_lock = &local->internal_lock;
 
-        afr_trace_inodelk_out (frame, AFR_INODELK_NB_TRANSACTION,
+        AFR_TRACE_INODELK_OUT (frame, this, AFR_INODELK_NB_TRANSACTION,
                                AFR_LOCK_OP, NULL, op_ret,
                                op_errno, (long) cookie);
 
-        LOCK (&frame->lock);
-        {
-                call_count = --int_lock->lk_call_count;
-        }
-        UNLOCK (&frame->lock);
-
         if (op_ret < 0) {
                 if (op_errno == ENOSYS) {
                         /* return ENOTSUP */
@@ -1351,14 +1367,16 @@ afr_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         int_lock->lock_op_errno      = op_errno;
                         local->op_errno              = op_errno;
                 }
+                if (local->transaction.eager_lock)
+                        local->transaction.eager_lock[child_index] = 0;
         } else {
                 int_lock->inode_locked_nodes[child_index]
                         |= LOCKED_YES;
                 int_lock->inodelk_lock_count++;
 
-                if (priv->eager_lock && local->fd) {
+                if (local->transaction.eager_lock &&
+                    local->transaction.eager_lock[child_index] && local->fd) {
                         fd_ctx = afr_fd_ctx_get (local->fd, this);
-                        local->transaction.eager_lock[child_index] = 1;
                         /* piggybacked */
 
                         if (op_ret == 1) {
@@ -1374,6 +1392,11 @@ afr_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 }
         }
 
+        LOCK (&frame->lock);
+        {
+                call_count = --int_lock->lk_call_count;
+        }
+        UNLOCK (&frame->lock);
         if (call_count == 0) {
                 gf_log (this->name, GF_LOG_TRACE,
                         "Last inode locking reply received");
@@ -1406,13 +1429,14 @@ afr_nonblocking_inodelk (call_frame_t *frame, xlator_t *this)
         afr_local_t         *local    = NULL;
         afr_private_t       *priv     = NULL;
         afr_fd_ctx_t        *fd_ctx   = NULL;
-        int32_t  call_count = 0;
-        int      i          = 0;
-        int      ret        = 0;
-        struct gf_flock flock = {0,};
-        struct gf_flock full_flock = {0,};
-        struct gf_flock *flock_use = &flock;
-        int     piggyback = 0;
+        int32_t             call_count = 0;
+        int                 i          = 0;
+        int                 ret        = 0;
+        struct              gf_flock flock = {0,};
+        struct              gf_flock full_flock = {0,};
+        struct              gf_flock *flock_use = NULL;
+        int                 piggyback = 0;
+        gf_boolean_t        fd_lock_owner = _gf_false;
 
         local    = frame->local;
         int_lock = &local->internal_lock;
@@ -1422,10 +1446,6 @@ afr_nonblocking_inodelk (call_frame_t *frame, xlator_t *this)
         flock.l_len   = int_lock->lk_flock.l_len;
         flock.l_type  = int_lock->lk_flock.l_type;
 
-        gf_log (this->name, GF_LOG_DEBUG, "attempting data lock range %"PRIu64
-                " %"PRIu64" by %s", flock.l_start, flock.l_len,
-                lkowner_utoa (&frame->root->lk_owner));
-
         full_flock.l_type = int_lock->lk_flock.l_type;
 
         initialize_inodelk_variables (frame, this);
@@ -1464,11 +1484,18 @@ afr_nonblocking_inodelk (call_frame_t *frame, xlator_t *this)
                         if (!local->child_up[i] || !local->fd_open_on[i])
                                 continue;
 
-                        if (!priv->eager_lock)
+                        flock_use = &flock;
+                        if (!priv->eager_lock) {
+                                if (fd_lock_owner) {
+                                        afr_set_lk_owner (frame, this,
+                                                          frame->root);
+                                        fd_lock_owner = _gf_false;
+                                }
                                 goto wind;
+                        }
 
-                        flock_use = &full_flock;
                         piggyback = 0;
+                        local->transaction.eager_lock[i] = 1;
 
                         LOCK (&local->fd->lock);
                         {
@@ -1487,8 +1514,14 @@ afr_nonblocking_inodelk (call_frame_t *frame, xlator_t *this)
                                         break;
                                 continue;
                         }
+                        flock_use = &full_flock;
+                        if (!fd_lock_owner) {
+                                afr_set_lk_owner (frame, this, local->fd);
+                                fd_lock_owner = _gf_true;
+                        }
                 wind:
-                        afr_trace_inodelk_in (frame, AFR_INODELK_NB_TRANSACTION,
+                        AFR_TRACE_INODELK_IN (frame, this,
+                                              AFR_INODELK_NB_TRANSACTION,
                                               AFR_LOCK_OP, flock_use, F_SETLK, i);
 
                         STACK_WIND_COOKIE (frame, afr_nonblocking_inodelk_cbk,
@@ -1509,7 +1542,8 @@ afr_nonblocking_inodelk (call_frame_t *frame, xlator_t *this)
                 for (i = 0; i < priv->child_count; i++) {
                         if (!local->child_up[i])
                                 continue;
-                        afr_trace_inodelk_in (frame, AFR_INODELK_NB_TRANSACTION,
+                        AFR_TRACE_INODELK_IN (frame, this,
+                                              AFR_INODELK_NB_TRANSACTION,
                                               AFR_LOCK_OP, &flock, F_SETLK, i);
 
                         STACK_WIND_COOKIE (frame, afr_nonblocking_inodelk_cbk,
@@ -1602,7 +1636,8 @@ afr_unlock_lower_entrylk (call_frame_t *frame, xlator_t *this)
 
         for (i = 0; i < priv->child_count; i++) {
                 if (int_lock->lower_locked_nodes[i] & LOCKED_LOWER) {
-                        afr_trace_entrylk_in (frame, AFR_ENTRYLK_NB_TRANSACTION,
+                        AFR_TRACE_ENTRYLK_IN (frame, this,
+                                              AFR_ENTRYLK_NB_TRANSACTION,
                                               AFR_UNLOCK_OP, basename, i);
 
                         STACK_WIND_COOKIE (frame, afr_unlock_entrylk_cbk,
diff --git a/xlators/cluster/afr/src/afr-self-heal-common.c b/xlators/cluster/afr/src/afr-self-heal-common.c
index f4f74940c..f9958e768 100644
--- a/xlators/cluster/afr/src/afr-self-heal-common.c
+++ b/xlators/cluster/afr/src/afr-self-heal-common.c
@@ -2086,7 +2086,7 @@ 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_lk_owner (sh_frame, this, sh_frame->root);
         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 fde17209f..ecd8c4726 100644
--- a/xlators/cluster/afr/src/afr-self-heald.c
+++ b/xlators/cluster/afr/src/afr-self-heald.c
@@ -1041,7 +1041,7 @@ 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_lk_owner (frame, this, frame->root);
         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 36e2812f9..584ad7aed 100644
--- a/xlators/cluster/afr/src/afr-transaction.c
+++ b/xlators/cluster/afr/src/afr-transaction.c
@@ -1001,6 +1001,7 @@ afr_post_nonblocking_inodelk_cbk (call_frame_t *frame, xlator_t *this)
                 gf_log (this->name, GF_LOG_DEBUG,
                         "Non blocking inodelks failed. Proceeding to blocking");
                 int_lock->lock_cbk = afr_post_blocking_inodelk_cbk;
+                afr_set_lk_owner (frame, this, frame->root);
                 afr_blocking_lock (frame, this);
         } else {
 
@@ -1168,7 +1169,7 @@ afr_lock (call_frame_t *frame, xlator_t *this)
 
         frame->root->pid = (long) frame->root;
 
-        afr_set_lk_owner (frame, this);
+        afr_set_lk_owner (frame, this, frame->root);
 
         afr_set_lock_number (frame, this);
 
diff --git a/xlators/cluster/afr/src/afr.c b/xlators/cluster/afr/src/afr.c
index 0bd389ef9..5d4516190 100644
--- a/xlators/cluster/afr/src/afr.c
+++ b/xlators/cluster/afr/src/afr.c
@@ -168,6 +168,7 @@ reconfigure (xlator_t *this, dict_t *options)
                 priv->read_child = index;
         }
 
+        GF_OPTION_RECONF ("eager-lock", priv->eager_lock, options, bool, out);
         GF_OPTION_RECONF ("quorum-type", qtype, options, str, out);
         GF_OPTION_RECONF ("quorum-count", priv->quorum_count, options,
                           uint32, out);
@@ -294,6 +295,7 @@ init (xlator_t *this)
 
         GF_OPTION_INIT ("strict-readdir", priv->strict_readdir, bool, out);
 
+        GF_OPTION_INIT ("eager-lock", priv->eager_lock, bool, out);
         GF_OPTION_INIT ("quorum-type", qtype, str, out);
         GF_OPTION_INIT ("quorum-count", priv->quorum_count, uint32, out);
         fix_quorum_options(this,priv,qtype);
@@ -572,6 +574,10 @@ struct volume_options options[] = {
           .type = GF_OPTION_TYPE_BOOL,
           .default_value = "off",
         },
+        { .key = {"eager-lock"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .default_value = "off",
+        },
         { .key = {"self-heal-daemon"},
           .type = GF_OPTION_TYPE_BOOL,
           .default_value = "off",
diff --git a/xlators/cluster/afr/src/afr.h b/xlators/cluster/afr/src/afr.h
index 23b0a4e27..80a764212 100644
--- a/xlators/cluster/afr/src/afr.h
+++ b/xlators/cluster/afr/src/afr.h
@@ -778,7 +778,7 @@ 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, void *lk_owner);
 
 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 9c6f9a52a..21d10aff2 100644
--- a/xlators/cluster/afr/src/pump.c
+++ b/xlators/cluster/afr/src/pump.c
@@ -704,7 +704,7 @@ pump_start (call_frame_t *pump_frame, xlator_t *this)
 	priv = this->private;
         pump_priv = priv->pump_private;
 
-        afr_set_lk_owner (pump_frame, this);
+        afr_set_lk_owner (pump_frame, this, pump_frame->root);
 	pump_pid = (uint64_t) (unsigned long)pump_frame->root;
 
 	ret = synctask_new (pump_priv->env, pump_task,
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index 91696303a..85691e681 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -109,22 +109,23 @@ struct volopt_map_entry {
 
 static struct volopt_map_entry glusterd_volopt_map[] = {
 
-        {"cluster.lookup-unhashed",              "cluster/distribute", NULL, NULL, NO_DOC, 0},
-        {"cluster.min-free-disk",                "cluster/distribute", NULL, NULL, NO_DOC, 0},
-	{"cluster.min-free-inodes",              "cluster/distribute", NULL, NULL, NO_DOC, 0},
-
-        {"cluster.entry-change-log",             "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.read-subvolume",               "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.background-self-heal-count",   "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.metadata-self-heal",           "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.data-self-heal",               "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.entry-self-heal",              "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.self-heal-daemon",             "cluster/replicate",  "!self-heal-daemon", NULL, NO_DOC, 0},
-        {"cluster.strict-readdir",               "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.self-heal-window-size",        "cluster/replicate",  "data-self-heal-window-size", NULL, DOC, 0},
-        {"cluster.data-change-log",              "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.metadata-change-log",          "cluster/replicate",  NULL, NULL, NO_DOC, 0},
-        {"cluster.data-self-heal-algorithm",     "cluster/replicate",  "data-self-heal-algorithm", NULL,DOC, 0},
+        {"cluster.lookup-unhashed",              "cluster/distribute", NULL, NULL, NO_DOC, 0    },
+        {"cluster.min-free-disk",                "cluster/distribute", NULL, NULL, NO_DOC, 0    },
+        {"cluster.min-free-inodes",              "cluster/distribute", NULL, NULL, NO_DOC, 0    },
+
+        {"cluster.entry-change-log",             "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
+        {"cluster.read-subvolume",               "cluster/replicate",  NULL, NULL, NO_DOC, 0    },
+        {"cluster.background-self-heal-count",   "cluster/replicate",  NULL, NULL, NO_DOC, 0    },
+        {"cluster.metadata-self-heal",           "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
+        {"cluster.data-self-heal",               "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
+        {"cluster.entry-self-heal",              "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
+        {"cluster.self-heal-daemon",             "cluster/replicate",  "!self-heal-daemon" , NULL, NO_DOC, 0     },
+        {"cluster.strict-readdir",               "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
+        {"cluster.self-heal-window-size",        "cluster/replicate",         "data-self-heal-window-size", NULL, DOC, 0},
+        {"cluster.data-change-log",              "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
+        {"cluster.metadata-change-log",          "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
+        {"cluster.data-self-heal-algorithm",     "cluster/replicate",         "data-self-heal-algorithm", NULL,DOC, 0},
+        {"cluster.eager-lock",                   "cluster/replicate",  NULL, NULL, NO_DOC, 0     },
         {"cluster.quorum-type",                  "cluster/replicate",  "quorum-type", NULL, NO_DOC, 0},
         {"cluster.quorum-count",                 "cluster/replicate",  "quorum-count", NULL, NO_DOC, 0},
 
@@ -1962,7 +1963,6 @@ get_key_from_volopt ( struct volopt_map_entry *vme, char **key)
         return ret;
 }
 
-
 int
 glusterd_get_volopt_content (gf_boolean_t xml_out)
 {
@@ -3322,6 +3322,57 @@ out:
 }
 
 int
+validate_wb_eagerlock (glusterd_volinfo_t *volinfo, dict_t *val_dict,
+                       char **op_errstr)
+{
+        int          ret = -1;
+        gf_boolean_t wb_val = _gf_false;
+        gf_boolean_t el_val = _gf_false;
+        char         msg[2048] = {0};
+        char         *wb_key = NULL;
+        char         *el_key = NULL;
+
+        wb_key = "performance.write-behind";
+        el_key = "cluster.eager-lock";
+        ret = dict_get_str_boolean (val_dict, wb_key, -1);
+        if (ret < 0)
+                goto check_eager_lock;
+        wb_val = ret;
+        ret = glusterd_volinfo_get_boolean (volinfo, el_key);
+        if (ret < 0)
+                goto out;
+        el_val = ret;
+        goto done;
+
+check_eager_lock:
+        ret = dict_get_str_boolean (val_dict, el_key, -1);
+        if (ret < 0) {
+                ret = 0; //Keys of intereset to this fn are not present.
+                goto out;
+        }
+        el_val = ret;
+        ret = glusterd_volinfo_get_boolean (volinfo, wb_key);
+        if (ret < 0)
+                goto out;
+        wb_val = ret;
+        goto done;
+
+done:
+        ret = 0;
+        if (!wb_val && el_val) {
+                ret = -1;
+                snprintf (msg, sizeof (msg), "%s off and %s on is not "
+                          "valid configuration", wb_key, el_key);
+                gf_log ("glusterd", GF_LOG_ERROR, msg);
+                if (op_errstr)
+                        *op_errstr = gf_strdup (msg);
+                goto out;
+        }
+out:
+        return ret;
+}
+
+int
 validate_clientopts (glusterd_volinfo_t *volinfo,
                      dict_t *val_dict,
                      char **op_errstr)
@@ -3466,6 +3517,10 @@ glusterd_validate_reconfopts (glusterd_volinfo_t *volinfo, dict_t *val_dict,
                 goto out;
         }
 
+        ret = validate_wb_eagerlock (volinfo, val_dict, op_errstr);
+        if (ret)
+                goto out;
+
         ret = validate_clientopts (volinfo, val_dict, op_errstr);
         if (ret) {
                 gf_log ("", GF_LOG_DEBUG,