libgfapi: Add new api for supporting mandatory-locks

The current API for byte-range locks [glfs_posix_lock()] doesn't
allow applications to specify whether it is advisory or mandatory
type locks. This particular change is to introduce an extended
byte-range lock API with an additional argument for including
the byte-range lock mode to be one among advisory(default) or
mandatory. Patch also includes a gfapi test case which make use
of this new api to acquire mandatory locks.

Ref: https://github.com/gluster/glusterfs-specs/blob/master/done/GlusterFS%203.8/Mandatory%20Locks.md

Change-Id: Ia09042c755d891895d96da857321abc4ce03e20c
Updates #393
Signed-off-by: Anoop C S <anoopcs@redhat.com>

10 files changed, 694 insertions, 36 deletions

diff --git a/api/src/gfapi-messages.h b/api/src/gfapi-messages.h
index 3eedfa7d071..f231ce99a47 100644
--- a/api/src/gfapi-messages.h
+++ b/api/src/gfapi-messages.h
@@ -73,7 +73,9 @@ GLFS_MSGID(API,
         API_MSG_CREATE_HANDLE_FAILED,
         API_MSG_INODE_LINK_FAILED,
         API_MSG_STATEDUMP_FAILED,
-        API_MSG_XREADDIRP_R_FAILED
+        API_MSG_XREADDIRP_R_FAILED,
+        API_MSG_LOCK_INSERT_MERGE_FAILED,
+        API_MSG_SETTING_LOCK_TYPE_FAILED
 );
 
 #endif /* !_GFAPI_MESSAGES_H__ */
diff --git a/api/src/gfapi.aliases b/api/src/gfapi.aliases
index 41a01f842fd..88d361dc329 100644
--- a/api/src/gfapi.aliases
+++ b/api/src/gfapi.aliases
@@ -169,3 +169,4 @@ _pub_glfs_upcall_register _glfs_upcall_register$GFAPI_3.13.0
 _pub_glfs_upcall_unregister _glfs_upcall_unregister$GFAPI_3.13.0
 
 _pub_glfs_setfsleaseid _glfs_setfsleaseid$GFAPI_4.0.0
+_pub_glfs_file_lock _glfs_file_lock$GFAPI_4.0.0
diff --git a/api/src/gfapi.map b/api/src/gfapi.map
index aa30286c01b..fc47a3b8f42 100644
--- a/api/src/gfapi.map
+++ b/api/src/gfapi.map
@@ -222,5 +222,7 @@ GFAPI_3.13.0 {
 } GFAPI_PRIVATE_3.12.0;
 
 GFAPI_4.0.0 {
+        global:
                 glfs_setfsleaseid;
+                glfs_file_lock;
 } GFAPI_3.13.0;
diff --git a/api/src/glfs-fops.c b/api/src/glfs-fops.c
index 287326c4e4a..833ff336634 100644
--- a/api/src/glfs-fops.c
+++ b/api/src/glfs-fops.c
@@ -4250,36 +4250,46 @@ gf_flock_from_flock (struct gf_flock *gf_flock, struct flock *flock)
 	gf_flock->l_pid    = flock->l_pid;
 }
 
