locks: various fixes

- implement ref/unref of entry locks (and fix bad pointer deref crashes)
- code cleanup and deleted various data types
- fix improper read/write lock conflict detection in entrylk
- fix indefinite hang of blocked locks on disconnect
- register locks in client_t synchronously, fix crashes in disconnect path

Change-Id: Id273690c9111b8052139d1847060d1fb5a711924
BUG: 849630
Signed-off-by: Anand Avati <avati@redhat.com>
Reviewed-on: http://review.gluster.org/6695
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Kaleb KEITHLEY <kkeithle@redhat.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

10 files changed, 386 insertions, 627 deletions

diff --git a/libglusterfs/src/client_t.h b/libglusterfs/src/client_t.h
index f7812f8f07d..548081896c0 100644
--- a/libglusterfs/src/client_t.h
+++ b/libglusterfs/src/client_t.h
@@ -60,6 +60,7 @@ struct clienttable {
         gf_lock_t            lock;
         cliententry_t       *cliententries;
         int                  first_free;
+	client_t            *local;
 };
 typedef struct clienttable clienttable_t;
 
diff --git a/libglusterfs/src/glusterfs.h b/libglusterfs/src/glusterfs.h
index dfe443016fc..4f7a5235372 100644
--- a/libglusterfs/src/glusterfs.h
+++ b/libglusterfs/src/glusterfs.h
@@ -452,7 +452,6 @@ struct _glusterfs_ctx {
 
         int                 daemon_pipe[2];
 
-        struct client_disconnect *client_disconnect;
         struct clienttable *clienttable;
 };
 typedef struct _glusterfs_ctx glusterfs_ctx_t;
diff --git a/tests/bugs/bug-765564.t b/tests/bugs/bug-765564.t
index 0b8b8cd4f9e..6e4087f805c 100644
--- a/tests/bugs/bug-765564.t
+++ b/tests/bugs/bug-765564.t
@@ -3,6 +3,8 @@
 . $(dirname $0)/../include.rc
 . $(dirname $0)/../volume.rc
 
+cleanup;
+
 TEST glusterd
 TEST pidof glusterd
 
diff --git a/xlators/features/locks/src/clear.c b/xlators/features/locks/src/clear.c
index 124b9ad0feb..75593b8988c 100644
--- a/xlators/features/locks/src/clear.c
+++ b/xlators/features/locks/src/clear.c
@@ -338,9 +338,8 @@ blkd:
                                    elock->basename, ENTRYLK_LOCK, elock->type,
                                    -1, EAGAIN);
                 STACK_UNWIND_STRICT (entrylk, elock->frame, -1, EAGAIN, NULL);
-                GF_FREE ((char *) elock->basename);
-                GF_FREE (elock->connection_id);
-                GF_FREE (elock);
+
+		__pl_entrylk_unref (elock);
         }
 
         if (!(args->kind & CLRLK_GRANTED)) {
@@ -363,13 +362,13 @@ granted:
                         gcount++;
                         list_del_init (&elock->domain_list);
                         list_add_tail (&elock->domain_list, &removed);
+
+			__pl_entrylk_unref (elock);
                 }
         }
         pthread_mutex_unlock (&pl_inode->mutex);
 
-        list_for_each_entry_safe (elock, tmp, &removed, domain_list) {
-                grant_blocked_entry_locks (this, pl_inode, elock, dom);
-        }
+	grant_blocked_entry_locks (this, pl_inode, dom);
 
         ret = 0;
 out:
diff --git a/xlators/features/locks/src/common.c b/xlators/features/locks/src/common.c
index b3309580d3d..f6c71c1cf86 100644
--- a/xlators/features/locks/src/common.c
+++ b/xlators/features/locks/src/common.c
@@ -1099,123 +1099,3 @@ pl_getlk (pl_inode_t *pl_inode, posix_lock_t *lock)
         return conf;
 }
 
-
-struct _lock_table *
-pl_lock_table_new (void)
-{
-        struct _lock_table *new = NULL;
-
-        new = GF_CALLOC (1, sizeof (struct _lock_table), gf_common_mt_lock_table);
-        if (new == NULL) {
-                goto out;
-        }
-        INIT_LIST_HEAD (&new->entrylk_lockers);
-        INIT_LIST_HEAD (&new->inodelk_lockers);
-        LOCK_INIT (&new->lock);
-out:
-        return new;
-}
-
-
-int
-pl_add_locker (struct _lock_table *table, const char *volume,
-               loc_t *loc, fd_t *fd, pid_t pid, gf_lkowner_t *owner,
-               glusterfs_fop_t type)
-{
-        int32_t         ret = -1;
-        struct _locker *new = NULL;
-
-        GF_VALIDATE_OR_GOTO ("lock-table", table, out);
-        GF_VALIDATE_OR_GOTO ("lock-table", volume, out);
-
-        new = GF_CALLOC (1, sizeof (struct _locker), gf_common_mt_locker);
-        if (new == NULL) {
-                goto out;
-        }
-        INIT_LIST_HEAD (&new->lockers);
-
-        new->volume = gf_strdup (volume);
-
-        if (fd == NULL) {
-                loc_copy (&new->loc, loc);
-        } else {
-                new->fd = fd_ref (fd);
-        }
-
-        new->pid   = pid;
-        new->owner = *owner;
-
-        LOCK (&table->lock);
-        {
-                if (type == GF_FOP_ENTRYLK)
-                        list_add_tail (&new->lockers, &table->entrylk_lockers);
-                else
-                        list_add_tail (&new->lockers, &table->inodelk_lockers);
-        }
-        UNLOCK (&table->lock);
-out:
-        return ret;
-}
-
-int
-pl_del_locker (struct _lock_table *table, const char *volume,
-               loc_t *loc, fd_t *fd, gf_lkowner_t *owner, glusterfs_fop_t type)
-{
-        struct _locker    *locker = NULL;
-        struct _locker    *tmp = NULL;
-        int32_t            ret = -1;
-        struct list_head  *head = NULL;
-        struct list_head   del;
-
-        GF_VALIDATE_OR_GOTO ("lock-table", table, out);
-        GF_VALIDATE_OR_GOTO ("lock-table", volume, out);
-
-        INIT_LIST_HEAD (&del);
-
-        LOCK (&table->lock);
-        {
-                if (type == GF_FOP_ENTRYLK) {
-                        head = &table->entrylk_lockers;
-                } else {
-                        head = &table->inodelk_lockers;
-                }
-
-                list_for_each_entry_safe (locker, tmp, head, lockers) {
-                        if (!is_same_lkowner (&locker->owner, owner) ||
-                            strcmp (locker->volume, volume))
-                                continue;
-
-                        /*
-                         * It is possible for inodelk lock to come on anon-fd
-                         * and inodelk unlock to come on normal fd in case of
-                         * client re-opens. So don't check for fds to be equal.
-                         */
-                        if (locker->fd && fd)
-                                list_move_tail (&locker->lockers, &del);
-                        else if (locker->loc.inode && loc &&
-                                 (locker->loc.inode == loc->inode))
-                                list_move_tail (&locker->lockers, &del);
-                }
-        }
-        UNLOCK (&table->lock);
-
-        tmp = NULL;
-        locker = NULL;
-
-        list_for_each_entry_safe (locker, tmp, &del, lockers) {
-                list_del_init (&locker->lockers);
-                if (locker->fd)
-                        fd_unref (locker->fd);
-                else
-                        loc_wipe (&locker->loc);
-
-                GF_FREE (locker->volume);
-                GF_FREE (locker);
-        }
-
-        ret = 0;
-out:
-        return ret;
-
-}
-
diff --git a/xlators/features/locks/src/common.h b/xlators/features/locks/src/common.h
index db19ec978b4..5ec630ee857 100644
--- a/xlators/features/locks/src/common.h
+++ b/xlators/features/locks/src/common.h
@@ -32,20 +32,6 @@
 
 #define SET_FLOCK_PID(flock, lock) ((flock)->l_pid = lock->client_pid)
 
