Leases: Add a server side xlator to handle lease requests

Before this patch, there was an effort to implement leases
in upcall xlator, these patches by Soumya and me can be
found @ http://review.gluster.org/#/c/10084/

Change-Id: I926728c7ec690727a8971039b240655882d02059
BUG: 1319992
Signed-off-by: Poornima G <pgurusid@redhat.com>
Reviewed-on: http://review.gluster.org/11643
Smoke: Gluster Build System <jenkins@build.gluster.com>
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
CentOS-regression: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Raghavendra Talur <rtalur@redhat.com>
Reviewed-by: Rajesh Joseph <rjoseph@redhat.com>
Reviewed-by: Pranith Kumar Karampuri <pkarampu@redhat.com>

7 files changed, 2882 insertions, 0 deletions

diff --git a/xlators/features/leases/Makefile.am b/xlators/features/leases/Makefile.am
new file mode 100644
index 00000000000..a985f42a877
--- /dev/null
+++ b/xlators/features/leases/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES =
diff --git a/xlators/features/leases/src/Makefile.am b/xlators/features/leases/src/Makefile.am
new file mode 100644
index 00000000000..343f5c82425
--- /dev/null
+++ b/xlators/features/leases/src/Makefile.am
@@ -0,0 +1,17 @@
+xlator_LTLIBRARIES = leases.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+leases_la_LDFLAGS = -module $(GF_XLATOR_DEFAULT_LDFLAGS)
+
+leases_la_SOURCES = leases.c leases-internal.c
+
+leases_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
+
+noinst_HEADERS = leases.h leases-mem-types.h leases-messages.h
+
+AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
+               -I$(CONTRIBDIR)/timer-wheel
+
+AM_CFLAGS = -Wall $(GF_CFLAGS)
+
+CLEANFILES =
diff --git a/xlators/features/leases/src/leases-internal.c b/xlators/features/leases/src/leases-internal.c
new file mode 100644
index 00000000000..8ed0abcae82
--- /dev/null
+++ b/xlators/features/leases/src/leases-internal.c
@@ -0,0 +1,1333 @@
+/*
+   Copyright (c) 2015-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.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "leases.h"
+
+
+/* Mutex locks used in this xlator and their order of acquisition:
+ * Check lease conflict:
+ *         lease_ctx lock
+ *                 add_timer => internal timer locks
+ *         lease_ctx unlock
+ *
+ * Add/remove lease:
+ *         lease_ctx lock
+ *                 add_timer => internal timer locks
+ *                 OR
+ *                 priv lock => Adding/removing to/from the cleanup client list
+ *                 priv unlock
+ *         lease_ctx unlock
+ *
+ * Timer thread:
+ *         Timer internal lock
+ *                 priv lock => By timer handler
+ *                 priv unlock
+ *         Timer internal unlock
+ *
+ * Expired recall cleanup thread:
+ *         priv lock
+ *                 priv condwait
+ *         priv unlock
+ *         lease_ctx lock
+ *                 priv lock
+ *                 priv unlock
+ *         lease_ctx unlock
+ */
+
+/*
+ * Check if lease_lk is enabled
+ * Return Value:
+ * _gf_true  - lease lock option enabled
+ * _gf_false - lease lock option disabled
+ */
+gf_boolean_t
+is_leases_enabled (xlator_t *this)
+{
+        leases_private_t    *priv       = NULL;
+        gf_boolean_t         is_enabled = _gf_false;
+
+        GF_VALIDATE_OR_GOTO ("leases", this, out);
+
+        if (this->private) {
+                priv = (leases_private_t *)this->private;
+                is_enabled = priv->leases_enabled;
+        }
+out:
+        return is_enabled;
+}
+
+
+/*
+ * Get the recall_leaselk_timeout
+ * Return Value:
+ * timeout value(in seconds) set as an option to this xlator.
+ * -1 error case
+ */
+int32_t
+get_recall_lease_timeout (xlator_t *this)
+{
+        leases_private_t    *priv       = NULL;
+        int32_t              timeout    = -1;
+
+        GF_VALIDATE_OR_GOTO ("leases", this, out);
+
+        if (this->private) {
+                priv = (leases_private_t *)this->private;
+                timeout = priv->recall_lease_timeout;
+        }
+out:
+        return timeout;
+}
+
+
+static void
+__dump_leases_info (xlator_t *this, lease_inode_ctx_t *lease_ctx)
+{
+        lease_id_entry_t   *lease_entry     = NULL;
+        lease_id_entry_t   *tmp             = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", this, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+
+        gf_msg_debug (this->name, 0, "Lease held on this inode, lease_type: %d,"
+                      " lease_cnt:%"PRIu64", RD lease:%d, RW lease:%d, "
+                      "openfd cnt:%"PRIu64, lease_ctx->lease_type,
+                      lease_ctx->lease_cnt,
+                      lease_ctx->lease_type_cnt[GF_RD_LEASE],
+                      lease_ctx->lease_type_cnt[GF_RW_LEASE],
+                      lease_ctx->openfd_cnt);
+
+        list_for_each_entry_safe (lease_entry, tmp,
+                                  &lease_ctx->lease_id_list,
+                                  lease_id_list) {
+                gf_msg_debug (this->name, 0, "Leases held by client: %s, lease "
+                              "ID:%s, RD lease:%d, RW lease:%d, lease_type: %d, "
+                              "lease_cnt:%"PRIu64, lease_entry->client_uid,
+                              lease_entry->lease_id,
+                              lease_entry->lease_type_cnt[GF_RD_LEASE],
+                              lease_entry->lease_type_cnt[GF_RW_LEASE],
+                              lease_entry->lease_type, lease_entry->lease_cnt);
+        }
+out:
+        return;
+}
+
+
+static int
+__lease_ctx_set (inode_t *inode, xlator_t *this)
+{
+        lease_inode_ctx_t   *inode_ctx    = NULL;
+        int                  ret          = -1;
+        uint64_t             ctx          = 0;
+
+        GF_VALIDATE_OR_GOTO ("leases", inode, out);
+        GF_VALIDATE_OR_GOTO ("leases", this, out);
+
+        ret = __inode_ctx_get (inode, this, &ctx);
+        if (!ret) {
+                gf_msg (this->name, GF_LOG_ERROR, 0, LEASE_MSG_INVAL_INODE_CTX,
+                        "inode_ctx_get failed");
+                goto out;
+        }
+
+        inode_ctx = GF_CALLOC (1, sizeof (*inode_ctx),
+                               gf_leases_mt_lease_inode_ctx_t);
+        GF_CHECK_ALLOC (inode_ctx, ret, out);
+
+        pthread_mutex_init (&inode_ctx->lock, NULL);
+        INIT_LIST_HEAD (&inode_ctx->lease_id_list);
+        INIT_LIST_HEAD (&inode_ctx->blocked_list);
+
+        inode_ctx->lease_cnt = 0;
+
+        ret = __inode_ctx_set (inode, this, (uint64_t *) inode_ctx);
+        if (ret) {
+                GF_FREE (inode_ctx);
+                gf_msg (this->name, GF_LOG_INFO, 0, LEASE_MSG_INVAL_INODE_CTX,
+                        "failed to set inode ctx (%p)", inode);
+        }
+out:
+        return ret;
+}
+
+
+static lease_inode_ctx_t *
+__lease_ctx_get (inode_t *inode, xlator_t *this)
+{
+        lease_inode_ctx_t *inode_ctx    = NULL;
+        uint64_t           ctx          = 0;
+        int                ret          = 0;
+
+        GF_VALIDATE_OR_GOTO ("leases", inode, out);
+        GF_VALIDATE_OR_GOTO ("leases", this, out);
+
+        ret = __inode_ctx_get (inode, this, &ctx);
+        if (ret < 0) {
+                ret = __lease_ctx_set (inode, this);
+                if (ret < 0)
+                        goto out;
+
+                ret = __inode_ctx_get (inode, this, &ctx);
+                if (ret < 0) {
+                        gf_msg (this->name, GF_LOG_WARNING, 0, LEASE_MSG_INVAL_INODE_CTX,
+                                "failed to get inode ctx (%p)", inode);
+                        goto out;
+                }
+        }
+
+        inode_ctx = (lease_inode_ctx_t *)(long) ctx;
+out:
+        return inode_ctx;
+}
+
+
+lease_inode_ctx_t *
+lease_ctx_get (inode_t *inode, xlator_t *this)
+{
+        lease_inode_ctx_t *inode_ctx = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", inode, out);
+        GF_VALIDATE_OR_GOTO ("leases", this, out);
+
+        LOCK (&inode->lock);
+        {
+                inode_ctx = __lease_ctx_get (inode, this);
+        }
+        UNLOCK (&inode->lock);
+out:
+        return inode_ctx;
+}
+
+
+static lease_id_entry_t *
+new_lease_id_entry (call_frame_t *frame, const char *lease_id)
+{
+        lease_id_entry_t *lease_entry = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", frame, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_id, out);
+
+        lease_entry = GF_CALLOC (1, sizeof (*lease_entry),
+                                  gf_leases_mt_lease_id_entry_t);
+        if (!lease_entry) {
+                gf_msg (frame->this->name, GF_LOG_ERROR, ENOMEM, LEASE_MSG_NO_MEM,
+                        "Memory allocation for lease_entry failed");
+                return NULL;
+        }
+
+        INIT_LIST_HEAD (&lease_entry->lease_id_list);
+        lease_entry->lease_type = NONE;
+        lease_entry->lease_cnt = 0;
+        lease_entry->recall_time =
+                        get_recall_lease_timeout (frame->this);
+        lease_entry->client_uid = gf_strdup (frame->root->client->client_uid);
+        if (!lease_entry->client_uid) {
+                gf_msg (frame->this->name, GF_LOG_ERROR, ENOMEM, LEASE_MSG_NO_MEM,
+                        "Memory allocation for client_uid failed");
+                GF_FREE (lease_entry);
+                lease_entry = NULL;
+                goto out;
+        }
+
+        memcpy (lease_entry->lease_id, lease_id, LEASE_ID_SIZE);
+out:
+        return lease_entry;
+}
+
+
+static void
+__destroy_lease_id_entry (lease_id_entry_t *lease_entry)
+{
+        GF_VALIDATE_OR_GOTO ("leases", lease_entry, out);
+
+        list_del_init (&lease_entry->lease_id_list);
+        GF_FREE (lease_entry->client_uid);
+        GF_FREE (lease_entry);
+out:
+        return;
+}
+
+
+static inline gf_boolean_t
+__is_same_lease_id (const char *k1, const char *k2)
+{
+        if (memcmp(k1, k2, LEASE_ID_SIZE) == 0)
+                return _gf_true;
+
+        return _gf_false;
+}
+
+
+/* Checks if there are any leases, other than the leases taken
+ * by the given lease_id
+ */
+static gf_boolean_t
+__another_lease_found (lease_inode_ctx_t *lease_ctx, const char *lease_id)
+{
+        lease_id_entry_t   *lease_entry     = NULL;
+        lease_id_entry_t   *tmp             = NULL;
+        gf_boolean_t        found_lease     = _gf_false;
+
+        GF_VALIDATE_OR_GOTO ("leases", lease_id, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+
+        list_for_each_entry_safe (lease_entry, tmp,
+                                  &lease_ctx->lease_id_list,
+                                  lease_id_list) {
+
+                if (!__is_same_lease_id (lease_id, lease_entry->lease_id)) {
+                        if (lease_entry->lease_cnt > 0) {
+                                found_lease = _gf_true;
+                                break;
+                        }
+                }
+        }
+out:
+        return found_lease;
+}
+
+
+/* Returns the lease_id_entry for a given lease_id and a given inode.
+ * Return values:
+ * NULL - If no client entry found
+ * lease_id_entry_t* - a pointer to the client entry if found
+ */
+static lease_id_entry_t *
+__get_lease_id_entry (lease_inode_ctx_t *lease_ctx, const char *lease_id)
+{
+        lease_id_entry_t   *lease_entry    = NULL;
+        lease_id_entry_t   *tmp            = NULL;
+        lease_id_entry_t   *found          = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", lease_id, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+
+        list_for_each_entry_safe (lease_entry, tmp,
+                                  &lease_ctx->lease_id_list,
+                                  lease_id_list) {
+
+                if (__is_same_lease_id (lease_id, lease_entry->lease_id)) {
+                        found = lease_entry;
+                        gf_msg_debug ("leases", 0, "lease ID entry found "
+                                      "Client UID:%s, lease id:%s",
+                                      lease_entry->client_uid,
+                                      leaseid_utoa (lease_entry->lease_id));
+                        break;
+                }
+        }
+out:
+        return found;
+}
+
+
+/* Returns the lease_id_entry for a given lease_id and a given inode,
+ * if none found creates one.
+ * Return values:
+ * lease_id_entry_t* - a pointer to the client entry
+ */
+static lease_id_entry_t *
+__get_or_new_lease_entry (call_frame_t *frame, const char *lease_id,
+                           lease_inode_ctx_t *lease_ctx)
+{
+        lease_id_entry_t *lease_entry    = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", frame, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_id, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+
+        lease_entry = __get_lease_id_entry (lease_ctx, lease_id);
+        if (!lease_entry) { /* create one */
+                lease_entry = new_lease_id_entry (frame, lease_id);
+                if (!lease_entry)
+                        goto out;
+
+                list_add_tail (&lease_entry->lease_id_list,
+                               &lease_ctx->lease_id_list);
+
+                gf_msg_debug (frame->this->name, 0, "lease ID entry added,"
+                              " Client UID:%s, lease id:%s",
+                              lease_entry->client_uid,
+                              leaseid_utoa (lease_entry->lease_id));
+        }
+out:
+        return lease_entry;
+}
+
+
+static lease_inode_t *
+new_lease_inode (inode_t *inode)
+{
+        lease_inode_t *l_inode = NULL;
+
+        l_inode = GF_CALLOC (1, sizeof (*l_inode), gf_leases_mt_lease_inode_t);
+        if (!l_inode)
+                goto out;
+
+        INIT_LIST_HEAD (&l_inode->list);
+        l_inode->inode = inode_ref (inode);
+out:
+        return l_inode;
+}
+
+
+static void
+__destroy_lease_inode (lease_inode_t *l_inode)
+{
+        list_del_init (&l_inode->list);
+        inode_unref (l_inode->inode);
+        GF_FREE (l_inode);
+}
+
+
+static lease_client_t *
+new_lease_client (const char *client_uid)
+{
+        lease_client_t  *clnt = NULL;
+
+        clnt = GF_CALLOC (1, sizeof (*clnt), gf_leases_mt_lease_client_t);
+        if (!clnt)
+                goto out;
+
+        INIT_LIST_HEAD (&clnt->client_list);
+        INIT_LIST_HEAD (&clnt->inode_list);
+        clnt->client_uid = gf_strdup (client_uid);
+out:
+        return clnt;
+}
+
+
+static void
+__destroy_lease_client (lease_client_t *clnt)
+{
+         list_del_init (&clnt->inode_list);
+         list_del_init (&clnt->client_list);
+         GF_FREE (clnt);
+
+         return;
+}
+
+
+static lease_client_t *
+__get_lease_client (xlator_t *this, leases_private_t *priv,
+                    const char *client_uid)
+{
+        lease_client_t  *clnt  = NULL;
+        lease_client_t  *tmp   = NULL;
+        lease_client_t  *found = NULL;
+
+        list_for_each_entry_safe (clnt, tmp, &priv->client_list, client_list) {
+                if ((strcmp (clnt->client_uid, client_uid) == 0)) {
+                        found = clnt;
+                        gf_msg_debug (this->name, 0, "Client:%s already found "
+                                      "in the cleanup list", client_uid);
+                        break;
+                }
+        }
+        return found;
+}
+
+
+static lease_client_t *
+__get_or_new_lease_client (xlator_t *this, leases_private_t *priv,
+                           const char *client_uid)
+{
+        lease_client_t  *found = NULL;
+
+        found = __get_lease_client (this, priv, client_uid);
+        if (!found) {
+                found = new_lease_client (client_uid);
+                if (!found)
+                        goto out;
+                list_add_tail (&found->client_list, &priv->client_list);
+                gf_msg_debug (this->name, 0, "Adding a new client:%s entry "
+                              "to the cleanup list", client_uid);
+        }
+out:
+        return found;
+}
+
+
+static int
+add_inode_to_client_list (xlator_t *this, inode_t *inode, const char *client_uid)
+{
+        int               ret         = 0;
+        leases_private_t *priv        = NULL;
+        lease_client_t   *clnt        = NULL;
+        lease_inode_t    *lease_inode = NULL;
+
+        priv = this->private;
+        pthread_mutex_lock (&priv->mutex);
+        {
+                clnt = __get_or_new_lease_client (this, priv, client_uid);
+                GF_CHECK_ALLOC (clnt, ret, out);
+
+                lease_inode = new_lease_inode (inode);
+                GF_CHECK_ALLOC (lease_inode, ret, out);
+
+                list_add_tail (&clnt->inode_list, &lease_inode->list);
+                gf_msg_debug (this->name, 0,
+                              "Added a new inode:%p to the client(%s) "
+                              "cleanup list, gfid(%s)", inode, client_uid,
+                              uuid_utoa (inode->gfid));
+        }
+        pthread_mutex_unlock (&priv->mutex);
+out:
+        return ret;
+}
+
+
+/* Add lease entry to the corresponding client entry.
+ * Return values:
+ * 0 Success
+ * -1 Failure
+ */
+static int
+__add_lease (call_frame_t *frame, inode_t *inode, lease_inode_ctx_t *lease_ctx,
+             const char *client_uid, struct gf_lease *lease)
+{
+        lease_id_entry_t  *lease_entry  = NULL;
+        int                ret          = -1;
+
+        GF_VALIDATE_OR_GOTO ("leases", frame, out);
+        GF_VALIDATE_OR_GOTO ("leases", client_uid, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+        GF_VALIDATE_OR_GOTO ("leases", inode, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease, out);
+
+        gf_msg_trace (frame->this->name, 0,
+                      "Granting lease lock to client %s with lease id %s"
+                      " on gfid(%s)", client_uid, leaseid_utoa (lease->lease_id),
+                      uuid_utoa (inode->gfid));
+
+        lease_entry = __get_or_new_lease_entry (frame, lease->lease_id, lease_ctx);
+        if (!lease_entry) {
+                errno = ENOMEM;
+                goto out;
+        }
+
+        lease_entry->lease_type_cnt[lease->lease_type]++;
+        lease_entry->lease_cnt++;
+        lease_entry->lease_type |= lease->lease_type;
+        /* If this is the first lease taken by the client on the file, then
+         * add this inode/file to the client disconnect cleanup list
+         */
+        if (lease_entry->lease_cnt == 1) {
+                add_inode_to_client_list (frame->this, inode, client_uid);
+        }
+
+        lease_ctx->lease_cnt++;
+        lease_ctx->lease_type_cnt[lease->lease_type]++;
+        lease_ctx->lease_type |= lease->lease_type;
+
+        /* Take a ref for the first lock taken on this inode. Corresponding
+         * unref when all the leases are unlocked or during DISCONNECT
+         * Ref is required because the inode on which lease is acquired should
+         * not be deleted when lru cleanup kicks in*/
+        if (lease_ctx->lease_cnt == 1) {
+                lease_ctx->inode = inode_ref (inode);
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+
+static gf_boolean_t
+__is_clnt_lease_none (const char *client_uid, lease_inode_ctx_t *lease_ctx)
+{
+        gf_boolean_t      lease_none  = _gf_true;
+        lease_id_entry_t *lease_entry = NULL;
+        lease_id_entry_t *tmp         = NULL;
+
+        list_for_each_entry_safe (lease_entry, tmp,
+                                  &lease_ctx->lease_id_list,
+                                  lease_id_list) {
+                if ((strcmp (client_uid, lease_entry->client_uid) == 0)
+                    && (lease_entry->lease_cnt != 0)) {
+                        lease_none = _gf_false;
+                        break;
+                }
+        }
+
+        return lease_none;
+}
+
+static int
+__remove_inode_from_clnt_list (xlator_t *this, lease_client_t *clnt,
+                               inode_t *inode)
+{
+        int               ret     = -1;
+        lease_inode_t    *l_inode = NULL;
+        lease_inode_t    *tmp1    = NULL;
+
+        list_for_each_entry_safe (l_inode, tmp1,
+                                  &clnt->inode_list,
+                                  list) {
+                if (l_inode->inode == inode) {
+                        __destroy_lease_inode (l_inode);
+                        gf_msg_debug (this->name, 0,
+                                      "Removed the inode from the client cleanup list");
+                        ret = 0;
+                }
+        }
+        /* TODO: Remove the client entry from the cleanup list */
+
+        return ret;
+}
+
+
+static int
+remove_from_clnt_list (xlator_t *this, const char *client_uid, inode_t *inode)
+{
+        leases_private_t *priv    = NULL;
+        int               ret     = -1;
+        lease_client_t   *clnt    = NULL;
+
+        priv = this->private;
+        if (!priv)
+                goto out;
+
+        pthread_mutex_lock (&priv->mutex);
+        {
+                clnt = __get_lease_client (this, priv, client_uid);
+                if (!clnt) {
+                        gf_msg (this->name, GF_LOG_ERROR, 0, LEASE_MSG_CLNT_NOTFOUND,
+                                "There is no client entry found in the cleanup list");
+                        pthread_mutex_unlock (&priv->mutex);
+                        goto out;
+                }
+                ret = __remove_inode_from_clnt_list (this, clnt, inode);
+                if (ret) {
+                        gf_msg (this->name, GF_LOG_ERROR, 0, LEASE_MSG_INODE_NOTFOUND,
+                                "There is no inode entry found in the cleanup list");
+                }
+        }
+        pthread_mutex_unlock (&priv->mutex);
+out:
+        return ret;
+}
+
+
+/* Remove lease entry in the corresponding client entry.
+ */
+static int
+__remove_lease (xlator_t *this, inode_t *inode, lease_inode_ctx_t *lease_ctx,
+                const char *client_uid, struct gf_lease *lease)
+{
+        lease_id_entry_t    *lease_entry     = NULL;
+        int                  ret             = 0;
+        int32_t              lease_type      = 0;
+        leases_private_t    *priv            = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease, out);
+
+        priv = this->private;
+
+        gf_msg_trace (this->name, 0, "Removing lease entry for client: %s, "
+                      "lease type:%d, lease id:%s", client_uid, lease->lease_type,
+                      leaseid_utoa (lease->lease_id));
+
+        lease_entry = __get_lease_id_entry (lease_ctx, lease->lease_id);
+        if (!lease_entry) {
+                gf_msg (this->name, GF_LOG_INFO, 0, LEASE_MSG_INVAL_UNLK_LEASE,
+                        "Got unlock lease request from client:%s, but has no "
+                        "corresponding lock", client_uid);
+                ret = -EINVAL;
+                errno = EINVAL;
+                goto out;
+        }
+
+        lease_type = lease->lease_type;
+        lease_entry->lease_type_cnt[lease_type]--;
+        lease_entry->lease_cnt--;
+
+        lease_ctx->lease_type_cnt[lease_type]--;
+        lease_ctx->lease_cnt--;
+
+        if (lease_entry->lease_type_cnt[lease_type] == 0)
+                lease_entry->lease_type = lease_entry->lease_type & (~lease_type);
+
+        if (lease_ctx->lease_type_cnt[lease_type] == 0)
+                lease_ctx->lease_type = lease_ctx->lease_type & (~lease_type);
+
+        if (lease_entry->lease_cnt == 0) {
+                if (__is_clnt_lease_none (client_uid, lease_ctx)) {
+                        gf_msg_debug (this->name, 0, "Client(%s) has no leases"
+                                      " on gfid (%s), hence removing the inode"
+                                      " from the client cleanup list",
+                                      client_uid, uuid_utoa (inode->gfid));
+                        remove_from_clnt_list (this, client_uid, lease_ctx->inode);
+                }
+                __destroy_lease_id_entry (lease_entry);
+        }
+
+        if (lease_ctx->lease_cnt == 0 && lease_ctx->timer) {
+                ret = gf_tw_del_timer (priv->timer_wheel, lease_ctx->timer);
+                lease_ctx->recall_in_progress = _gf_false;
+        }
+out:
+        return ret;
+}
+
+
+static gf_boolean_t
+__is_lease_grantable (xlator_t *this, lease_inode_ctx_t *lease_ctx,
+                      struct gf_lease *lease, inode_t *inode)
+{
+        uint32_t        fd_count     = 0;
+        int32_t         flags        = 0;
+        fd_t           *iter_fd      = NULL;
+        gf_boolean_t    grant        = _gf_false;
+        int             ret          = 0;
+        lease_fd_ctx_t *fd_ctx       = NULL;
+        uint64_t        ctx          = 0;
+
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease, out);
+        GF_VALIDATE_OR_GOTO ("leases", inode, out);
+
+        if (lease_ctx->recall_in_progress) {
+                gf_msg_debug (this->name, 0, "Recall in progress, hence "
+                              "failing the lease request");
+                grant = _gf_false;
+                goto out;
+        }
+
+        LOCK (&inode->lock);
+        {
+                list_for_each_entry (iter_fd, &inode->fd_list, inode_list) {
+                        ret = fd_ctx_get (iter_fd, this, &ctx);
+                        if (ret < 0) {
+                                grant = _gf_false;
+                                UNLOCK (&inode->lock);
+                                gf_msg (this->name, GF_LOG_ERROR, 0,
+                                        LEASE_MSG_INVAL_FD_CTX,
+                                        "Unable to get fd ctx");
+                                goto out;
+                        }
+                        fd_ctx = (lease_fd_ctx_t *)(long) ctx;
+
+                        /* Check for open fd conflict, note that open fds from
+                         * the same lease id is not checked for conflict, as it is
+                         * lease id based lease.
+                         */
+                        if (!__is_same_lease_id (fd_ctx->lease_id, lease->lease_id)) {
+                                fd_count++;
+                                flags |= iter_fd->flags;
+                        }
+                }
+        }
+        UNLOCK (&inode->lock);
+
+        gf_msg_debug (this->name, 0, "open fd count:%d flags:%d",
+                      fd_count, flags);
+
+        __dump_leases_info (this, lease_ctx);
+
+        switch (lease->lease_type) {
+        case GF_RD_LEASE:
+                /* check open fd conflict */
+                if ((fd_count > 0) && ((flags & O_WRONLY) || (flags & O_RDWR))) {
+                        grant = _gf_false;
+                        break;
+                }
+
+                /* check for conflict with existing leases */
+                if (lease_ctx->lease_type == NONE || lease_ctx->lease_type == GF_RD_LEASE)
+                        grant = _gf_true;
+                else
+                        grant = _gf_false;
+                break;
+
+        case GF_RW_LEASE:
+                /* check open fd conflict; conflict if there are any fds open
+                 * other than the client on which the lease is requested. */
+                if (fd_count > 0) {
+                        grant = _gf_false;
+                        break;
+                }
+
+                /* check existing lease conflict */
+                if (lease_ctx->lease_type == NONE ||
+                    !(__another_lease_found (lease_ctx, lease->lease_id)))
+                        grant = _gf_true;
+                else
+                        grant = _gf_false;
+                break;
+
+        default:
+                gf_msg (this->name, GF_LOG_ERROR, EINVAL, LEASE_MSG_INVAL_LEASE_TYPE,
+                        "Invalid lease type specified");
+                break;
+        }
+out:
+        return grant;
+}
+
+
+static void
+do_blocked_fops (xlator_t *this, lease_inode_ctx_t *lease_ctx)
+{
+        struct list_head   wind_list;
+        fop_stub_t        *blk_fop  = NULL;
+        fop_stub_t        *tmp      = NULL;
+
+        INIT_LIST_HEAD (&wind_list);
+
+        pthread_mutex_lock (&lease_ctx->lock);
+        {
+                list_for_each_entry_safe (blk_fop, tmp,
+                                          &lease_ctx->blocked_list, list) {
+                        list_del_init (&blk_fop->list);
+                        list_add_tail (&blk_fop->list, &wind_list);
+                }
+        }
+        pthread_mutex_unlock (&lease_ctx->lock);
+
+        gf_msg_trace (this->name, 0, "Executing the blocked stubs on gfid(%s)",
+                      uuid_utoa (lease_ctx->inode->gfid));
+
+        list_for_each_entry_safe (blk_fop, tmp, &wind_list, list) {
+                list_del_init (&blk_fop->list);
+                gf_msg_trace (this->name, 0, "Executing fop:%d", blk_fop->stub->fop);
+                call_resume (blk_fop->stub);
+                GF_FREE (blk_fop);
+        }
+
+        pthread_mutex_lock (&lease_ctx->lock);
+        {
+                lease_ctx->lease_type = NONE;
+                inode_unref (lease_ctx->inode);
+                lease_ctx->inode = NULL;
+        }
+        pthread_mutex_unlock (&lease_ctx->lock);
+
+        return;
+}
+
+
+void
+recall_lease_timer_handler (struct gf_tw_timer_list *timer,
+                            void *data, unsigned long calltime)
+{
+        inode_t          *inode       = NULL;
+        lease_inode_t    *lease_inode = NULL;
+        leases_private_t *priv        = NULL;
+
+        priv = THIS->private;
+        inode = (inode_t *)data;
+        pthread_mutex_lock (&priv->mutex);
+        {
+                lease_inode = new_lease_inode (inode);
+                if (!lease_inode) {
+                        errno = ENOMEM;
+                        goto out;
+                }
+                list_add_tail (&lease_inode->list, &priv->recall_list);
+                pthread_cond_broadcast (&priv->cond);
+        }
+out:
+        pthread_mutex_unlock (&priv->mutex);
+
+        GF_FREE (timer);
+}
+
+
+static void
+__recall_lease (xlator_t *this, lease_inode_ctx_t *lease_ctx)
+{
+        lease_id_entry_t                  *lease_entry   = NULL;
+        lease_id_entry_t                  *tmp           = NULL;
+        struct gf_upcall                   up_req        = {0,};
+        struct gf_upcall_recall_lease      recall_req    = {0,};
+        int                                notify_ret    = -1;
+        struct gf_tw_timer_list           *timer         = NULL;
+        leases_private_t                  *priv          = NULL;
+
+        if (lease_ctx->recall_in_progress) {
+                gf_msg_debug (this->name, 0, "Lease recall is already in "
+                              "progress, hence not sending another recall");
+                goto out;
+        }
+
+        priv = this->private;
+        list_for_each_entry_safe (lease_entry, tmp,
+                                  &lease_ctx->lease_id_list,
+                                  lease_id_list) {
+                gf_uuid_copy (up_req.gfid, lease_ctx->inode->gfid);
+                up_req.client_uid = lease_entry->client_uid;
+                up_req.event_type = GF_UPCALL_RECALL_LEASE;
+                up_req.data = &recall_req;
+
+                notify_ret = this->notify (this, GF_EVENT_UPCALL, &up_req);
+                if (notify_ret < 0) {
+                        gf_msg (this->name, GF_LOG_ERROR, 0, LEASE_MSG_RECALL_FAIL,
+                                "Recall notification to client: %s failed",
+                                lease_entry->client_uid);
+                        /* Do not return from here, continue registering the timer,
+                           this is required mostly o keep replicas in sync*/
+                } else {
+                        gf_msg_debug (this->name, 0, "Recall lease (all)"
+                                      "notification sent to client %s",
+                                      lease_entry->client_uid);
+                }
+
+                lease_ctx->recall_in_progress = _gf_true;
+                lease_entry->recall_time = time (NULL);
+        }
+        timer = GF_CALLOC (1, sizeof (*timer),
+                           gf_common_mt_tw_timer_list);
+        if (!timer) {
+                goto out;
+        }
+        INIT_LIST_HEAD (&timer->entry);
+        timer->data = inode_ref (lease_ctx->inode);
+        timer->expires = get_recall_lease_timeout (this);
+        timer->function = recall_lease_timer_handler;
+        lease_ctx->timer = timer;
+        gf_tw_add_timer (priv->timer_wheel, timer);
+        gf_msg_trace (this->name, 0, "Registering timer " "%p, after "
+                      "sending recall", timer);
+out:
+        return;
+}
+
+
+/* ret = 0; STACK_UNWIND Success
+ * ret = -1; STACK_UNWIND failure
+ */
+int
+process_lease_req (call_frame_t *frame, xlator_t *this,
+                   inode_t *inode, struct gf_lease *lease)
+{
+        int                 ret             = 0;
+        char               *client_uid      = NULL;
+        lease_inode_ctx_t  *lease_ctx       = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", frame, out);
+        GF_VALIDATE_OR_GOTO ("leases", this, out);
+        GF_VALIDATE_OR_GOTO ("leases", inode, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease, out);
+
+        client_uid = frame->root->client->client_uid;
+
+        if (!is_valid_lease_id (lease->lease_id)) {
+                gf_msg (this->name, GF_LOG_ERROR, EINVAL,
+                        LEASE_MSG_INVAL_LEASE_ID, "Invalid lease id, from"
+                        "client:%s", client_uid);
+                ret = -EINVAL;
+                errno = EINVAL;
+                goto out;
+        }
+
+        lease_ctx = lease_ctx_get (inode, this);
+        if (!lease_ctx) {
+                gf_msg (this->name, GF_LOG_WARNING, ENOMEM,
+                        LEASE_MSG_NO_MEM, "Unable to create/get inode ctx, "
+                        "inode:%p", inode);
+                ret = -ENOMEM;
+                errno = ENOMEM;
+                goto out;
+        }
+
+        gf_msg_debug (this->name, 0, "Lease request from client: %s, "
+                      "lease type:%d, lease cmd:%d, lease ID:%s, gfid:%s",
+                      client_uid, lease->lease_type, lease->cmd,
+                      leaseid_utoa (lease->lease_id), uuid_utoa (inode->gfid));
+
+        pthread_mutex_lock (&lease_ctx->lock);
+        {
+                switch (lease->cmd) {
+                case GF_GET_LEASE:
+                        lease->lease_type = lease_ctx->lease_type;
+                        gf_msg_debug (this->name, 0, "Get lease, existing lease"
+                                      "type: %d", lease_ctx->lease_type);
+                        /*TODO:Should it consider lease id or client_uid?*/
+                        break;
+
+                case GF_SET_LEASE:
+                        if (__is_lease_grantable (this, lease_ctx, lease, inode)) {
+                                __add_lease (frame, inode, lease_ctx,
+                                             client_uid, lease);
+                                ret = 0;
+                        } else {
+                                gf_msg_debug (this->name, GF_LOG_DEBUG,
+                                              "Not granting the conflicting lease"
+                                              " request from %s on gfid(%s)",
+                                              client_uid, uuid_utoa (inode->gfid));
+                                __recall_lease (this, lease_ctx);
+                                ret = -1;
+                        }
+                        break;
+                case GF_UNLK_LEASE:
+                        ret = __remove_lease (this, inode, lease_ctx,
+                                              client_uid, lease);
+                        if ((ret == 0) && (lease_ctx->lease_cnt == 0)) {
+                                pthread_mutex_unlock (&lease_ctx->lock);
+                                goto unblock;
+                        }
+                        break;
+                default:
+                        ret = -EINVAL;
+                        break;
+                }
+        }
+        pthread_mutex_unlock (&lease_ctx->lock);
+
+        return ret;
+
+unblock:
+        do_blocked_fops (this, lease_ctx);
+out:
+        return ret;
+}
+
+
+/* ret = 1 conflict
+ * ret = 0 no conflict
+ */
+gf_boolean_t
+__check_lease_conflict (call_frame_t *frame, lease_inode_ctx_t *lease_ctx,
+                        const char *lease_id, gf_boolean_t is_write)
+{
+        gf_lease_types_t   lease_type      = {0,};
+        gf_boolean_t       conflicts       = _gf_false;
+        lease_id_entry_t  *lease_entry     = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", frame, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_ctx, out);
+        GF_VALIDATE_OR_GOTO ("leases", lease_id, out);
+
+        lease_type = lease_ctx->lease_type;
+
+        /* If the fop is rename or unlink conflict the lease even if its
+         * from the same client??
+         */
+        if ((frame->op == GF_FOP_RENAME) || (frame->op == GF_FOP_UNLINK)) {
+                conflicts = _gf_true;
+                goto recall;
+        }
+
+        /* TODO: If lease_id is not sent, fall back to client uid conflict check?
+         * Or set conflicts = true if lease_id is 0 when there is an existing
+         * lease */
+        switch (lease_type) {
+        case (GF_RW_LEASE | GF_RD_LEASE):
+        case GF_RW_LEASE:
+                lease_entry = __get_lease_id_entry (lease_ctx, lease_id);
+                if (lease_entry && (lease_entry->lease_type & GF_RW_LEASE))
+                        conflicts = _gf_false;
+                else
+                        conflicts = _gf_true;
+                break;
+        case GF_RD_LEASE:
+                if (is_write && __another_lease_found(lease_ctx, lease_id))
+                        conflicts = _gf_true;
+                else
+                        conflicts = _gf_false;
+                break;
+        default:
+                break;
+        }
+
+recall:
+        /* If there is a conflict found and recall is not already sent to all
+         * the clients, then send recall to each of the client holding lease.
+         */
+        if (conflicts)
+                __recall_lease (frame->this, lease_ctx);
+out:
+        return conflicts;
+}
+
+
+/* Return values:
+ * -1 : error, unwind the fop
+ * WIND_FOP: No conflict, wind the fop
+ * BLOCK_FOP: Found a conflicting lease, block the fop
+ */
+int
+check_lease_conflict (call_frame_t *frame, inode_t *inode,
+                      const char *lease_id, uint32_t fop_flags)
+{
+        lease_inode_ctx_t       *lease_ctx  = NULL;
+        gf_boolean_t       is_blocking_fop  = _gf_false;
+        gf_boolean_t       is_write_fop     = _gf_false;
+        gf_boolean_t       conflicts        = _gf_false;
+        int                ret              = -1;
+
+        lease_ctx = lease_ctx_get (inode, frame->this);
+        if (!lease_ctx) {
+                gf_msg (frame->this->name, GF_LOG_WARNING, ENOMEM,
+                        LEASE_MSG_NO_MEM,
+                        "Unable to create/get inode ctx");
+                ret = -1;
+                errno = ENOMEM;
+                goto out;
+        }
+
+        is_blocking_fop = fop_flags & BLOCKING_FOP;
+        is_write_fop = fop_flags & DATA_MODIFY_FOP;
+
+        pthread_mutex_lock (&lease_ctx->lock);
+        {
+                if (lease_ctx->lease_type == NONE) {
+                        gf_msg_debug (frame->this->name, 0,
+                                        "No leases found continuing with the"
+                                        " fop:%s", gf_fop_list[frame->op]);
+                        ret = WIND_FOP;
+                        goto unlock;
+                }
+                conflicts = __check_lease_conflict (frame, lease_ctx,
+                                lease_id, is_write_fop);
+                if (conflicts) {
+                        if (is_blocking_fop) {
+                                gf_msg_debug (frame->this->name, 0, "Fop: %s "
+                                              "conflicting existing "
+                                              "lease: %d, blocking the"
+                                              "fop", gf_fop_list[frame->op],
+                                              lease_ctx->lease_type);
+                                ret = BLOCK_FOP;
+                        } else {
+                                gf_msg_debug (frame->this->name, 0, "Fop: %s "
+                                              "conflicting existing "
+                                              "lease: %d, sending "
+                                              "EAGAIN", gf_fop_list[frame->op],
+                                              lease_ctx->lease_type);
+                                errno = EAGAIN;
+                                ret = -1;
+                        }
+                }
+        }
+unlock:
+        pthread_mutex_unlock (&lease_ctx->lock);
+out:
+        return ret;
+}
+
+
+static int
+remove_clnt_leases (const char *client_uid, inode_t *inode, xlator_t *this)
+{
+        lease_inode_ctx_t  *lease_ctx      = NULL;
+        lease_id_entry_t   *lease_entry    = NULL;
+        lease_id_entry_t   *tmp            = NULL;
+        int                 ret            = 0;
+        int                 i              = 0;
+
+        lease_ctx = lease_ctx_get (inode, this);
+        if (!lease_ctx) {
+                gf_msg (this->name, GF_LOG_WARNING, ENOMEM,
+                        LEASE_MSG_INVAL_INODE_CTX,
+                        "Unable to create/get inode ctx");
+                ret = -1;
+                errno = ENOMEM;
+                goto out;
+        }
+
+        pthread_mutex_lock (&lease_ctx->lock);
+        {
+                list_for_each_entry_safe (lease_entry, tmp,
+                                          &lease_ctx->lease_id_list,
+                                          lease_id_list) {
+                        if (strcmp (client_uid, lease_entry->client_uid) == 0) {
+                                for (i = 0; i < GF_LEASE_MAX_TYPE; i++) {
+                                        lease_ctx->lease_type_cnt[i] -= lease_entry->lease_type_cnt[i];
+                                }
+                                lease_ctx->lease_cnt -= lease_entry->lease_cnt;
+                                __destroy_lease_id_entry (lease_entry);
+                                if (lease_ctx->lease_cnt == 0) {
+                                        pthread_mutex_unlock (&lease_ctx->lock);
+                                        goto unblock;
+                                }
+                        }
+                }
+        }
+        pthread_mutex_unlock (&lease_ctx->lock);
+out:
+        return ret;
+
+unblock:
+        do_blocked_fops (this, lease_ctx);
+        return ret;
+}
+
+
+int
+cleanup_client_leases (xlator_t *this, const char *client_uid)
+{
+        lease_client_t    *clnt         = NULL;
+        lease_client_t    *tmp          = NULL;
+        struct list_head   cleanup_list = {0, };
+        lease_inode_t     *l_inode      = NULL;
+        lease_inode_t     *tmp1         = NULL;
+        leases_private_t  *priv         = NULL;
+        int                ret          = 0;
+
+        priv = this->private;
+        if (!priv) {
+                ret = -1;
+                errno = EINVAL;
+                goto out;
+        }
+
+        INIT_LIST_HEAD (&cleanup_list);
+        pthread_mutex_lock (&priv->mutex);
+        {
+                list_for_each_entry_safe (clnt, tmp, &priv->client_list, client_list) {
+                        if ((strcmp (clnt->client_uid, client_uid) == 0)) {
+                                list_for_each_entry_safe (l_inode, tmp1,
+                                                          &clnt->inode_list, list) {
+                                        list_del_init (&l_inode->list);
+                                        list_add_tail (&l_inode->list, &cleanup_list);
+                                }
+                                break;
+                        }
+                        __destroy_lease_client (clnt);
+                }
+        }
+        pthread_mutex_unlock (&priv->mutex);
+
+        l_inode = tmp1 = NULL;
+        list_for_each_entry_safe (l_inode, tmp1, &cleanup_list, list) {
+                remove_clnt_leases (client_uid, l_inode->inode, this);
+        }
+out:
+        return ret;
+}
+
+
+static void
+__remove_all_leases (xlator_t *this, lease_inode_ctx_t *lease_ctx)
+{
+        int                 i              = 0;
+        lease_id_entry_t   *lease_entry    = NULL;
+        lease_id_entry_t   *tmp            = NULL;
+
+        __dump_leases_info (this, lease_ctx);
+
+        list_for_each_entry_safe (lease_entry, tmp,
+                                  &lease_ctx->lease_id_list,
+                                  lease_id_list) {
+                lease_entry->lease_cnt = 0;
+                remove_from_clnt_list (this, lease_entry->client_uid, lease_ctx->inode);
+                __destroy_lease_id_entry (lease_entry);
+        }
+        INIT_LIST_HEAD (&lease_ctx->lease_id_list);
+        for (i = 0; i <= GF_LEASE_MAX_TYPE; i++)
+                lease_ctx->lease_type_cnt[i] = 0;
+        lease_ctx->lease_type = 0;
+        lease_ctx->lease_cnt = 0;
+        lease_ctx->recall_in_progress = _gf_false;
+        inode_unref (lease_ctx->inode);
+        lease_ctx->timer = NULL;
+
+        /* TODO:
+         * - Mark the corresponding fd bad. Could be done on client side
+         * as a result of recall
+         * - Free the lease_ctx
+         */
+        return;
+}
+
+
+static int
+remove_all_leases (xlator_t *this, inode_t *inode)
+{
+        lease_inode_ctx_t  *lease_ctx       = NULL;
+        int                 ret             = 0;
+
+        GF_VALIDATE_OR_GOTO ("leases", inode, out);
+
+        lease_ctx = lease_ctx_get (inode, this);
+        if (!lease_ctx) {
+                gf_msg (this->name, GF_LOG_WARNING, ENOMEM,
+                        LEASE_MSG_INVAL_INODE_CTX,
+                        "Unable to create/get inode ctx");
+                ret = -1;
+                errno = ENOMEM;
+                goto out;
+        }
+
+        pthread_mutex_lock (&lease_ctx->lock);
+        {
+                __remove_all_leases (this, lease_ctx);
+        }
+        pthread_mutex_unlock (&lease_ctx->lock);
+
+        do_blocked_fops (this, lease_ctx);
+out:
+        return ret;
+}
+
+
+void *
+expired_recall_cleanup (void *data)
+{
+        struct timespec      sleep_till      = {0, };
+        struct list_head     recall_cleanup_list;
+        lease_inode_t       *recall_entry    = NULL;
+        lease_inode_t       *tmp             = NULL;
+        leases_private_t    *priv            = NULL;
+        xlator_t            *this            = NULL;
+
+        GF_VALIDATE_OR_GOTO ("leases", data, out);
+
+        this = data;
+        priv = this->private;
+
+        gf_msg_debug (this->name, 0, "Started the expired_recall_cleanup thread");
+
+        while (1) {
+                pthread_mutex_lock (&priv->mutex);
+                {
+                        if (priv->fini) {
+                                pthread_mutex_unlock (&priv->mutex);
+                                goto out;
+                        }
+                        INIT_LIST_HEAD (&recall_cleanup_list);
+                        if (list_empty (&priv->recall_list)) {
+                                sleep_till.tv_sec = time (NULL) + 600;
+                                pthread_cond_timedwait (&priv->cond, &priv->mutex,
+                                                        &sleep_till);
+                        }
+                        if (!list_empty (&priv->recall_list)) {
+                                gf_msg_debug (this->name, 0, "Found expired recalls");
+                                list_for_each_entry_safe (recall_entry, tmp,
+                                                          &priv->recall_list, list) {
+                                        list_del_init (&recall_entry->list);
+                                        list_add_tail (&recall_entry->list, &recall_cleanup_list);
+                                }
+                        }
+                }
+                pthread_mutex_unlock (&priv->mutex);
+
+                recall_entry = tmp = NULL;
+                list_for_each_entry_safe (recall_entry, tmp, &recall_cleanup_list, list) {
+                        gf_msg_debug (this->name, 0, "Recall lease was sent on"
+                                      " inode:%p, recall timer has expired"
+                                      " and clients haven't unlocked the lease"
+                                      " hence cleaning up leases on the inode",
+                                      recall_entry->inode);
+                        remove_all_leases (this, recall_entry->inode);
+                        list_del_init (&recall_entry->list);
+                }
+        }
+
+out:
+        return NULL;
+}
diff --git a/xlators/features/leases/src/leases-mem-types.h b/xlators/features/leases/src/leases-mem-types.h
new file mode 100644
index 00000000000..055da45be5d
--- /dev/null
+++ b/xlators/features/leases/src/leases-mem-types.h
@@ -0,0 +1,27 @@
+/*
+   Copyright (c) 2015-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.
+*/
+
+#ifndef __LEASES_MEM_TYPES_H__
+#define __LEASES_MEM_TYPES_H__
+
+#include "mem-types.h"
+
+enum gf_leases_mem_types_ {
+        gf_leases_mt_conf_t = gf_common_mt_end + 1,
+        gf_leases_mt_private_t,
+        gf_leases_mt_lease_client_t,
+        gf_leases_mt_lease_inode_t,
+        gf_leases_mt_fd_ctx_t,
+        gf_leases_mt_lease_inode_ctx_t,
+        gf_leases_mt_lease_id_entry_t,
+        gf_leases_mt_fop_stub_t,
+        gf_leases_mt_end
+};
+#endif
diff --git a/xlators/features/leases/src/leases-messages.h b/xlators/features/leases/src/leases-messages.h
new file mode 100644
index 00000000000..62df4395a59
--- /dev/null
+++ b/xlators/features/leases/src/leases-messages.h
@@ -0,0 +1,129 @@
+/*
+ Copyright (c) 2015-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.
+ */
+
+#ifndef _LEASES_MESSAGES_H_
+#define _LEASES_MESSAGES_H_
+
+#include "glfs-message-id.h"
+
+/* NOTE: Rules for message additions
+ * 1) Each instance of a message is _better_ left with a unique message ID, even
+ * if the message format is the same. Reasoning is that, if the message
+ * format needs to change in one instance, the other instances are not
+ * impacted or the new change does not change the ID of the instance being
+ * modified.
+ * 2) Addition of a message,
+ * - Should increment the GLFS_NUM_MESSAGES
+ * - Append to the list of messages defined, towards the end
+ * - Retain macro naming as glfs_msg_X (for redability across developers)
+ * NOTE: Rules for message format modifications
+ * 3) Check across the code if the message ID macro in question is reused
+ * anywhere. If reused then then the modifications should ensure correctness
+ * everywhere, or needs a new message ID as (1) above was not adhered to. If
+ * not used anywhere, proceed with the required modification.
+ * NOTE: Rules for message deletion
+ * 4) Check (3) and if used anywhere else, then cannot be deleted. If not used
+ * anywhere, then can be deleted, but will leave a hole by design, as
+ * addition rules specify modification to the end of the list and not filling
+ * holes.
+ */
+
+#define LEASES_COMP_BASE        GLFS_MSGID_COMP_LEASES
+#define GLFS_NUM_MESSAGES       11
+#define GLFS_MSGID_END          (LEASES_COMP_BASE + GLFS_NUM_MESSAGES + 1)
+
+#define glfs_msg_start_x LEASES_COMP_BASE, "Invalid: Start of messages"
+/*------------*/
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_NO_MEM                        (LEASES_COMP_BASE + 1)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_RECALL_FAIL                   (LEASES_COMP_BASE + 2)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_INVAL_LEASE_ID                (LEASES_COMP_BASE + 3)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_INVAL_UNLK_LEASE              (LEASES_COMP_BASE + 4)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_INVAL_INODE_CTX               (LEASES_COMP_BASE + 5)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_NOT_ENABLED                   (LEASES_COMP_BASE + 6)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_NO_TIMER_WHEEL                (LEASES_COMP_BASE + 7)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_CLNT_NOTFOUND                 (LEASES_COMP_BASE + 8)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_INODE_NOTFOUND                (LEASES_COMP_BASE + 9)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_INVAL_FD_CTX                  (LEASES_COMP_BASE + 10)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define LEASE_MSG_INVAL_LEASE_TYPE              (LEASES_COMP_BASE + 11)
+
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ */
+#define glfs_msg_end_x GLFS_MSGID_END, "Invalid: End of messages"
+
+#endif /* !_LEASES_MESSAGES_H_ */
diff --git a/xlators/features/leases/src/leases.c b/xlators/features/leases/src/leases.c
new file mode 100644
index 00000000000..8783a5ce9c2
--- /dev/null
+++ b/xlators/features/leases/src/leases.c
@@ -0,0 +1,1134 @@
+/*
+   Copyright (c) 2015-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.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "leases.h"
+
+int32_t
+leases_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                 int32_t op_ret, int32_t op_errno, fd_t *fd, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (open, frame, op_ret, op_errno, fd, xdata);
+
+        return 0;
+}
+
+
+int32_t
+leases_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
+             fd_t *fd, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = 0;
+        int              ret             = 0;
+        lease_fd_ctx_t  *fd_ctx          = NULL;
+        char            *lease_id        = NULL;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        fd_ctx = GF_CALLOC (1, sizeof (*fd_ctx), gf_leases_mt_fd_ctx_t);
+
+        fd_ctx->client_uid = gf_strdup (frame->root->client->client_uid);
+        if (!fd_ctx->client_uid) {
+                op_errno = ENOMEM;
+                goto err;
+        }
+
+        GET_FLAGS (frame->op, flags);
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        if (lease_id != NULL)
+                memcpy (fd_ctx->lease_id, lease_id, LEASE_ID_SIZE);
+        else
+                memset (fd_ctx->lease_id, 0, LEASE_ID_SIZE);
+
+        ret = fd_ctx_set (fd, this, (uint64_t)fd_ctx);
+        if (ret) {
+                op_errno = ENOMEM;
+                goto err;
+        }
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, open, frame, this,
+                         loc, flags, fd, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_open_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->open,
+                    loc, flags, fd, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (open, frame, -1, op_errno, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int op_ret, int op_errno, struct iatt *prebuf,
+                   struct iatt *postbuf, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (writev, frame, op_ret, op_errno,
+                             prebuf, postbuf, xdata);
+
+        return 0;
+}
+
+
+int32_t
+leases_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
+               struct iovec *vector, int count, off_t off, uint32_t flags,
+               struct iobref *iobref, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, writev, frame, this, fd, vector, count,
+                         off, flags, iobref, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_writev_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->writev,
+                    fd, vector, count, off, flags, iobref, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (writev, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+
+int32_t
+leases_readv_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                  int op_ret, int op_errno,
+                  struct iovec *vector, int count, struct iatt *stbuf,
+                  struct iobref *iobref, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (readv, frame, op_ret, op_errno, vector,
+                             count, stbuf, iobref, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_readv (call_frame_t *frame, xlator_t *this,
+              fd_t *fd, size_t size, off_t offset,
+              uint32_t flags, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, readv, frame, this,
+                         fd, size, offset, flags, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_readv_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readv,
+                    fd, size, offset, flags, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (readv, frame, -1, op_errno, NULL, 0,
+                             NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_lk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+               int32_t op_ret, int32_t op_errno, struct gf_flock *lock,
+               dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (lk, frame, op_ret, op_errno, lock, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_lk (call_frame_t *frame, xlator_t *this,
+           fd_t *fd, int32_t cmd, struct gf_flock *flock, dict_t *xdata)
+{
+        int32_t         op_errno         = 0;
+        uint32_t        fop_flags        = 0;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS_LK (cmd, flock->l_type, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, lk, frame, this,
+                         fd, cmd, flock, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_lk_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->lk,
+                    fd, cmd, flock, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (lk, frame, -1, op_errno, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_lease (call_frame_t *frame, xlator_t *this,
+             loc_t *loc, struct gf_lease *lease, dict_t *xdata)
+{
+        int32_t         op_errno         = 0;
+        int             ret              = 0;
+        struct gf_lease nullease         = {0, };
+        int32_t         op_ret           = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        ret = process_lease_req (frame, this, loc->inode, lease);
+        if (ret < 0) {
+                op_errno = -ret;
+                op_ret = -1;
+        }
+        goto unwind;
+
+out:
+        gf_msg (this->name, GF_LOG_ERROR, EINVAL, LEASE_MSG_NOT_ENABLED,
+                "\"features/leases\" translator is not enabled. "
+                "You need to enable it for proper functioning of your "
+                "application");
+        op_errno = ENOSYS;
+        op_ret = -1;
+
+unwind:
+        STACK_UNWIND_STRICT (lease, frame, op_ret, op_errno,
+                             (op_errno == ENOSYS) ? &nullease : lease, xdata);
+        return 0;
+}
+
+int32_t
+leases_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                     int op_ret, int op_errno, struct iatt *prebuf,
+                     struct iatt *postbuf, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (truncate, frame, op_ret, op_errno,
+                             prebuf, postbuf, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
+                 dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, 0);
+
+        ret = check_lease_conflict (frame, loc->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (loc->inode, truncate, frame, this,
+                         loc, offset, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_truncate_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->truncate,
+                    loc, offset, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (truncate, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int op_ret, int op_errno, struct iatt *statpre,
+                    struct iatt *statpost, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (setattr, frame, op_ret, op_errno,
+                             statpre, statpost, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
+                struct iatt *stbuf, int32_t valid, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, 0);
+
+        ret = check_lease_conflict (frame, loc->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (loc->inode, setattr, frame, this,
+                         loc, stbuf, valid, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_setattr_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->setattr,
+                    loc, stbuf, valid, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (setattr, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, struct iatt *stbuf,
+                   struct iatt *preoldparent, struct iatt *postoldparent,
+                   struct iatt *prenewparent, struct iatt *postnewparent,
+                   dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (rename, frame, op_ret, op_errno,
+                             stbuf, preoldparent, postoldparent,
+                             prenewparent, postnewparent, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_rename (call_frame_t *frame, xlator_t *this,
+               loc_t *oldloc, loc_t *newloc, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        /* should the lease be also checked for newloc */
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, 0);
+
+        ret = check_lease_conflict (frame, oldloc->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (oldloc->inode, rename, frame, this,
+                         oldloc, newloc, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_rename_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->rename,
+                    oldloc, newloc, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (rename, frame, -1, op_errno, NULL,
+                             NULL, NULL, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int op_ret, int op_errno, struct iatt *preparent,
+                   struct iatt *postparent, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (unlink, frame, op_ret, op_errno,
+                             preparent, postparent, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
+               dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, 0);
+
+        ret = check_lease_conflict (frame, loc->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (loc->inode, unlink, frame, this,
+                         loc, xflag, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_unlink_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->unlink,
+                    loc, xflag, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (unlink, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                 int op_ret, int op_errno, inode_t *inode, struct iatt *stbuf,
+                 struct iatt *preparent, struct iatt *postparent, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (link, frame, op_ret, op_errno,
+                             inode, stbuf, preparent, postparent, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
+             loc_t *newloc, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, 0);
+
+        ret = check_lease_conflict (frame, oldloc->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (oldloc->inode, link, frame, this,
+                         oldloc, newloc, xdata);
+        return 0;
+out:
+        STACK_WIND (frame, leases_link_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->link,
+                    oldloc, newloc, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (link, frame, -1, op_errno, NULL,
+                             NULL, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int op_ret, int op_errno, fd_t *fd, inode_t *inode,
+                   struct iatt *stbuf, struct iatt *preparent,
+                   struct iatt *postparent, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (create, frame, op_ret, op_errno, fd,
+                             inode, stbuf, preparent, postparent, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_create (call_frame_t *frame, xlator_t *this,
+               loc_t *loc, int32_t flags, mode_t mode,
+               mode_t umask, fd_t *fd, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, create, frame, this,
+                         loc, flags, mode, umask, fd, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_create_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->create,
+                    loc, flags, mode, umask, fd, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (create, frame, -1, op_errno, NULL, NULL, NULL,
+                             NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                  int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                  struct iatt *postbuf,
+                  dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (fsync, frame, op_ret, op_errno, prebuf, postbuf,
+                             xdata);
+        return 0;
+}
+
+int32_t
+leases_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd,
+              int32_t flags, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, fsync, frame, this,
+                         fd, flags, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_fsync_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->fsync, fd, flags, xdata);
+        return 0;
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (fsync, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                      int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                      struct iatt *postbuf,
+                      dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (ftruncate, frame, op_ret, op_errno, prebuf,
+                             postbuf, xdata);
+        return 0;
+}
+
+int32_t
+leases_ftruncate (call_frame_t *frame, xlator_t *this,
+                  fd_t *fd, off_t offset, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, 0); /* TODO:fd->flags?*/
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, ftruncate, frame, this,
+                         fd, offset, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_ftruncate_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->ftruncate,
+                    fd, offset, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (ftruncate, frame, -1, op_errno, NULL,
+                             NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_fsetattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                     int32_t op_ret, int32_t op_errno, struct iatt *statpre,
+                     struct iatt *statpost, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (fsetattr, frame, op_ret, op_errno,
+                             statpre, statpost, xdata);
+        return 0;
+}
+
+int32_t
+leases_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
+                 struct iatt  *stbuf, int32_t valid, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, fsetattr, frame, this,
+                         fd, stbuf, valid, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_fsetattr_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fsetattr,
+                    fd, stbuf, valid, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (fsetattr, frame, -1, op_errno, NULL,
+                             NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_fallocate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                      int32_t op_ret, int32_t op_errno, struct iatt *pre,
+                      struct iatt *post, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (fallocate, frame, op_ret, op_errno, pre,
+                             post, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_fallocate (call_frame_t *frame, xlator_t *this, fd_t *fd,
+                  int32_t mode, off_t offset, size_t len, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, fallocate, frame, this,
+                         fd, mode, offset, len, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_fallocate_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fallocate,
+                    fd, mode, offset, len, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (fallocate, frame, -1, op_errno, NULL,
+                             NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_discard_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, struct iatt *pre,
+                    struct iatt *post, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (discard, frame, op_ret, op_errno, pre,
+                             post, xdata);
+
+        return 0;
+}
+
+int32_t
+leases_discard (call_frame_t *frame, xlator_t *this, fd_t *fd,
+                off_t offset, size_t len, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, discard, frame, this,
+                         fd, offset, len, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_discard_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->discard,
+                    fd, offset, len, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (discard, frame, -1, op_errno, NULL,
+                             NULL, NULL);
+        return 0;
+}
+
+int32_t
+leases_zerofill_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                     int32_t op_ret, int32_t op_errno, struct iatt *pre,
+                     struct iatt *post, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (zerofill, frame, op_ret, op_errno, pre,
+                             post, xdata);
+
+        return 0;
+}
+
+int
+leases_zerofill (call_frame_t *frame, xlator_t *this, fd_t *fd,
+                 off_t offset, off_t len, dict_t *xdata)
+{
+        uint32_t         fop_flags       = 0;
+        int32_t          op_errno        = -1;
+        char            *lease_id        = NULL;
+        int              ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, zerofill, frame, this,
+                         fd, offset, len, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_zerofill_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->zerofill,
+                    fd, offset, len, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (zerofill, frame, -1, op_errno, NULL,
+                             NULL, NULL);
+        return 0;
+}
+
+int
+leases_flush_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+              int32_t op_ret, int32_t op_errno, dict_t *xdata)
+{
+        STACK_UNWIND_STRICT (flush, frame, op_ret, op_errno, xdata);
+
+        return 0;
+}
+
+int
+leases_flush (call_frame_t *frame, xlator_t *this,
+              fd_t *fd, dict_t *xdata)
+{
+        int32_t       op_errno       = -1;
+        uint32_t      fop_flags      = 0;
+        char         *lease_id       = NULL;
+        int          ret             = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        GET_LEASE_ID (xdata, lease_id, frame->root->client->client_uid);
+        GET_FLAGS (frame->op, fd->flags);
+
+        ret = check_lease_conflict (frame, fd->inode, lease_id, fop_flags);
+        if (ret < 0)
+                goto err;
+        else if (ret == BLOCK_FOP)
+                goto block;
+        else if (ret == WIND_FOP)
+                goto out;
+
+block:
+        LEASE_BLOCK_FOP (fd->inode, flush, frame, this,
+                         fd, xdata);
+        return 0;
+
+out:
+        STACK_WIND (frame, leases_flush_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->flush, fd, xdata);
+        return 0;
+
+err:
+        op_errno = (op_errno == -1) ? errno : op_errno;
+        STACK_UNWIND_STRICT (create, frame, -1, op_errno, NULL,
+                             NULL, NULL, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+mem_acct_init (xlator_t *this)
+{
+        int     ret = -1;
+
+        if (!this)
+                return ret;
+
+        ret = xlator_mem_acct_init (this, gf_leases_mt_end + 1);
+
+        if (ret != 0) {
+                gf_msg (this->name, GF_LOG_WARNING, ENOMEM, LEASE_MSG_NO_MEM,
+                        "mem account init failed");
+                return ret;
+        }
+
+        return ret;
+}
+
+int
+reconfigure (xlator_t *this, dict_t *options)
+{
+        leases_private_t *priv                   = NULL;
+        int               ret                    = -1;
+
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        GF_OPTION_RECONF ("leases", priv->leases_enabled,
+                          options, bool, out);
+        GF_OPTION_RECONF ("lease-lock-recall-timeout",
+                          priv->recall_lease_timeout,
+                          options, int32, out);
+
+        ret = 0;
+out:
+        return ret;
+}
+
+int
+init (xlator_t *this)
+{
+        int                       ret        = -1;
+        leases_private_t         *priv       = NULL;
+
+        priv = GF_CALLOC (1, sizeof (*priv),
+                          gf_leases_mt_private_t);
+        if (!priv) {
+                gf_msg (this->name, GF_LOG_WARNING, ENOMEM, LEASE_MSG_NO_MEM,
+                        "Leases init failed");
+                goto out;
+        }
+
+        GF_OPTION_INIT ("leases", priv->leases_enabled,
+                        bool, out);
+        GF_OPTION_INIT ("lease-lock-recall-timeout",
+                        priv->recall_lease_timeout, int32, out);
+        pthread_mutex_init (&priv->mutex, NULL);
+        INIT_LIST_HEAD (&priv->client_list);
+        INIT_LIST_HEAD (&priv->recall_list);
+
+        priv->timer_wheel = glusterfs_global_timer_wheel (this);
+        if (!priv->timer_wheel) {
+                gf_msg (this->name, GF_LOG_ERROR, 0, LEASE_MSG_NO_TIMER_WHEEL,
+                        "Initing the global timer wheel");
+                ret = glusterfs_global_timer_wheel_init (this->ctx);
+                if (ret) {
+                        gf_msg (this->name, GF_LOG_INFO, 0, LEASE_MSG_NO_TIMER_WHEEL,
+                                "Initing the global timer wheel failed");
+                        goto out;
+                }
+        }
+
+        pthread_create (&priv->recall_thr, NULL, expired_recall_cleanup, this);
+
+        this->private = priv;
+        ret = 0;
+
+out:
+        if (ret) {
+                GF_FREE (priv);
+        }
+
+        return ret;
+}
+
+int
+fini (xlator_t *this)
+{
+        leases_private_t *priv = NULL;
+
+        priv = this->private;
+        if (!priv) {
+                return 0;
+        }
+        this->private = NULL;
+
+        priv->fini = _gf_false;
+        pthread_join (priv->recall_thr, NULL);
+
+        GF_FREE (priv);
+
+        return 0;
+}
+
+static int
+leases_forget (xlator_t *this, inode_t *inode)
+{
+        /* TODO:leases_cleanup_inode_ctx (this, inode); */
+        return 0;
+}
+
+static int
+leases_release (xlator_t *this, fd_t *fd)
+{
+        /* TODO:cleanup fd_ctx */
+        return 0;
+}
+
+static int
+leases_clnt_disconnect_cbk (xlator_t *this, client_t *client)
+{
+        int ret = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        ret = cleanup_client_leases (this, client->client_uid);
+out:
+        return ret;
+}
+
+int
+notify (xlator_t *this, int32_t event, void *data, ...)
+{
+        int ret = 0;
+
+        EXIT_IF_LEASES_OFF (this, out);
+
+        ret = default_notify (this, event, data);
+out:
+        return ret;
+}
+
+struct xlator_fops fops = {
+        /* Metadata modifying fops */
+        .fsetattr    = leases_fsetattr,
+        .setattr     = leases_setattr,
+
+        /* File Data reading fops */
+        .open        = leases_open,
+        .readv       = leases_readv,
+
+        /* File Data modifying fops */
+        .truncate    = leases_truncate,
+        .ftruncate   = leases_ftruncate,
+        .writev      = leases_writev,
+        .zerofill    = leases_zerofill,
+        .fallocate   = leases_fallocate,
+        .discard     = leases_discard,
+        .lk          = leases_lk,
+        .fsync       = leases_fsync,
+        .flush       = leases_flush,
+        .lease       = leases_lease,
+
+        /* Directory Data modifying fops */
+        .create      = leases_create,
+        .rename      = leases_rename,
+        .unlink      = leases_unlink,
+        .link        = leases_link,
+
+#ifdef NOT_SUPPORTED
+        /* internal lk fops */
+        .inodelk     = leases_inodelk,
+        .finodelk    = leases_finodelk,
+        .entrylk     = leases_entrylk,
+        .fentrylk    = leases_fentrylk,
+
+        /* Internal special fops*/
+        .xattrop     = leases_xattrop,
+        .fxattrop    = leases_fxattrop,
+#endif
+};
+
+struct xlator_cbks cbks = {
+        .forget            = leases_forget,
+        .release           = leases_release,
+        .client_disconnect = leases_clnt_disconnect_cbk,
+};
+
+struct volume_options options[] = {
+        { .key  = {"leases"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .default_value = "off",
+          .description = "When \"on\", enables leases support"
+        },
+        { .key  = {"lease-lock-recall-timeout"},
+          .type = GF_OPTION_TYPE_INT,
+          .default_value = RECALL_LEASE_LK_TIMEOUT,
+          .description = "After 'timeout' seconds since the recall_lease"
+                         " request has been sent to the client, the lease lock"
+                         " will be forcefully purged by the server."
+        },
+        { .key = {NULL} },
+};
diff --git a/xlators/features/leases/src/leases.h b/xlators/features/leases/src/leases.h
new file mode 100644
index 00000000000..16143691075
--- /dev/null
+++ b/xlators/features/leases/src/leases.h
@@ -0,0 +1,239 @@
+/*
+  Copyright (c) 2015-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.
+*/
+
+#ifndef _LEASES_H
+#define _LEASES_H
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "common-utils.h"
+#include "glusterfs.h"
+#include "xlator.h"
+#include "inode.h"
+#include "call-stub.h"
+#include "logging.h"
+#include "client_t.h"
+#include "lkowner.h"
+#include "locking.h"
+#include "upcall-utils.h"
+#include "tw.h"
+#include "timer-wheel.h"
+#include "leases-mem-types.h"
+#include "leases-messages.h"
+
+/* The time period for which a client lease lock will be stored after its been
+ * recalled for the first time. */
+#define RECALL_LEASE_LK_TIMEOUT "60"
+
+#define DATA_MODIFY_FOP 0x0001
+#define BLOCKING_FOP 0x0002
+
+#define BLOCK_FOP 0x0001
+#define WIND_FOP 0x0002
+
+#define EXIT_IF_LEASES_OFF(this, label) do {                                   \
+        if (!is_leases_enabled(this))                                          \
+                goto label;                                                    \
+} while (0)
+
+#define GET_LEASE_ID(xdata, lease_id, client_uid) do {                         \
+        int   ret_val = -1;                                                    \
+        ret_val = dict_get_bin (xdata, "lease-id", (void **)&lease_id);        \
+        if (ret_val) {                                                         \
+                ret_val = 0;                                                   \
+                gf_msg_debug ("leases", 0, "Lease id is not set for client:%s", client_uid); \
+        }                                                                      \
+} while (0)
+
+#define GET_FLAGS(fop, fd_flags)                                               \
+do {                                                                           \
+        if (fd_flags & (O_WRONLY | O_RDWR) && fop == GF_FOP_OPEN)              \
+                fop_flags = DATA_MODIFY_FOP;                                   \
+                                                                               \
+        if (fop == GF_FOP_UNLINK || fop == GF_FOP_RENAME ||                    \
+            fop == GF_FOP_TRUNCATE || fop == GF_FOP_FTRUNCATE ||               \
+            fop == GF_FOP_FLUSH || fop == GF_FOP_FSYNC ||                      \
+            fop == GF_FOP_WRITE || fop == GF_FOP_FALLOCATE ||                  \
+            fop == GF_FOP_DISCARD || fop == GF_FOP_ZEROFILL ||                 \
+            fop == GF_FOP_SETATTR || fop == GF_FOP_FSETATTR ||                 \
+            fop == GF_FOP_LINK)                                                \
+                fop_flags = DATA_MODIFY_FOP;                                   \
+                                                                               \
+        if (fd_flags & (O_NONBLOCK | O_NDELAY))                                \
+                fop_flags |= BLOCKING_FOP;                                     \
+                                                                               \
+} while (0)                                                                    \
+
+
+#define GET_FLAGS_LK(cmd, l_type, fd_flags)                                    \
+do {                                                                           \
+        /* TODO: handle F_RESLK_LCK and other glusterfs_lk_recovery_cmds_t */  \
+        if ((cmd == F_SETLKW || cmd == F_SETLKW64 ||                           \
+             cmd == F_SETLK || cmd == F_SETLK64) &&                            \
+            l_type == F_WRLCK)                                                 \
+                fop_flags = DATA_MODIFY_FOP;                                   \
+                                                                               \
+        if (fd_flags & (O_NONBLOCK | O_NDELAY) &&                              \
+            (cmd == F_SETLKW || cmd == F_SETLKW64))                            \
+                fop_flags |= BLOCKING_FOP;                                     \
+                                                                               \
+} while (0)                                                                    \
+
+#define LEASE_BLOCK_FOP(inode, fop_name, frame, this, params ...)              \
+do {                                                                           \
+        call_stub_t             *__stub     = NULL;                            \
+        fop_stub_t              *blk_fop    = NULL;                            \
+        lease_inode_ctx_t       *lease_ctx  = NULL;                            \
+        int                      __ret      = 0;                               \
+                                                                               \
+        __stub = fop_##fop_name##_stub (frame, default_##fop_name##_resume,    \
+                                        params);                               \
+        if (!__stub) {                                                         \
+                ret = -ENOMEM;                                                 \
+                goto __out;                                                    \
+        }                                                                      \
+                                                                               \
+        blk_fop = GF_CALLOC (1, sizeof (*blk_fop),                             \
+                             gf_leases_mt_fop_stub_t);                         \
+        if (blk_fop) {                                                         \
+                ret = -ENOMEM;                                                 \
+                goto __out;                                                    \
+        }                                                                      \
+                                                                               \
+        lease_ctx = lease_ctx_get (inode, this);                               \
+        if (!lease_ctx) {                                                      \
+                gf_msg (this->name, GF_LOG_WARNING, ENOMEM,                    \
+                        LEASE_MSG_NO_MEM,                                      \
+                        "Unable to create/get inode ctx");                     \
+                op_errno = ENOMEM;                                             \
+                goto __out;                                                    \
+        }                                                                      \
+                                                                               \
+        pthread_mutex_lock (&lease_ctx->lock);                                 \
+        {                                                                      \
+                /*TODO: If the lease is unlocked btw check lease conflict and  \
+                 * by now, then this fop shouldn't be add to the blocked fop   \
+                 * list, can use generation number for the same?*/             \
+                list_add_tail (&blk_fop->list, &lease_ctx->blocked_list);      \
+        }                                                                      \
+        pthread_mutex_unlock (&lease_ctx->lock);                               \
+                                                                               \
+__out:                                                                         \
+        if (ret < 0) {                                                         \
+                gf_msg (this->name, GF_LOG_WARNING, ENOMEM, LEASE_MSG_NO_MEM,  \
+                        "Unable to create stub for blocking the fop:%s (%s)",  \
+                         gf_fop_list[frame->op], strerror(ENOMEM));            \
+                default_##fop_name##_failure_cbk (frame, -__ret);              \
+                if (__stub != NULL) {                                          \
+                        call_stub_destroy (__stub);                            \
+                }                                                              \
+                GF_FREE (blk_fop);                                             \
+                goto err;                                                      \
+        }                                                                      \
+} while (0)                                                                    \
+
+struct _leases_private {
+        gf_boolean_t      leases_enabled;
+        int32_t           recall_lease_timeout;
+        struct list_head  client_list;
+        struct list_head  recall_list;
+        struct tvec_base *timer_wheel;    /* timer wheel where the recall request
+                                             is qued and waits for unlock/expiry */
+        gf_boolean_t      fini;
+        pthread_t         recall_thr;
+        pthread_mutex_t   mutex;
+        pthread_cond_t    cond;
+};
+typedef struct _leases_private leases_private_t;
+
+struct _lease_client {
+        char             *client_uid;
+        struct list_head  client_list;
+        struct list_head  inode_list;
+};
+typedef struct _lease_client lease_client_t;
+
+struct _lease_inode {
+        inode_t          *inode;
+        struct list_head  list;  /* This can be part of both inode_list and recall_list */
+};
+typedef struct _lease_inode lease_inode_t;
+
+struct _lease_fd_ctx {
+        char             *client_uid;
+        char              lease_id[LEASE_ID_SIZE];
+};
+typedef struct _lease_fd_ctx lease_fd_ctx_t;
+
+struct _lease_inode_ctx {
+        struct list_head  lease_id_list;  /* clients that have taken leases */
+        int               lease_type_cnt[GF_LEASE_MAX_TYPE+1];
+        int               lease_type;   /* Types of leases acquired */
+        uint64_t          lease_cnt;    /* Total number of leases on this inode */
+        uint64_t          openfd_cnt;   /* number of fds open */
+        gf_boolean_t      recall_in_progress;  /* if lease recall is sent on this inode */
+        struct list_head  blocked_list; /* List of fops blocked until the
+                                           lease recall is complete */
+        inode_t          *inode;        /* this represents the inode on which the
+                                           lock was taken, required mainly during
+                                           disconnect cleanup */
+        struct gf_tw_timer_list *timer;
+        pthread_mutex_t   lock;
+};
+typedef struct _lease_inode_ctx lease_inode_ctx_t;
+
+struct _lease_id_entry {
+        struct list_head    lease_id_list;
+        char                lease_id[LEASE_ID_SIZE];
+        char               *client_uid;  /* uid of the client that has
+                                            taken the lease */
+        int                 lease_type_cnt[GF_LEASE_MAX_TYPE+1]; /* count of each lease type */
+        int                 lease_type;  /* Union of all the leases taken
+                                            under the given lease id */
+        uint64_t            lease_cnt;   /* Number of leases taken under the
+                                            given lease id */
+        time_t              recall_time; /* time @ which recall was sent */
+};
+typedef struct _lease_id_entry lease_id_entry_t;
+
+/* Required? as stub itself will have list */
+struct __fop_stub {
+        struct list_head      list;
+        call_stub_t          *stub;
+};
+typedef struct __fop_stub fop_stub_t;
+
+gf_boolean_t
+is_leases_enabled (xlator_t *this);
+
+int32_t
+get_recall_lease_timeout (xlator_t *this);
+
+lease_inode_ctx_t *
+lease_ctx_get (inode_t *inode, xlator_t *this);
+
+int
+process_lease_req (call_frame_t *frame, xlator_t *this,
+                   inode_t *inode, struct gf_lease *lease);
+
+int
+check_lease_conflict (call_frame_t *frame, inode_t *inode,
+                      const char *lease_id, uint32_t fop_flags);
+
+int
+cleanup_client_leases (xlator_t *this, const char *client_uid);
+
+void *
+expired_recall_cleanup (void *data);
+
+#endif /* _LEASES_H */