Merge branch 'server-cluster-locking-api' of git://linux-nfs.org/~bfields/linux

* 'server-cluster-locking-api' of git://linux-nfs.org/~bfields/linux:
  gfs2: nfs lock support for gfs2
  lockd: add code to handle deferred lock requests
  lockd: always preallocate block in nlmsvc_lock()
  lockd: handle test_lock deferrals
  lockd: pass cookie in nlmsvc_testlock
  lockd: handle fl_grant callbacks
  lockd: save lock state on deferral
  locks: add fl_grant callback for asynchronous lock return
  nfsd4: Convert NFSv4 to new lock interface
  locks: add lock cancel command
  locks: allow {vfs,posix}_lock_file to return conflicting lock
  locks: factor out generic/filesystem switch from setlock code
  locks: factor out generic/filesystem switch from test_lock
  locks: give posix_test_lock same interface as ->lock
  locks: make ->lock release private data before returning in GETLK case
  locks: create posix-to-flock helper functions
  locks: trivial removal of unnecessary parentheses

4 files changed, 240 insertions, 50 deletions

diff --git a/fs/lockd/svc4proc.c b/fs/lockd/svc4proc.c
index 47a66aa5d55b..bf27b6c6cb6b 100644
--- a/fs/lockd/svc4proc.c
+++ b/fs/lockd/svc4proc.c
@@ -99,7 +99,9 @@ nlm4svc_proc_test(struct svc_rqst *rqstp, struct nlm_args *argp,
                 return resp->status == nlm_drop_reply ? rpc_drop_reply :rpc_success;
 
         /* Now check for conflicting locks */
-        resp->status = nlmsvc_testlock(file, &argp->lock, &resp->lock);
+        resp->status = nlmsvc_testlock(rqstp, file, &argp->lock, &resp->lock, &resp->cookie);
+        if (resp->status == nlm_drop_reply)
+                return rpc_drop_reply;
 
         dprintk("lockd: TEST4          status %d\n", ntohl(resp->status));
         nlm_release_host(host);
@@ -143,6 +145,8 @@ nlm4svc_proc_lock(struct svc_rqst *rqstp, struct nlm_args *argp,
         /* Now try to lock the file */
         resp->status = nlmsvc_lock(rqstp, file, &argp->lock,
                                         argp->block, &argp->cookie);
+        if (resp->status == nlm_drop_reply)
+                return rpc_drop_reply;
 
         dprintk("lockd: LOCK          status %d\n", ntohl(resp->status));
         nlm_release_host(host);
diff --git a/fs/lockd/svclock.c b/fs/lockd/svclock.c
index cf51f849e76c..b3efa4536cc5 100644
--- a/fs/lockd/svclock.c
+++ b/fs/lockd/svclock.c
@@ -173,7 +173,7 @@ found:
  */
 static inline struct nlm_block *
 nlmsvc_create_block(struct svc_rqst *rqstp, struct nlm_file *file,
-                                struct nlm_lock *lock, struct nlm_cookie *cookie)
+                struct nlm_lock *lock, struct nlm_cookie *cookie)
 {
         struct nlm_block        *block;
         struct nlm_host         *host;
@@ -210,6 +210,7 @@ nlmsvc_create_block(struct svc_rqst *rqstp, struct nlm_file *file,
         block->b_daemon = rqstp->rq_server;
         block->b_host   = host;
         block->b_file   = file;
+        block->b_fl = NULL;
         file->f_count++;
 
         /* Add to file's list of blocks */
@@ -261,6 +262,7 @@ static void nlmsvc_free_block(struct kref *kref)
         nlmsvc_freegrantargs(block->b_call);
         nlm_release_call(block->b_call);
         nlm_release_file(block->b_file);
+        kfree(block->b_fl);
         kfree(block);
 }
 
@@ -331,6 +333,31 @@ static void nlmsvc_freegrantargs(struct nlm_rqst *call)
 }
 
 /*
+ * Deferred lock request handling for non-blocking lock
+ */
+static u32
+nlmsvc_defer_lock_rqst(struct svc_rqst *rqstp, struct nlm_block *block)
+{
+        u32 status = nlm_lck_denied_nolocks;
+
+        block->b_flags |= B_QUEUED;
+
+        nlmsvc_insert_block(block, NLM_TIMEOUT);
+
+        block->b_cache_req = &rqstp->rq_chandle;
+        if (rqstp->rq_chandle.defer) {
+                block->b_deferred_req =
+                        rqstp->rq_chandle.defer(block->b_cache_req);
+                if (block->b_deferred_req != NULL)
+                        status = nlm_drop_reply;
+        }
+        dprintk("lockd: nlmsvc_defer_lock_rqst block %p flags %d status %d\n",
+                block, block->b_flags, status);
+
+        return status;
+}
+
+/*
  * Attempt to establish a lock, and if it can't be granted, block it
  * if required.
  */
@@ -338,7 +365,7 @@ __be32
 nlmsvc_lock(struct svc_rqst *rqstp, struct nlm_file *file,
                         struct nlm_lock *lock, int wait, struct nlm_cookie *cookie)
 {
-        struct nlm_block        *block, *newblock = NULL;
+        struct nlm_block        *block = NULL;
         int                     error;
         __be32                  ret;
 
@@ -351,29 +378,58 @@ nlmsvc_lock(struct svc_rqst *rqstp, struct nlm_file *file,
                                 wait);
 
 
-        lock->fl.fl_flags &= ~FL_SLEEP;
-again:
         /* Lock file against concurrent access */
         mutex_lock(&file->f_mutex);
-        /* Get existing block (in case client is busy-waiting) */
+        /* Get existing block (in case client is busy-waiting)
+         * or create new block
+         */
         block = nlmsvc_lookup_block(file, lock);
         if (block == NULL) {
-                if (newblock != NULL)
-                        lock = &newblock->b_call->a_args.lock;
-        } else
+                block = nlmsvc_create_block(rqstp, file, lock, cookie);
+                ret = nlm_lck_denied_nolocks;
+                if (block == NULL)
+                        goto out;
                 lock = &block->b_call->a_args.lock;
+        } else
+                lock->fl.fl_flags &= ~FL_SLEEP;
 
-        error = posix_lock_file(file->f_file, &lock->fl);
-        lock->fl.fl_flags &= ~FL_SLEEP;
+        if (block->b_flags & B_QUEUED) {
+                dprintk("lockd: nlmsvc_lock deferred block %p flags %d\n",
+                                                        block, block->b_flags);
+                if (block->b_granted) {
+                        nlmsvc_unlink_block(block);
+                        ret = nlm_granted;
+                        goto out;
+                }
+                if (block->b_flags & B_TIMED_OUT) {
+                        nlmsvc_unlink_block(block);
+                        ret = nlm_lck_denied;
+                        goto out;
+                }
+                ret = nlm_drop_reply;
+                goto out;
+        }
 
-        dprintk("lockd: posix_lock_file returned %d\n", error);
+        if (!wait)
+                lock->fl.fl_flags &= ~FL_SLEEP;
+        error = vfs_lock_file(file->f_file, F_SETLK, &lock->fl, NULL);
+        lock->fl.fl_flags &= ~FL_SLEEP;
 
+        dprintk("lockd: vfs_lock_file returned %d\n", error);
         switch(error) {
                 case 0:
                         ret = nlm_granted;
                         goto out;
                 case -EAGAIN:
+                        ret = nlm_lck_denied;
                         break;
+                case -EINPROGRESS:
+                        if (wait)
+                                break;
+                        /* Filesystem lock operation is in progress
+                           Add it to the queue waiting for callback */
+                        ret = nlmsvc_defer_lock_rqst(rqstp, block);
+                        goto out;
                 case -EDEADLK:
                         ret = nlm_deadlock;
                         goto out;
@@ -387,26 +443,11 @@ again:
                 goto out;
 
         ret = nlm_lck_blocked;
-        if (block != NULL)
-                goto out;
-
-        /* If we don't have a block, create and initialize it. Then
-         * retry because we may have slept in kmalloc. */
-        /* We have to release f_mutex as nlmsvc_create_block may try to
-         * to claim it while doing host garbage collection */
-        if (newblock == NULL) {
-                mutex_unlock(&file->f_mutex);
-                dprintk("lockd: blocking on this lock (allocating).\n");
-                if (!(newblock = nlmsvc_create_block(rqstp, file, lock, cookie)))
-                        return nlm_lck_denied_nolocks;
-                goto again;
-        }
 
         /* Append to list of blocked */
-        nlmsvc_insert_block(newblock, NLM_NEVER);
+        nlmsvc_insert_block(block, NLM_NEVER);
 out:
         mutex_unlock(&file->f_mutex);
-        nlmsvc_release_block(newblock);
         nlmsvc_release_block(block);
         dprintk("lockd: nlmsvc_lock returned %u\n", ret);
         return ret;
@@ -416,9 +457,14 @@ out:
  * Test for presence of a conflicting lock.
  */
 __be32
-nlmsvc_testlock(struct nlm_file *file, struct nlm_lock *lock,
-                                       struct nlm_lock *conflock)
+nlmsvc_testlock(struct svc_rqst *rqstp, struct nlm_file *file,
+                struct nlm_lock *lock, struct nlm_lock *conflock,
+                struct nlm_cookie *cookie)
 {
+        struct nlm_block        *block = NULL;
+        int                     error;
+        __be32                  ret;
+
         dprintk("lockd: nlmsvc_testlock(%s/%ld, ty=%d, %Ld-%Ld)\n",
                                 file->f_file->f_path.dentry->d_inode->i_sb->s_id,
                                 file->f_file->f_path.dentry->d_inode->i_ino,
@@ -426,19 +472,70 @@ nlmsvc_testlock(struct nlm_file *file, struct nlm_lock *lock,
                                 (long long)lock->fl.fl_start,
                                 (long long)lock->fl.fl_end);
 
-        if (posix_test_lock(file->f_file, &lock->fl, &conflock->fl)) {
-                dprintk("lockd: conflicting lock(ty=%d, %Ld-%Ld)\n",
-                                conflock->fl.fl_type,
-                                (long long)conflock->fl.fl_start,
-                                (long long)conflock->fl.fl_end);
-                conflock->caller = "somehost";  /* FIXME */
-                conflock->len = strlen(conflock->caller);
-                conflock->oh.len = 0;           /* don't return OH info */
-                conflock->svid = conflock->fl.fl_pid;
-                return nlm_lck_denied;
+        /* Get existing block (in case client is busy-waiting) */
+        block = nlmsvc_lookup_block(file, lock);
+
+        if (block == NULL) {
+                struct file_lock *conf = kzalloc(sizeof(*conf), GFP_KERNEL);
+
+                if (conf == NULL)
+                        return nlm_granted;
+                block = nlmsvc_create_block(rqstp, file, lock, cookie);
+                if (block == NULL) {
+                        kfree(conf);
+                        return nlm_granted;
+                }
+                block->b_fl = conf;
+        }
+        if (block->b_flags & B_QUEUED) {
+                dprintk("lockd: nlmsvc_testlock deferred block %p flags %d fl %p\n",
+                        block, block->b_flags, block->b_fl);
+                if (block->b_flags & B_TIMED_OUT) {
+                        nlmsvc_unlink_block(block);
+                        return nlm_lck_denied;
+                }
+                if (block->b_flags & B_GOT_CALLBACK) {
+                        if (block->b_fl != NULL
+                                        && block->b_fl->fl_type != F_UNLCK) {
+                                lock->fl = *block->b_fl;
+                                goto conf_lock;
+                        }
+                        else {
+                                nlmsvc_unlink_block(block);
+                                return nlm_granted;
+                        }
+                }
+                return nlm_drop_reply;
         }
 
-        return nlm_granted;
+        error = vfs_test_lock(file->f_file, &lock->fl);
+        if (error == -EINPROGRESS)
+                return nlmsvc_defer_lock_rqst(rqstp, block);
+        if (error) {
+                ret = nlm_lck_denied_nolocks;
+                goto out;
+        }
+        if (lock->fl.fl_type == F_UNLCK) {
+                ret = nlm_granted;
+                goto out;
+        }
+
+conf_lock:
+        dprintk("lockd: conflicting lock(ty=%d, %Ld-%Ld)\n",
+                lock->fl.fl_type, (long long)lock->fl.fl_start,
+                (long long)lock->fl.fl_end);
+        conflock->caller = "somehost";  /* FIXME */
+        conflock->len = strlen(conflock->caller);
+        conflock->oh.len = 0;           /* don't return OH info */
+        conflock->svid = lock->fl.fl_pid;
+        conflock->fl.fl_type = lock->fl.fl_type;
+        conflock->fl.fl_start = lock->fl.fl_start;
+        conflock->fl.fl_end = lock->fl.fl_end;
+        ret = nlm_lck_denied;
+out:
+        if (block)
+                nlmsvc_release_block(block);
+        return ret;
 }
 
 /*
@@ -464,7 +561,7 @@ nlmsvc_unlock(struct nlm_file *file, struct nlm_lock *lock)
         nlmsvc_cancel_blocked(file, lock);
 
         lock->fl.fl_type = F_UNLCK;
-        error = posix_lock_file(file->f_file, &lock->fl);
+        error = vfs_lock_file(file->f_file, F_SETLK, &lock->fl, NULL);
 
         return (error < 0)? nlm_lck_denied_nolocks : nlm_granted;
 }
@@ -493,6 +590,8 @@ nlmsvc_cancel_blocked(struct nlm_file *file, struct nlm_lock *lock)
         block = nlmsvc_lookup_block(file, lock);
         mutex_unlock(&file->f_mutex);
         if (block != NULL) {
+                vfs_cancel_lock(block->b_file->f_file,
+                                &block->b_call->a_args.lock.fl);
                 status = nlmsvc_unlink_block(block);
                 nlmsvc_release_block(block);
         }
@@ -500,6 +599,63 @@ nlmsvc_cancel_blocked(struct nlm_file *file, struct nlm_lock *lock)
 }
 
 /*
+ * This is a callback from the filesystem for VFS file lock requests.
+ * It will be used if fl_grant is defined and the filesystem can not
+ * respond to the request immediately.
+ * For GETLK request it will copy the reply to the nlm_block.
+ * For SETLK or SETLKW request it will get the local posix lock.
+ * In all cases it will move the block to the head of nlm_blocked q where
+ * nlmsvc_retry_blocked() can send back a reply for SETLKW or revisit the
+ * deferred rpc for GETLK and SETLK.
+ */
+static void
+nlmsvc_update_deferred_block(struct nlm_block *block, struct file_lock *conf,
+                             int result)
+{
+        block->b_flags |= B_GOT_CALLBACK;
+        if (result == 0)
+                block->b_granted = 1;
+        else
+                block->b_flags |= B_TIMED_OUT;
+        if (conf) {
+                if (block->b_fl)
+                        locks_copy_lock(block->b_fl, conf);
+        }
+}
+
+static int nlmsvc_grant_deferred(struct file_lock *fl, struct file_lock *conf,
+                                        int result)
+{
+        struct nlm_block *block;
+        int rc = -ENOENT;
+
+        lock_kernel();
+        list_for_each_entry(block, &nlm_blocked, b_list) {
+                if (nlm_compare_locks(&block->b_call->a_args.lock.fl, fl)) {
+                        dprintk("lockd: nlmsvc_notify_blocked block %p flags %d\n",
+                                                        block, block->b_flags);
+                        if (block->b_flags & B_QUEUED) {
+                                if (block->b_flags & B_TIMED_OUT) {
+                                        rc = -ENOLCK;
+                                        break;
+                                }
+                                nlmsvc_update_deferred_block(block, conf, result);
+                        } else if (result == 0)
+                                block->b_granted = 1;
+
+                        nlmsvc_insert_block(block, 0);
+                        svc_wake_up(block->b_daemon);
+                        rc = 0;
+                        break;
+                }
+        }
+        unlock_kernel();
+        if (rc == -ENOENT)
+                printk(KERN_WARNING "lockd: grant for unknown block\n");
+        return rc;
+}
+
+/*
  * Unblock a blocked lock request. This is a callback invoked from the
  * VFS layer when a lock on which we blocked is removed.
  *
@@ -531,6 +687,7 @@ static int nlmsvc_same_owner(struct file_lock *fl1, struct file_lock *fl2)
 struct lock_manager_operations nlmsvc_lock_operations = {
         .fl_compare_owner = nlmsvc_same_owner,
         .fl_notify = nlmsvc_notify_blocked,
+        .fl_grant = nlmsvc_grant_deferred,
 };
 
 /*
@@ -553,6 +710,8 @@ nlmsvc_grant_blocked(struct nlm_block *block)
 
         dprintk("lockd: grant blocked lock %p\n", block);
 
+        kref_get(&block->b_count);
+
         /* Unlink block request from list */
         nlmsvc_unlink_block(block);
 
@@ -566,20 +725,23 @@ nlmsvc_grant_blocked(struct nlm_block *block)
 
         /* Try the lock operation again */
         lock->fl.fl_flags |= FL_SLEEP;
-        error = posix_lock_file(file->f_file, &lock->fl);
+        error = vfs_lock_file(file->f_file, F_SETLK, &lock->fl, NULL);
         lock->fl.fl_flags &= ~FL_SLEEP;
 
         switch (error) {
         case 0:
                 break;
         case -EAGAIN:
-                dprintk("lockd: lock still blocked\n");
+        case -EINPROGRESS:
+                dprintk("lockd: lock still blocked error %d\n", error);
                 nlmsvc_insert_block(block, NLM_NEVER);
+                nlmsvc_release_block(block);
                 return;
         default:
                 printk(KERN_WARNING "lockd: unexpected error %d in %s!\n",
                                 -error, __FUNCTION__);
                 nlmsvc_insert_block(block, 10 * HZ);
+                nlmsvc_release_block(block);
                 return;
         }
 
@@ -592,7 +754,6 @@ callback:
         nlmsvc_insert_block(block, 30 * HZ);
 
         /* Call the client */
-        kref_get(&block->b_count);
         nlm_async_call(block->b_call, NLMPROC_GRANTED_MSG, &nlmsvc_grant_ops);
 }
 
@@ -665,6 +826,23 @@ nlmsvc_grant_reply(struct nlm_cookie *cookie, __be32 status)
         nlmsvc_release_block(block);
 }
 
+/* Helper function to handle retry of a deferred block.
+ * If it is a blocking lock, call grant_blocked.
+ * For a non-blocking lock or test lock, revisit the request.
+ */
+static void
+retry_deferred_block(struct nlm_block *block)
+{
+        if (!(block->b_flags & B_GOT_CALLBACK))
+                block->b_flags |= B_TIMED_OUT;
+        nlmsvc_insert_block(block, NLM_TIMEOUT);
+        dprintk("revisit block %p flags %d\n",  block, block->b_flags);
+        if (block->b_deferred_req) {
+                block->b_deferred_req->revisit(block->b_deferred_req, 0);
+                block->b_deferred_req = NULL;
+        }
+}
+
 /*
  * Retry all blocked locks that have been notified. This is where lockd
  * picks up locks that can be granted, or grant notifications that must
@@ -688,9 +866,12 @@ nlmsvc_retry_blocked(void)
 
                 dprintk("nlmsvc_retry_blocked(%p, when=%ld)\n",
                         block, block->b_when);
-                kref_get(&block->b_count);
-                nlmsvc_grant_blocked(block);
-                nlmsvc_release_block(block);
+                if (block->b_flags & B_QUEUED) {
+                        dprintk("nlmsvc_retry_blocked delete block (%p, granted=%d, flags=%d)\n",
+                                block, block->b_granted, block->b_flags);
+                        retry_deferred_block(block);
+                } else
+                        nlmsvc_grant_blocked(block);
         }
 
         return timeout;
diff --git a/fs/lockd/svcproc.c b/fs/lockd/svcproc.c
index 31cb48425733..9cd5c8b37593 100644
--- a/fs/lockd/svcproc.c
+++ b/fs/lockd/svcproc.c
@@ -33,6 +33,7 @@ cast_to_nlm(__be32 status, u32 vers)
                 case nlm_lck_denied_nolocks:
                 case nlm_lck_blocked:
                 case nlm_lck_denied_grace_period:
+                case nlm_drop_reply:
                         break;
                 case nlm4_deadlock:
                         status = nlm_lck_denied;
@@ -127,7 +128,9 @@ nlmsvc_proc_test(struct svc_rqst *rqstp, struct nlm_args *argp,
                 return resp->status == nlm_drop_reply ? rpc_drop_reply :rpc_success;
 
         /* Now check for conflicting locks */
-        resp->status = cast_status(nlmsvc_testlock(file, &argp->lock, &resp->lock));
+        resp->status = cast_status(nlmsvc_testlock(rqstp, file, &argp->lock, &resp->lock, &resp->cookie));
+        if (resp->status == nlm_drop_reply)
+                return rpc_drop_reply;
 
         dprintk("lockd: TEST          status %d vers %d\n",
                 ntohl(resp->status), rqstp->rq_vers);
@@ -172,6 +175,8 @@ nlmsvc_proc_lock(struct svc_rqst *rqstp, struct nlm_args *argp,
         /* Now try to lock the file */
         resp->status = cast_status(nlmsvc_lock(rqstp, file, &argp->lock,
                                                argp->block, &argp->cookie));
+        if (resp->status == nlm_drop_reply)
+                return rpc_drop_reply;
 
         dprintk("lockd: LOCK          status %d\n", ntohl(resp->status));
         nlm_release_host(host);
diff --git a/fs/lockd/svcsubs.c b/fs/lockd/svcsubs.c
index c0df00c74ce3..84ebba33b98d 100644
--- a/fs/lockd/svcsubs.c
+++ b/fs/lockd/svcsubs.c
@@ -182,7 +182,7 @@ again:
                         lock.fl_type  = F_UNLCK;
                         lock.fl_start = 0;
                         lock.fl_end   = OFFSET_MAX;
-                        if (posix_lock_file(file->f_file, &lock) < 0) {
+                        if (vfs_lock_file(file->f_file, F_SETLK, &lock, NULL) < 0) {
                                 printk("lockd: unlock failure in %s:%d\n",
                                                 __FILE__, __LINE__);
                                 return 1;