-struct _locker {
-        struct list_head  lockers;
-        char             *volume;
-        loc_t             loc;
-        fd_t             *fd;
-        gf_lkowner_t      owner;
-        pid_t             pid;
-};
-
-struct _lock_table {
-        struct list_head  inodelk_lockers;
-        struct list_head  entrylk_lockers;
-        gf_lock_t         lock;
-};
 
 posix_lock_t *
 new_posix_lock (struct gf_flock *flock, client_t *client, pid_t client_pid,
@@ -92,7 +78,7 @@ __pl_inodelk_unref (pl_inode_lock_t *lock);
 
 void
 grant_blocked_entry_locks (xlator_t *this, pl_inode_t *pl_inode,
-                           pl_entry_lock_t *unlocked, pl_dom_list_t *dom);
+                           pl_dom_list_t *dom);
 
 void pl_update_refkeeper (xlator_t *this, inode_t *inode);
 
@@ -166,22 +152,7 @@ pl_reserve_unlock (xlator_t *this, pl_inode_t *pl_inode, posix_lock_t *reqlock);
 uint32_t
 check_entrylk_on_basename (xlator_t *this, inode_t *parent, char *basename);
 
-int32_t
-pl_add_locker (struct _lock_table *table, const char *volume,
-               loc_t *loc,
-               fd_t *fd,
-               pid_t pid,
-               gf_lkowner_t *owner,
-               glusterfs_fop_t type);
-
-int32_t
-pl_del_locker (struct _lock_table *table, const char *volume,
-               loc_t *loc,
-               fd_t *fd,
-               gf_lkowner_t *owner,
-               glusterfs_fop_t type);
-
-struct _lock_table *
-pl_lock_table_new (void);
+void __pl_inodelk_unref (pl_inode_lock_t *lock);
+void __pl_entrylk_unref (pl_entry_lock_t *lock);
 
 #endif /* __COMMON_H__ */
diff --git a/xlators/features/locks/src/entrylk.c b/xlators/features/locks/src/entrylk.c
index 0785dc547fc..208bc140e7a 100644
--- a/xlators/features/locks/src/entrylk.c
+++ b/xlators/features/locks/src/entrylk.c
@@ -23,11 +23,29 @@
 #include "locks.h"
 #include "common.h"
 
+
+void
+__pl_entrylk_unref (pl_entry_lock_t *lock)
+{
+        lock->ref--;
+        if (!lock->ref) {
+		GF_FREE ((char *)lock->basename);
+                GF_FREE (lock->connection_id);
+                GF_FREE (lock);
+        }
+}
+
+
+static void
+__pl_entrylk_ref (pl_entry_lock_t *lock)
+{
+        lock->ref++;
+}
+
+
 static pl_entry_lock_t *
 new_entrylk_lock (pl_inode_t *pinode, const char *basename, entrylk_type type,
-                  client_t *client, pid_t client_pid, gf_lkowner_t *owner,
-                  const char *volume)
-
+		  const char *domain, call_frame_t *frame, char *conn_id)
 {
         pl_entry_lock_t *newlock = NULL;
 
@@ -39,14 +57,21 @@ new_entrylk_lock (pl_inode_t *pinode, const char *basename, entrylk_type type,
 
         newlock->basename   = basename ? gf_strdup (basename) : NULL;
         newlock->type       = type;
-        newlock->trans      = client;
-        newlock->volume     = volume;
-        newlock->client_pid = client_pid;
-        newlock->owner      = *owner;
+        newlock->client     = frame->root->client;
+        newlock->client_pid = frame->root->pid;
+        newlock->volume     = domain;
+        newlock->owner      = frame->root->lk_owner;
+	newlock->frame      = frame;
+
+        if (conn_id) {
+                newlock->connection_id = gf_strdup (conn_id);
+        }
 
         INIT_LIST_HEAD (&newlock->domain_list);
         INIT_LIST_HEAD (&newlock->blocked_locks);
+	INIT_LIST_HEAD (&newlock->client_list);
 
+	__pl_entrylk_ref (newlock);
 out:
         return newlock;
 }
@@ -77,42 +102,42 @@ __same_entrylk_owner (pl_entry_lock_t *l1, pl_entry_lock_t *l2)
 {
 
         return (is_same_lkowner (&l1->owner, &l2->owner) &&
-                (l1->trans  == l2->trans));
+                (l1->client  == l2->client));
 }
 
 
 /**
- * lock_grantable - is this lock grantable?
+ * entrylk_grantable - is this lock grantable?
  * @inode: inode in which to look
  * @basename: name we're trying to lock
  * @type: type of lock
  */
 static pl_entry_lock_t *
