NLM - Network Lock Manger V4

Change-Id: Ic31b8bb10a28408da2a623f4ecc0c60af01c64af
BUG: 795421
Signed-off-by: Krishna Srinivas <ksriniva@redhat.com>
Reviewed-on: http://review.gluster.com/2711
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Amar Tumballi <amarts@redhat.com>

13 files changed, 2077 insertions, 9 deletions

diff --git a/xlators/nfs/server/src/Makefile.am b/xlators/nfs/server/src/Makefile.am
index 8fd28638506..a2c76a5021d 100644
--- a/xlators/nfs/server/src/Makefile.am
+++ b/xlators/nfs/server/src/Makefile.am
@@ -2,10 +2,10 @@ xlator_LTLIBRARIES = server.la
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/nfs
 nfsrpclibdir = $(top_srcdir)/rpc/rpc-lib/src
 server_la_LDFLAGS = -module -avoidversion
-server_la_SOURCES = nfs.c nfs-common.c nfs-fops.c nfs-inodes.c nfs-generics.c mount3.c nfs3-fh.c nfs3.c nfs3-helpers.c
+server_la_SOURCES = nfs.c nfs-common.c nfs-fops.c nfs-inodes.c nfs-generics.c mount3.c nfs3-fh.c nfs3.c nfs3-helpers.c nlm4.c nlmcbk_svc.c
 server_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
-noinst_HEADERS = nfs.h nfs-common.h nfs-fops.h nfs-inodes.h nfs-generics.h mount3.h nfs3-fh.h nfs3.h nfs3-helpers.h nfs-mem-types.h
+noinst_HEADERS = nfs.h nfs-common.h nfs-fops.h nfs-inodes.h nfs-generics.h mount3.h nfs3-fh.h nfs3.h nfs3-helpers.h nfs-mem-types.h nlm4.h
 
 AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -D$(GF_HOST_OS)\
 	-DLIBDIR=\"$(libdir)/glusterfs/$(PACKAGE_VERSION)/auth\" \
diff --git a/xlators/nfs/server/src/nfs-fops.c b/xlators/nfs/server/src/nfs-fops.c
index 8774da07581..2624c50e378 100644
--- a/xlators/nfs/server/src/nfs-fops.c
+++ b/xlators/nfs/server/src/nfs-fops.c
@@ -121,6 +121,7 @@ nfs_create_frame (xlator_t *xl, nfs_user_t *nfu)
         frame->root->pid = NFS_PID;
         frame->root->uid = nfu->uid;
         frame->root->gid = nfu->gids[NFS_PRIMGID_IDX];
+        frame->root->lk_owner = nfu->lk_owner;
         if (nfu->ngrps == 1)
                 goto err;       /* Done, we only got primary gid */
 
@@ -133,15 +134,13 @@ nfs_create_frame (xlator_t *xl, nfs_user_t *nfu)
                 frame->root->groups[y] = nfu->gids[x];
         }
 
-        set_lk_owner_from_uint64 (&frame->root->lk_owner, nfs_frame_getctr ());
-
 err:
         return frame;
 }
 
 #define nfs_fop_handle_frame_create(fram, xla, nfuser, retval, errlabel)      \
         do {                                                                  \
-                fram = nfs_create_frame (xla, (nfuser));                      \
+                fram = nfs_create_frame (xla, (nfuser));                \
                 if (!fram) {                                                  \
                         retval = (-ENOMEM);                                   \
                         gf_log (GF_NFS, GF_LOG_ERROR,"Frame creation failed");\
@@ -1362,6 +1361,49 @@ err:
         return ret;
 }
 
+int32_t
+nfs_fop_lk_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                int32_t op_ret, int32_t op_errno, struct gf_flock *flock)
+{
+        struct nfs_fop_local    *nfl = NULL;
+        fop_lk_cbk_t         progcbk = NULL;
+
+        nfl_to_prog_data (nfl, progcbk, frame);
+
+        if (progcbk)
+                progcbk (frame, cookie, this, op_ret, op_errno, flock);
+
+        nfs_stack_destroy (nfl, frame);
+        return 0;
+}
+
+
+int
+nfs_fop_lk (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, fd_t *fd,
+            int cmd, struct gf_flock *flock, fop_lk_cbk_t cbk, void *local)
+{
+        call_frame_t            *frame = NULL;
+        int                     ret = -EFAULT;
+        struct nfs_fop_local    *nfl = NULL;
+
+        if ((!xl) || (!fd) || (!nfu))
+                return ret;
+
+        nfs_fop_handle_frame_create (frame, nfsx, nfu, ret, err);
+        nfs_fop_handle_local_init (frame, nfsx, nfl, cbk, local, ret, err);
+
+        STACK_WIND_COOKIE (frame, nfs_fop_lk_cbk, xl, xl, xl->fops->lk,
+                           fd, cmd, flock);
+        ret = 0;
+err:
+        if (ret < 0) {
+                if (frame)
+                        nfs_stack_destroy (nfl, frame);
+        }
+
+        return ret;
+}
+
 
 int32_t
 nfs_fop_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
