Merge tag 'nfs-for-3.15-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs

Pull NFS client updates from Trond Myklebust:
 "Highlights include:

   - Stable fix for a use after free issue in the NFSv4.1 open code
   - Fix the SUNRPC bi-directional RPC code to account for TCP segmentation
   - Optimise usage of readdirplus when confronted with 'ls -l' situations
   - Soft mount bugfixes
   - NFS over RDMA bugfixes
   - NFSv4 close locking fixes
   - Various NFSv4.x client state management optimisations
   - Rename/unlink code cleanups"

* tag 'nfs-for-3.15-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs: (28 commits)
  nfs: pass string length to pr_notice message about readdir loops
  NFSv4: Fix a use-after-free problem in open()
  SUNRPC: rpc_restart_call/rpc_restart_call_prepare should clear task->tk_status
  SUNRPC: Don't let rpc_delay() clobber non-timeout errors
  SUNRPC: Ensure call_connect_status() deals correctly with SOFTCONN tasks
  SUNRPC: Ensure call_status() deals correctly with SOFTCONN tasks
  NFSv4: Ensure we respect soft mount timeouts during trunking discovery
  NFSv4: Schedule recovery if nfs40_walk_client_list() is interrupted
  NFS: advertise only supported callback netids
  SUNRPC: remove KERN_INFO from dprintk() call sites
  SUNRPC: Fix large reads on NFS/RDMA
  NFS: Clean up: revert increase in READDIR RPC buffer max size
  SUNRPC: Ensure that call_bind times out correctly
  SUNRPC: Ensure that call_connect times out correctly
  nfs: emit a fsnotify_nameremove call in sillyrename codepath
  nfs: remove synchronous rename code
  nfs: convert nfs_rename to use async_rename infrastructure
  nfs: make nfs_async_rename non-static
  nfs: abstract out code needed to complete a sillyrename
  NFSv4: Clear the open state flags if the new stateid does not match
  ...

6 files changed, 93 insertions, 68 deletions

diff --git a/net/sunrpc/backchannel_rqst.c b/net/sunrpc/backchannel_rqst.c
index e860d4f7ed2a..3513d559bc45 100644
--- a/net/sunrpc/backchannel_rqst.c
+++ b/net/sunrpc/backchannel_rqst.c
@@ -212,39 +212,23 @@ out:
 }
 EXPORT_SYMBOL_GPL(xprt_destroy_backchannel);
 
-/*
- * One or more rpc_rqst structure have been preallocated during the
- * backchannel setup.  Buffer space for the send and private XDR buffers
- * has been preallocated as well.  Use xprt_alloc_bc_request to allocate
- * to this request.  Use xprt_free_bc_request to return it.
- *
- * We know that we're called in soft interrupt context, grab the spin_lock
- * since there is no need to grab the bottom half spin_lock.
- *
- * Return an available rpc_rqst, otherwise NULL if non are available.
- */
-struct rpc_rqst *xprt_alloc_bc_request(struct rpc_xprt *xprt)
+static struct rpc_rqst *xprt_alloc_bc_request(struct rpc_xprt *xprt, __be32 xid)
 {
-        struct rpc_rqst *req;
+        struct rpc_rqst *req = NULL;
 
         dprintk("RPC:       allocate a backchannel request\n");
-        spin_lock(&xprt->bc_pa_lock);
-        if (!list_empty(&xprt->bc_pa_list)) {
-                req = list_first_entry(&xprt->bc_pa_list, struct rpc_rqst,
-                                rq_bc_pa_list);
-                list_del(&req->rq_bc_pa_list);
-        } else {
-                req = NULL;
-        }
-        spin_unlock(&xprt->bc_pa_lock);
+        if (list_empty(&xprt->bc_pa_list))
+                goto not_found;
 
-        if (req != NULL) {
-                set_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state);
-                req->rq_reply_bytes_recvd = 0;
-                req->rq_bytes_sent = 0;
-                memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
+        req = list_first_entry(&xprt->bc_pa_list, struct rpc_rqst,
+                                rq_bc_pa_list);
+        req->rq_reply_bytes_recvd = 0;
+        req->rq_bytes_sent = 0;
+        memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
                         sizeof(req->rq_private_buf));
-        }
+        req->rq_xid = xid;
+        req->rq_connect_cookie = xprt->connect_cookie;
+not_found:
         dprintk("RPC:       backchannel req=%p\n", req);
         return req;
 }
@@ -259,6 +243,7 @@ void xprt_free_bc_request(struct rpc_rqst *req)
 
         dprintk("RPC:       free backchannel req=%p\n", req);
 
+        req->rq_connect_cookie = xprt->connect_cookie - 1;
         smp_mb__before_clear_bit();
         WARN_ON_ONCE(!test_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state));
         clear_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state);
@@ -281,7 +266,57 @@ void xprt_free_bc_request(struct rpc_rqst *req)
          * may be reused by a new callback request.
          */
         spin_lock_bh(&xprt->bc_pa_lock);