-__lock_grantable (pl_dom_list_t *dom, const char *basename, entrylk_type type)
+__entrylk_grantable (pl_dom_list_t *dom, pl_entry_lock_t *lock)
 {
-        pl_entry_lock_t *lock = NULL;
+        pl_entry_lock_t *tmp = NULL;
 
         if (list_empty (&dom->entrylk_list))
                 return NULL;
 
-        list_for_each_entry (lock, &dom->entrylk_list, domain_list) {
-                if (names_conflict (lock->basename, basename))
-                        return lock;
+        list_for_each_entry (tmp, &dom->entrylk_list, domain_list) {
+                if (names_conflict (tmp->basename, lock->basename))
+                        return tmp;
         }
 
         return NULL;
 }
 
 static pl_entry_lock_t *
-__blocked_lock_conflict (pl_dom_list_t *dom, const char *basename, entrylk_type type)
+__blocked_entrylk_conflict (pl_dom_list_t *dom, pl_entry_lock_t *lock)
 {
-        pl_entry_lock_t *lock = NULL;
+        pl_entry_lock_t *tmp = NULL;
 
         if (list_empty (&dom->blocked_entrylks))
                 return NULL;
 
-        list_for_each_entry (lock, &dom->blocked_entrylks, blocked_locks) {
-                if (names_conflict (lock->basename, basename))
+        list_for_each_entry (tmp, &dom->blocked_entrylks, blocked_locks) {
+                if (names_conflict (tmp->basename, lock->basename))
                         return lock;
         }
 
@@ -293,7 +318,7 @@ __find_most_matching_lock (pl_dom_list_t *dom, const char *basename)
 }
 
 /**
- * __lock_name - lock a name in a directory
+ * __lock_entrylk - lock a name in a directory
  * @inode: inode for the directory in which to lock
  * @basename: name of the entry to lock
  *            if null, lock the entire directory
@@ -304,89 +329,49 @@ __find_most_matching_lock (pl_dom_list_t *dom, const char *basename)
  */
 
 int
-__lock_name (pl_inode_t *pinode, const char *basename, entrylk_type type,
-             call_frame_t *frame, pl_dom_list_t *dom, xlator_t *this,
-             int nonblock, char *conn_id)
+__lock_entrylk (xlator_t *this, pl_inode_t *pinode, pl_entry_lock_t *lock,
+		int nonblock, pl_dom_list_t *dom)
 {
-        pl_entry_lock_t *lock       = NULL;
-        pl_entry_lock_t *conf       = NULL;
-        int              ret        = -EINVAL;
-
-        lock = new_entrylk_lock (pinode, basename, type,
-                                 frame->root->client, frame->root->pid,
-                                 &frame->root->lk_owner, dom->domain);
-        if (!lock) {
-                ret = -ENOMEM;
-                goto out;
-        }
-
-        lock->frame   = frame;
-        lock->this    = this;
-        lock->trans   = frame->root->client;
+        pl_entry_lock_t *conf = NULL;
+        int              ret  = -EAGAIN;
 
-        if (conn_id) {
-                lock->connection_id = gf_strdup (conn_id);
-        }
-
-        conf = __lock_grantable (dom, basename, type);
+        conf = __entrylk_grantable (dom, lock);
         if (conf) {
                 ret = -EAGAIN;
-                if (nonblock){
-                        GF_FREE (lock->connection_id);
-                        GF_FREE ((char *)lock->basename);
-                        GF_FREE (lock);
+                if (nonblock)
                         goto out;
 
-                }
-
                 gettimeofday (&lock->blkd_time, NULL);
                 list_add_tail (&lock->blocked_locks, &dom->blocked_entrylks);
 
                 gf_log (this->name, GF_LOG_TRACE,
                         "Blocking lock: {pinode=%p, basename=%s}",
-                        pinode, basename);
+                        pinode, lock->basename);
 
                 goto out;
         }
 
-        if ( __blocked_lock_conflict (dom, basename, type) && !(__owner_has_lock (dom, lock))) {
+        if (__blocked_entrylk_conflict (dom, lock) && !(__owner_has_lock (dom, lock))) {
                 ret = -EAGAIN;
-                if (nonblock) {
-                        GF_FREE (lock->connection_id);
-                        GF_FREE ((char *) lock->basename);
-                        GF_FREE (lock);
+                if (nonblock)
                         goto out;
 
-                }
-                lock->frame     = frame;
-                lock->this      = this;
-
                 gettimeofday (&lock->blkd_time, NULL);
                 list_add_tail (&lock->blocked_locks, &dom->blocked_entrylks);
 
-                gf_log (this->name, GF_LOG_TRACE,
+                gf_log (this->name, GF_LOG_DEBUG,
                         "Lock is grantable, but blocking to prevent starvation");
                 gf_log (this->name, GF_LOG_TRACE,
                         "Blocking lock: {pinode=%p, basename=%s}",
-                        pinode, basename);
+                        pinode, lock->basename);
 
-                ret = -EAGAIN;
                 goto out;
         }
-        switch (type) {
 
-        case ENTRYLK_WRLCK:
-                gettimeofday (&lock->granted_time, NULL);
-                list_add_tail (&lock->domain_list, &dom->entrylk_list);
-                break;
-
-        default:
-
-                gf_log (this->name, GF_LOG_DEBUG,
-                        "Invalid type for entrylk specified: %d", type);
-                ret = -EINVAL;
-                goto out;
-        }
+        __pl_entrylk_ref (lock);
+        gettimeofday (&lock->granted_time, NULL);
+        list_add (&lock->domain_list, &dom->entrylk_list);
+	lock->frame = NULL;
 
         ret = 0;
 out:
@@ -394,37 +379,36 @@ out:
 }
 
 /**
- * __unlock_name - unlock a name in a directory
+ * __unlock_entrylk - unlock a name in a directory
  * @inode: inode for the directory to unlock in
  * @basename: name of the entry to unlock
  *            if null, unlock the entire directory
  */
 
 pl_entry_lock_t *
-__unlock_name (pl_dom_list_t *dom, const char *basename, entrylk_type type)
+__unlock_entrylk (pl_dom_list_t *dom, pl_entry_lock_t *lock)
 {
-        pl_entry_lock_t *lock = NULL;
+        pl_entry_lock_t *tmp = NULL;
         pl_entry_lock_t *ret_lock = NULL;
 
-        lock = __find_most_matching_lock (dom, basename);
+        tmp = __find_most_matching_lock (dom, lock->basename);
 
-        if (!lock) {
-                gf_log ("locks", GF_LOG_DEBUG,
+        if (!tmp) {
+                gf_log ("locks", GF_LOG_ERROR,
                         "unlock on %s (type=ENTRYLK_WRLCK) attempted but no matching lock found",
-                        basename);
+                        lock->basename);
                 goto out;
         }
 
-        if (names_equal (lock->basename, basename)
-            && lock->type == type) {
+        if (names_equal (tmp->basename, lock->basename)
+            && tmp->type == lock->type) {
+
+		list_del_init (&tmp->domain_list);
+		ret_lock = tmp;
 
-                if (type == ENTRYLK_WRLCK) {
-                        list_del_init (&lock->domain_list);
-                        ret_lock = lock;
-                }
         } else {
-                gf_log ("locks", GF_LOG_DEBUG,
-                        "Unlock for a non-existing lock!");
+                gf_log ("locks", GF_LOG_ERROR,
+                        "Unlock on %s for a non-existing lock!", lock->basename);
                 goto out;
         }
 
@@ -446,7 +430,7 @@ check_entrylk_on_basename (xlator_t *this, inode_t *parent, char *basename)
         pthread_mutex_lock (&pinode->mutex);
         {
                 list_for_each_entry (dom, &pinode->dom_list, inode_list) {
-                        conf = __lock_grantable (dom, basename, ENTRYLK_WRLCK);
+                        conf = __find_most_matching_lock (dom, basename);
                         if (conf && conf->basename) {
                                 entrylk = 1;
                                 break;
@@ -472,28 +456,14 @@ __grant_blocked_entry_locks (xlator_t *this, pl_inode_t *pl_inode,
         INIT_LIST_HEAD (&blocked_list);
         list_splice_init (&dom->blocked_entrylks, &blocked_list);
 
-        list_for_each_entry_safe (bl, tmp, &blocked_list,
-                                  blocked_locks) {
+        list_for_each_entry_safe (bl, tmp, &blocked_list, blocked_locks) {
 
                 list_del_init (&bl->blocked_locks);
 
-
-                gf_log ("locks", GF_LOG_TRACE,
-                        "Trying to unblock: {pinode=%p, basename=%s}",
-                        pl_inode, bl->basename);
-
-                bl_ret = __lock_name (pl_inode, bl->basename, bl->type,
-                                      bl->frame, dom, bl->this, 0,
-                                      bl->connection_id);
+                bl_ret = __lock_entrylk (bl->this, pl_inode, bl, 0, dom);
 
                 if (bl_ret == 0) {
                         list_add (&bl->blocked_locks, granted);
-                } else {
-                        gf_log (this->name, GF_LOG_DEBUG,
-                                "should never happen");
-                        GF_FREE (bl->connection_id);
-                        GF_FREE ((char *)bl->basename);
-                        GF_FREE (bl);
                 }
         }
         return;
@@ -502,7 +472,7 @@ __grant_blocked_entry_locks (xlator_t *this, pl_inode_t *pl_inode,
 /* Grants locks if possible which are blocked on a lock */
 void
 grant_blocked_entry_locks (xlator_t *this, pl_inode_t *pl_inode,
-                           pl_entry_lock_t *unlocked, pl_dom_list_t *dom)
+			   pl_dom_list_t *dom)
 {
         struct list_head  granted_list;
         pl_entry_lock_t  *tmp = NULL;
@@ -518,105 +488,26 @@ grant_blocked_entry_locks (xlator_t *this, pl_inode_t *pl_inode,
         pthread_mutex_unlock (&pl_inode->mutex);
 
         list_for_each_entry_safe (lock, tmp, &granted_list, blocked_locks) {
-                list_del_init (&lock->blocked_locks);
-
                 entrylk_trace_out (this, lock->frame, NULL, NULL, NULL,
                                    lock->basename, ENTRYLK_LOCK, lock->type,
                                    0, 0);
 
                 STACK_UNWIND_STRICT (entrylk, lock->frame, 0, 0, NULL);
+		lock->frame = NULL;
+	}
 
-                GF_FREE (lock->connection_id);
-                GF_FREE ((char *)lock->basename);
-                GF_FREE (lock);
-        }
-
-        GF_FREE ((char *)unlocked->basename);
-        GF_FREE (unlocked->connection_id);
-        GF_FREE (unlocked);
+        pthread_mutex_lock (&pl_inode->mutex);
+        {
+		list_for_each_entry_safe (lock, tmp, &granted_list, blocked_locks) {
+			list_del_init (&lock->blocked_locks);
+			__pl_entrylk_unref (lock);
+		}
+	}
+        pthread_mutex_unlock (&pl_inode->mutex);
 
         return;
 }
 
-/**
- * release_entry_locks_for_client: release all entry locks from this
- * client for this loc_t
- */
-
-static int
-release_entry_locks_for_client (xlator_t *this, pl_inode_t *pinode,
-                                pl_dom_list_t *dom, client_t *client)
-{
-        pl_entry_lock_t  *lock = NULL;
-        pl_entry_lock_t  *tmp = NULL;
-        struct list_head  granted;
-        struct list_head  released;
-
-        INIT_LIST_HEAD (&granted);
-        INIT_LIST_HEAD (&released);
-
-        pthread_mutex_lock (&pinode->mutex);
-        {
-                list_for_each_entry_safe (lock, tmp, &dom->blocked_entrylks,
-                                          blocked_locks) {
-                        if (lock->trans != client)
-                                continue;
-
-                        list_del_init (&lock->blocked_locks);
-
-                        gf_log (this->name, GF_LOG_TRACE,
-                                "releasing lock on  held by "
-                                "{client=%p}", client);
-
-                        list_add (&lock->blocked_locks, &released);
-
-                }
-
-                list_for_each_entry_safe (lock, tmp, &dom->entrylk_list,
-                                          domain_list) {
-                        if (lock->trans != client)
-                                continue;
-
-                        list_del_init (&lock->domain_list);
-
-                        gf_log (this->name, GF_LOG_TRACE,
-                                "releasing lock on  held by "
-                                "{client=%p}", client);
-
-                        GF_FREE ((char *)lock->basename);
-                        GF_FREE (lock->connection_id);
-                        GF_FREE (lock);
-                }
-
-                __grant_blocked_entry_locks (this, pinode, dom, &granted);
-
-        }
-
-        pthread_mutex_unlock (&pinode->mutex);
-
-        list_for_each_entry_safe (lock, tmp, &released, blocked_locks) {
-                list_del_init (&lock->blocked_locks);
-
-                STACK_UNWIND_STRICT (entrylk, lock->frame, -1, EAGAIN, NULL);
-
-                GF_FREE ((char *)lock->basename);
-                GF_FREE (lock->connection_id);
-                GF_FREE (lock);
-
-        }
-
-        list_for_each_entry_safe (lock, tmp, &granted, blocked_locks) {
-                list_del_init (&lock->blocked_locks);
-
-                STACK_UNWIND_STRICT (entrylk, lock->frame, 0, 0, NULL);
-
-                GF_FREE ((char *)lock->basename);
-                GF_FREE (lock->connection_id);
-                GF_FREE (lock);
-        }
-
-        return 0;
-}
 
 /* Common entrylk code called by pl_entrylk and pl_fentrylk */
 int
@@ -632,10 +523,12 @@ pl_common_entrylk (call_frame_t *frame, xlator_t *this,
         char             unwind   = 1;
         GF_UNUSED int    dict_ret = -1;
         pl_inode_t      *pinode   = NULL;
+        pl_entry_lock_t *reqlock  = NULL;
         pl_entry_lock_t *unlocked = NULL;
         pl_dom_list_t   *dom      = NULL;
         char            *conn_id  = NULL;
         pl_ctx_t        *ctx      = NULL;
+	int              nonblock = 0;
 
         if (xdata)
                 dict_ret = dict_get_str (xdata, "connection-id", &conn_id);
@@ -646,6 +539,15 @@ pl_common_entrylk (call_frame_t *frame, xlator_t *this,
                 goto out;
         }
 
+	if (frame->root->client) {
+		ctx = pl_ctx_get (frame->root->client, this);
+		if (!ctx) {
+			op_errno = ENOMEM;
+			gf_log (this->name, GF_LOG_INFO, "pl_ctx_get() failed");
+			goto unwind;
+		}
+	}
+
         dom = get_domain (pinode, volume);
         if (!dom){
                 op_errno = ENOMEM;
@@ -654,72 +556,64 @@ pl_common_entrylk (call_frame_t *frame, xlator_t *this,
 
         entrylk_trace_in (this, frame, volume, fd, loc, basename, cmd, type);
 
-        if (frame->root->lk_owner.len == 0) {
-                /*
-                  this is a special case that means release
-                  all locks from this client
-                */
-
-                gf_log (this->name, GF_LOG_TRACE,
-                        "Releasing locks for client %p", frame->root->client);
-
-                release_entry_locks_for_client (this, pinode, dom,
-                                                frame->root->client);
-                op_ret = 0;
-
-                goto out;
+        reqlock = new_entrylk_lock (pinode, basename, type, dom->domain, frame,
+				    conn_id);
+        if (!reqlock) {
+                op_ret = -1;
+                op_errno = ENOMEM;
+                goto unwind;
         }
 
         switch (cmd) {
-        case ENTRYLK_LOCK:
-                pthread_mutex_lock (&pinode->mutex);
-                {
-                        ret = __lock_name (pinode, basename, type,
-                                           frame, dom, this, 0, conn_id);
-                }
-                pthread_mutex_unlock (&pinode->mutex);
-
-                op_errno = -ret;
-                if (ret < 0) {
-                        if (ret == -EAGAIN)
-                                unwind = 0;
-                        else
-                                unwind = 1;
-                        goto out;
-                } else {
-                        op_ret = 0;
-                        op_errno = 0;
-                        unwind = 1;
-                        goto out;
-                }
-
-                break;
-
         case ENTRYLK_LOCK_NB:
-                unwind = 1;
+		nonblock = 1;
+		/* fall through */
+	case ENTRYLK_LOCK:
+		if (ctx)
+			pthread_mutex_lock (&ctx->lock);
                 pthread_mutex_lock (&pinode->mutex);
                 {
-                        ret = __lock_name (pinode, basename, type,
-                                           frame, dom, this, 1, conn_id);
+			reqlock->pinode = pinode;
+
+                        ret = __lock_entrylk (this, pinode, reqlock, nonblock, dom);
+			if (ret == 0)
+				op_ret = 0;
+			else
+				op_errno = -ret;
+
+			if (ctx && (!ret || !nonblock))
+				list_add (&reqlock->client_list,
+					  &ctx->entrylk_lockers);
+
+			if (ret == -EAGAIN && !nonblock) {
+				/* blocked */
+				unwind = 0;
+			} else {
+				__pl_entrylk_unref (reqlock);
+			}
                 }
                 pthread_mutex_unlock (&pinode->mutex);
-
-                if (ret < 0) {
-                        op_errno = -ret;
-                        goto out;
-                }
-
-                break;
+		if (ctx)
+			pthread_mutex_unlock (&ctx->lock);
+		break;
 
         case ENTRYLK_UNLOCK:
+		if (ctx)
+			pthread_mutex_lock (&ctx->lock);
                 pthread_mutex_lock (&pinode->mutex);
                 {
-                        unlocked = __unlock_name (dom, basename, type);
+                        unlocked = __unlock_entrylk (dom, reqlock);
+			if (unlocked) {
+				list_del_init (&unlocked->client_list);
+				__pl_entrylk_unref (unlocked);
+			}
+			__pl_entrylk_unref (reqlock);
                 }
                 pthread_mutex_unlock (&pinode->mutex);
+		if (ctx)
+			pthread_mutex_unlock (&ctx->lock);
 
-                if (unlocked)
-                        grant_blocked_entry_locks (this, pinode, unlocked, dom);
+		grant_blocked_entry_locks (this, pinode, dom);
 
                 break;
 
@@ -733,27 +627,10 @@ pl_common_entrylk (call_frame_t *frame, xlator_t *this,
         op_ret = 0;
 out:
         pl_update_refkeeper (this, inode);
+
         if (unwind) {
                 entrylk_trace_out (this, frame, volume, fd, loc, basename,
                                    cmd, type, op_ret, op_errno);
-
-                ctx = pl_ctx_get (frame->root->client, this);
-
-                if (ctx == NULL) {
-                        gf_log (this->name, GF_LOG_INFO, "pl_ctx_get() failed");
-                        goto unwind;
-                }
-
-                if (cmd == ENTRYLK_UNLOCK)
-                        pl_del_locker (ctx->ltable, volume, loc, fd,
-                                       &frame->root->lk_owner,
-                                       GF_FOP_ENTRYLK);
-                else
-                        pl_add_locker (ctx->ltable, volume, loc, fd,
-                                       frame->root->pid,
-                                       &frame->root->lk_owner,
-                                       GF_FOP_ENTRYLK);
-
 unwind:
                 STACK_UNWIND_STRICT (entrylk, frame, op_ret, op_errno, NULL);
         } else {
@@ -761,7 +638,6 @@ unwind:
                                      cmd, type);
         }
 
-
         return 0;
 }
 
@@ -801,6 +677,88 @@ pl_fentrylk (call_frame_t *frame, xlator_t *this,
 }
 
 
+static void
+pl_entrylk_log_cleanup (pl_entry_lock_t *lock)
+{
+	pl_inode_t *pinode = NULL;
+        char *path = NULL;
+        char *file = NULL;
+
+	pinode = lock->pinode;
+
+	inode_path (pinode->refkeeper, NULL, &path);
+
+	if (path)
+		file = path;
+	else
+		file = uuid_utoa (pinode->refkeeper->gfid);
+
+	gf_log (THIS->name, GF_LOG_WARNING,
+		"releasing lock on %s held by "
+		"{client=%p, pid=%"PRId64" lk-owner=%s}",
+		file, lock->client, (uint64_t) lock->client_pid,
+		lkowner_utoa (&lock->owner));
+	GF_FREE (path);
+}
+
+
+/* Release all entrylks from this client */
+int
+pl_entrylk_client_cleanup (xlator_t *this, pl_ctx_t *ctx)
+{
+        pl_entry_lock_t *tmp = NULL;
+        pl_entry_lock_t *l = NULL;
+	pl_dom_list_t *dom = NULL;
+        pl_inode_t *pinode = NULL;
+
+        struct list_head released;
+
+        INIT_LIST_HEAD (&released);
+
+	pthread_mutex_lock (&ctx->lock);
+        {
+                list_for_each_entry_safe (l, tmp, &ctx->entrylk_lockers,
+					  client_list) {
+                        list_del_init (&l->client_list);
+			list_add_tail (&l->client_list, &released);
+
+			pl_entrylk_log_cleanup (l);
+
+			pinode = l->pinode;
+
+			pthread_mutex_lock (&pinode->mutex);
+			{
+				list_del_init (&l->domain_list);
+                        }
+			pthread_mutex_unlock (&pinode->mutex);
+                }
+	}
+        pthread_mutex_unlock (&ctx->lock);
+
+        list_for_each_entry_safe (l, tmp, &released, client_list) {
+                list_del_init (&l->client_list);
+
+		if (l->frame)
+			STACK_UNWIND_STRICT (entrylk, l->frame, -1, EAGAIN,
+					     NULL);
+
+		pinode = l->pinode;
+
+		dom = get_domain (pinode, l->volume);
+
+		grant_blocked_inode_locks (this, pinode, dom);
+
+		pthread_mutex_lock (&pinode->mutex);
+		{
+			__pl_entrylk_unref (l);
+		}
+		pthread_mutex_unlock (&pinode->mutex);
+        }
+
+        return 0;
+}
+
+
 int32_t
 __get_entrylk_count (xlator_t *this, pl_inode_t *pl_inode)
 {
diff --git a/xlators/features/locks/src/inodelk.c b/xlators/features/locks/src/inodelk.c
index 508523e1106..969b67a61db 100644
--- a/xlators/features/locks/src/inodelk.c
+++ b/xlators/features/locks/src/inodelk.c
@@ -35,7 +35,7 @@ __pl_inodelk_ref (pl_inode_lock_t *lock)
         lock->ref++;
 }
 
-inline void
+void
 __pl_inodelk_unref (pl_inode_lock_t *lock)
 {
         lock->ref--;
@@ -204,7 +204,7 @@ __lock_inodelk (xlator_t *this, pl_inode_t *pl_inode, pl_inode_lock_t *lock,
         int ret = -EINVAL;
 
         conf = __inodelk_grantable (dom, lock);
-        if (conf){
+        if (conf) {
                 ret = -EAGAIN;
                 if (can_block == 0)
                         goto out;
@@ -232,7 +232,7 @@ __lock_inodelk (xlator_t *this, pl_inode_t *pl_inode, pl_inode_lock_t *lock,
                 gettimeofday (&lock->blkd_time, NULL);
                 list_add_tail (&lock->blocked_locks, &dom->blocked_inodelks);
 
-                gf_log (this->name, GF_LOG_TRACE,
+                gf_log (this->name, GF_LOG_DEBUG,
                         "Lock is grantable, but blocking to prevent starvation");
                 gf_log (this->name, GF_LOG_TRACE,
                         "%s (pid=%d) (lk-owner=%s) %"PRId64" - %"PRId64" => Blocked",
@@ -307,6 +307,8 @@ __inode_unlock_lock (xlator_t *this, pl_inode_lock_t *lock, pl_dom_list_t *dom)
 out:
         return conf;
 }
+
+
 static void
 __grant_blocked_inode_locks (xlator_t *this, pl_inode_t *pl_inode,
                              struct list_head *granted, pl_dom_list_t *dom)
@@ -363,6 +365,7 @@ grant_blocked_inode_locks (xlator_t *this, pl_inode_t *pl_inode,
                               &lock->user_flock, 0, 0, lock->volume);
 
                 STACK_UNWIND_STRICT (inodelk, lock->frame, 0, 0, NULL);
+		lock->frame = NULL;
         }
 
         pthread_mutex_lock (&pl_inode->mutex);
@@ -375,103 +378,101 @@ grant_blocked_inode_locks (xlator_t *this, pl_inode_t *pl_inode,
         pthread_mutex_unlock (&pl_inode->mutex);
 }
 
-/* Release all inodelks from this client */
-static int
-release_inode_locks_of_client (xlator_t *this, pl_dom_list_t *dom,
-                                  inode_t *inode, client_t *client)
+
+static void
+pl_inodelk_log_cleanup (pl_inode_lock_t *lock)
 {
-        pl_inode_lock_t *tmp = NULL;
-        pl_inode_lock_t *l = NULL;
+	pl_inode_t *pl_inode = NULL;
+        char *path = NULL;
+        char *file = NULL;
 
-        pl_inode_t * pinode = NULL;
+	pl_inode = lock->pl_inode;
 
-        struct list_head released;
+	inode_path (pl_inode->refkeeper, NULL, &path);
 
-        char *path = NULL;
-        char *file = NULL;
+	if (path)
+		file = path;
+	else
+		file = uuid_utoa (pl_inode->refkeeper->gfid);
 
-        INIT_LIST_HEAD (&released);
+	gf_log (THIS->name, GF_LOG_WARNING,
+		"releasing lock on %s held by "
+		"{client=%p, pid=%"PRId64" lk-owner=%s}",
+		file, lock->client, (uint64_t) lock->client_pid,
+		lkowner_utoa (&lock->owner));
+	GF_FREE (path);
+}
 
-        pinode = pl_inode_get (this, inode);
 
-        pthread_mutex_lock (&pinode->mutex);
-        {
+/* Release all entrylks from this client */
+int
+pl_inodelk_client_cleanup (xlator_t *this, pl_ctx_t *ctx)
+{
+        pl_inode_lock_t *tmp = NULL;
+        pl_inode_lock_t *l = NULL;
+	pl_dom_list_t *dom = NULL;
+        pl_inode_t *pl_inode = NULL;
+
+        struct list_head released;
 
-                list_for_each_entry_safe (l, tmp, &dom->blocked_inodelks, blocked_locks) {
-                        if (l->client != client)
-                                continue;
+        INIT_LIST_HEAD (&released);
 
-                        list_del_init (&l->blocked_locks);
+	pthread_mutex_lock (&ctx->lock);
+        {
+                list_for_each_entry_safe (l, tmp, &ctx->inodelk_lockers,
+					  client_list) {
+                        list_del_init (&l->client_list);
+			list_add_tail (&l->client_list, &released);
 
-                        inode_path (inode, NULL, &path);
-                        if (path)
-                                file = path;
-                        else
-                                file = uuid_utoa (inode->gfid);
+			pl_inodelk_log_cleanup (l);
 
-                        gf_log (this->name, GF_LOG_DEBUG,
-                                "releasing blocking lock on %s held by "
-                                "{client=%p, pid=%"PRId64" lk-owner=%s}",
-                                file, client, (uint64_t) l->client_pid,
-                                lkowner_utoa (&l->owner));
+			pl_inode = l->pl_inode;
 
-                        list_add (&l->blocked_locks, &released);
-                        if (path) {
-                                GF_FREE (path);
-                                path = NULL;
+			pthread_mutex_lock (&pl_inode->mutex);
+			{
+				__delete_inode_lock (l);
                         }
+			pthread_mutex_unlock (&pl_inode->mutex);
                 }
+	}
+        pthread_mutex_unlock (&ctx->lock);
 
-                list_for_each_entry_safe (l, tmp, &dom->inodelk_list, list) {
-                        if (l->client != client)
-                                continue;
-
-                        inode_path (inode, NULL, &path);
-                        if (path)
-                                file = path;
-                        else
-                                file = uuid_utoa (inode->gfid);
-
-                        gf_log (this->name, GF_LOG_DEBUG,
-                                "releasing granted lock on %s held by "
-                                "{client=%p, pid=%"PRId64" lk-owner=%s}",
-                                file, client, (uint64_t) l->client_pid,
-                                lkowner_utoa (&l->owner));
-
-                        if (path) {
-                                GF_FREE (path);
-                                path = NULL;
-                        }
+        list_for_each_entry_safe (l, tmp, &released, client_list) {
+                list_del_init (&l->client_list);
 
-                        __delete_inode_lock (l);
-                        __pl_inodelk_unref (l);
-                }
-        }
-        GF_FREE (path);
+		if (l->frame)
+			STACK_UNWIND_STRICT (inodelk, l->frame, -1, EAGAIN,
+					     NULL);
+
+		pl_inode = l->pl_inode;
 
-        pthread_mutex_unlock (&pinode->mutex);
+		dom = get_domain (pl_inode, l->volume);
 
-        list_for_each_entry_safe (l, tmp, &released, blocked_locks) {
-                list_del_init (&l->blocked_locks);
+		grant_blocked_inode_locks (this, pl_inode, dom);
 
-                STACK_UNWIND_STRICT (inodelk, l->frame, -1, EAGAIN, NULL);
-                //No need to take lock as the locks are only in one list
-                __pl_inodelk_unref (l);
+		pthread_mutex_lock (&pl_inode->mutex);
+		{
+			__pl_inodelk_unref (l);
+		}
+		pthread_mutex_unlock (&pl_inode->mutex);
         }
 
-        grant_blocked_inode_locks (this, pinode, dom);
         return 0;
 }
 
 
 static int
-pl_inode_setlk (xlator_t *this, pl_inode_t *pl_inode, pl_inode_lock_t *lock,
-                int can_block,  pl_dom_list_t *dom)
+pl_inode_setlk (xlator_t *this, pl_ctx_t *ctx, pl_inode_t *pl_inode,
+		pl_inode_lock_t *lock, int can_block, pl_dom_list_t *dom)
 {
         int ret = -EINVAL;
         pl_inode_lock_t *retlock = NULL;
         gf_boolean_t    unref = _gf_true;
 
+	lock->pl_inode = pl_inode;
+
+	if (ctx)
+		pthread_mutex_lock (&ctx->lock);
         pthread_mutex_lock (&pl_inode->mutex);
         {
                 if (lock->fl_type != F_UNLCK) {
@@ -495,6 +496,10 @@ pl_inode_setlk (xlator_t *this, pl_inode_t *pl_inode, pl_inode_lock_t *lock,
                                 if (can_block)
                                         unref = _gf_false;
                         }
+
+			if (ctx && (!ret || can_block))
+				list_add_tail (&lock->client_list,
+					       &ctx->inodelk_lockers);
                 } else {
                         retlock = __inode_unlock_lock (this, lock, dom);
                         if (!retlock) {
@@ -503,16 +508,21 @@ pl_inode_setlk (xlator_t *this, pl_inode_t *pl_inode, pl_inode_lock_t *lock,
                                 ret = -EINVAL;
                                 goto out;
                         }
-                        __pl_inodelk_unref (retlock);
+			list_del_init (&retlock->client_list);
+			__pl_inodelk_unref (retlock);
 
                         ret = 0;
                 }
-        }
 out:
-        if (unref)
-                __pl_inodelk_unref (lock);
+		if (unref)
+			__pl_inodelk_unref (lock);
+        }
         pthread_mutex_unlock (&pl_inode->mutex);
+	if (ctx)
+		pthread_mutex_unlock (&ctx->lock);
+
         grant_blocked_inode_locks (this, pl_inode, dom);
+
         return ret;
 }
 
@@ -552,6 +562,7 @@ new_inode_lock (struct gf_flock *flock, client_t *client, pid_t client_pid,
 
         INIT_LIST_HEAD (&lock->list);
         INIT_LIST_HEAD (&lock->blocked_locks);
+	INIT_LIST_HEAD (&lock->client_list);
         __pl_inodelk_ref (lock);
 
         return lock;
@@ -627,6 +638,15 @@ pl_common_inodelk (call_frame_t *frame, xlator_t *this,
 
         pl_trace_in (this, frame, fd, loc, cmd, flock, volume);
 
+	if (frame->root->client) {
+		ctx = pl_ctx_get (frame->root->client, this);
+		if (!ctx) {
+			op_errno = ENOMEM;
+			gf_log (this->name, GF_LOG_INFO, "pl_ctx_get() failed");
+			goto unwind;
+		}
+	}
+
         pinode = pl_inode_get (this, inode);
         if (!pinode) {
                 op_errno = ENOMEM;
@@ -639,27 +659,6 @@ pl_common_inodelk (call_frame_t *frame, xlator_t *this,
                 goto unwind;
         }
 
-        if (frame->root->lk_owner.len == 0) {
-                /*
-                  special case: this means release all locks
-                  from this client
-                */
-                gf_log (this->name, GF_LOG_TRACE,
-                        "Releasing all locks from client %p", frame->root->client);
-
-                release_inode_locks_of_client (this, dom, inode, frame->root->client);
-                _pl_convert_volume (volume, &res1);
-                if (res1) {
-                        dom = get_domain (pinode, res1);
-                        if (dom)
-                                release_inode_locks_of_client (this, dom,
-                                                        inode, frame->root->client);
-                }
-
-                op_ret = 0;
-                goto unwind;
-        }
-
         reqlock = new_inode_lock (flock, frame->root->client, frame->root->pid,
                                   frame, this, volume, conn_id);
 
@@ -678,8 +677,8 @@ pl_common_inodelk (call_frame_t *frame, xlator_t *this,
 
         case F_SETLK:
                 memcpy (&reqlock->user_flock, flock, sizeof (struct gf_flock));
-                ret = pl_inode_setlk (this, pinode, reqlock,
-                                      can_block, dom);
+                ret = pl_inode_setlk (this, ctx, pinode, reqlock, can_block,
+				      dom);
 
                 if (ret < 0) {
                         if ((can_block) && (F_UNLCK != flock->l_type)) {
@@ -704,23 +703,6 @@ pl_common_inodelk (call_frame_t *frame, xlator_t *this,
 
         op_ret = 0;
 
-        ctx = pl_ctx_get (frame->root->client, this);
-
-        if (ctx == NULL) {
-                gf_log (this->name, GF_LOG_INFO, "pl_ctx_get() failed");
-                goto unwind;
-        }
-
-        if (flock->l_type == F_UNLCK)
-                pl_del_locker (ctx->ltable, volume, loc, fd,
-                               &frame->root->lk_owner,
-                               GF_FOP_INODELK);
-        else
-                pl_add_locker (ctx->ltable, volume, loc, fd,
-                               frame->root->pid,
-                               &frame->root->lk_owner,
-                               GF_FOP_INODELK);
-
 unwind:
         if ((inode != NULL) && (flock !=NULL)) {
                 pl_update_refkeeper (this, inode);
diff --git a/xlators/features/locks/src/locks.h b/xlators/features/locks/src/locks.h
index 76fc941d74c..8c2a6f867ee 100644
--- a/xlators/features/locks/src/locks.h
+++ b/xlators/features/locks/src/locks.h
@@ -65,7 +65,7 @@ struct __pl_inode_lock {
 
         struct gf_flock    user_flock; /* the flock supplied by the user */
         xlator_t          *this;       /* required for blocked locks */
-        fd_t              *fd;
+	struct __pl_inode *pl_inode;
 
         call_frame_t      *frame;
 
@@ -80,6 +80,8 @@ struct __pl_inode_lock {
         pid_t              client_pid;    /* pid of client process */
 
         char              *connection_id; /* stores the client connection id */
+
+	struct list_head   client_list; /* list of all locks from a client */
 };
 typedef struct __pl_inode_lock pl_inode_lock_t;
 
@@ -103,9 +105,11 @@ typedef struct __pl_dom_list_t pl_dom_list_t;
 struct __entry_lock {
         struct list_head  domain_list;    /* list_head back to pl_dom_list_t */
         struct list_head  blocked_locks; /* list_head back to blocked_entrylks */
+	int ref;
 
         call_frame_t     *frame;
         xlator_t         *this;
+	struct __pl_inode *pinode;
 
         const char       *volume;
 
@@ -115,11 +119,13 @@ struct __entry_lock {
         struct timeval     blkd_time;   /*time at which lock was queued into blkd list*/
         struct timeval     granted_time; /*time at which lock was queued into active list*/
 
-        void             *trans;
+        void             *client;
         gf_lkowner_t      owner;
 	pid_t             client_pid;    /* pid of client process */
 
         char             *connection_id; /* stores the client connection id */
+
+	struct list_head   client_list; /* list of all locks from a client */
 };
 typedef struct __entry_lock pl_entry_lock_t;
 
@@ -144,12 +150,6 @@ struct __pl_inode {
 typedef struct __pl_inode pl_inode_t;
 
 
-struct __pl_fd {
-        gf_boolean_t nonblocking;       /* whether O_NONBLOCK has been set */
-};
-typedef struct __pl_fd pl_fd_t;
-
-
 typedef struct {
         gf_boolean_t    mandatory;      /* if mandatory locking is enabled */
         gf_boolean_t    trace;          /* trace lock requests in and out */
@@ -178,15 +178,27 @@ typedef struct {
 } pl_fdctx_t;
 
 
+struct _locker {
+        struct list_head  lockers;
+        char             *volume;
+        inode_t          *inode;
+        gf_lkowner_t      owner;
+};
+
 typedef struct _locks_ctx {
-        gf_lock_t            ltable_lock; /* only for replace, 
-                                             ltable has its own internal 
-                                             lock for operations */
-        struct _lock_table  *ltable;
+        pthread_mutex_t      lock;
+        struct list_head     inodelk_lockers;
+        struct list_head     entrylk_lockers;
 } pl_ctx_t;
 
 
 pl_ctx_t *
 pl_ctx_get (client_t *client, xlator_t *xlator);
 
+int
+pl_inodelk_client_cleanup (xlator_t *this, pl_ctx_t *ctx);
+
+int
+pl_entrylk_client_cleanup (xlator_t *this, pl_ctx_t *ctx);
+
 #endif /* __POSIX_LOCKS_H__ */
diff --git a/xlators/features/locks/src/posix.c b/xlators/features/locks/src/posix.c
index 7bfb38a51ac..fce0d509f17 100644
--- a/xlators/features/locks/src/posix.c
+++ b/xlators/features/locks/src/posix.c
@@ -2243,7 +2243,7 @@ __dump_entrylks (pl_inode_t *pl_inode)
                                           lock->type == ENTRYLK_RDLCK ? "ENTRYLK_RDLCK" :
                                           "ENTRYLK_WRLCK", lock->basename,
                                           (unsigned long long) lock->client_pid,
-                                          lkowner_utoa (&lock->owner), lock->trans,
+                                          lkowner_utoa (&lock->owner), lock->client,
                                           lock->connection_id,
                                           ctime_r (&lock->granted_time.tv_sec, granted));
                         } else {
@@ -2251,7 +2251,7 @@ __dump_entrylks (pl_inode_t *pl_inode)
                                           lock->type == ENTRYLK_RDLCK ? "ENTRYLK_RDLCK" :
                                           "ENTRYLK_WRLCK", lock->basename,
                                           (unsigned long long) lock->client_pid,
-                                          lkowner_utoa (&lock->owner), lock->trans,
+                                          lkowner_utoa (&lock->owner), lock->client,
                                           lock->connection_id,
                                           ctime_r (&lock->blkd_time.tv_sec, blocked),
                                           ctime_r (&lock->granted_time.tv_sec, granted));
@@ -2271,7 +2271,7 @@ __dump_entrylks (pl_inode_t *pl_inode)
                                   lock->type == ENTRYLK_RDLCK ? "ENTRYLK_RDLCK" :
                                   "ENTRYLK_WRLCK", lock->basename,
                                   (unsigned long long) lock->client_pid,
-                                  lkowner_utoa (&lock->owner), lock->trans,
+                                  lkowner_utoa (&lock->owner), lock->client,
                                   lock->connection_id,
                                   ctime_r (&lock->blkd_time.tv_sec, blocked));
 
@@ -2524,19 +2524,12 @@ pl_ctx_get (client_t *client, xlator_t *xlator)
         if (ctx == NULL)
                 goto out;
 
-        ctx->ltable = pl_lock_table_new();
-
-        if (ctx->ltable == NULL) {
-                GF_FREE (ctx);
-                ctx = NULL;
-                goto out;
-        }
-
-        LOCK_INIT (&ctx->ltable_lock);
+        pthread_mutex_init (&ctx->lock, NULL);
+	INIT_LIST_HEAD (&ctx->inodelk_lockers);
+	INIT_LIST_HEAD (&ctx->entrylk_lockers);
 
         if (client_ctx_set (client, xlator, ctx) != 0) {
-                LOCK_DESTROY (&ctx->ltable_lock);
-                GF_FREE (ctx->ltable);
+                pthread_mutex_destroy (&ctx->lock);
                 GF_FREE (ctx);
                 ctx = NULL;
         }
@@ -2544,82 +2537,44 @@ out:
         return ctx;
 }
 
-static void
-ltable_delete_locks (struct _lock_table *ltable)
+
+static int
+pl_client_disconnect_cbk (xlator_t *this, client_t *client)
 {
-        struct _locker *locker = NULL;
-        struct _locker *tmp    = NULL;
+        pl_ctx_t *pl_ctx = NULL;
 
-        list_for_each_entry_safe (locker, tmp, &ltable->inodelk_lockers, lockers) {
-                if (locker->fd)
-                        pl_del_locker (ltable, locker->volume, &locker->loc,
-                                       locker->fd, &locker->owner,
-                                       GF_FOP_INODELK);
-                GF_FREE (locker->volume);
-                GF_FREE (locker);
-        }
+	pl_ctx = pl_ctx_get (client, this);
 
-        list_for_each_entry_safe (locker, tmp, &ltable->entrylk_lockers, lockers) {
-                if (locker->fd)
-                        pl_del_locker (ltable, locker->volume, &locker->loc,
-                                       locker->fd, &locker->owner,
-                                       GF_FOP_ENTRYLK);
-                GF_FREE (locker->volume);
-                GF_FREE (locker);
-        }
-        GF_FREE (ltable);
+	pl_inodelk_client_cleanup (this, pl_ctx);
+
+	pl_entrylk_client_cleanup (this, pl_ctx);
+
+	return 0;
 }
 
 
-static int32_t
-destroy_cbk (xlator_t *this, client_t *client)
+static int
+pl_client_destroy_cbk (xlator_t *this, client_t *client)
 {
-        void     *tmp       = NULL;
-        pl_ctx_t *locks_ctx = NULL;
+        void     *tmp    = NULL;
+        pl_ctx_t *pl_ctx = NULL;
+
+	pl_client_disconnect_cbk (this, client);
 
         client_ctx_del (client, this, &tmp);
 
         if (tmp == NULL)
-                return 0
-;
-        locks_ctx = tmp;
-        if (locks_ctx->ltable)
-                ltable_delete_locks (locks_ctx->ltable);
-
-        LOCK_DESTROY (&locks_ctx->ltable_lock);
-        GF_FREE (locks_ctx);
-
-        return 0;
-}
+                return 0;
 
+        pl_ctx = tmp;
 
-static int32_t
-disconnect_cbk (xlator_t *this, client_t *client)
-{
-        int32_t              ret       = 0;
-        pl_ctx_t            *locks_ctx = NULL;
-        struct _lock_table  *ltable    = NULL;
+	GF_ASSERT (list_empty(&pl_ctx->inodelk_lockers));
+	GF_ASSERT (list_empty(&pl_ctx->entrylk_lockers));
 
-        locks_ctx = pl_ctx_get (client, this);
-        if (locks_ctx == NULL) {
-                gf_log (this->name, GF_LOG_INFO, "pl_ctx_get() failed");
-                goto out;
-        }
+	pthread_mutex_destroy (&pl_ctx->lock);
+        GF_FREE (pl_ctx);
 
-        LOCK (&locks_ctx->ltable_lock);
-        {
-                if (locks_ctx->ltable) {
-                        ltable = locks_ctx->ltable;
-                        locks_ctx->ltable = pl_lock_table_new ();
-                }
-        }
-        UNLOCK (&locks_ctx->ltable_lock);
-
-        if (ltable)
-                ltable_delete_locks (ltable);
-
-out:
-        return ret;
+        return 0;
 }
 
 
@@ -2756,8 +2711,8 @@ struct xlator_cbks cbks = {
         .forget            = pl_forget,
         .release           = pl_release,
         .releasedir        = pl_releasedir,
-        .client_destroy    = destroy_cbk,
-        .client_disconnect = disconnect_cbk,
+        .client_destroy    = pl_client_destroy_cbk,
+        .client_disconnect = pl_client_disconnect_cbk,
 };