-
-int
-pub_glfs_posix_lock (struct glfs_fd *glfd, int cmd, struct flock *flock)
+static int
+glfs_lock_common (struct glfs_fd *glfd, int cmd, struct flock *flock,
+                  dict_t *xdata)
 {
-	int              ret = -1;
-	xlator_t        *subvol = NULL;
-	struct gf_flock  gf_flock = {0, };
-	struct gf_flock  saved_flock = {0, };
-	fd_t            *fd = NULL;
+        int              ret = -1;
+        xlator_t        *subvol = NULL;
+        struct gf_flock  gf_flock = {0, };
+        struct gf_flock  saved_flock = {0, };
+        fd_t            *fd = NULL;
 
         DECLARE_OLD_THIS;
-	__GLFS_ENTRY_VALIDATE_FD (glfd, invalid_fs);
+        __GLFS_ENTRY_VALIDATE_FD (glfd, invalid_fs);
+
+        if (!flock) {
+                errno = EINVAL;
+                goto out;
+        }
 
         GF_REF_GET (glfd);
-	subvol = glfs_active_subvol (glfd->fs);
-	if (!subvol) {
-		ret = -1;
-		errno = EIO;
-		goto out;
-	}
+        subvol = glfs_active_subvol (glfd->fs);
+        if (!subvol) {
+                ret = -1;
+                errno = EIO;
+                goto out;
+        }
 
-	fd = glfs_resolve_fd (glfd->fs, subvol, glfd);
-	if (!fd) {
-		ret = -1;
-		errno = EBADFD;
-		goto out;
-	}
+        fd = glfs_resolve_fd (glfd->fs, subvol, glfd);
+        if (!fd) {
+                ret = -1;
+                errno = EBADFD;
+                goto out;
+        }
+
+        /* Generate glusterfs flock structure from client flock
+         * structure to be processed by server */
+        gf_flock_from_flock (&gf_flock, flock);
 
-	gf_flock_from_flock (&gf_flock, flock);
-	gf_flock_from_flock (&saved_flock, flock);
+        /* Keep another copy of flock for split/merge of locks
+         * at client side */
+        gf_flock_from_flock (&saved_flock, flock);
 
         if (glfd->lk_owner.len != 0) {
                 ret = syncopctx_setfslkowner (&glfd->lk_owner);
@@ -4288,24 +4298,80 @@ pub_glfs_posix_lock (struct glfs_fd *glfd, int cmd, struct flock *flock)
                         goto out;
         }
 
-	ret = syncop_lk (subvol, fd, cmd, &gf_flock, NULL, NULL);
+        ret = syncop_lk (subvol, fd, cmd, &gf_flock, xdata, NULL);
         DECODE_SYNCOP_ERR (ret);
-	gf_flock_to_flock (&gf_flock, flock);
 
-	if (ret == 0 && (cmd == F_SETLK || cmd == F_SETLKW))
-		fd_lk_insert_and_merge (fd, cmd, &saved_flock);
+        /* Convert back from gf_flock to flock as expected by application */
+        gf_flock_to_flock (&gf_flock, flock);
+
+        if (ret == 0 && (cmd == F_SETLK || cmd == F_SETLKW)) {
+                ret = fd_lk_insert_and_merge (fd, cmd, &saved_flock);
+                if (ret) {
+                        gf_msg (THIS->name, GF_LOG_ERROR, 0,
+                                API_MSG_LOCK_INSERT_MERGE_FAILED,
+                                "Lock insertion and splitting/merging failed "
+                                "on gfid %s", uuid_utoa (fd->inode->gfid));
+                        ret = 0;
+                }
+        }
+
 out:
-	if (fd)
-		fd_unref (fd);
+        if (fd)
+                fd_unref (fd);
         if (glfd)
                 GF_REF_PUT (glfd);
 
-	glfs_subvol_done (glfd->fs, subvol);
+        glfs_subvol_done (glfd->fs, subvol);
 
         __GLFS_EXIT_FS;
 
 invalid_fs:
-	return ret;
+        return ret;
+}
+
+int
+pub_glfs_file_lock (struct glfs_fd *glfd, int cmd, struct flock *flock,
+                    enum glfs_lock_mode_t lk_mode)
+{
+        int              ret            = -1;
+        dict_t          *xdata_in       = NULL;
+
+        if (lk_mode == GLFS_LK_MANDATORY) {
+                /* Create a new dictionary */
+                xdata_in = dict_new ();
+                if (xdata_in == NULL) {
+                        ret = -1;
+                        errno = ENOMEM;
+                        goto out;
+                }
+
+                /* Set GF_LK_MANDATORY internally within dictionary to map
+                 * GLFS_LK_MANDATORY */
+                ret = dict_set_uint32 (xdata_in, GF_LOCK_MODE, GF_LK_MANDATORY);
+                if (ret) {
+                        gf_msg (THIS->name, GF_LOG_ERROR, 0,
+                                API_MSG_SETTING_LOCK_TYPE_FAILED,
+                                "Setting lock type failed");
+                        ret = -1;
+                        errno = ENOMEM;
+                        goto out;
+                }
+        }
+
+        ret = glfs_lock_common (glfd, cmd, flock, xdata_in);
+out:
+        if (xdata_in)
+                dict_unref (xdata_in);
+
+        return ret;
+}
+
+GFAPI_SYMVER_PUBLIC_DEFAULT(glfs_file_lock, 4.0.0);
+
+int
+pub_glfs_posix_lock (struct glfs_fd *glfd, int cmd, struct flock *flock)
+{
+        return glfs_lock_common (glfd, cmd, flock, NULL);
 }
 
 GFAPI_SYMVER_PUBLIC_DEFAULT(glfs_posix_lock, 3.4.0);
