feature/dht: Directory synchronization

Design doc: https://review.gluster.org/16876

Directory creation is now synchronized with blocking inodelk of the
parent on the hashed subvolume followed by the entrylk on the hashed
subvolume between dht_mkdir, dht_rmdir, dht_rename_dir and lookup
selfheal mkdir.

To maintain internal consistency of directories across all subvols of
dht, we need locks. Specifically we are interested in:

 1. Consistency of layout of a directory. Only one writer should modify
    the layout at a time. A writer (layout setting during directory heal
    as part of lookup) shouldn't modify the layout while there are
    readers (all other fops like create, mkdir etc., which consume
    layout) and readers shouldn't read the layout while a writer is in
    progress. Readers can read the layout simultaneously. Writer takes
    a WRITE inodelk on the directory (whose layout is being modified)
    across ALL subvols. Reader takes a READ inodelk on the directory
    (whose layout is being read) on ANY subvol.

 2. Consistency of directory namespace across subvols. The path and
    associated gfid should be same on all subvols. A gfid should not be
    associated with more than one path on any subvol. All fops that can
    change directory names (mkdir, rmdir, renamedir, directory creation
    phase in lookup-heal) takes an entrylk on hashed subvol of the
    directory.

 NOTE1: In point 2 above, since dht takes entrylk on hashed subvol of a
        directory, the transaction itself is a consumer of layout on
        parent directory. So, the transaction is a reader of parent
        layout and does an inodelk on parent directory just like any
        other layout reader. So a mkdir (dir/subdir) would:

     > Acquire a READ inodelk on "dir" on any subvol.
     > Acquire an entrylk (dir, "subdir") on hashed subvol of "subdir".
     > creates directory on hashed subvol and possibly on non-hashed subvols.
     > UNLOCK (entrylk)
     > UNLOCK (inodelk)

 NOTE2: mkdir fop while setting the layout of the directory being created
        is considered as a reader, but NOT a writer. The reason is for
        a fop which can consume the layout of a directory to come either
        of the following conditions has to be true:

     > mkdir syscall from application has to complete. In this case no
       need of synchronization.
     > A lookup issued on the directory racing with mkdir has to complete.
       Since layout setting by a lookup is considered as a writer, only
       one of either mkdir or lookup will set the layout.

Code re-organization:
   All the lock related routines are moved to "dht-lock.c" file.
   New wrapper function is introduced to take blocking inodelk
   followed by entrylk 'dht_protect_namespace'

Updates #191
Change-Id: I01569094dfbe1852de6f586475be79c1ba965a31
Signed-off-by: Kotresh HR <khiremat@redhat.com>
BUG: 1443373
Reviewed-on: https://review.gluster.org/15472
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
CentOS-regression: Gluster Build System <jenkins@build.gluster.org>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
Smoke: Gluster Build System <jenkins@build.gluster.org>

1 files changed, 1383 insertions, 0 deletions