diff --git a/xlators/nfs/server/src/nfs-fops.h b/xlators/nfs/server/src/nfs-fops.h
index 48565215e38..2801da2c319 100644
--- a/xlators/nfs/server/src/nfs-fops.h
+++ b/xlators/nfs/server/src/nfs-fops.h
@@ -134,7 +134,7 @@ nfs_fop_local_wipe (xlator_t *xl, struct nfs_fop_local *l);
 
 #define nfs_fop_handle_local_init(fram,nfx, nfloc, cbck,prgloc,retval,lab)  \
         do {                                                                \
-                prog_data_to_nfl (nfx, nfloc, fram, cbck, prgloc);          \
+                prog_data_to_nfl (nfx, nfloc, fram, cbck, prgloc);      \
                 if (!nfloc) {                                               \
                         gf_log (GF_NFS,GF_LOG_ERROR,"Failed to init local");\
                         retval = -ENOMEM;                                   \
@@ -236,4 +236,9 @@ nfs_fop_stat (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, loc_t *loc,
 extern int
 nfs_fop_access (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, loc_t *loc,
                 int32_t accesstest, fop_access_cbk_t cbk, void *local);
+
+extern int
+nfs_fop_lk (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, fd_t *fd,
+            int cmd, struct gf_flock *flock, fop_lk_cbk_t cbk, void *local);
+
 #endif
diff --git a/xlators/nfs/server/src/nfs-generics.c b/xlators/nfs/server/src/nfs-generics.c
index 642879b8afc..dbb79fc9c2f 100644
--- a/xlators/nfs/server/src/nfs-generics.c
+++ b/xlators/nfs/server/src/nfs-generics.c
@@ -151,7 +151,6 @@ nfs_truncate (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, loc_t *pathloc,
         return ret;
 }
 
-
 int
 nfs_read (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, fd_t *fd, size_t size,
           off_t offset, fop_readv_cbk_t cbk, void *local)
@@ -159,6 +158,12 @@ nfs_read (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, fd_t *fd, size_t size,
         return nfs_fop_read (nfsx, xl, nfu, fd, size, offset, cbk, local);
 }
 
+int
+nfs_lk (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, fd_t *fd,
+        int cmd, struct gf_flock *flock, fop_lk_cbk_t cbk, void *local)
+{
+        return nfs_fop_lk ( nfsx, xl, nfu, fd, cmd, flock, cbk, local);
+}
 
 int
 nfs_fsync (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, fd_t *fd,
@@ -330,3 +335,4 @@ nfs_opendir (xlator_t *nfsx, xlator_t *fopxl, nfs_user_t *nfu, loc_t *pathloc,
 
         return nfs_inode_opendir (nfsx, fopxl, nfu, pathloc, cbk, local);
 }
+
diff --git a/xlators/nfs/server/src/nfs-generics.h b/xlators/nfs/server/src/nfs-generics.h
index 237a0ad09bc..11f191f5fd9 100644
--- a/xlators/nfs/server/src/nfs-generics.h
+++ b/xlators/nfs/server/src/nfs-generics.h
@@ -161,4 +161,7 @@ nfs_opendir (xlator_t *nfsx, xlator_t *fopxl, nfs_user_t *nfu, loc_t *pathloc,
 extern int
 nfs_access (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, loc_t *pathloc,
             int32_t accesstest, fop_access_cbk_t cbk, void *local);
+extern int
+nfs_lk (xlator_t *nfsx, xlator_t *xl, nfs_user_t *nfu, fd_t *fd,
+        int cmd, struct gf_flock *flock, fop_lk_cbk_t cbk, void *local);
 #endif
diff --git a/xlators/nfs/server/src/nfs-mem-types.h b/xlators/nfs/server/src/nfs-mem-types.h
index 57f5f90b2fe..e2dae4a3403 100644
--- a/xlators/nfs/server/src/nfs-mem-types.h
+++ b/xlators/nfs/server/src/nfs-mem-types.h
@@ -43,6 +43,10 @@ enum gf_nfs_mem_types_ {
         gf_nfs_mt_mnt3_resolve,
         gf_nfs_mt_mnt3_export,
         gf_nfs_mt_inode_q,
+	gf_nfs_mt_nlm4_state,
+	gf_nfs_mt_nlm4_cm,
+	gf_nfs_mt_nlm4_fde,
+        gf_nfs_mt_nlm4_nlmclnt,
         gf_nfs_mt_end
 };
 #endif
diff --git a/xlators/nfs/server/src/nfs.c b/xlators/nfs/server/src/nfs.c
index d92a0e126e6..67efe530746 100644
--- a/xlators/nfs/server/src/nfs.c
+++ b/xlators/nfs/server/src/nfs.c
@@ -40,6 +40,7 @@
 #include "nfs3.h"
 #include "nfs-mem-types.h"
 #include "nfs3-helpers.h"
+#include "nlm4.h"
 
 /* Every NFS version must call this function with the init function
  * for its particular version.
@@ -172,6 +173,13 @@ nfs_add_all_initiators (struct nfs_state *nfs)
                 goto ret;
         }
 
+        ret = nfs_add_initer (&nfs->versions, nlm4svc_init);
+        if (ret == -1) {
+                gf_log (GF_NFS, GF_LOG_ERROR, "Failed to add protocol"
+                        " initializer");
+                goto ret;
+        }
+
         ret = 0;
 ret:
         return ret;
@@ -747,6 +755,13 @@ init (xlator_t *this) {
                 goto err;
         }
 
+        ret = nlm4_init_state (this);
+        if (ret == -1) {
+                gf_log (GF_NFS, GF_LOG_CRITICAL, "Failed to init NLM"
+                        "state");
+                goto err;
+        }
+
         ret = nfs_init_versions (nfs, this);
         if (ret == -1) {
                 gf_log (GF_NFS, GF_LOG_ERROR, "Failed to initialize "
diff --git a/xlators/nfs/server/src/nfs.h b/xlators/nfs/server/src/nfs.h
index 3113e32cc0b..706cba86f3e 100644
--- a/xlators/nfs/server/src/nfs.h
+++ b/xlators/nfs/server/src/nfs.h
@@ -28,6 +28,7 @@
 #include "rpcsvc.h"
 #include "dict.h"
 #include "xlator.h"
+#include "lkowner.h"
 
 #define GF_NFS                  "nfs"
 
@@ -69,6 +70,8 @@ struct nfs_state {
         rpcsvc_t                *rpcsvc;
         struct list_head        versions;
         struct mount3_state     *mstate;
+        struct nfs3_state       *nfs3state;
+        struct nlm4_state       *nlm4state;
         struct mem_pool         *foppool;
         unsigned int            memfactor;
         xlator_list_t           *subvols;
@@ -82,6 +85,7 @@ struct nfs_state {
         int                     enable_ino32;
         unsigned int            override_portnum;
         int                     allow_insecure;
+        struct rpc_clnt         *rpc_clnt;
 };
 
 #define gf_nfs_dvm_on(nfsstt)   (((struct nfs_state *)nfsstt)->dynamicvolumes == GF_NFS_DVM_ON)
@@ -102,6 +106,7 @@ typedef struct nfs_user_info {
         uid_t   uid;
         gid_t   gids[NFS_NGROUPS];
         int     ngrps;
+        gf_lkowner_t lk_owner;
 } nfs_user_t;
 
 extern int
diff --git a/xlators/nfs/server/src/nfs3.c b/xlators/nfs/server/src/nfs3.c
index 527101de22b..1cb8d07e656 100644
--- a/xlators/nfs/server/src/nfs3.c
+++ b/xlators/nfs/server/src/nfs3.c
@@ -450,6 +450,8 @@ nfs3_call_state_wipe (nfs3_call_state_t *cs)
                 iobuf_unref (cs->iob);
         if (cs->iobref)
                 iobref_unref (cs->iobref);
+        if (cs->trans)
+                rpc_transport_unref (cs->trans);
         memset (cs, 0, sizeof (*cs));
         mem_put (cs);
         /* Already refd by fd_lookup, so no need to ref again. */
@@ -5483,6 +5485,7 @@ nfs3_init_state (xlator_t *nfsx)
                 goto free_localpool;
         }
 
+        nfs->nfs3state = nfs3;
         ret = 0;
 
 free_localpool:
diff --git a/xlators/nfs/server/src/nfs3.h b/xlators/nfs/server/src/nfs3.h
index 3e99184c7ad..4a595e5ef22 100644
--- a/xlators/nfs/server/src/nfs3.h
+++ b/xlators/nfs/server/src/nfs3.h
@@ -34,6 +34,7 @@
 #include "nfs-common.h"
 #include "xdr-nfs3.h"
 #include "mem-pool.h"
+#include "nlm4.h"
 
 #include <sys/statvfs.h>
 
@@ -94,7 +95,7 @@ struct nfs3_export {
 #define GF_NFS3_DEFAULT_VOLACCESS       (GF_NFS3_VOLACCESS_RW)
 
 /* The NFSv3 protocol state */
-struct nfs3_state {
+typedef struct nfs3_state {
 
         /* The NFS xlator pointer. The NFS xlator can be running
          * multiple versions of the NFS protocol.
@@ -133,13 +134,30 @@ struct nfs3_state {
         struct list_head        fdlru;
         gf_lock_t               fdlrulock;
         int                     fdcount;
-};
+} nfs3_state_t;
 
 typedef enum nfs3_lookup_type {
         GF_NFS3_REVALIDATE = 1,
         GF_NFS3_FRESH,
 } nfs3_lookup_type_t;
 
+typedef union args_ {
+        nlm4_stat nlm4_stat;
+        nlm4_holder nlm4_holder;
+        nlm4_lock nlm4_lock;
+        nlm4_share nlm4_share;
+        nlm4_testrply nlm4_testrply;
+        nlm4_testres nlm4_testres;
+        nlm4_testargs nlm4_testargs;
+        nlm4_res nlm4_res;
+        nlm4_lockargs nlm4_lockargs;
+        nlm4_cancargs nlm4_cancargs;
+        nlm4_unlockargs nlm4_unlockargs;
+        nlm4_shareargs nlm4_shareargs;
+        nlm4_shareres nlm4_shareres;
+} args;
+
+
 typedef int (*nfs3_resume_fn_t) (void *cs);
 /* Structure used to communicate state between a fop and its callback.
  * Not all members are used at all times. Usage is fop and NFS request
@@ -208,6 +226,13 @@ struct nfs3_local {
         gf_dirent_t             *hashmatch;
         gf_dirent_t             *entrymatch;
         off_t                   lastentryoffset;
+        struct flock            flock;
+        args                    args;
+        nlm4_lkowner_t          lkowner;
+        char                    cookiebytes[1024];
+        struct nfs3_fh          lockfh;
+        rpc_transport_t         *trans;
+        call_frame_t            *frame;
 };
 
 #define nfs3_is_revalidate_lookup(cst) ((cst)->lookuptype == GF_NFS3_REVALIDATE)
diff --git a/xlators/nfs/server/src/nlm4.c b/xlators/nfs/server/src/nlm4.c
new file mode 100644
index 00000000000..6e8494ce0ad
--- /dev/null
+++ b/xlators/nfs/server/src/nlm4.c
@@ -0,0 +1,1760 @@
+/*
+  Copyright (c) 2012 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3 of the License,
+  or (at your option) any later version.
+
+  GlusterFS is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see
+  <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "defaults.h"
+#include "rpcsvc.h"
+#include "dict.h"
+#include "xlator.h"
+#include "nfs.h"
+#include "mem-pool.h"
+#include "logging.h"
+#include "nfs-fops.h"
+#include "inode.h"
+#include "mount3.h"
+#include "nfs3.h"
+#include "nfs-mem-types.h"
+#include "nfs3-helpers.h"
+#include "nfs3-fh.h"
+#include "nlm4.h"
+#include "nlm4-xdr.h"
+#include "msg-nfs3.h"
+#include "nfs-generics.h"
+#include "rpc-clnt.h"
+#include "nsm-xdr.h"
+#include "nlmcbk-xdr.h"
+#include <unistd.h>
+#include <rpc/pmap_clnt.h>
+#include <rpc/rpc.h>
+#include <rpc/xdr.h>
+
+/* TODO:
+ * 1) 2 opens racing .. creating an fd leak.
+ * 2) use mempool for nlmclnt - destroy if no fd exists, create during 1st call
+ */
+
+typedef ssize_t (*nlm4_serializer) (struct iovec outmsg, void *args);
+
+extern void nfs3_call_state_wipe (nfs3_call_state_t *cs);
+
+struct list_head nlm_client_list;
+gf_lock_t nlm_client_list_lk;
+
+/* race on this is harmless */
+int nlm_grace_period = 50;
+
+#define nlm4_validate_nfs3_state(request, state, status, label, retval) \
+        do      {                                                       \
+                state = rpcsvc_request_program_private (request);       \
+                if (!state) {                                           \
+                        gf_log (GF_NLM, GF_LOG_ERROR, "NFSv3 state "    \
+                                "missing from RPC request");            \
+                        rpcsvc_request_seterr (req, SYSTEM_ERR);        \
+                        status = nlm4_failed;                           \
+                        goto label;                                     \
+                }                                                       \
+        } while (0);                                                    \
+
+nfs3_call_state_t *
+nfs3_call_state_init (struct nfs3_state *s, rpcsvc_request_t *req, xlator_t *v);
+
+#define nlm4_handle_call_state_init(nfs3state, calls, rq, opstat, errlabel)\
+        do {                                                            \
+                calls = nlm4_call_state_init ((nfs3state), (rq));       \
+                if (!calls) {                                           \
+                        gf_log (GF_NLM, GF_LOG_ERROR, "Failed to "      \
+                                "init call state");                     \
+                        opstat = nlm4_failed;                           \
+                        rpcsvc_request_seterr (req, SYSTEM_ERR);        \
+                        goto errlabel;                                  \
+                }                                                       \
+        } while (0)                                                     \
+
+#define nlm4_validate_gluster_fh(handle, status, errlabel)              \
+        do {                                                            \
+                if (!nfs3_fh_validate (handle)) {                       \
+                        status = nlm4_stale_fh;                         \
+                        goto errlabel;                                  \
+                }                                                       \
+        } while (0)                                                     \
+
+xlator_t *
+nfs3_fh_to_xlator (struct nfs3_state *nfs3, struct nfs3_fh *fh);
+
+#define nlm4_map_fh_to_volume(nfs3state, handle, rqst, volume, status, label) \
+        do {                                                            \
+                volume = nfs3_fh_to_xlator ((nfs3state), handle);       \
+                if (!volume) {                                          \
+                        gf_log (GF_NLM, GF_LOG_ERROR, "Failed to map "  \
+                                "FH to vol");                           \
+                        status = nlm4_stale_fh;                         \
+                        goto label;                                     \
+                } else {                                                \
+                        gf_log (GF_NLM, GF_LOG_TRACE, "FH to Volume: %s"\
+                                ,volume->name);                         \
+                        rpcsvc_request_set_private (req, volume);       \
+                }                                                       \
+        } while (0);                                                    \
+
+#define nlm4_volume_started_check(nfs3state, vlm, rtval, erlbl)         \
+        do {                                                            \
+              if ((!nfs_subvolume_started (nfs_state (nfs3state->nfsx), vlm))){\
+                      gf_log (GF_NLM, GF_LOG_ERROR, "Volume is disabled: %s",\
+                              vlm->name);                               \
+                      rtval = RPCSVC_ACTOR_IGNORE;                      \
+                      goto erlbl;                                       \
+              }                                                         \
+        } while (0)                                                     \
+
+#define nlm4_check_fh_resolve_status(cst, _stat, erlabl)               \
+        do {                                                            \
+                if ((cst)->resolve_ret < 0) {                           \
+                        _stat = nlm4_errno_to_nlm4stat (cst->resolve_errno);\
+                        goto erlabl;                                    \
+                }                                                       \
+        } while (0)                                                     \
+
+
+void
+nlm4_prep_nlm4_testargs (nlm4_testargs *args, struct nfs3_fh *fh,
+                         nlm4_lkowner_t *oh, char *cookiebytes)
+{
+        memset (args, 0, sizeof (*args));
+        args->alock.fh.n_bytes = (void *)fh;
+        args->alock.oh.n_bytes = (void *)oh;
+        args->cookie.n_bytes = (void *)cookiebytes;
+}
+
+void
+nlm4_prep_nlm4_lockargs (nlm4_lockargs *args, struct nfs3_fh *fh,
+                         nlm4_lkowner_t *oh, char *cookiebytes)
+{
+        memset (args, 0, sizeof (*args));
+        args->alock.fh.n_bytes = (void *)fh;
+        args->alock.oh.n_bytes = (void *)oh;
+        args->cookie.n_bytes = (void *)cookiebytes;
+}
+
+void
+nlm4_prep_nlm4_cancargs (nlm4_cancargs *args, struct nfs3_fh *fh,
+                           nlm4_lkowner_t *oh, char *cookiebytes)
+{
+        memset (args, 0, sizeof (*args));
+        args->alock.fh.n_bytes = (void *)fh;
+        args->alock.oh.n_bytes = (void *)oh;
+        args->cookie.n_bytes = (void *)cookiebytes;
+}
+
+void
+nlm4_prep_nlm4_unlockargs (nlm4_unlockargs *args, struct nfs3_fh *fh,
+                           nlm4_lkowner_t *oh, char *cookiebytes)
+{
+        memset (args, 0, sizeof (*args));
+        args->alock.fh.n_bytes = (void *)fh;
+        args->alock.oh.n_bytes = (void *)oh;
+        args->cookie.n_bytes = (void *)cookiebytes;
+}
+
+void
+nlm_copy_lkowner (gf_lkowner_t *dst, netobj *src)
+{
+        dst->len = src->n_len;
+        memcpy (dst->data, src->n_bytes, dst->len);
+}
+
+
+nfsstat3
+nlm4_errno_to_nlm4stat (int errnum)
+{
+        nlm4_stats        stat = nlm4_denied;
+
+        switch (errnum) {
+        case 0:
+                stat = nlm4_granted;
+                break;
+        case EROFS:
+                stat = nlm4_rofs;
+                break;
+        case ESTALE:
+                stat = nlm4_stale_fh;
+                break;
+        case ENOLCK:
+                stat = nlm4_failed;
+                break;
+        default:
+                stat = nlm4_denied;
+                break;
+        }
+
+        return stat;
+}
+
+nfs3_call_state_t *
+nlm4_call_state_init (struct nfs3_state *s, rpcsvc_request_t *req)
+{
+        nfs3_call_state_t       *cs = NULL;
+
+        if ((!s) || (!req))
+                return NULL;
+
+        cs = (nfs3_call_state_t *) mem_get (s->localpool);
+        if (!cs)
+                return NULL;
+
+        memset (cs, 0, sizeof (*cs));
+        INIT_LIST_HEAD (&cs->entries.list);
+        INIT_LIST_HEAD (&cs->openwait_q);
+        cs->operrno = EINVAL;
+        cs->req = req;
+        cs->nfsx = s->nfsx;
+        cs->nfs3state = s;
+
+        return cs;
+}
+
+rpc_clnt_t *
+nlm_get_rpc_clnt (char *caller_name)
+{
+        nlm_client_t *nlmclnt = NULL;
+        int nlmclnt_found = 0;
+        rpc_clnt_t *rpc_clnt = NULL;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt, &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+        if (!nlmclnt_found)
+                goto ret;
+        if (nlmclnt->rpc_clnt)
+                rpc_clnt = rpc_clnt_ref (nlmclnt->rpc_clnt);
+ret:
+        UNLOCK (&nlm_client_list_lk);
+        return rpc_clnt;
+}
+
+int
+nlm_set_rpc_clnt (rpc_clnt_t *rpc_clnt, char *caller_name)
+{
+        nlm_client_t *nlmclnt = NULL;
+        int nlmclnt_found = 0;
+        int ret = -1;
+        rpc_clnt_t *rpc_clnt_old = NULL;
+        char *old_name = NULL;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt, &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+        if (!nlmclnt_found) {
+                nlmclnt = GF_CALLOC (1, sizeof(*nlmclnt),
+                                     gf_nfs_mt_nlm4_nlmclnt);
+                if (nlmclnt == NULL) {
+                        gf_log (GF_NLM, GF_LOG_DEBUG, "malloc error");
+                        goto ret;
+                }
+
+                INIT_LIST_HEAD(&nlmclnt->fdes);
+                INIT_LIST_HEAD(&nlmclnt->nlm_clients);
+
+                list_add (&nlmclnt->nlm_clients, &nlm_client_list);
+        }
+        rpc_clnt_old = nlmclnt->rpc_clnt;
+        old_name = nlmclnt->caller_name;
+        if (rpc_clnt)
+                nlmclnt->rpc_clnt = rpc_clnt_ref (rpc_clnt);
+        nlmclnt->caller_name = gf_strdup (caller_name);
+
+        ret = 0;
+ret:
+        UNLOCK (&nlm_client_list_lk);
+        if (rpc_clnt_old)
+                rpc_clnt_unref (rpc_clnt_old);
+        if (old_name)
+                GF_FREE (old_name);
+        return ret;
+}
+
+int
+nlm_unset_rpc_clnt (rpc_clnt_t *rpc)
+{
+        nlm_client_t *nlmclnt = NULL;
+        rpc_clnt_t *rpc_clnt = NULL;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt, &nlm_client_list, nlm_clients) {
+                if (rpc == nlmclnt->rpc_clnt) {
+                        rpc_clnt = nlmclnt->rpc_clnt;
+                        nlmclnt->rpc_clnt = NULL;
+                        break;
+                }
+        }
+        UNLOCK (&nlm_client_list_lk);
+        if (rpc_clnt == NULL) {
+                return -1;
+        }
+        if (rpc_clnt)
+                rpc_clnt_unref (rpc_clnt);
+        return 0;
+}
+
+int
+nlm_add_nlmclnt (char *caller_name)
+{
+        nlm_client_t *nlmclnt = NULL;
+        int nlmclnt_found = 0;
+        int ret = -1;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt, &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+        if (!nlmclnt_found) {
+                nlmclnt = GF_CALLOC (1, sizeof(*nlmclnt),
+                                     gf_nfs_mt_nlm4_nlmclnt);
+                if (nlmclnt == NULL) {
+                        gf_log (GF_NLM, GF_LOG_DEBUG, "malloc error");
+                        goto ret;
+                }
+
+                INIT_LIST_HEAD(&nlmclnt->fdes);
+                INIT_LIST_HEAD(&nlmclnt->nlm_clients);
+
+                list_add (&nlmclnt->nlm_clients, &nlm_client_list);
+                nlmclnt->caller_name = gf_strdup (caller_name);
+        }
+        ret = 0;
+ret:
+        UNLOCK (&nlm_client_list_lk);
+        return ret;
+}
+
+int
+nlm4svc_submit_reply (rpcsvc_request_t *req, void *arg, nlm4_serializer sfunc)
+{
+        struct iovec            outmsg = {0, };
+        struct iobuf            *iob = NULL;
+        struct nfs3_state       *nfs3 = NULL;
+        int                     ret = -1;
+        struct iobref           *iobref = NULL;
+
+        if (!req)
+                return -1;
+
+        nfs3 = (struct nfs3_state *)rpcsvc_request_program_private (req);
+        if (!nfs3) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "mount state not found");
+                goto ret;
+        }
+
+        /* First, get the io buffer into which the reply in arg will
+         * be serialized.
+         */
+        iob = iobuf_get (nfs3->iobpool);
+        if (!iob) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Failed to get iobuf");
+                goto ret;
+        }
+
+        iobuf_to_iovec (iob, &outmsg);
+        /* Use the given serializer to translate the give C structure in arg
+         * to XDR format which will be written into the buffer in outmsg.
+         */
+        outmsg.iov_len = sfunc (outmsg, arg);
+
+        iobref = iobref_new ();
+        if (iobref == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Failed to get iobref");
+                goto ret;
+        }
+
+        iobref_add (iobref, iob);
+
+        /* Then, submit the message for transmission. */
+        ret = rpcsvc_submit_message (req, &outmsg, 1, NULL, 0, iobref);
+        iobuf_unref (iob);
+        iobref_unref (iobref);
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Reply submission failed");
+                goto ret;
+        }
+
+        ret = 0;
+ret:
+        return ret;
+}
+
+typedef int (*nlm4_resume_fn_t) (void *cs);
+
+int32_t
+nlm4_file_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, fd_t *fd)
+{
+        nfs3_call_state_t *cs = frame->local;
+
+        if (op_ret == 0)
+                fd_bind (cs->fd);
+        cs->resolve_ret = op_ret;
+        cs->resume_fn (cs);
+        frame->local = NULL;
+        STACK_DESTROY (frame->root);
+        return 0;
+}
+
+int nsm_monitor(char *host)
+{
+        CLIENT *clnt = NULL;
+        enum clnt_stat ret;
+        struct mon nsm_mon;
+        struct sm_stat_res res;
+        struct timeval tout = { 5, 0 };
+        int retstat = -1;
+
+        nsm_mon.mon_id.mon_name = strdup(host);
+        nsm_mon.mon_id.my_id.my_name = strdup("localhost");
+        nsm_mon.mon_id.my_id.my_prog = NLMCBK_PROGRAM;
+        nsm_mon.mon_id.my_id.my_vers = NLMCBK_V1;
+        nsm_mon.mon_id.my_id.my_proc = NLMCBK_SM_NOTIFY;
+        /* nothing to put in the private data */
+#define SM_PROG 100024
+#define SM_VERS 1
+#define SM_MON 2
+
+        /* create a connection to nsm on the localhost */
+        clnt = clnt_create("localhost", SM_PROG, SM_VERS, "tcp");
+        if(!clnt)
+        {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Clnt_create()");
+                goto out;
+        }
+
+        ret = clnt_call(clnt, SM_MON,
+                        (xdrproc_t) xdr_mon, (caddr_t) & nsm_mon,
+                        (xdrproc_t) xdr_sm_stat_res, (caddr_t) & res, tout);
+        if(ret != RPC_SUCCESS)
+        {
+                gf_log (GF_NLM, GF_LOG_ERROR, "clnt_call(): %s",
+                        clnt_sperrno(ret));
+                goto out;
+        }
+        if(res.res_stat != STAT_SUCC)
+        {
+                gf_log (GF_NLM, GF_LOG_ERROR, "clnt_call(): %s",
+                        clnt_sperrno(ret));
+                goto out;
+        }
+        retstat = 0;
+out:
+        free(nsm_mon.mon_id.mon_name);
+        free(nsm_mon.mon_id.my_id.my_name);
+        clnt_destroy(clnt);
+        return retstat;
+}
+
+nlm_client_t *
+nlm_get_uniq (char *caller_name)
+{
+        nlm_client_t *nlmclnt = NULL;
+        int nlmclnt_found = 0;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt, &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+        UNLOCK (&nlm_client_list_lk);
+        if (nlmclnt_found)
+                return nlmclnt;
+        else
+                return NULL;
+}
+
+
+int
+nlm4_file_open_and_resume(nfs3_call_state_t *cs, nlm4_resume_fn_t resume)
+{
+        fd_t *fd = NULL;
+        int ret = -1;
+        nlm_client_t *nlmclnt = NULL;
+        call_frame_t *frame = NULL;
+
+        nlmclnt = nlm_get_uniq (cs->args.nlm4_lockargs.alock.caller_name);
+        if (nlmclnt == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "nlm_get_uniq() returned NULL");
+                cs->resolve_ret = -1;
+                cs->resume_fn(cs);
+                ret = -1;
+                goto err;
+        }
+        cs->resume_fn = resume;
+        fd = fd_lookup_uint64 (cs->resolvedloc.inode, (uint64_t)nlmclnt);
+        if (fd) {
+                cs->fd = fd;
+                cs->resolve_ret = 0;
+                cs->resume_fn(cs);
+                ret = 0;
+                goto err;
+        }
+
+        fd = fd_create_uint64 (cs->resolvedloc.inode, (uint64_t)nlmclnt);
+        if (fd == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "fd_create_uint64() returned NULL");
+                cs->resolve_ret = -1;
+                cs->resume_fn(cs);
+                ret = -1;
+                goto err;
+        }
+
+        cs->fd = fd;
+
+        frame = create_frame (cs->nfsx, cs->nfsx->ctx->pool);
+        frame->root->pid = NFS_PID;
+        frame->root->uid = 0;
+        frame->root->gid = 0;
+        frame->local = cs;
+        STACK_WIND_COOKIE(frame, nlm4_file_open_cbk, cs->nfsx, cs->nfsx,
+                          cs->nfsx->fops->open, &cs->resolvedloc, O_RDWR,
+                          cs->fd, GF_OPEN_NOWB);
+        ret = 0;
+err:
+        return ret;
+}
+
+int
+nlm4_generic_reply (rpcsvc_request_t *req, netobj cookie, nlm4_stats stat)
+{
+        nlm4_res res;
+
+        memset (&res, 0, sizeof (res));
+        res.cookie = cookie;
+        res.stat.stat = stat;
+
+        nlm4svc_submit_reply (req, (void *)&res,
+                              (nlm4_serializer)xdr_serialize_nlm4_res);
+        return 0;
+}
+
+int
+nlm4svc_null (rpcsvc_request_t *req)
+{
+        struct iovec    dummyvec = {0, };
+
+        if (!req) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Got NULL request!");
+                return 0;
+        }
+        rpcsvc_submit_generic (req, &dummyvec, 1,  NULL, 0, NULL);
+        return 0;
+}
+
+int
+nlm4_gf_flock_to_holder (nlm4_holder *holder, struct gf_flock *flock)
+{
+        switch (flock->l_type) {
+        case GF_LK_F_WRLCK:
+                holder->exclusive = 1;
+                break;
+        }
+
+        holder->svid = flock->l_pid;
+        holder->l_offset = flock->l_start;
+        holder->l_len = flock->l_len;
+        return 0;
+}
+
+int
+nlm4_lock_to_gf_flock (struct gf_flock *flock, nlm4_lock *lock, int excl)
+{
+        flock->l_pid = lock->svid;
+        flock->l_start = lock->l_offset;
+        flock->l_len = lock->l_len;
+        if (excl)
+                flock->l_type = F_WRLCK;
+        else
+                flock->l_type = F_RDLCK;
+        flock->l_whence = SEEK_SET;
+        nlm_copy_lkowner (&flock->l_owner, &lock->oh);
+        return 0;
+}
+
+rpc_clnt_procedure_t nlm4_clnt_actors[NLM4_PROC_COUNT] = {
+        [NLM4_NULL] = {"NULL", NULL},
+        [NLM4_GRANTED] = {"GRANTED", NULL},
+};
+
+char *nlm4_clnt_names[NLM4_PROC_COUNT] = {
+        [NLM4_NULL] = "NULL",
+        [NLM4_GRANTED] = "GRANTED",
+};
+
+rpc_clnt_prog_t nlm4clntprog = {
+        .progname = "NLMv4",
+        .prognum = NLM_PROGRAM,
+        .progver = NLM_V4,
+        .numproc = NLM4_PROC_COUNT,
+        .proctable = nlm4_clnt_actors,
+        .procnames = nlm4_clnt_names,
+};
+
+int
+nlm4_test_reply (nfs3_call_state_t *cs, nlm4_stats stat, struct gf_flock *flock)
+{
+        nlm4_testres res;
+
+        memset (&res, 0, sizeof (res));
+        res.cookie = cs->args.nlm4_testargs.cookie;
+        res.stat.stat = stat;
+        if (stat == nlm4_denied)
+                nlm4_gf_flock_to_holder (&res.stat.nlm4_testrply_u.holder,
+                                         flock);
+
+        nlm4svc_submit_reply (cs->req, (void *)&res,
+                              (nlm4_serializer)xdr_serialize_nlm4_testres);
+        return 0;
+}
+
+int
+nlm4svc_test_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                  int32_t op_ret, int32_t op_errno, struct gf_flock *flock)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        nfs3_call_state_t              *cs = NULL;
+
+        cs = frame->local;
+        if (op_ret == -1) {
+                stat = nlm4_errno_to_nlm4stat (op_errno);
+                goto err;
+        } else if (flock->l_type == F_UNLCK)
+                stat = nlm4_granted;
+
+err:
+        nlm4_test_reply (cs, stat, flock);
+        nfs3_call_state_wipe (cs);
+        return 0;
+}
+
+int
+nlm4_test_fd_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        int                             ret = -EFAULT;
+        nfs_user_t                      nfu = {0, };
+        nfs3_call_state_t               *cs = NULL;
+        struct gf_flock                 flock = {0, };
+
+        if (!carg)
+                return ret;
+
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+        nfs_request_user_init (&nfu, cs->req);
+        nlm4_lock_to_gf_flock (&flock, &cs->args.nlm4_testargs.alock,
+                               cs->args.nlm4_testargs.exclusive);
+        nlm_copy_lkowner (&nfu.lk_owner, &cs->args.nlm4_testargs.alock.oh);
+        ret = nfs_lk (cs->nfsx, cs->vol, &nfu, cs->fd, F_GETLK, &flock,
+                      nlm4svc_test_cbk, cs);
+        if (ret < 0)
+                stat = nlm4_errno_to_nlm4stat (-ret);
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Unable to call lk()");
+                nlm4_test_reply (cs, stat, &flock);
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+
+int
+nlm4_test_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_failed;
+        int                             ret = -1;
+        nfs3_call_state_t               *cs = NULL;
+        fd_t                            *fd = NULL;
+
+        if (!carg)
+                return ret;
+
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+        fd = fd_anonymous (cs->resolvedloc.inode);
+        cs->fd = fd;
+        ret = nlm4_test_fd_resume (cs);
+        if (ret < 0)
+                stat = nlm4_errno_to_nlm4stat (-ret);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to open_and_resume");
+                nlm4_test_reply (cs, stat, NULL);
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+int
+nlm4svc_test (rpcsvc_request_t *req)
+{
+        xlator_t                        *vol = NULL;
+        nlm4_stats                      stat = nlm4_failed;
+        struct nfs_state               *nfs = NULL;
+        nfs3_state_t                   *nfs3 = NULL;
+        nfs3_call_state_t               *cs = NULL;
+        int                             ret = RPCSVC_ACTOR_ERROR;
+        struct nfs3_fh                  fh = {{0}, };
+
+        if (!req)
+                return ret;
+
+        nlm4_validate_nfs3_state (req, nfs3, stat, rpcerr, ret);
+        nfs = nfs_state (nfs3->nfsx);
+        nlm4_handle_call_state_init (nfs->nfs3state, cs, req,
+                                     stat, rpcerr);
+
+        nlm4_prep_nlm4_testargs (&cs->args.nlm4_testargs, &fh, &cs->lkowner,
+                                 cs->cookiebytes);
+        if (xdr_to_nlm4_testargs(req->msg[0], &cs->args.nlm4_testargs) <= 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Error decoding args");
+                rpcsvc_request_seterr (req, GARBAGE_ARGS);
+                goto rpcerr;
+        }
+
+        nlm4_validate_gluster_fh (&fh, stat, nlm4err);
+        nlm4_map_fh_to_volume (cs->nfs3state, &fh, req, vol, stat, nlm4err);
+
+        if (nlm_grace_period) {
+                gf_log (GF_NLM, GF_LOG_WARNING, "NLM in grace period");
+                stat = nlm4_denied_grace_period;
+                nlm4_test_reply (cs, stat, NULL);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+        cs->vol = vol;
+        nlm4_volume_started_check (nfs3, vol, ret, rpcerr);
+
+        ret = nfs3_fh_resolve_and_resume (cs, &fh,
+                                          NULL, nlm4_test_resume);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to resolve and resume");
+                nlm4_test_reply (cs, stat, NULL);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+rpcerr:
+        if (ret < 0) {
+                nfs3_call_state_wipe (cs);
+        }
+        return ret;
+}
+
+int
+nlm4svc_send_granted_cbk (struct rpc_req *req, struct iovec *iov, int count,
+                          void *myframe)
+{
+        STACK_DESTROY (((call_frame_t*)myframe)->root);
+        return 0;
+}
+
+int nlm_rpcclnt_notify (struct rpc_clnt *rpc, void *mydata,
+                        rpc_clnt_event_t fn, void *data)
+{
+        nlm_condmutex_t *cm = NULL;
+        int ret;
+        cm = mydata;
+        switch (fn) {
+        case RPC_CLNT_CONNECT:
+                ret = pthread_cond_broadcast (&cm->cond);
+                if (ret!=0)
+                        gf_log (GF_NLM, GF_LOG_ERROR, "cond_broadcast error %s",
+                                strerror (errno));
+                break;
+        case RPC_CLNT_MSG:
+                break;
+        case RPC_CLNT_DISCONNECT:
+                nlm_unset_rpc_clnt(rpc);
+                break;
+        }
+        return 0;
+}
+
+void
+nlm4svc_send_granted (nfs3_call_state_t *cs);
+
+void *
+nlm4_establish_callback (void *csarg)
+{
+        nfs3_call_state_t *cs = NULL;
+        struct sockaddr_storage sa;
+        struct sockaddr *sockaddr = NULL;
+        dict_t *options = NULL;
+        char peerip[INET6_ADDRSTRLEN+1], *portstr = NULL;
+        rpc_clnt_t *rpc_clnt = NULL;
+        int port = -1;
+        int ret = -1;
+        char *caller_name = NULL;
+
+        cs = (nfs3_call_state_t *) csarg;
+
+        caller_name = cs->args.nlm4_lockargs.alock.caller_name;
+        nsm_monitor (caller_name);
+
+        rpc_transport_get_peeraddr (cs->trans, NULL, 0, &sa, sizeof (sa));
+        sockaddr = (struct sockaddr*) &sa;
+        switch (sockaddr->sa_family) {
+        case AF_INET6:
+                inet_ntop (AF_INET6,
+                           &((struct sockaddr_in6 *)sockaddr)->sin6_addr,
+                           peerip, INET6_ADDRSTRLEN+1);
+                break;
+        case AF_INET:
+                inet_ntop (AF_INET,
+                           &((struct sockaddr_in *)sockaddr)->sin_addr,
+                           peerip, INET6_ADDRSTRLEN+1);
+        default:
+                break;
+                /* FIXME: handle the error */
+        }
+
+        /* looks like libc rpc supports only ipv4 */
+        port = pmap_getport ((struct sockaddr_in*)sockaddr, NLM_PROGRAM,
+                             NLM_V4, IPPROTO_TCP);
+
+        options = dict_new();
+        ret = dict_set_str (options, "transport-type", "socket");
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_str error");
+                goto err;
+        }
+
+        ret = dict_set_dynstr (options, "remote-host", strdup (peerip));
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_str error");
+                goto err;
+        }
+
+        ret = gf_asprintf (&portstr, "%d", port);
+        if (ret == -1)
+                goto err;
+
+        ret = dict_set_dynstr (options, "remote-port",
+                               portstr);
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_dynstr error");
+                goto err;
+        }
+        ret = dict_set_str (options, "auth-null", "on");
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_dynstr error");
+                goto err;
+        }
+
+        rpc_clnt = rpc_clnt_new (options, cs->nfsx->ctx, "NLM-client");
+        if (rpc_clnt == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "rpc_clnt NULL");
+                goto err;
+        }
+        nlm_condmutex_t *cm;
+        cm = GF_CALLOC (1, sizeof(*cm), gf_nfs_mt_nlm4_cm);
+        pthread_mutex_init (&cm->mutex, NULL);
+        pthread_cond_init (&cm->cond, NULL);
+        ret = rpc_clnt_register_notify (rpc_clnt, nlm_rpcclnt_notify,
+                                        cm);
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR,"rpc_clnt_register_connect error");
+                goto err;
+        }
+        ret = rpc_transport_connect (rpc_clnt->conn.trans, port);
+        pthread_cond_wait (&cm->cond, &cm->mutex);
+        pthread_mutex_destroy (&cm->mutex);
+        GF_FREE (cm);
+        rpc_clnt_set_connected (&rpc_clnt->conn);
+        ret = nlm_set_rpc_clnt (rpc_clnt, caller_name);
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_ptr error");
+                goto err;
+        }
+        nlm4svc_send_granted (cs);
+err:
+        rpc_clnt_unref (rpc_clnt);
+        return NULL;
+}
+
+void
+nlm4svc_send_granted (nfs3_call_state_t *cs)
+{
+        int ret = -1;
+        rpc_clnt_t *rpc_clnt = NULL;
+        struct iovec            outmsg = {0, };
+        nlm4_testargs testargs;
+        struct iobuf *iobuf = NULL;
+        struct iobref *iobref = NULL;
+        struct nfs_state *nfs = NULL;
+        char peerip[INET6_ADDRSTRLEN+1];
+        pthread_t thr;
+        struct sockaddr_storage sa;
+        struct sockaddr *sockaddr = NULL;
+
+        nfs = cs->nfsx->private;
+
+        rpc_transport_get_peeraddr (cs->trans, NULL, 0, &sa, sizeof (sa));
+        sockaddr = (struct sockaddr*) &sa;
+        switch (sockaddr->sa_family) {
+        case AF_INET6:
+                inet_ntop (AF_INET6,
+                           &((struct sockaddr_in6 *)sockaddr)->sin6_addr,
+                           peerip, INET6_ADDRSTRLEN+1);
+                break;
+        case AF_INET:
+                inet_ntop (AF_INET,
+                           &((struct sockaddr_in *)sockaddr)->sin_addr,
+                           peerip, INET6_ADDRSTRLEN+1);
+        default:
+                break;
+                /* FIXME: handle the error */
+        }
+
+        rpc_clnt = nlm_get_rpc_clnt (cs->args.nlm4_lockargs.alock.caller_name);
+        if (rpc_clnt == NULL) {
+                pthread_create (&thr, NULL, nlm4_establish_callback, (void*)cs);
+                return;
+        }
+
+        testargs.cookie = cs->args.nlm4_lockargs.cookie;
+        testargs.exclusive = cs->args.nlm4_lockargs.exclusive;
+        testargs.alock = cs->args.nlm4_lockargs.alock;
+
+        iobuf = iobuf_get (cs->nfs3state->iobpool);
+        if (!iobuf) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Failed to get iobuf");
+                goto ret;
+        }
+
+        iobuf_to_iovec (iobuf, &outmsg);
+        /* Use the given serializer to translate the give C structure in arg
+         * to XDR format which will be written into the buffer in outmsg.
+         */
+        outmsg.iov_len = xdr_serialize_nlm4_testargs (outmsg, &testargs);
+
+        iobref = iobref_new ();
+        if (iobref == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Failed to get iobref");
+                goto ret;
+        }
+
+        iobref_add (iobref, iobuf);
+
+        ret = rpc_clnt_submit (rpc_clnt, &nlm4clntprog, NLM4_GRANTED,
+                               nlm4svc_send_granted_cbk,
+                               &outmsg, 1,
+                               NULL,
+                               0, iobref, cs->frame, NULL, 0,
+                               NULL, 0, NULL);
+
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "rpc_clnt_submit error");
+                goto ret;
+        }
+ret:
+        rpc_clnt_unref (rpc_clnt);
+        nfs3_call_state_wipe (cs);
+        return;
+}
+
+int
+nlm_cleanup_fds (char *caller_name)
+{
+        int nlmclnt_found = 0;
+        nlm_fde_t *fde = NULL, *tmp = NULL;
+        nlm_client_t *nlmclnt = NULL;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt,
+                             &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+
+        if (!nlmclnt_found)
+                goto ret;
+
+        if (list_empty (&nlmclnt->fdes))
+                goto ret;
+
+        list_for_each_entry_safe (fde, tmp, &nlmclnt->fdes, fde_list) {
+                fd_unref (fde->fd);
+                list_del (&fde->fde_list);
+                GF_FREE (fde);
+        }
+
+ret:
+        UNLOCK (&nlm_client_list_lk);
+        return 0;
+}
+
+void
+nlm_search_and_delete (fd_t *fd, char *caller_name)
+{
+        nlm_fde_t *fde = NULL;
+        nlm_client_t *nlmclnt = NULL;
+        int nlmclnt_found = 0;
+        int fde_found = 0;
+        int transit_cnt = 0;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt,
+                             &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+
+        if (!nlmclnt_found)
+                goto ret;
+
+        list_for_each_entry (fde, &nlmclnt->fdes, fde_list) {
+                if (fde->fd == fd) {
+                        fde_found = 1;
+                        break;
+                }
+        }
+
+        if (!fde_found)
+                goto ret;
+        transit_cnt = fde->transit_cnt;
+        if (transit_cnt)
+                goto ret;
+        list_del (&fde->fde_list);
+
+ret:
+        UNLOCK (&nlm_client_list_lk);
+
+        if (fde_found && !transit_cnt) {
+                fd_unref (fde->fd);
+                GF_FREE (fde);
+        }
+        return;
+}
+
+int
+nlm_dec_transit_count (fd_t *fd, char *caller_name)
+{
+        nlm_fde_t *fde = NULL;
+        nlm_client_t *nlmclnt = NULL;
+        int nlmclnt_found = 0;
+        int fde_found = 0;
+        int transit_cnt = -1;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt,
+                             &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+
+        if (!nlmclnt_found) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "nlmclnt not found");
+                nlmclnt = NULL;
+                goto ret;
+        }
+
+        list_for_each_entry (fde, &nlmclnt->fdes, fde_list) {
+                if (fde->fd == fd) {
+                        fde_found = 1;
+                        break;
+                }
+        }
+
+        if (fde_found) {
+                transit_cnt = --fde->transit_cnt;
+                goto ret;
+        }
+ret:
+
+        UNLOCK (&nlm_client_list_lk);
+        return transit_cnt;
+}
+
+
+nlm_client_t *
+nlm_search_and_add (fd_t *fd, char *caller_name)
+{
+        nlm_fde_t *fde = NULL;
+        nlm_client_t *nlmclnt = NULL;
+        int nlmclnt_found = 0;
+        int fde_found = 0;
+
+        LOCK (&nlm_client_list_lk);
+        list_for_each_entry (nlmclnt,
+                             &nlm_client_list, nlm_clients) {
+                if (!strcmp(caller_name, nlmclnt->caller_name)) {
+                        nlmclnt_found = 1;
+                        break;
+                }
+        }
+
+        if (!nlmclnt_found) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "nlmclnt not found");
+                nlmclnt = NULL;
+                goto ret;
+        }
+
+        list_for_each_entry (fde, &nlmclnt->fdes, fde_list) {
+                if (fde->fd == fd) {
+                        fde_found = 1;
+                        break;
+                }
+        }
+
+        if (fde_found)
+                goto ret;
+
+        fde = GF_CALLOC (1, sizeof (*fde), gf_nfs_mt_nlm4_fde);
+
+        fde->fd = fd_ref (fd);
+        list_add (&fde->fde_list, &nlmclnt->fdes);
+ret:
+        if (nlmclnt_found && fde)
+                fde->transit_cnt++;
+        UNLOCK (&nlm_client_list_lk);
+        return nlmclnt;
+}
+
+int
+nlm4svc_lock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                  int32_t op_ret, int32_t op_errno, struct gf_flock *flock)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        nfs3_call_state_t              *cs = NULL;
+        int                             transit_cnt = -1;
+
+        cs = frame->local;
+
+        transit_cnt = nlm_dec_transit_count (cs->fd,
+                                             cs->args.nlm4_lockargs.alock.caller_name);
+        if (op_ret == -1) {
+                if (transit_cnt == 0)
+                        nlm_search_and_delete (cs->fd,
+                                               cs->args.nlm4_lockargs.alock.caller_name);
+                stat = nlm4_errno_to_nlm4stat (op_errno);
+                goto err;
+        } else
+                stat = nlm4_granted;
+
+err:
+        if (cs->args.nlm4_lockargs.block) {
+                cs->frame = copy_frame (frame);
+                nlm4svc_send_granted (cs);
+        } else {
+                nlm4_generic_reply (cs->req, cs->args.nlm4_lockargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+        }
+        return 0;
+}
+
+int
+nlm4_lock_fd_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        int                             ret = -EFAULT;
+        nfs_user_t                      nfu = {0, };
+        nfs3_call_state_t               *cs = NULL;
+        struct gf_flock                 flock = {0, };
+        nlm_client_t                    *nlmclnt = NULL;
+
+        if (!carg)
+                return ret;
+
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+
+        nlmclnt = nlm_search_and_add (cs->fd,
+                                      cs->args.nlm4_lockargs.alock.caller_name);
+        nfs_request_user_init (&nfu, cs->req);
+        nlm4_lock_to_gf_flock (&flock, &cs->args.nlm4_lockargs.alock,
+                               cs->args.nlm4_lockargs.exclusive);
+        nlm_copy_lkowner (&nfu.lk_owner, &cs->args.nlm4_lockargs.alock.oh);
+        if (cs->args.nlm4_lockargs.block) {
+                nlm4_generic_reply (cs->req, cs->args.nlm4_lockargs.cookie,
+                                    nlm4_blocked);
+                ret = nfs_lk (cs->nfsx, cs->vol, &nfu, cs->fd, F_SETLKW,
+                              &flock, nlm4svc_lock_cbk, cs);
+        } else
+                ret = nfs_lk (cs->nfsx, cs->vol, &nfu, cs->fd, F_SETLK,
+                              &flock, nlm4svc_lock_cbk, cs);
+
+        if (ret < 0)
+                stat = nlm4_errno_to_nlm4stat (-ret);
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to call lk()");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_lockargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+
+int
+nlm4_lock_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_failed;
+        int                             ret = -1;
+        nfs3_call_state_t               *cs = NULL;
+
+        if (!carg)
+                return ret;
+
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+        ret = nlm4_file_open_and_resume (cs, nlm4_lock_fd_resume);
+        if (ret < 0)
+                stat = nlm4_errno_to_nlm4stat (-ret);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to open and resume");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_lockargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+
+int
+nlm4svc_lock (rpcsvc_request_t *req)
+{
+        xlator_t                        *vol = NULL;
+        nlm4_stats                      stat = nlm4_failed;
+        struct nfs_state               *nfs = NULL;
+        nfs3_state_t                   *nfs3 = NULL;
+        nfs3_call_state_t               *cs = NULL;
+        int                             ret = RPCSVC_ACTOR_ERROR;
+        struct nfs3_fh                  fh = {{0}, };
+
+        if (!req)
+                return ret;
+
+        nlm4_validate_nfs3_state (req, nfs3, stat, rpcerr, ret);
+        nfs = nfs_state (nfs3->nfsx);
+        nlm4_handle_call_state_init (nfs->nfs3state, cs, req,
+                                     stat, rpcerr);
+
+        nlm4_prep_nlm4_lockargs (&cs->args.nlm4_lockargs, &cs->lockfh,
+                                 &cs->lkowner, cs->cookiebytes);
+        if (xdr_to_nlm4_lockargs(req->msg[0], &cs->args.nlm4_lockargs) <= 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Error decoding args");
+                rpcsvc_request_seterr (req, GARBAGE_ARGS);
+                goto rpcerr;
+        }
+        fh = cs->lockfh;
+        nlm4_validate_gluster_fh (&fh, stat, nlm4err);
+        nlm4_map_fh_to_volume (cs->nfs3state, &fh, req, vol, stat, nlm4err);
+
+        if (nlm_grace_period && !cs->args.nlm4_lockargs.reclaim) {
+                gf_log (GF_NLM, GF_LOG_WARNING, "NLM in grace period");
+                stat = nlm4_denied_grace_period;
+                nlm4_generic_reply (req, cs->args.nlm4_unlockargs.cookie, stat);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+        cs->vol = vol;
+        cs->trans = rpcsvc_request_transport_ref(req);
+        nlm4_volume_started_check (nfs3, vol, ret, rpcerr);
+
+        ret = nlm_add_nlmclnt (cs->args.nlm4_lockargs.alock.caller_name);
+
+        ret = nfs3_fh_resolve_and_resume (cs, &fh,
+                                          NULL, nlm4_lock_resume);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to resolve and resume");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_lockargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+rpcerr:
+        if (ret < 0) {
+                nfs3_call_state_wipe (cs);
+        }
+        return ret;
+}
+
+int
+nlm4svc_cancel_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, struct gf_flock *flock)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        nfs3_call_state_t              *cs = NULL;
+
+        cs = frame->local;
+        if (op_ret == -1) {
+                stat = nlm4_errno_to_nlm4stat (op_errno);
+                goto err;
+        } else
+                stat = nlm4_granted;
+
+err:
+        nlm4_generic_reply (cs->req, cs->args.nlm4_cancargs.cookie,
+                            stat);
+        nfs3_call_state_wipe (cs);
+        return 0;
+}
+
+int
+nlm4_cancel_fd_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        int                             ret = -EFAULT;
+        nfs_user_t                      nfu = {0, };
+        nfs3_call_state_t               *cs = NULL;
+        struct gf_flock                 flock = {0, };
+
+        if (!carg)
+                return ret;
+
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+        nfs_request_user_init (&nfu, cs->req);
+        nlm4_lock_to_gf_flock (&flock, &cs->args.nlm4_cancargs.alock,
+                               cs->args.nlm4_cancargs.exclusive);
+        nlm_copy_lkowner (&nfu.lk_owner, &cs->args.nlm4_cancargs.alock.oh);
+        flock.l_type = F_UNLCK;
+        ret = nfs_lk (cs->nfsx, cs->vol, &nfu, cs->fd, F_SETLK,
+                      &flock, nlm4svc_cancel_cbk, cs);
+
+        if (ret < 0)
+                stat = nlm4_errno_to_nlm4stat (-ret);
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to call lk()");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_cancargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+int
+nlm4svc_unlock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, struct gf_flock *flock)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        nfs3_call_state_t              *cs = NULL;
+
+        cs = frame->local;
+        if (op_ret == -1) {
+                stat = nlm4_errno_to_nlm4stat (op_errno);
+                goto err;
+        } else {
+                stat = nlm4_granted;
+                if (flock->l_type == F_UNLCK)
+                        nlm_search_and_delete (cs->fd,
+                                               cs->args.nlm4_unlockargs.alock.caller_name);
+        }
+
+err:
+        nlm4_generic_reply (cs->req, cs->args.nlm4_unlockargs.cookie, stat);
+        nfs3_call_state_wipe (cs);
+        return 0;
+}
+
+int
+nlm4_unlock_fd_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_denied;
+        int                             ret = -EFAULT;
+        nfs_user_t                      nfu = {0, };
+        nfs3_call_state_t               *cs = NULL;
+        struct gf_flock                 flock = {0, };
+
+        if (!carg)
+                return ret;
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+        nfs_request_user_init (&nfu, cs->req);
+        nlm4_lock_to_gf_flock (&flock, &cs->args.nlm4_unlockargs.alock, 0);
+        nlm_copy_lkowner (&nfu.lk_owner, &cs->args.nlm4_unlockargs.alock.oh);
+        flock.l_type = F_UNLCK;
+        ret = nfs_lk (cs->nfsx, cs->vol, &nfu, cs->fd, F_SETLK,
+                      &flock, nlm4svc_unlock_cbk, cs);
+
+        if (ret < 0)
+                stat = nlm4_errno_to_nlm4stat (-ret);
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to call lk()");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_unlockargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+int
+nlm4_cancel_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_failed;
+        int                             ret = -1;
+        nfs3_call_state_t               *cs = NULL;
+        nlm_client_t                    *nlmclnt = NULL;
+
+        if (!carg)
+                return ret;
+
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+
+        nlmclnt = nlm_get_uniq (cs->args.nlm4_cancargs.alock.caller_name);
+        if (nlmclnt == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "nlm_get_uniq() returned NULL");
+                goto nlm4err;
+        }
+        cs->fd = fd_lookup_uint64 (cs->resolvedloc.inode, (uint64_t)nlmclnt);
+        if (cs->fd == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "nlm_get_uniq() returned NULL");
+                goto nlm4err;
+        }
+        ret = nlm4_unlock_fd_resume (cs);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to unlock_fd_resume()");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_cancargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+int
+nlm4svc_cancel (rpcsvc_request_t *req)
+{
+        xlator_t                        *vol = NULL;
+        nlm4_stats                      stat = nlm4_failed;
+        struct nfs_state               *nfs = NULL;
+        nfs3_state_t                   *nfs3 = NULL;
+        nfs3_call_state_t               *cs = NULL;
+        int                             ret = RPCSVC_ACTOR_ERROR;
+        struct nfs3_fh                  fh = {{0}, };
+
+        if (!req)
+                return ret;
+
+        nlm4_validate_nfs3_state (req, nfs3, stat, rpcerr, ret);
+        nfs = nfs_state (nfs3->nfsx);
+        nlm4_handle_call_state_init (nfs->nfs3state, cs, req,
+                                     stat, rpcerr);
+
+        nlm4_prep_nlm4_cancargs (&cs->args.nlm4_cancargs, &fh, &cs->lkowner,
+                                 cs->cookiebytes);
+        if (xdr_to_nlm4_cancelargs(req->msg[0], &cs->args.nlm4_cancargs) <= 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Error decoding args");
+                rpcsvc_request_seterr (req, GARBAGE_ARGS);
+                goto rpcerr;
+        }
+
+        nlm4_validate_gluster_fh (&fh, stat, nlm4err);
+        nlm4_map_fh_to_volume (cs->nfs3state, &fh, req, vol, stat, nlm4err);
+
+        if (nlm_grace_period) {
+                gf_log (GF_NLM, GF_LOG_WARNING, "NLM in grace period");
+                stat = nlm4_denied_grace_period;
+                nlm4_generic_reply (req, cs->args.nlm4_unlockargs.cookie, stat);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+        cs->vol = vol;
+        nlm4_volume_started_check (nfs3, vol, ret, rpcerr);
+
+        ret = nfs3_fh_resolve_and_resume (cs, &fh,
+                                          NULL, nlm4_cancel_resume);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to resolve and resume");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_cancargs.cookie,
+                                    stat);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+rpcerr:
+        if (ret < 0) {
+                nfs3_call_state_wipe (cs);
+        }
+        return ret;
+}
+
+int
+nlm4_unlock_resume (void *carg)
+{
+        nlm4_stats                      stat = nlm4_failed;
+        int                             ret = -1;
+        nfs3_call_state_t               *cs = NULL;
+        nlm_client_t                    *nlmclnt = NULL;
+
+        if (!carg)
+                return ret;
+
+        cs = (nfs3_call_state_t *)carg;
+        nlm4_check_fh_resolve_status (cs, stat, nlm4err);
+
+        nlmclnt = nlm_get_uniq (cs->args.nlm4_unlockargs.alock.caller_name);
+        if (nlmclnt == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "nlm_get_uniq() returned NULL");
+                goto nlm4err;
+        }
+        cs->fd = fd_lookup_uint64 (cs->resolvedloc.inode, (uint64_t)nlmclnt);
+        if (cs->fd == NULL) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "fd_lookup_uint64() returned NULL");
+                goto nlm4err;
+        }
+        ret = nlm4_unlock_fd_resume (cs);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to unlock_fd_resume");
+                nlm4_generic_reply (cs->req, cs->args.nlm4_unlockargs.cookie,
+                                    stat);
+
+                nfs3_call_state_wipe (cs);
+        }
+
+        return ret;
+}
+
+int
+nlm4svc_unlock (rpcsvc_request_t *req)
+{
+        xlator_t                        *vol = NULL;
+        nlm4_stats                      stat = nlm4_failed;
+        struct nfs_state               *nfs = NULL;
+        nfs3_state_t                   *nfs3 = NULL;
+        nfs3_call_state_t               *cs = NULL;
+        int                             ret = RPCSVC_ACTOR_ERROR;
+        struct nfs3_fh                  fh = {{0}, };
+
+        if (!req)
+                return ret;
+
+        nlm4_validate_nfs3_state (req, nfs3, stat, rpcerr, ret);
+        nfs = nfs_state (nfs3->nfsx);
+        nlm4_handle_call_state_init (nfs->nfs3state, cs, req,
+                                     stat, rpcerr);
+
+        nlm4_prep_nlm4_unlockargs (&cs->args.nlm4_unlockargs, &fh, &cs->lkowner,
+                                   cs->cookiebytes);
+        if (xdr_to_nlm4_unlockargs(req->msg[0], &cs->args.nlm4_unlockargs) <= 0)
+        {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Error decoding args");
+                rpcsvc_request_seterr (req, GARBAGE_ARGS);
+                goto rpcerr;
+        }
+
+        nlm4_validate_gluster_fh (&fh, stat, nlm4err);
+        nlm4_map_fh_to_volume (cs->nfs3state, &fh, req, vol, stat, nlm4err);
+
+        if (nlm_grace_period) {
+                gf_log (GF_NLM, GF_LOG_WARNING, "NLM in grace period");
+                stat = nlm4_denied_grace_period;
+                nlm4_generic_reply (req, cs->args.nlm4_unlockargs.cookie, stat);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+        cs->vol = vol;
+        cs->trans = rpcsvc_request_transport_ref(req);
+        nlm4_volume_started_check (nfs3, vol, ret, rpcerr);
+
+        ret = nfs3_fh_resolve_and_resume (cs, &fh,
+                                          NULL, nlm4_unlock_resume);
+
+nlm4err:
+        if (ret < 0) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "unable to resolve and resume");
+                nlm4_generic_reply (req, cs->args.nlm4_unlockargs.cookie, stat);
+                nfs3_call_state_wipe (cs);
+                return 0;
+        }
+
+rpcerr:
+        if (ret < 0) {
+                nfs3_call_state_wipe (cs);
+        }
+        return ret;
+}
+
+void
+nlm4svc_sm_notify (struct nlm_sm_status *status)
+{
+        gf_log (GF_NLM, GF_LOG_INFO, "sm_notify: %s, state: %d",
+                status->mon_name,
+                status->state);
+        nlm_cleanup_fds (status->mon_name);
+}
+
+rpcsvc_actor_t  nlm4svc_actors[NLM4_PROC_COUNT] = {
+        {"NULL", NLM4_NULL, nlm4svc_null, NULL, NULL},
+        {"TEST", NLM4_TEST, nlm4svc_test, NULL, NULL},
+        {"LOCK", NLM4_LOCK, nlm4svc_lock, NULL, NULL},
+        {"CANCEL", NLM4_CANCEL, nlm4svc_cancel, NULL, NULL},
+        {"UNLOCK", NLM4_UNLOCK, nlm4svc_unlock, NULL, NULL},
+        {"GRANTED", NLM4_GRANTED, NULL, NULL, NULL},
+        {"TEST", NLM4_TEST_MSG, NULL, NULL, NULL},
+        {"LOCK", NLM4_LOCK_MSG, NULL, NULL, NULL},
+        {"CANCEL", NLM4_CANCEL_MSG, NULL, NULL, NULL},
+        {"UNLOCK", NLM4_UNLOCK_MSG, NULL, NULL, NULL},
+        {"GRANTED", NLM4_GRANTED_MSG, NULL, NULL, NULL},
+        {"TEST", NLM4_TEST_RES, NULL, NULL, NULL},
+        {"LOCK", NLM4_LOCK_RES, NULL, NULL, NULL},
+        {"CANCEL", NLM4_CANCEL_RES, NULL, NULL, NULL},
+        {"UNLOCK", NLM4_UNLOCK_RES, NULL, NULL, NULL},
+        {"GRANTED", NLM4_GRANTED_RES, NULL, NULL, NULL},
+        {"SM_NOTIFY", NLM4_SM_NOTIFY, NULL, NULL, NULL},
+};
+
+rpcsvc_program_t        nlm4prog = {
+        .progname       = "NLM4",
+        .prognum        = NLM_PROGRAM,
+        .progver        = NLM_V4,
+        .progport       = GF_NLM4_PORT,
+        .actors         = nlm4svc_actors,
+        .numactors      = NLM4_PROC_COUNT,
+        .min_auth       = AUTH_NULL,
+};
+
+
+int
+nlm4_init_state (xlator_t *nfsx)
+{
+        return 0;
+}
+
+extern void *nsm_thread (void *argv);
+
+void nlm_grace_period_over(void *arg)
+{
+        nlm_grace_period = 0;
+}
+
+rpcsvc_program_t *
+nlm4svc_init(xlator_t *nfsx)
+{
+        struct nfs3_state *ns = NULL;
+        struct nfs_state *nfs = NULL;
+        dict_t *options = NULL;
+        int ret = -1;
+        char *portstr = NULL;
+        pthread_t thr;
+        struct timeval timeout = {0,};
+
+        nfs = (struct nfs_state*)nfsx->private;
+
+        ns = nfs->nfs3state;
+        if (!ns) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "NLM4 init failed");
+                goto err;
+        }
+        nlm4prog.private = ns;
+
+        options = dict_new ();
+
+        ret = gf_asprintf (&portstr, "%d", GF_NLM4_PORT);
+        if (ret == -1)
+                goto err;
+
+        ret = dict_set_dynstr (options, "transport.socket.listen-port",
+                               portstr);
+        if (ret == -1)
+                goto err;
+        ret = dict_set_str (options, "transport-type", "socket");
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_str error");
+                goto err;
+        }
+
+        if (nfs->allow_insecure) {
+                ret = dict_set_str (options, "rpc-auth-allow-insecure", "on");
+                if (ret == -1) {
+                        gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_str error");
+                        goto err;
+                }
+                ret = dict_set_str (options, "rpc-auth.ports.insecure", "on");
+                if (ret == -1) {
+                        gf_log (GF_NLM, GF_LOG_ERROR, "dict_set_str error");
+                        goto err;
+                }
+        }
+
+        rpcsvc_create_listeners (nfs->rpcsvc, options, "NLM");
+        if (ret == -1) {
+                gf_log (GF_NLM, GF_LOG_ERROR, "Unable to create listeners");
+                dict_unref (options);
+                goto err;
+        }
+        INIT_LIST_HEAD(&nlm_client_list);
+        LOCK_INIT (&nlm_client_list_lk);
+
+        pthread_create (&thr, NULL, nsm_thread, (void*)NULL);
+
+        timeout.tv_sec = nlm_grace_period;
+        gf_timer_call_after (nfsx->ctx, timeout, nlm_grace_period_over, NULL);
+        return &nlm4prog;
+err:
+        return NULL;
+}
diff --git a/xlators/nfs/server/src/nlm4.h b/xlators/nfs/server/src/nlm4.h
new file mode 100644
index 00000000000..3c00186d971
--- /dev/null
+++ b/xlators/nfs/server/src/nlm4.h
@@ -0,0 +1,78 @@
+/*
+  Copyright (c) 2012 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3 of the License,
+  or (at your option) any later version.
+
+  GlusterFS is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see
+  <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _NLM4_H_
+#define _NLM4_H_
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "rpcsvc.h"
+#include "dict.h"
+#include "xlator.h"
+#include "iobuf.h"
+#include "nfs.h"
+#include "list.h"
+#include "xdr-nfs3.h"
+#include "locking.h"
+#include "nfs3-fh.h"
+#include "uuid.h"
+#include "nlm4-xdr.h"
+#include "lkowner.h"
+
+/* Registered with portmap */
+#define GF_NLM4_PORT            38468
+#define GF_NLM                  GF_NFS"-NLM"
+
+extern rpcsvc_program_t *
+nlm4svc_init (xlator_t *nfsx);
+
+extern int
+nlm4_init_state (xlator_t *nfsx);
+
+#define NLM_PROGRAM 100021
+#define NLM_V4 4
+
+typedef struct nlm4_lwowner {
+        char temp[1024];
+} nlm4_lkowner_t;
+
+typedef struct nlm_client {
+        struct sockaddr_storage sa;
+        pid_t uniq;
+        struct list_head nlm_clients;
+        struct list_head fdes;
+        struct rpc_clnt *rpc_clnt;
+        char *caller_name;
+} nlm_client_t;
+
+typedef struct nlm_fde {
+        struct list_head fde_list;
+        fd_t *fd;
+        int transit_cnt;
+} nlm_fde_t;
+
+typedef struct {
+        pthread_cond_t cond;
+        pthread_mutex_t mutex;
+} nlm_condmutex_t;
+
+#endif
diff --git a/xlators/nfs/server/src/nlmcbk_svc.c b/xlators/nfs/server/src/nlmcbk_svc.c
new file mode 100644
index 00000000000..5401dc39b96
--- /dev/null
+++ b/xlators/nfs/server/src/nlmcbk_svc.c
@@ -0,0 +1,122 @@
+/*
+  Copyright (c) 2012 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published
+  by the Free Software Foundation; either version 3 of the License,
+  or (at your option) any later version.
+
+  GlusterFS is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see
+  <http://www.gnu.org/licenses/>.
+*/
+
+/*
+ * Please do not edit this file.
+ * It was generated using rpcgen.
+ */
+
+#include "nlmcbk-xdr.h"
+#include "nlm4.h"
+#include "logging.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <rpc/pmap_clnt.h>
+#include <string.h>
+#include <memory.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+
+#ifndef SIG_PF
+#define SIG_PF void(*)(int)
+#endif
+
+void
+nlm4svc_sm_notify (struct nlm_sm_status *status);
+
+void *nlmcbk_sm_notify_0_svc(struct nlm_sm_status *status, struct svc_req *req)
+{
+        nlm4svc_sm_notify (status);
+        return NULL;
+}
+
+static void
+nlmcbk_program_0(struct svc_req *rqstp, register SVCXPRT *transp)
+{
+	union {
+		struct nlm_sm_status nlmcbk_sm_notify_0_arg;
+	} argument;
+	char *result;
+	xdrproc_t _xdr_argument, _xdr_result;
+	char *(*local)(char *, struct svc_req *);
+
+	switch (rqstp->rq_proc) {
+	case NULLPROC:
+		(void) svc_sendreply (transp, (xdrproc_t) xdr_void, (char *)NULL);
+		return;
+
+	case NLMCBK_SM_NOTIFY:
+		_xdr_argument = (xdrproc_t) xdr_nlm_sm_status;
+		_xdr_result = (xdrproc_t) xdr_void;
+		local = (char *(*)(char *, struct svc_req *)) nlmcbk_sm_notify_0_svc;
+		break;
+
+	default:
+		svcerr_noproc (transp);
+		return;
+	}
+	memset ((char *)&argument, 0, sizeof (argument));
+	if (!svc_getargs (transp, (xdrproc_t) _xdr_argument, (caddr_t) &argument)) {
+		svcerr_decode (transp);
+		return;
+	}
+	result = (*local)((char *)&argument, rqstp);
+	if (!svc_sendreply(transp, (xdrproc_t) _xdr_result, result)) {
+		svcerr_systemerr (transp);
+	}
+
+	if (!svc_freeargs (transp, (xdrproc_t) _xdr_argument, (caddr_t) &argument)) {
+		gf_log (GF_NLM, GF_LOG_ERROR, "unable to free arguments");
+                return;
+	}
+	return;
+}
+
+void *
+nsm_thread (void *argv)
+{
+	register SVCXPRT *transp;
+
+	pmap_unset (NLMCBK_PROGRAM, NLMCBK_V1);
+
+	transp = svcudp_create(RPC_ANYSOCK);
+	if (transp == NULL) {
+		gf_log (GF_NLM, GF_LOG_ERROR, "cannot create udp service.");
+                return NULL;
+	}
+	if (!svc_register(transp, NLMCBK_PROGRAM, NLMCBK_V1, nlmcbk_program_0, IPPROTO_UDP)) {
+		gf_log (GF_NLM, GF_LOG_ERROR, "unable to register (NLMCBK_PROGRAM, NLMCBK_V0, udp).");
+                return NULL;
+	}
+
+	transp = svctcp_create(RPC_ANYSOCK, 0, 0);
+	if (transp == NULL) {
+		gf_log (GF_NLM, GF_LOG_ERROR, "cannot create tcp service.");
+                return NULL;
+	}
+	if (!svc_register(transp, NLMCBK_PROGRAM, NLMCBK_V1, nlmcbk_program_0, IPPROTO_TCP)) {
+		gf_log (GF_NLM, GF_LOG_ERROR, "unable to register (NLMCBK_PROGRAM, NLMCBK_V0, tcp).");
+                return NULL;
+	}
+
+	svc_run ();
+	gf_log (GF_NLM, GF_LOG_ERROR, "svc_run returned");
+        return NULL;
+	/* NOTREACHED */
+}