diff --git a/api/src/glfs.h b/api/src/glfs.h
index 42cbfe7ab6c..5ffed1e0853 100644
--- a/api/src/glfs.h
+++ b/api/src/glfs.h
@@ -774,6 +774,50 @@ char *glfs_realpath (glfs_t *fs, const char *path, char *resolved_path) __THROW
 int glfs_posix_lock (glfs_fd_t *fd, int cmd, struct flock *flock) __THROW
         GFAPI_PUBLIC(glfs_posix_lock, 3.4.0);
 
+/*
+  SYNOPSIS
+
+  glfs_file_lock: Request extended byte range lock on a file
+
+  DESCRIPTION
+
+  This function is capable of requesting either advisory or mandatory type
+  byte range locks on a file.
+
+  Note: To set a unique owner key for locks based on a particular file
+  descriptor, make use of glfs_fd_set_lkowner() api to do so before
+  requesting lock via this api. This owner key will be further consumed
+  by other incoming data modifying file operations via the same file
+  descriptor.
+
+  PARAMETERS
+
+  @fd: File descriptor
+
+  @cmd: As specified in man fcntl(2).
+
+  @flock: As specified in man fcntl(2).
+
+  @lk_mode: Required lock type from options available with the
+            enum glfs_lock_mode_t defined below.
+
+  RETURN VALUES
+
+  0   : Success. Lock has been granted.
+  -1  : Failure. @errno will be set indicating the type of failure.
+
+ */
+
+/* Lock modes used by glfs_file_lock() */
+enum glfs_lock_mode_t {
+        GLFS_LK_ADVISORY = 0,
+        GLFS_LK_MANDATORY
+};
+
+int glfs_file_lock (glfs_fd_t *fd, int cmd, struct flock *flock,
+                    enum glfs_lock_mode_t lk_mode) __THROW
+        GFAPI_PUBLIC(glfs_file_lock, 3.13.0);
+
 glfs_fd_t *glfs_dup (glfs_fd_t *fd) __THROW
         GFAPI_PUBLIC(glfs_dup, 3.4.0);
 
diff --git a/libglusterfs/src/glusterfs.h b/libglusterfs/src/glusterfs.h
index db4a5ec4029..56be7487ce7 100644
--- a/libglusterfs/src/glusterfs.h
+++ b/libglusterfs/src/glusterfs.h
@@ -301,8 +301,9 @@
 #define GF_BACKTRACE_LEN        4096
 #define GF_BACKTRACE_FRAME_COUNT 7
 
-#define GF_LK_ADVISORY 0
-#define GF_LK_MANDATORY 1
+#define GF_LK_ADVISORY 0 /* maps to GLFS_LK_ADVISORY from libgfapi*/
+#define GF_LK_MANDATORY 1 /* maps to GLFS_LK_MANDATORY from libgfapi*/
+#define GF_LOCK_MODE "glusterfs.lk.lkmode"
 
 const char *fop_enum_to_pri_string (glusterfs_fop_t fop);
 
diff --git a/tests/basic/gfapi/Makefile b/tests/basic/gfapi/Makefile
index e30fefea5b9..1c5cf03ca3d 100644
--- a/tests/basic/gfapi/Makefile
+++ b/tests/basic/gfapi/Makefile
@@ -5,7 +5,8 @@ CFLAGS   = -Wall -g $(shell pkg-config --cflags glusterfs-api)
 LDFLAGS  = $(shell pkg-config --libs glusterfs-api)
 
 BINARIES = upcall-cache-invalidate libgfapi-fini-hang anonymous_fd seek \
-	bug1283983 bug1291259 gfapi-ssl-test gfapi-load-volfile
+	bug1283983 bug1291259 gfapi-ssl-test gfapi-load-volfile \
+        mandatory-lock-optimal
 
 %: %.c
 	$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^