-        list_add(&req->rq_bc_pa_list, &xprt->bc_pa_list);
+        list_add_tail(&req->rq_bc_pa_list, &xprt->bc_pa_list);
         spin_unlock_bh(&xprt->bc_pa_lock);
 }
 
+/*
+ * One or more rpc_rqst structure have been preallocated during the
+ * backchannel setup.  Buffer space for the send and private XDR buffers
+ * has been preallocated as well.  Use xprt_alloc_bc_request to allocate
+ * to this request.  Use xprt_free_bc_request to return it.
+ *
+ * We know that we're called in soft interrupt context, grab the spin_lock
+ * since there is no need to grab the bottom half spin_lock.
+ *
+ * Return an available rpc_rqst, otherwise NULL if non are available.
+ */
+struct rpc_rqst *xprt_lookup_bc_request(struct rpc_xprt *xprt, __be32 xid)
+{
+        struct rpc_rqst *req;
+
+        spin_lock(&xprt->bc_pa_lock);
+        list_for_each_entry(req, &xprt->bc_pa_list, rq_bc_pa_list) {
+                if (req->rq_connect_cookie != xprt->connect_cookie)
+                        continue;
+                if (req->rq_xid == xid)
+                        goto found;
+        }
+        req = xprt_alloc_bc_request(xprt, xid);
+found:
+        spin_unlock(&xprt->bc_pa_lock);
+        return req;
+}
+
+/*
+ * Add callback request to callback list.  The callback
+ * service sleeps on the sv_cb_waitq waiting for new
+ * requests.  Wake it up after adding enqueing the
+ * request.
+ */
+void xprt_complete_bc_request(struct rpc_rqst *req, uint32_t copied)
+{
+        struct rpc_xprt *xprt = req->rq_xprt;
+        struct svc_serv *bc_serv = xprt->bc_serv;
+
+        req->rq_private_buf.len = copied;
+        set_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state);
+
+        dprintk("RPC:       add callback request to list\n");
+        spin_lock(&bc_serv->sv_cb_lock);
+        list_del(&req->rq_bc_pa_list);
+        list_add(&req->rq_bc_list, &bc_serv->sv_cb_list);
+        wake_up(&bc_serv->sv_cb_waitq);
+        spin_unlock(&bc_serv->sv_cb_lock);
+}
+
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index 0edada973434..f400445d1a44 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -1363,6 +1363,7 @@ rpc_restart_call_prepare(struct rpc_task *task)
         if (RPC_ASSASSINATED(task))
                 return 0;
         task->tk_action = call_start;
+        task->tk_status = 0;
         if (task->tk_ops->rpc_call_prepare != NULL)
                 task->tk_action = rpc_prepare_task;
         return 1;
@@ -1379,6 +1380,7 @@ rpc_restart_call(struct rpc_task *task)
         if (RPC_ASSASSINATED(task))
                 return 0;
         task->tk_action = call_start;
+        task->tk_status = 0;
         return 1;
 }
 EXPORT_SYMBOL_GPL(rpc_restart_call);
@@ -1728,9 +1730,7 @@ call_bind_status(struct rpc_task *task)
         case -EPROTONOSUPPORT:
                 dprintk("RPC: %5u remote rpcbind version unavailable, retrying\n",
                                 task->tk_pid);
-                task->tk_status = 0;
-                task->tk_action = call_bind;
-                return;
+                goto retry_timeout;
         case -ECONNREFUSED:             /* connection problems */
         case -ECONNRESET:
         case -ECONNABORTED:
@@ -1756,6 +1756,7 @@ call_bind_status(struct rpc_task *task)
         return;
 
 retry_timeout:
+        task->tk_status = 0;
         task->tk_action = call_timeout;
 }
 