diff --git a/xlators/cluster/dht/src/dht-lock.c b/xlators/cluster/dht/src/dht-lock.c
new file mode 100644
index 00000000000..0a198a17db4
--- /dev/null
+++ b/xlators/cluster/dht/src/dht-lock.c
@@ -0,0 +1,1383 @@
+/*
+  Copyright (c) 2016 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
+*/
+
+#include "dht-lock.h"
+
+static char *
+dht_lock_asprintf (dht_lock_t *lock)
+{
+        char *lk_buf                = NULL;
+        char gfid[GF_UUID_BUF_SIZE] = {0, };
+
+        if (lock == NULL)
+                goto out;
+
+        uuid_utoa_r (lock->loc.gfid, gfid);
+
+        gf_asprintf (&lk_buf, "%s:%s", lock->xl->name, gfid);
+
+out:
+        return lk_buf;
+}
+
+static void
+dht_log_lk_array (char *name, gf_loglevel_t log_level, dht_lock_t **lk_array,
+                  int count)
+{
+        int   i      = 0;
+        char *lk_buf = NULL;
+
+        if ((lk_array == NULL) || (count == 0))
+                goto out;
+
+        for (i = 0; i < count; i++) {
+                lk_buf = dht_lock_asprintf (lk_array[i]);
+                if (!lk_buf)
+                        goto out;
+
+                gf_msg (name, log_level, 0, DHT_MSG_LK_ARRAY_INFO,
+                        "%d. %s", i, lk_buf);
+                GF_FREE (lk_buf);
+        }
+
+out:
+        return;
+}
+
+static void
+dht_lock_stack_destroy (call_frame_t *lock_frame, dht_lock_type_t lk)
+{
+        dht_local_t *local = NULL;
+
+        local = lock_frame->local;
+
+        if (lk == DHT_INODELK) {
+                local->lock[0].layout.my_layout.locks = NULL;
+                local->lock[0].layout.my_layout.lk_count = 0;
+        } else {
+                local->lock[0].ns.directory_ns.locks = NULL;
+                local->lock[0].ns.directory_ns.lk_count = 0;
+        }
+
+        DHT_STACK_DESTROY (lock_frame);
+        return;
+}
+
+static void
+dht_lock_free (dht_lock_t *lock)
+{
+        if (lock == NULL)
+                goto out;
+
+        loc_wipe (&lock->loc);
+        GF_FREE (lock->domain);
+        GF_FREE (lock->basename);
+        mem_put (lock);
+
+out:
+        return;
+}
+
+static void
+dht_set_lkowner (dht_lock_t **lk_array, int count, gf_lkowner_t *lkowner)
+{
+        int i = 0;
+
+        if (!lk_array || !lkowner)
+                goto out;
+
+        for (i = 0; i < count; i++) {
+                lk_array[i]->lk_owner = *lkowner;
+        }
+
+out:
+        return;
+}
+
+static int
+dht_lock_request_cmp (const void *val1, const void *val2)
+{
+        dht_lock_t *lock1 = NULL;
+        dht_lock_t *lock2 = NULL;
+        int         ret   = -1;
+
+        lock1 = *(dht_lock_t **)val1;
+        lock2 = *(dht_lock_t **)val2;
+
+        GF_VALIDATE_OR_GOTO ("dht-locks", lock1, out);
+        GF_VALIDATE_OR_GOTO ("dht-locks", lock2, out);
+
+        ret = strcmp (lock1->xl->name, lock2->xl->name);
+
+        if (ret == 0) {
+                ret = gf_uuid_compare (lock1->loc.gfid, lock2->loc.gfid);
+        }
+
+out:
+        return ret;
+}
+
+static int
+dht_lock_order_requests (dht_lock_t **locks, int count)
+{
+        int        ret     = -1;
+
+        if (!locks || !count)
+                goto out;
+
+        qsort (locks, count, sizeof (*locks), dht_lock_request_cmp);
+        ret = 0;
+
+out:
+        return ret;
+}
+
+void
+dht_lock_array_free (dht_lock_t **lk_array, int count)
+{
+        int            i       = 0;
+        dht_lock_t    *lock    = NULL;
+
+        if (lk_array == NULL)
+                goto out;
+
+        for (i = 0; i < count; i++) {
+                lock = lk_array[i];
+                lk_array[i] = NULL;
+                dht_lock_free (lock);
+        }
+
+out:
+        return;
+}
+
+int32_t
+dht_lock_count (dht_lock_t **lk_array, int lk_count)
+{
+        int i = 0, locked = 0;
+
+        if ((lk_array == NULL) || (lk_count == 0))
+                goto out;
+
+        for (i = 0; i < lk_count; i++) {
+                if (lk_array[i]->locked)
+                        locked++;
+        }
+out:
+        return locked;
+}
+
+static call_frame_t *
+dht_lock_frame (call_frame_t *parent_frame)
+{
+        call_frame_t *lock_frame = NULL;
+
+        lock_frame = copy_frame (parent_frame);
+        if (lock_frame == NULL)
+                goto out;
+
+        set_lk_owner_from_ptr (&lock_frame->root->lk_owner, parent_frame->root);
+
+out:
+        return lock_frame;
+}
+
+dht_lock_t *
+dht_lock_new (xlator_t *this, xlator_t *xl, loc_t *loc, short type,
+              const char *domain, const char *basename)
+{
+        dht_conf_t *conf = NULL;
+        dht_lock_t *lock = NULL;
+
+        conf = this->private;
+
+        lock = mem_get0 (conf->lock_pool);
+        if (lock == NULL)
+                goto out;
+
+        lock->xl = xl;
+        lock->type = type;
+
+        lock->domain = gf_strdup (domain);
+        if (lock->domain == NULL) {
+                dht_lock_free (lock);
+                lock = NULL;
+                goto out;
+        }
+
+        if (basename) {
+                lock->basename = gf_strdup (basename);
+                if (lock->basename == NULL) {
+                        dht_lock_free (lock);
+                        lock = NULL;
+                        goto out;
+                }
+        }
+
+        /* Fill only inode and gfid.
+           posix and protocol/server give preference to pargfid/basename over
+           gfid/inode for resolution if all the three parameters of loc_t are
+           present. I want to avoid the following hypothetical situation:
+
+           1. rebalance did a lookup on a dentry and got a gfid.
+           2. rebalance acquires lock on loc_t which was filled with gfid and
+              path (pargfid/bname) from step 1.
+           3. somebody deleted and recreated the same file
+           4. rename on the same path acquires lock on loc_t which now points
+              to a different inode (and hence gets the lock).
+           5. rebalance continues to migrate file (note that not all fops done
+              by rebalance during migration are inode/gfid based Eg., unlink)
+           6. rename continues.
+        */
+        lock->loc.inode = inode_ref (loc->inode);
+        loc_gfid (loc, lock->loc.gfid);
+
+out:
+        return lock;
+}
+
+static int
+dht_local_entrylk_init (call_frame_t *frame, dht_lock_t **lk_array,
+                     int lk_count, fop_entrylk_cbk_t entrylk_cbk)
+{
+        int          ret   = -1;
+        dht_local_t *local = NULL;
+
+        local = frame->local;
+
+        if (local == NULL) {
+                local = dht_local_init (frame, NULL, NULL, 0);
+        }
+
+        if (local == NULL) {
+                goto out;
+        }
+
+        local->lock[0].ns.directory_ns.entrylk_cbk = entrylk_cbk;
+        local->lock[0].ns.directory_ns.locks = lk_array;
+        local->lock[0].ns.directory_ns.lk_count = lk_count;
+
+        ret = dht_lock_order_requests (local->lock[0].ns.directory_ns.locks,
+                                       local->lock[0].ns.directory_ns.lk_count);
+        if (ret < 0)
+                goto out;
+
+        ret = 0;
+out:
+        return ret;
+}
+
+static void
+dht_entrylk_done (call_frame_t *lock_frame)
+{
+        fop_entrylk_cbk_t  entrylk_cbk = NULL;
+        call_frame_t      *main_frame  = NULL;
+        dht_local_t       *local       = NULL;
+
+        local = lock_frame->local;
+        main_frame = local->main_frame;
+
+        local->lock[0].ns.directory_ns.locks = NULL;
+        local->lock[0].ns.directory_ns.lk_count = 0;
+
+        entrylk_cbk = local->lock[0].ns.directory_ns.entrylk_cbk;
+        local->lock[0].ns.directory_ns.entrylk_cbk = NULL;
+
+        entrylk_cbk (main_frame, NULL, main_frame->this,
+                     local->lock[0].ns.directory_ns.op_ret,
+                     local->lock[0].ns.directory_ns.op_errno, NULL);
+
+        dht_lock_stack_destroy (lock_frame, DHT_ENTRYLK);
+        return;
+}
+
+static int32_t
+dht_unlock_entrylk_done (call_frame_t *frame, void *cookie, xlator_t *this,
+                          int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t *local                   = NULL;
+        char          gfid[GF_UUID_BUF_SIZE] = {0};
+
+        local = frame->local;
+        gf_uuid_unparse (local->lock[0].ns.directory_ns.locks[0]->loc.inode->gfid, gfid);
+
+        if (op_ret < 0) {
+                gf_msg (this->name, GF_LOG_WARNING, op_errno,
+                        DHT_MSG_PARENT_LAYOUT_CHANGED,
+                        "unlock failed on gfid: %s, stale lock might be left "
+                        "in DHT_LAYOUT_HEAL_DOMAIN", gfid);
+        }
+
+        DHT_STACK_DESTROY (frame);
+        return 0;
+}
+
+static int32_t
+dht_unlock_entrylk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                        int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t *local                  = NULL;
+        int          lk_index               = 0, call_cnt = 0;
+        char         gfid[GF_UUID_BUF_SIZE] = {0};
+
+        lk_index = (long) cookie;
+
+        local = frame->local;
+
+        uuid_utoa_r (local->lock[0].ns.directory_ns.locks[lk_index]->loc.gfid, gfid);
+
+        if (op_ret < 0) {
+                gf_msg (this->name, GF_LOG_WARNING, op_errno,
+                        DHT_MSG_UNLOCKING_FAILED,
+                        "unlocking failed on %s:%s",
+                        local->lock[0].ns.directory_ns.locks[lk_index]->xl->name,
+                        gfid);
+        } else {
+                local->lock[0].ns.directory_ns.locks[lk_index]->locked = 0;
+        }
+
+        call_cnt = dht_frame_return (frame);
+        if (is_last_call (call_cnt)) {
+                dht_entrylk_done (frame);
+        }
+
+        return 0;
+}
+
+static int32_t
+dht_unlock_entrylk (call_frame_t *frame, dht_lock_t **lk_array, int lk_count,
+                     fop_entrylk_cbk_t entrylk_cbk)
+{
+        dht_local_t     *local      = NULL;
+        int              ret        = -1 , i = 0;
+        call_frame_t    *lock_frame = NULL;
+        int              call_cnt   = 0;
+
+        GF_VALIDATE_OR_GOTO ("dht-locks", frame, done);
+        GF_VALIDATE_OR_GOTO (frame->this->name, lk_array, done);
+        GF_VALIDATE_OR_GOTO (frame->this->name, entrylk_cbk, done);
+
+        call_cnt = dht_lock_count (lk_array, lk_count);
+        if (call_cnt == 0) {
+                ret = 0;
+                goto done;
+        }
+
+        lock_frame = dht_lock_frame (frame);
+        if (lock_frame == NULL) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, 0,
+                        DHT_MSG_UNLOCKING_FAILED,
+                        "cannot allocate a frame, not unlocking following "
+                        "entrylks:");
+
+                dht_log_lk_array (frame->this->name, GF_LOG_WARNING, lk_array,
+                                  lk_count);
+                goto done;
+        }
+
+        ret = dht_local_entrylk_init (lock_frame, lk_array, lk_count,
+                                      entrylk_cbk);
+        if (ret < 0) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, 0,
+                        DHT_MSG_UNLOCKING_FAILED,
+                        "storing locks in local failed, not unlocking "
+                        "following entrylks:");
+
+                dht_log_lk_array (frame->this->name, GF_LOG_WARNING, lk_array,
+                                  lk_count);
+
+                goto done;
+        }
+
+        local = lock_frame->local;
+        local->main_frame = frame;
+        local->call_cnt = call_cnt;
+
+        for (i = 0; i < local->lock[0].ns.directory_ns.lk_count; i++) {
+                if (!local->lock[0].ns.directory_ns.locks[i]->locked)
+                        continue;
+
+                lock_frame->root->lk_owner = local->lock[0].ns.directory_ns.locks[i]->lk_owner;
+                STACK_WIND_COOKIE (lock_frame, dht_unlock_entrylk_cbk,
+                                   (void *)(long)i,
+                                   local->lock[0].ns.directory_ns.locks[i]->xl,
+                                   local->lock[0].ns.directory_ns.locks[i]->xl->fops->entrylk,
+                                   local->lock[0].ns.directory_ns.locks[i]->domain,
+                                   &local->lock[0].ns.directory_ns.locks[i]->loc,
+                                   local->lock[0].ns.directory_ns.locks[i]->basename,
+                                   ENTRYLK_UNLOCK, ENTRYLK_WRLCK, NULL);
+                if (!--call_cnt)
+                        break;
+        }
+
+        return 0;
+
+done:
+        if (lock_frame)
+                dht_lock_stack_destroy (lock_frame, DHT_ENTRYLK);
+
+        /* no locks acquired, invoke entrylk_cbk */
+        if (ret == 0)
+                entrylk_cbk (frame, NULL, frame->this, 0, 0, NULL);
+
+        return ret;
+}
+
+int32_t
+dht_unlock_entrylk_wrapper (call_frame_t *frame, dht_elock_wrap_t *entrylk)
+{
+        dht_local_t  *local                   = NULL, *lock_local = NULL;
+        call_frame_t *lock_frame              = NULL;
+        char          pgfid[GF_UUID_BUF_SIZE] = {0};
+        int           ret                     = 0;
+
+        local = frame->local;
+
+        if (!entrylk || !entrylk->locks)
+                goto out;
+
+        gf_uuid_unparse (local->loc.parent->gfid, pgfid);
+
+        lock_frame = copy_frame (frame);
+        if (lock_frame == NULL) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, ENOMEM,
+                        DHT_MSG_PARENT_LAYOUT_CHANGED,
+                        "mkdir (%s/%s) (path: %s): "
+                        "copy frame failed", pgfid, local->loc.name,
+                        local->loc.path);
+                goto done;
+        }
+
+        lock_local = mem_get0 (THIS->local_pool);
+        if (lock_local == NULL) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, ENOMEM,
+                        DHT_MSG_PARENT_LAYOUT_CHANGED,
+                        "mkdir (%s/%s) (path: %s): "
+                        "local creation failed", pgfid, local->loc.name,
+                        local->loc.path);
+                goto done;
+        }
+
+        lock_frame->local = lock_local;
+
+        lock_local->lock[0].ns.directory_ns.locks = entrylk->locks;
+        lock_local->lock[0].ns.directory_ns.lk_count = entrylk->lk_count;
+        entrylk->locks = NULL;
+        entrylk->lk_count = 0;
+
+        ret = dht_unlock_entrylk (lock_frame,
+                                  lock_local->lock[0].ns.directory_ns.locks,
+                                  lock_local->lock[0].ns.directory_ns.lk_count,
+                                  dht_unlock_entrylk_done);
+        if (ret)
+                goto done;
+
+        lock_frame = NULL;
+
+done:
+        if (lock_frame != NULL) {
+                DHT_STACK_DESTROY (lock_frame);
+        }
+
+out:
+        return 0;
+}
+
+static int
+dht_entrylk_cleanup_cbk (call_frame_t *frame, void *cookie,
+                         xlator_t *this, int32_t op_ret, int32_t op_errno,
+                         dict_t *xdata)
+{
+        dht_entrylk_done (frame);
+        return 0;
+}
+
+static void
+dht_entrylk_cleanup (call_frame_t *lock_frame)
+{
+        dht_lock_t  **lk_array = NULL;
+        int           lk_count = 0, lk_acquired = 0;
+        dht_local_t  *local    = NULL;
+
+        local = lock_frame->local;
+
+        lk_array = local->lock[0].ns.directory_ns.locks;
+        lk_count = local->lock[0].ns.directory_ns.lk_count;
+
+        lk_acquired = dht_lock_count (lk_array, lk_count);
+        if (lk_acquired != 0) {
+                dht_unlock_entrylk (lock_frame, lk_array, lk_count,
+                                     dht_entrylk_cleanup_cbk);
+        } else {
+                dht_entrylk_done (lock_frame);
+        }
+
+        return;
+}
+
+
+static int32_t
+dht_blocking_entrylk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                          int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        int          lk_index = 0;
+        int          i        = 0;
+        dht_local_t *local    = NULL;
+
+        lk_index = (long) cookie;
+
+        local = frame->local;
+        if (op_ret == 0) {
+                local->lock[0].ns.directory_ns.locks[lk_index]->locked = _gf_true;
+        } else {
+                switch (op_errno) {
+                case ESTALE:
+                case ENOENT:
+                        if (local->lock[0].ns.directory_ns.reaction != IGNORE_ENOENT_ESTALE) {
+                                local->lock[0].ns.directory_ns.op_ret = -1;
+                                local->lock[0].ns.directory_ns.op_errno = op_errno;
+                                goto cleanup;
+                        }
+                        break;
+                default:
+                        local->lock[0].ns.directory_ns.op_ret = -1;
+                        local->lock[0].ns.directory_ns.op_errno = op_errno;
+                        goto cleanup;
+                }
+        }
+
+        if (lk_index == (local->lock[0].ns.directory_ns.lk_count - 1)) {
+                for (i = 0; (i < local->lock[0].ns.directory_ns.lk_count) &&
+                     (!local->lock[0].ns.directory_ns.locks[i]->locked); i++)
+                        ;
+
+                if (i == local->lock[0].ns.directory_ns.lk_count) {
+                        local->lock[0].ns.directory_ns.op_ret = -1;
+                        local->lock[0].ns.directory_ns.op_errno = op_errno;
+                }
+
+                dht_entrylk_done (frame);
+        } else {
+                dht_blocking_entrylk_rec (frame, ++lk_index);
+        }
+
+        return 0;
+
+cleanup:
+        dht_entrylk_cleanup (frame);
+
+        return 0;
+}
+
+void
+dht_blocking_entrylk_rec (call_frame_t *frame, int i)
+{
+        dht_local_t     *local = NULL;
+
+        local = frame->local;
+
+        STACK_WIND_COOKIE (frame, dht_blocking_entrylk_cbk,
+                           (void *) (long) i,
+                           local->lock[0].ns.directory_ns.locks[i]->xl,
+                           local->lock[0].ns.directory_ns.locks[i]->xl->fops->entrylk,
+                           local->lock[0].ns.directory_ns.locks[i]->domain,
+                           &local->lock[0].ns.directory_ns.locks[i]->loc,
+                           local->lock[0].ns.directory_ns.locks[i]->basename,
+                           ENTRYLK_LOCK, ENTRYLK_WRLCK, NULL);
+
+        return;
+}
+
+int
+dht_blocking_entrylk (call_frame_t *frame, dht_lock_t **lk_array,
+                      int lk_count, dht_reaction_type_t reaction,
+                      fop_entrylk_cbk_t entrylk_cbk)
+{
+        int           ret        = -1;
+        call_frame_t *lock_frame = NULL;
+        dht_local_t  *local      = NULL;
+
+        GF_VALIDATE_OR_GOTO ("dht-locks", frame, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, lk_array, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, entrylk_cbk, out);
+
+        lock_frame = dht_lock_frame (frame);
+        if (lock_frame == NULL)
+                goto out;
+
+        ret = dht_local_entrylk_init (lock_frame, lk_array, lk_count,
+                                      entrylk_cbk);
+        if (ret < 0) {
+                goto out;
+        }
+
+        dht_set_lkowner (lk_array, lk_count, &lock_frame->root->lk_owner);
+
+        local = lock_frame->local;
+        local->lock[0].ns.directory_ns.reaction = reaction;
+        local->main_frame = frame;
+
+        dht_blocking_entrylk_rec (lock_frame, 0);
+
+        return 0;
+out:
+        if (lock_frame)
+                dht_lock_stack_destroy (lock_frame, DHT_ENTRYLK);
+
+        return -1;
+}
+
+static int
+dht_local_inodelk_init (call_frame_t *frame, dht_lock_t **lk_array,
+                     int lk_count, fop_inodelk_cbk_t inodelk_cbk)
+{
+        int          ret   = -1;
+        dht_local_t *local = NULL;
+
+        local = frame->local;
+
+        if (local == NULL) {
+                local = dht_local_init (frame, NULL, NULL, 0);
+        }
+
+        if (local == NULL) {
+                goto out;
+        }
+
+        local->lock[0].layout.my_layout.inodelk_cbk = inodelk_cbk;
+        local->lock[0].layout.my_layout.locks = lk_array;
+        local->lock[0].layout.my_layout.lk_count = lk_count;
+
+        ret = dht_lock_order_requests (local->lock[0].layout.my_layout.locks,
+                                       local->lock[0].layout.my_layout.lk_count);
+        if (ret < 0)
+                goto out;
+
+        ret = 0;
+out:
+        return ret;
+}
+
+static void
+dht_inodelk_done (call_frame_t *lock_frame)
+{
+        fop_inodelk_cbk_t  inodelk_cbk = NULL;
+        call_frame_t      *main_frame  = NULL;
+        dht_local_t       *local       = NULL;
+
+        local = lock_frame->local;
+        main_frame = local->main_frame;
+
+        local->lock[0].layout.my_layout.locks = NULL;
+        local->lock[0].layout.my_layout.lk_count = 0;
+
+        inodelk_cbk = local->lock[0].layout.my_layout.inodelk_cbk;
+        local->lock[0].layout.my_layout.inodelk_cbk = NULL;
+
+        inodelk_cbk (main_frame, NULL, main_frame->this,
+                     local->lock[0].layout.my_layout.op_ret,
+                     local->lock[0].layout.my_layout.op_errno, NULL);
+
+        dht_lock_stack_destroy (lock_frame, DHT_INODELK);
+        return;
+}
+
+static int32_t
+dht_unlock_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                        int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t *local                  = NULL;
+        int          lk_index               = 0, call_cnt = 0;
+        char         gfid[GF_UUID_BUF_SIZE] = {0};
+
+        lk_index = (long) cookie;
+
+        local = frame->local;
+        if (op_ret < 0) {
+                uuid_utoa_r (local->lock[0].layout.my_layout.locks[lk_index]->loc.gfid,
+                             gfid);
+
+                gf_msg (this->name, GF_LOG_WARNING, op_errno,
+                        DHT_MSG_UNLOCKING_FAILED,
+                        "unlocking failed on %s:%s",
+                        local->lock[0].layout.my_layout.locks[lk_index]->xl->name,
+                        gfid);
+        } else {
+                local->lock[0].layout.my_layout.locks[lk_index]->locked = 0;
+        }
+
+        call_cnt = dht_frame_return (frame);
+        if (is_last_call (call_cnt)) {
+                dht_inodelk_done (frame);
+        }
+
+        return 0;
+}
+
+static int32_t
+dht_unlock_inodelk_done (call_frame_t *frame, void *cookie, xlator_t *this,
+                          int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t *local                   = NULL;
+        char          gfid[GF_UUID_BUF_SIZE] = {0};
+
+        local = frame->local;
+        gf_uuid_unparse (local->lock[0].layout.my_layout.locks[0]->loc.inode->gfid, gfid);
+
+        if (op_ret < 0) {
+                gf_msg (this->name, GF_LOG_WARNING, op_errno,
+                        DHT_MSG_PARENT_LAYOUT_CHANGED,
+                        "unlock failed on gfid: %s, stale lock might be left "
+                        "in DHT_LAYOUT_HEAL_DOMAIN", gfid);
+        }
+
+        DHT_STACK_DESTROY (frame);
+        return 0;
+}
+
+int32_t
+dht_unlock_inodelk (call_frame_t *frame, dht_lock_t **lk_array, int lk_count,
+                     fop_inodelk_cbk_t inodelk_cbk)
+{
+        dht_local_t     *local      = NULL;
+        struct gf_flock  flock      = {0,};
+        int              ret        = -1 , i = 0;
+        call_frame_t    *lock_frame = NULL;
+        int              call_cnt   = 0;
+
+        GF_VALIDATE_OR_GOTO ("dht-locks", frame, done);
+        GF_VALIDATE_OR_GOTO (frame->this->name, lk_array, done);
+        GF_VALIDATE_OR_GOTO (frame->this->name, inodelk_cbk, done);
+
+        call_cnt = dht_lock_count (lk_array, lk_count);
+        if (call_cnt == 0) {
+                ret = 0;
+                goto done;
+        }
+
+        lock_frame = dht_lock_frame (frame);
+        if (lock_frame == NULL) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, 0,
+                        DHT_MSG_UNLOCKING_FAILED,
+                        "cannot allocate a frame, not unlocking following "
+                        "locks:");
+
+                dht_log_lk_array (frame->this->name, GF_LOG_WARNING, lk_array,
+                                  lk_count);
+                goto done;
+        }
+
+        ret = dht_local_inodelk_init (lock_frame, lk_array, lk_count,
+                                      inodelk_cbk);
+        if (ret < 0) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, 0,
+                        DHT_MSG_UNLOCKING_FAILED,
+                        "storing locks in local failed, not unlocking "
+                        "following locks:");
+
+                dht_log_lk_array (frame->this->name, GF_LOG_WARNING, lk_array,
+                                  lk_count);
+
+                goto done;
+        }
+
+        local = lock_frame->local;
+        local->main_frame = frame;
+        local->call_cnt = call_cnt;
+
+        flock.l_type = F_UNLCK;
+
+        for (i = 0; i < local->lock[0].layout.my_layout.lk_count; i++) {
+                if (!local->lock[0].layout.my_layout.locks[i]->locked)
+                        continue;
+
+                lock_frame->root->lk_owner = local->lock[0].layout.my_layout.locks[i]->lk_owner;
+                STACK_WIND_COOKIE (lock_frame, dht_unlock_inodelk_cbk,
+                                   (void *)(long)i,
+                                   local->lock[0].layout.my_layout.locks[i]->xl,
+                                   local->lock[0].layout.my_layout.locks[i]->xl->fops->inodelk,
+                                   local->lock[0].layout.my_layout.locks[i]->domain,
+                                   &local->lock[0].layout.my_layout.locks[i]->loc, F_SETLK,
+                                   &flock, NULL);
+                if (!--call_cnt)
+                        break;
+        }
+
+        return 0;
+
+done:
+        if (lock_frame)
+                dht_lock_stack_destroy (lock_frame, DHT_INODELK);
+
+        /* no locks acquired, invoke inodelk_cbk */
+        if (ret == 0)
+                inodelk_cbk (frame, NULL, frame->this, 0, 0, NULL);
+
+        return ret;
+}
+
+int32_t
+dht_unlock_inodelk_wrapper (call_frame_t *frame, dht_ilock_wrap_t *inodelk)
+{
+        dht_local_t  *local                   = NULL, *lock_local = NULL;
+        call_frame_t *lock_frame              = NULL;
+        char          pgfid[GF_UUID_BUF_SIZE] = {0};
+        int           ret                     = 0;
+
+        local = frame->local;
+
+        if (!inodelk || !inodelk->locks)
+                goto out;
+
+        gf_uuid_unparse (local->loc.parent->gfid, pgfid);
+
+        lock_frame = copy_frame (frame);
+        if (lock_frame == NULL) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, ENOMEM,
+                        DHT_MSG_PARENT_LAYOUT_CHANGED,
+                        "mkdir (%s/%s) (path: %s): "
+                        "copy frame failed", pgfid, local->loc.name,
+                        local->loc.path);
+                goto done;
+        }
+
+        lock_local = mem_get0 (THIS->local_pool);
+        if (lock_local == NULL) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, ENOMEM,
+                        DHT_MSG_PARENT_LAYOUT_CHANGED,
+                        "mkdir (%s/%s) (path: %s): "
+                        "local creation failed", pgfid, local->loc.name,
+                        local->loc.path);
+                goto done;
+        }
+
+        lock_frame->local = lock_local;
+
+        lock_local->lock[0].layout.my_layout.locks = inodelk->locks;
+        lock_local->lock[0].layout.my_layout.lk_count = inodelk->lk_count;
+        inodelk->locks = NULL;
+        inodelk->lk_count = 0;
+
+        ret = dht_unlock_inodelk (lock_frame,
+                                  lock_local->lock[0].layout.my_layout.locks,
+                                  lock_local->lock[0].layout.my_layout.lk_count,
+                                  dht_unlock_inodelk_done);
+
+        if (ret)
+                goto done;
+
+        lock_frame = NULL;
+
+done:
+        if (lock_frame != NULL) {
+                DHT_STACK_DESTROY (lock_frame);
+        }
+out:
+        return 0;
+}
+
+static int
+dht_inodelk_cleanup_cbk (call_frame_t *frame, void *cookie,
+                         xlator_t *this, int32_t op_ret, int32_t op_errno,
+                         dict_t *xdata)
+{
+        dht_inodelk_done (frame);
+        return 0;
+}
+
+static void
+dht_inodelk_cleanup (call_frame_t *lock_frame)
+{
+        dht_lock_t  **lk_array = NULL;
+        int           lk_count = 0, lk_acquired = 0;
+        dht_local_t  *local    = NULL;
+
+        local = lock_frame->local;
+
+        lk_array = local->lock[0].layout.my_layout.locks;
+        lk_count = local->lock[0].layout.my_layout.lk_count;
+
+        lk_acquired = dht_lock_count (lk_array, lk_count);
+        if (lk_acquired != 0) {
+                dht_unlock_inodelk (lock_frame, lk_array, lk_count,
+                                     dht_inodelk_cleanup_cbk);
+        } else {
+                dht_inodelk_done (lock_frame);
+        }
+
+        return;
+}
+
+static int32_t
+dht_nonblocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                             int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t *local                   = NULL;
+        int          lk_index               = 0, call_cnt = 0;
+        char          gfid[GF_UUID_BUF_SIZE] = {0};
+
+        local = frame->local;
+        lk_index = (long) cookie;
+
+        if (op_ret == -1) {
+                local->lock[0].layout.my_layout.op_ret = -1;
+                local->lock[0].layout.my_layout.op_errno = op_errno;
+
+                if (local && local->lock[0].layout.my_layout.locks[lk_index]) {
+                        uuid_utoa_r (local->lock[0].layout.my_layout.locks[lk_index]->loc.inode->gfid,
+                                     gfid);
+
+                        gf_msg_debug (this->name, op_errno,
+                                      "inodelk failed on gfid: %s "
+                                      "subvolume: %s", gfid,
+                                      local->lock[0].layout.my_layout.locks[lk_index]->xl->name);
+                }
+
+                goto out;
+        }
+
+        local->lock[0].layout.my_layout.locks[lk_index]->locked = _gf_true;
+
+out:
+        call_cnt = dht_frame_return (frame);
+        if (is_last_call (call_cnt)) {
+                if (local->lock[0].layout.my_layout.op_ret < 0) {
+                        dht_inodelk_cleanup (frame);
+                        return 0;
+                }
+
+                dht_inodelk_done (frame);
+        }
+
+        return 0;
+}
+
+int
+dht_nonblocking_inodelk (call_frame_t *frame, dht_lock_t **lk_array,
+                         int lk_count, fop_inodelk_cbk_t inodelk_cbk)
+{
+        struct gf_flock  flock      = {0,};
+        int              i          = 0, ret = 0;
+        dht_local_t     *local      = NULL;
+        call_frame_t    *lock_frame = NULL;
+
+        GF_VALIDATE_OR_GOTO ("dht-locks", frame, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, lk_array, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, inodelk_cbk, out);
+
+        lock_frame = dht_lock_frame (frame);
+        if (lock_frame == NULL)
+                goto out;
+
+        ret = dht_local_inodelk_init (lock_frame, lk_array, lk_count,
+                                      inodelk_cbk);
+        if (ret < 0) {
+                goto out;
+        }
+
+        dht_set_lkowner (lk_array, lk_count, &lock_frame->root->lk_owner);
+
+        local = lock_frame->local;
+        local->main_frame = frame;
+
+        local->call_cnt = lk_count;
+
+        for (i = 0; i < lk_count; i++) {
+                flock.l_type = local->lock[0].layout.my_layout.locks[i]->type;
+
+                STACK_WIND_COOKIE (lock_frame, dht_nonblocking_inodelk_cbk,
+                                   (void *) (long) i,
+                                   local->lock[0].layout.my_layout.locks[i]->xl,
+                                   local->lock[0].layout.my_layout.locks[i]->xl->fops->inodelk,
+                                   local->lock[0].layout.my_layout.locks[i]->domain,
+                                   &local->lock[0].layout.my_layout.locks[i]->loc,
+                                   F_SETLK,
+                                   &flock, NULL);
+        }
+
+        return 0;
+
+out:
+        if (lock_frame)
+                dht_lock_stack_destroy (lock_frame, DHT_INODELK);
+
+        return -1;
+}
+
+static int32_t
+dht_blocking_inodelk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                          int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        int          lk_index                   = 0;
+        int          i                          = 0;
+        dht_local_t *local                      = NULL;
+        char         gfid[GF_UUID_BUF_SIZE]     = {0,};
+
+        lk_index = (long) cookie;
+
+        local = frame->local;
+        if (op_ret == 0) {
+                local->lock[0].layout.my_layout.locks[lk_index]->locked = _gf_true;
+        } else {
+                switch (op_errno) {
+                case ESTALE:
+                case ENOENT:
+                        if (local->lock[0].layout.my_layout.reaction != IGNORE_ENOENT_ESTALE) {
+                                gf_uuid_unparse (local->lock[0].layout.my_layout.locks[lk_index]->loc.gfid, gfid);
+                                local->lock[0].layout.my_layout.op_ret = -1;
+                                local->lock[0].layout.my_layout.op_errno = op_errno;
+                                gf_msg (this->name, GF_LOG_ERROR, op_errno,
+                                        DHT_MSG_INODELK_FAILED,
+                                        "inodelk failed on subvol %s. gfid:%s",
+                                        local->lock[0].layout.my_layout.locks[lk_index]->xl->name,
+                                        gfid);
+                                goto cleanup;
+                        }
+                        break;
+                default:
+                        gf_uuid_unparse (local->lock[0].layout.my_layout.locks[lk_index]->loc.gfid, gfid);
+                        local->lock[0].layout.my_layout.op_ret = -1;
+                        local->lock[0].layout.my_layout.op_errno = op_errno;
+                        gf_msg (this->name, GF_LOG_ERROR, op_errno,
+                                DHT_MSG_INODELK_FAILED,
+                                "inodelk failed on subvol %s, gfid:%s",
+                                local->lock[0].layout.my_layout.locks[lk_index]->xl->name, gfid);
+                        goto cleanup;
+                }
+        }
+
+        if (lk_index == (local->lock[0].layout.my_layout.lk_count - 1)) {
+                for (i = 0; (i < local->lock[0].layout.my_layout.lk_count) &&
+                     (!local->lock[0].layout.my_layout.locks[i]->locked); i++)
+                        ;
+
+                if (i == local->lock[0].layout.my_layout.lk_count) {
+                        local->lock[0].layout.my_layout.op_ret = -1;
+                        local->lock[0].layout.my_layout.op_errno = op_errno;
+                }
+
+                dht_inodelk_done (frame);
+        } else {
+                dht_blocking_inodelk_rec (frame, ++lk_index);
+        }
+
+        return 0;
+
+cleanup:
+        dht_inodelk_cleanup (frame);
+
+        return 0;
+}
+
+void
+dht_blocking_inodelk_rec (call_frame_t *frame, int i)
+{
+        dht_local_t     *local = NULL;
+        struct gf_flock  flock = {0,};
+
+        local = frame->local;
+
+        flock.l_type = local->lock[0].layout.my_layout.locks[i]->type;
+
+        STACK_WIND_COOKIE (frame, dht_blocking_inodelk_cbk,
+                           (void *) (long) i,
+                           local->lock[0].layout.my_layout.locks[i]->xl,
+                           local->lock[0].layout.my_layout.locks[i]->xl->fops->inodelk,
+                           local->lock[0].layout.my_layout.locks[i]->domain,
+                           &local->lock[0].layout.my_layout.locks[i]->loc,
+                           F_SETLKW,
+                           &flock, NULL);
+
+        return;
+}
+
+int
+dht_blocking_inodelk (call_frame_t *frame, dht_lock_t **lk_array,
+                      int lk_count, dht_reaction_type_t reaction,
+                      fop_inodelk_cbk_t inodelk_cbk)
+{
+        int           ret                       = -1;
+        call_frame_t *lock_frame                = NULL;
+        dht_local_t  *local                     = NULL;
+        dht_local_t  *tmp_local                 = NULL;
+        char          gfid[GF_UUID_BUF_SIZE]    = {0,};
+
+        GF_VALIDATE_OR_GOTO ("dht-locks", frame, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, lk_array, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, inodelk_cbk, out);
+
+        tmp_local = frame->local;
+
+        lock_frame = dht_lock_frame (frame);
+        if (lock_frame == NULL) {
+                gf_uuid_unparse (tmp_local->loc.gfid, gfid);
+                gf_msg ("dht", GF_LOG_ERROR, ENOMEM,
+                        DHT_MSG_LOCK_FRAME_FAILED,
+                        "memory allocation failed for lock_frame. gfid:%s"
+                        " path:%s", gfid, tmp_local->loc.path);
+                goto out;
+        }
+
+        ret = dht_local_inodelk_init (lock_frame, lk_array, lk_count,
+                                      inodelk_cbk);
+        if (ret < 0) {
+                gf_uuid_unparse (tmp_local->loc.gfid, gfid);
+                gf_msg ("dht", GF_LOG_ERROR, ENOMEM,
+                        DHT_MSG_LOCAL_LOCK_INIT_FAILED,
+                        "dht_local_lock_init failed, gfid: %s path:%s", gfid,
+                        tmp_local->loc.path);
+                goto out;
+        }
+
+        dht_set_lkowner (lk_array, lk_count, &lock_frame->root->lk_owner);
+
+        local = lock_frame->local;
+        local->lock[0].layout.my_layout.reaction = reaction;
+        local->main_frame = frame;
+
+        dht_blocking_inodelk_rec (lock_frame, 0);
+
+        return 0;
+out:
+        if (lock_frame)
+                dht_lock_stack_destroy (lock_frame, DHT_INODELK);
+
+        return -1;
+}
+
+void
+dht_unlock_namespace (call_frame_t *frame, dht_dir_transaction_t *lock)
+{
+        GF_VALIDATE_OR_GOTO ("dht-locks", frame, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, lock, out);
+
+        dht_unlock_entrylk_wrapper (frame, &lock->ns.directory_ns);
+        dht_unlock_inodelk_wrapper (frame, &lock->ns.parent_layout);
+
+out:
+        return;
+}
+
+static int32_t
+dht_protect_namespace_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                           int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t *local    = NULL;
+
+        local = frame->local;
+        if (op_ret != 0)
+                dht_unlock_inodelk_wrapper (frame,
+                                            &local->current->ns.parent_layout);
+
+        local->current->ns.ns_cbk (frame, cookie, this, op_ret, op_errno,
+                                   xdata);
+        return 0;
+}
+
+int32_t
+dht_blocking_entrylk_after_inodelk (call_frame_t *frame, void *cookie,
+                                    xlator_t *this, int32_t op_ret,
+                                    int32_t op_errno, dict_t *xdata)
+{
+        dht_local_t           *local                   = NULL;
+        int                    ret                     = -1;
+        loc_t                 *loc                     = NULL;
+        dht_lock_t           **lk_array                = NULL;
+        char                   pgfid[GF_UUID_BUF_SIZE] = {0};
+        int                    count                   = 0;
+        dht_elock_wrap_t      *entrylk                 = NULL;
+
+        local = frame->local;
+        entrylk = &local->current->ns.directory_ns;
+
+        if (op_ret < 0) {
+                local->op_ret = -1;
+                local->op_errno = op_errno;
+                goto err;
+        }
+
+        loc = &entrylk->locks[0]->loc;
+        gf_uuid_unparse (loc->gfid, pgfid);
+
+        local->op_ret = 0;
+        lk_array = entrylk->locks;
+        count = entrylk->lk_count;
+
+        ret = dht_blocking_entrylk (frame, lk_array, count, FAIL_ON_ANY_ERROR,
+                                    dht_protect_namespace_cbk);
+
+        if (ret < 0) {
+                local->op_ret = -1;
+                local->op_errno = EIO;
+                gf_msg (this->name, GF_LOG_WARNING, local->op_errno,
+                        DHT_MSG_ENTRYLK_ERROR,
+                        "%s (%s/%s): "
+                        "dht_blocking_entrylk failed after taking inodelk",
+                        gf_fop_list[local->fop], pgfid,
+                        entrylk->locks[0]->basename);
+                goto err;
+        }
+
+        return 0;
+
+err:
+        if (lk_array != NULL) {
+                dht_lock_array_free (lk_array, count);
+                GF_FREE (lk_array);
+                entrylk->locks = NULL;
+                entrylk->lk_count = 0;
+        }
+
+        /* Unlock inodelk. No harm calling unlock twice */
+        dht_unlock_inodelk_wrapper (frame, &local->current->ns.parent_layout);
+        /* Call ns_cbk. It will take care of unwinding */
+        local->current->ns.ns_cbk (frame, NULL, this, local->op_ret,
+                                   local->op_errno, NULL);
+        return 0;
+}
+
+/* Given the loc and the subvol, this routine takes the inodelk on
+ * the parent inode and entrylk on (parent, loc->name). This routine
+ * is specific as it supports only one subvol on which it takes inodelk
+ * and then entrylk serially.
+ */
+int
+dht_protect_namespace (call_frame_t *frame, loc_t *loc,
+                       xlator_t *subvol,
+                       struct dht_namespace *ns,
+                       fop_entrylk_cbk_t ns_cbk)
+{
+        dht_ilock_wrap_t  *inodelk                 = NULL;
+        dht_elock_wrap_t  *entrylk                 = NULL;
+        dht_lock_t       **lk_array                = NULL;
+        dht_local_t       *local                   = NULL;
+        xlator_t          *this                    = NULL;
+        loc_t              parent                  = {0,};
+        int                ret                     = -1;
+        char               pgfid[GF_UUID_BUF_SIZE] = {0};
+        int32_t            op_errno                = 0;
+        int                count                   = 1;
+
+        GF_VALIDATE_OR_GOTO ("dht-locks", frame, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, loc, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, loc->parent, out);
+        GF_VALIDATE_OR_GOTO (frame->this->name, subvol, out);
+
+        local = frame->local;
+        this = frame->this;
+
+        inodelk = &ns->parent_layout;
+        entrylk = &ns->directory_ns;
+
+        /* Initialize entrylk_cbk and parent loc */
+        ns->ns_cbk = ns_cbk;
+
+        ret = dht_build_parent_loc (this, &parent, loc, &op_errno);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_ERROR, op_errno,
+                        DHT_MSG_LOC_FAILED, "gfid:%s (name:%s) (path: %s): "
+                        "parent loc build failed", loc->gfid, loc->name,
+                         loc->path);
+                goto out;
+        }
+        gf_uuid_unparse (parent.gfid, pgfid);
+
+        /* Alloc inodelk */
+        inodelk->locks = GF_CALLOC (count, sizeof (*lk_array),
+                                    gf_common_mt_pointer);
+        if (inodelk->locks == NULL) {
+                local->op_errno = ENOMEM;
+                gf_msg (this->name, GF_LOG_WARNING, local->op_errno,
+                        DHT_MSG_NO_MEMORY,
+                        "%s (%s/%s) (path: %s): "
+                        "calloc failure",
+                        gf_fop_list[local->fop], pgfid, loc->name, loc->path);
+                goto out;
+        }
+
+        inodelk->locks[0] = dht_lock_new (this, subvol, &parent, F_RDLCK,
+                                        DHT_LAYOUT_HEAL_DOMAIN, NULL);
+        if (inodelk->locks[0] == NULL) {
+                local->op_errno = ENOMEM;
+                gf_msg (this->name, GF_LOG_WARNING, local->op_errno,
+                        DHT_MSG_NO_MEMORY,
+                        "%s (%s/%s) (path: %s): "
+                        "inodelk: lock allocation failed",
+                        gf_fop_list[local->fop], pgfid, loc->name, loc->path);
+                goto err;
+        }
+        inodelk->lk_count = count;
+
+        /* Allock entrylk */
+        entrylk->locks = GF_CALLOC (count, sizeof (*lk_array),
+                                    gf_common_mt_pointer);
+        if (entrylk->locks == NULL) {
+                local->op_errno = ENOMEM;
+                gf_msg (this->name, GF_LOG_WARNING, local->op_errno,
+                        DHT_MSG_NO_MEMORY,
+                        "%s (%s/%s) (path: %s): "
+                        "entrylk: calloc failure",
+                        gf_fop_list[local->fop], pgfid, loc->name, loc->path);
+
+                goto err;
+        }
+
+        entrylk->locks[0] = dht_lock_new (this, subvol, &parent, F_WRLCK,
+                                          DHT_ENTRY_SYNC_DOMAIN, loc->name);
+        if (entrylk->locks[0] == NULL) {
+                local->op_errno = ENOMEM;
+                gf_msg (this->name, GF_LOG_WARNING, local->op_errno,
+                        DHT_MSG_NO_MEMORY,
+                        "%s (%s/%s) (path: %s): "
+                        "entrylk: lock allocation failed",
+                        gf_fop_list[local->fop], pgfid, loc->name, loc->path);
+
+                goto err;
+        }
+        entrylk->lk_count = count;
+
+        /* Take read inodelk on parent. If it is successful, take write entrylk
+         * on name in cbk.
+         */
+        lk_array = inodelk->locks;
+        ret = dht_blocking_inodelk (frame, lk_array, count, FAIL_ON_ANY_ERROR,
+                                    dht_blocking_entrylk_after_inodelk);
+        if (ret < 0) {
+                local->op_errno = EIO;
+                gf_msg (this->name, GF_LOG_WARNING, local->op_errno,
+                        DHT_MSG_INODELK_ERROR,
+                        "%s (%s/%s) (path: %s): "
+                        "dht_blocking_inodelk failed",
+                        gf_fop_list[local->fop], pgfid, loc->name, loc->path);
+                goto err;
+        }
+
+        loc_wipe (&parent);
+
+        return 0;
+err:
+        if (entrylk->locks != NULL) {
+                dht_lock_array_free (entrylk->locks, count);
+                GF_FREE (entrylk->locks);
+                entrylk->locks = NULL;
+                entrylk->lk_count = 0;
+        }
+
+        if (inodelk->locks != NULL) {
+                dht_lock_array_free (inodelk->locks, count);
+                GF_FREE (inodelk->locks);
+                inodelk->locks = NULL;
+                inodelk->lk_count = 0;
+        }
+
+        loc_wipe (&parent);
+out:
+        return -1;
+}