diff --git a/tests/basic/gfapi/mandatory-lock-optimal.c b/tests/basic/gfapi/mandatory-lock-optimal.c
new file mode 100644
index 00000000000..6c62f437a0f
--- /dev/null
+++ b/tests/basic/gfapi/mandatory-lock-optimal.c
@@ -0,0 +1,503 @@
+/* Pre-requisites:-
+ *
+ * 1. Make sure that peformance translators are switched off while running this test.
+ * 2. Perform the following volume set operation:
+ *    # gluster volume set <VOLNAME> locks.mandatory-locking optimal
+ * 3. For installation under non-standard paths, export LD_LIBRARY_PATH to
+ *    automatically load exact libgfapi.so and compile this C file as follows:
+ *    $ gcc mandatory-lock-optimal.c -lgfapi -I <include path for api/glfs.h> -L <include path for libgfapi shared library>
+ */
+
+#include <errno.h>
+#include <stdio.h>
+#include <fcntl.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <glusterfs/api/glfs.h>
+
+#define TOTAL_TEST_COUNT 8
+
+/* C1 = Client 1 : C2 = Client 2 : C3 = Client 3 :
+ * fs1, fd1 are associated with C1. Similarly fs2, fd2 for C2
+ * and fs3, fd3 for C3 */
+
+FILE *fp;
+glfs_t *fs1, *fs2, *fs3;
+glfs_fd_t *fd, *fd1, *fd2, *fd3;
+struct flock lock;
+char buf1[10], *buf2 = "ten bytes!", *fname = "/mand.lock";
+int ret, test_count;
+off_t offset;
+
+/* run_test_1 ()   : C1 takes byte range mandatory read lock.
+                     C2 attempts to read from a conflicting range.
+   Expected result : Read from C2 should pass.
+
+ * run_test_2 ()   : C1 takes byte range mandatory read lock.
+                     C2 attempts write to a conflicting range.
+   Expected result : Write from C2 should fail with EAGAIN.
+
+ * run_test_3 ()   : C1 takes byte range advisory write lock.
+                     C2 attempts to read from a conflicting range.
+   Expected result : Read from C2 should pass.
+
+ * run_test_4 ()   : C1 takes byte range advisory write lock.
+                     C2 attempts write to a conflicting range.
+   Expected result : Write from C2 should pass.
+
+ * run_test_5 ()   : C1 takes byte range advisory read lock.
+                     C2 attempts to open the same file with O_TRUNC.
+   Expected result : Open from C2 should pass.
+
+ * run_test_6 ()   : C1 takes byte range mandatory read lock.
+                     C2 attempts to open the same file with O_TRUNC.
+   Expected result : Open from C2 should fail with EAGAIN.
+
+ * run_test_7 ()   : C1 takes byte range mandatory read lock.
+                     C2 attempts ftruncate on a conflicting range.
+   Expected result : Write from C2 should fail with EAGAIN.
+
+ * run_test_8 ()   : C1 takes byte range advisory read lock.
+                     C2 takes byte range mandatory read lock
+                     within the byte range for which C1 already
+                     holds an advisory lock so as to perform a
+                     basic split/merge. C3 repositions fd3 to
+                     start of C2's byte range mandatory lock
+                     offset and attempts a write. Then it again
+                     repositions fd3 to one byte past C2's byte
+                     range mandatoy lock and again attempts a write.
+   Expected result : First write should fail with EAGAIN.
+                     Second write should pass. */
+
+#define LOG_ERR(func, err) do {                                                         \
+        if (!fp)                                                                        \
+                fprintf (stderr, "\n%s : returned error (%s)\n", func, strerror (err)); \
+        else                                                                            \
+                fprintf (fp, "\n%s : returned error (%s)\n", func, strerror (err));     \
+        cleanup_and_exit (err);                                                         \
+} while (0)
+
+void cleanup_and_exit (int exit_status) {
+        if (exit_status || test_count != TOTAL_TEST_COUNT) {
+                fprintf (fp, "\nAborting due to some test failures.\n");
+                exit_status = 1;
+        } else
+                fprintf (fp, "\nAll tests ran successfully.\n");
+        if (fp)
+                fclose (fp);
+        if (fd)
+                glfs_close (fd);
+        if (fd1)
+                glfs_close (fd1);
+        if (fd2)
+                glfs_close (fd2);
+
+        glfs_unlink (fs1, fname);
+
+        if (fs1)
+                glfs_fini (fs1);
+        if (fs2)
+                glfs_fini (fs2);
+
+        exit (exit_status);
+}
+
+glfs_t *new_client_create (char *hostname, char *volname, char *logfile_name) {
+        glfs_t *fs = NULL;
+
+        fs = glfs_new (volname);
+        if (!fs)
+                LOG_ERR ("glfs_new", errno);
+
+        ret = glfs_set_volfile_server (fs, "tcp", hostname, 24007);
+        if (ret)
+                LOG_ERR ("glfs_set_volfile_server", errno);
+
+        ret = glfs_set_logging (fs, logfile_name, 7);
+        if (ret)
+                LOG_ERR ("glfs_set_logging", errno);
+
+        ret = glfs_init (fs);
+        if (ret)
+                LOG_ERR ("glfs_init", errno);
+
+        return fs;
+}
+
+void run_test_1 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_RDLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 5L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_MANDATORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        /* On successful read, 0 is returned as there is no content inside the
+         * file
+         */
+        ret = glfs_read (fd2, buf1, 10, 0);
+        if (ret)
+                LOG_ERR ("glfs_read", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        ret = glfs_close (fd2);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd2 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+void run_test_2 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_RDLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 5L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_MANDATORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_WRONLY | O_NONBLOCK);
+        if (!fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        ret = glfs_write (fd2, buf2, 10, 0);
+        if (ret == 10 || errno != EAGAIN)
+                LOG_ERR ("glfs_write", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        ret = glfs_close (fd2);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd2 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+void run_test_3 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_WRONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_WRLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 5L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_ADVISORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        /* Still there is no content inside file. So following read should
+         * return 0
+         */
+        ret = glfs_read (fd2, buf1, 10, 0);
+        if (ret)
+                LOG_ERR ("glfs_read", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        ret = glfs_close (fd2);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd2 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+void run_test_4 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_WRONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_WRLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 5L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_ADVISORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_WRONLY | O_NONBLOCK);
+        if (!fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        ret = glfs_write (fd2, buf2, 10, 0);
+        if (ret != 10)
+                LOG_ERR ("glfs_write", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        ret = glfs_close (fd2);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd2 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+void run_test_5 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_RDLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 5L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_ADVISORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_RDONLY | O_NONBLOCK | O_TRUNC);
+        if (!fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        ret = glfs_close (fd2);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd2 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+void run_test_6 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_RDLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 5L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_MANDATORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_RDONLY | O_NONBLOCK | O_TRUNC);
+        if (fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+void run_test_7 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_RDLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 5L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_MANDATORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_RDWR | O_NONBLOCK);
+        if (!fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        ret = glfs_ftruncate (fd2, 4);
+        if (ret == 0 || errno != EAGAIN)
+                LOG_ERR ("glfs_ftruncate", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        ret = glfs_close (fd2);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd2 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+void run_test_8 (int i) {
+        fprintf (fp, "\nRunning Test-%d . . . ", i);
+
+        fd1 = glfs_open (fs1, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd1)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_RDLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0L;
+        lock.l_len = 10L;
+
+        ret = glfs_file_lock (fd1, F_SETLK, &lock, GLFS_LK_ADVISORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd2 = glfs_open (fs2, fname, O_RDONLY | O_NONBLOCK);
+        if (!fd2)
+                LOG_ERR ("glfs_open", errno);
+
+        lock.l_type = F_RDLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 5L;
+        lock.l_len = 2L;
+
+        ret = glfs_file_lock (fd2, F_SETLK, &lock, GLFS_LK_MANDATORY);
+        if (ret)
+                LOG_ERR ("glfs_file_lock", errno);
+
+        fd3 = glfs_open (fs3, fname, O_RDWR | O_NONBLOCK);
+        if (!fd3)
+                LOG_ERR ("glfs_open", errno);
+
+        offset = glfs_lseek (fd3, 5L, SEEK_SET);
+        if (offset != 5)
+                LOG_ERR ("glfs_lseek", errno);
+
+        ret = glfs_write (fd3, buf2, 10, 0);
+        if (ret == 10 || errno != EAGAIN)
+                LOG_ERR ("glfs_write", errno);
+
+        offset = glfs_lseek (fd3, 8L, SEEK_SET);
+        if (offset != 8)
+                LOG_ERR ("glfs_lseek", errno);
+
+        ret = glfs_write (fd3, buf2, 10, 0);
+        if (ret != 10)
+                LOG_ERR ("glfs_write", errno);
+
+        ret = glfs_close (fd1);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd1 = NULL;
+
+        ret = glfs_close (fd2);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd2 = NULL;
+
+        ret = glfs_close (fd3);
+        if (ret)
+                LOG_ERR ("glfs_close", errno);
+        fd3 = NULL;
+
+        test_count++;
+        fprintf (fp, "OK\n", i);
+}
+
+int main (int argc, char *argv[]) {
+        char logfile[50];
+
+        if (argc != 4) {
+                fprintf (stderr, "Usage: %s <server ip/hostname> <volume name> <test log directory>\n", argv[0]);
+                return 0;
+        }
+
+        sprintf (logfile, "%s/%s", argv[3], "mandatory-lock-optimal-test.log");
+        fp = fopen (logfile, "w");
+        if (!fp) {
+                fprintf (stderr, "\n%s\n", logfile);
+                LOG_ERR ("Log file creation", errno);
+        }
+
+        sprintf (logfile, "%s/%s", argv[3], "glfs-client-1.log");
+        fs1 = new_client_create (argv[1], argv[2], logfile);
+        if (!fs1)
+                LOG_ERR ("client-1 creation", EINVAL);
+
+        sprintf (logfile, "%s/%s", argv[3], "glfs-client-2.log");
+        fs2 = new_client_create (argv[1], argv[2], logfile);
+        if (!fs2)
+                LOG_ERR ("client-2 creation", EINVAL);
+
+        sprintf (logfile, "%s/%s", argv[3], "glfs-client-3.log");
+        fs3 = new_client_create (argv[1], argv[2], logfile);
+        if (!fs3)
+                LOG_ERR ("client-3 creation", EINVAL);
+
+        fd = glfs_creat (fs1, fname, O_RDWR, 0644);
+        if (!fd)
+                LOG_ERR ("glfs_creat", errno);
+
+        test_count = 0;
+
+        run_test_1 (1);
+        run_test_2 (2);
+        run_test_3 (3);
+        run_test_4 (4);
+        run_test_5 (5);
+        run_test_6 (6);
+        run_test_7 (7);
+        run_test_8 (8);
+
+        cleanup_and_exit (0);
+
+        return 0;
+}
diff --git a/tests/basic/gfapi/mandatory-lock-optimal.t b/tests/basic/gfapi/mandatory-lock-optimal.t
new file mode 100644
index 00000000000..27062e1f6c2
--- /dev/null
+++ b/tests/basic/gfapi/mandatory-lock-optimal.t
@@ -0,0 +1,38 @@
+#!/bin/bash
+
+. $(dirname $0)/../../include.rc
+. $(dirname $0)/../../volume.rc
+
+cleanup
+
+TEST glusterd
+
+# Create and start the volume
+TEST $CLI volume create $V0 $H0:$B0/${V0}1
+TEST $CLI volume start $V0
+
+logdir=`gluster --print-logdir`
+
+# Switch off performance translators
+TEST $CLI volume set $V0 performance.open-behind off
+TEST $CLI volume set $V0 performance.write-behind off
+TEST $CLI volume set $V0 performance.quick-read off
+TEST $CLI volume set $V0 performance.io-cache off
+TEST $CLI volume set $V0 performance.read-ahead off
+TEST $CLI volume set $V0 performance.readdir-ahead off
+
+# Enable optimal mandatory-locking mode and restart the volume
+TEST $CLI volume set $V0 locks.mandatory-locking optimal
+TEST $CLI volume stop $V0
+TEST $CLI volume start $V0
+
+# Compile and run the test program
+TEST build_tester $(dirname $0)/mandatory-lock-optimal.c -lgfapi
+TEST ./$(dirname $0)/mandatory-lock-optimal $H0 $V0 $logdir
+
+# Cleanup the environment
+cleanup_tester $(dirname $0)/mandatory-lock-optimal
+TEST $CLI volume stop $V0
+TEST $CLI volume delete $V0
+
+cleanup
diff --git a/xlators/features/locks/src/posix.c b/xlators/features/locks/src/posix.c
index 440b272b21e..1e1d9e5ae30 100644
--- a/xlators/features/locks/src/posix.c
+++ b/xlators/features/locks/src/posix.c
@@ -2239,7 +2239,7 @@ pl_lk (call_frame_t *frame, xlator_t *this,
 
         priv = this->private;
 
-        ret = dict_get_uint32 (xdata, "lkmode", &lk_flags);
+        ret = dict_get_uint32 (xdata, GF_LOCK_MODE, &lk_flags);
         if (ret == 0) {
                 if (priv->mandatory_mode == MLK_NONE)
                         gf_log (this->name, GF_LOG_DEBUG, "Lock flags received "