@@ -1798,21 +1799,19 @@ call_connect_status(struct rpc_task *task)
         trace_rpc_connect_status(task, status);
         task->tk_status = 0;
         switch (status) {
-                /* if soft mounted, test if we've timed out */
-        case -ETIMEDOUT:
-                task->tk_action = call_timeout;
-                return;
         case -ECONNREFUSED:
         case -ECONNRESET:
         case -ECONNABORTED:
         case -ENETUNREACH:
         case -EHOSTUNREACH:
-                /* retry with existing socket, after a delay */
-                rpc_delay(task, 3*HZ);
                 if (RPC_IS_SOFTCONN(task))
                         break;
+                /* retry with existing socket, after a delay */
+                rpc_delay(task, 3*HZ);
         case -EAGAIN:
-                task->tk_action = call_bind;
+                /* Check for timeouts before looping back to call_bind */
+        case -ETIMEDOUT:
+                task->tk_action = call_timeout;
                 return;
         case 0:
                 clnt->cl_stats->netreconn++;
@@ -2007,6 +2006,10 @@ call_status(struct rpc_task *task)
         case -EHOSTDOWN:
         case -EHOSTUNREACH:
         case -ENETUNREACH:
+                if (RPC_IS_SOFTCONN(task)) {
+                        rpc_exit(task, status);
+                        break;
+                }
                 /*
                  * Delay any retries for 3 seconds, then handle as if it
                  * were a timeout.
diff --git a/net/sunrpc/sched.c b/net/sunrpc/sched.c
index ff3cc4bf4b24..25578afe1548 100644
--- a/net/sunrpc/sched.c
+++ b/net/sunrpc/sched.c
@@ -637,7 +637,8 @@ static void __rpc_queue_timer_fn(unsigned long ptr)
 
 static void __rpc_atrun(struct rpc_task *task)
 {
-        task->tk_status = 0;
+        if (task->tk_status == -ETIMEDOUT)
+                task->tk_status = 0;
 }
 
 /*
diff --git a/net/sunrpc/xprtrdma/rpc_rdma.c b/net/sunrpc/xprtrdma/rpc_rdma.c
index e03725bfe2b8..96ead526b125 100644
--- a/net/sunrpc/xprtrdma/rpc_rdma.c
+++ b/net/sunrpc/xprtrdma/rpc_rdma.c
@@ -649,9 +649,7 @@ rpcrdma_inline_fixup(struct rpc_rqst *rqst, char *srcp, int copy_len, int pad)
                                 break;
                         page_base = 0;
                 }
-                rqst->rq_rcv_buf.page_len = olen - copy_len;
-        } else
-                rqst->rq_rcv_buf.page_len = 0;
+        }
 
         if (copy_len && rqst->rq_rcv_buf.tail[0].iov_len) {
                 curlen = copy_len;
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index 285dc0884115..1eb9c468d0c9 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -733,7 +733,7 @@ static void __exit xprt_rdma_cleanup(void)
 {
         int rc;
 
-        dprintk(KERN_INFO "RPCRDMA Module Removed, deregister RPC RDMA transport\n");
+        dprintk("RPCRDMA Module Removed, deregister RPC RDMA transport\n");
 #ifdef RPC_DEBUG
         if (sunrpc_table_header) {
                 unregister_sysctl_table(sunrpc_table_header);
@@ -755,14 +755,14 @@ static int __init xprt_rdma_init(void)
         if (rc)
                 return rc;
 
-        dprintk(KERN_INFO "RPCRDMA Module Init, register RPC RDMA transport\n");
+        dprintk("RPCRDMA Module Init, register RPC RDMA transport\n");
 
-        dprintk(KERN_INFO "Defaults:\n");
-        dprintk(KERN_INFO "\tSlots %d\n"
+        dprintk("Defaults:\n");
+        dprintk("\tSlots %d\n"
                 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n",
                 xprt_rdma_slot_table_entries,
                 xprt_rdma_max_inline_read, xprt_rdma_max_inline_write);
-        dprintk(KERN_INFO "\tPadding %d\n\tMemreg %d\n",
+        dprintk("\tPadding %d\n\tMemreg %d\n",
                 xprt_rdma_inline_write_padding, xprt_rdma_memreg_strategy);
 
 #ifdef RPC_DEBUG
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 0addefca8e77..966763d735e9 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -1306,41 +1306,29 @@ static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
  * If we're unable to obtain the rpc_rqst we schedule the closing of the
  * connection and return -1.
  */
-static inline int xs_tcp_read_callback(struct rpc_xprt *xprt,
+static int xs_tcp_read_callback(struct rpc_xprt *xprt,
                                        struct xdr_skb_reader *desc)
 {
         struct sock_xprt *transport =
                                 container_of(xprt, struct sock_xprt, xprt);
         struct rpc_rqst *req;
 
-        req = xprt_alloc_bc_request(xprt);
+        /* Look up and lock the request corresponding to the given XID */
+        spin_lock(&xprt->transport_lock);
+        req = xprt_lookup_bc_request(xprt, transport->tcp_xid);
         if (req == NULL) {
+                spin_unlock(&xprt->transport_lock);
                 printk(KERN_WARNING "Callback slot table overflowed\n");
                 xprt_force_disconnect(xprt);
                 return -1;
         }
 
-        req->rq_xid = transport->tcp_xid;
         dprintk("RPC:       read callback  XID %08x\n", ntohl(req->rq_xid));
         xs_tcp_read_common(xprt, desc, req);
 
-        if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) {
-                struct svc_serv *bc_serv = xprt->bc_serv;
-
-                /*
-                 * Add callback request to callback list.  The callback
-                 * service sleeps on the sv_cb_waitq waiting for new
-                 * requests.  Wake it up after adding enqueing the
-                 * request.
-                 */
-                dprintk("RPC:       add callback request to list\n");
-                spin_lock(&bc_serv->sv_cb_lock);
-                list_add(&req->rq_bc_list, &bc_serv->sv_cb_list);
-                spin_unlock(&bc_serv->sv_cb_lock);
-                wake_up(&bc_serv->sv_cb_waitq);
-        }
-
-        req->rq_private_buf.len = transport->tcp_copied;
+        if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
+                xprt_complete_bc_request(req, transport->tcp_copied);
+        spin_unlock(&xprt->transport_lock);
 
         return 0;
 }