Merge tag 'nfs-for-4.15-3' of git://git.linux-nfs.org/projects/anna/linux-nfs

Pull NFS client fixes from Anna Schumaker:
 "This has two stable bugfixes, one to fix a BUG_ON() when
  nfs_commit_inode() is called with no outstanding commit requests and
  another to fix a race in the SUNRPC receive codepath.

  Additionally, there are also fixes for an NFS client deadlock and an
  xprtrdma performance regression.

  Summary:

  Stable bugfixes:
   - NFS: Avoid a BUG_ON() in nfs_commit_inode() by not waiting for a
     commit in the case that there were no commit requests.
   - SUNRPC: Fix a race in the receive code path

  Other fixes:
   - NFS: Fix a deadlock in nfs client initialization
   - xprtrdma: Fix a performance regression for small IOs"

* tag 'nfs-for-4.15-3' of git://git.linux-nfs.org/projects/anna/linux-nfs:
  SUNRPC: Fix a race in the receive code path
  nfs: don't wait on commit in nfs_commit_inode() if there were no commit requests
  xprtrdma: Spread reply processing over more CPUs
  nfs: fix a deadlock in nfs client initialization

5 files changed, 24 insertions, 15 deletions

diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index 333b9d697ae5..33b74fd84051 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -1001,6 +1001,7 @@ void xprt_transmit(struct rpc_task *task)
 {
         struct rpc_rqst *req = task->tk_rqstp;
         struct rpc_xprt *xprt = req->rq_xprt;
+        unsigned int connect_cookie;
         int status, numreqs;
 
         dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
@@ -1024,6 +1025,7 @@ void xprt_transmit(struct rpc_task *task)
         } else if (!req->rq_bytes_sent)
                 return;
 
+        connect_cookie = xprt->connect_cookie;
         req->rq_xtime = ktime_get();
         status = xprt->ops->send_request(task);
         trace_xprt_transmit(xprt, req->rq_xid, status);
@@ -1047,20 +1049,28 @@ void xprt_transmit(struct rpc_task *task)
         xprt->stat.bklog_u += xprt->backlog.qlen;
         xprt->stat.sending_u += xprt->sending.qlen;
         xprt->stat.pending_u += xprt->pending.qlen;
+        spin_unlock_bh(&xprt->transport_lock);
 
-        /* Don't race with disconnect */
-        if (!xprt_connected(xprt))
-                task->tk_status = -ENOTCONN;
-        else {
+        req->rq_connect_cookie = connect_cookie;
+        if (rpc_reply_expected(task) && !READ_ONCE(req->rq_reply_bytes_recvd)) {
                 /*
-                 * Sleep on the pending queue since
-                 * we're expecting a reply.
+                 * Sleep on the pending queue if we're expecting a reply.
+                 * The spinlock ensures atomicity between the test of
+                 * req->rq_reply_bytes_recvd, and the call to rpc_sleep_on().
                  */
-                if (!req->rq_reply_bytes_recvd && rpc_reply_expected(task))
+                spin_lock(&xprt->recv_lock);
+                if (!req->rq_reply_bytes_recvd) {
                         rpc_sleep_on(&xprt->pending, task, xprt_timer);
-                req->rq_connect_cookie = xprt->connect_cookie;
+                        /*
+                         * Send an extra queue wakeup call if the
+                         * connection was dropped in case the call to
+                         * rpc_sleep_on() raced.
+                         */
+                        if (!xprt_connected(xprt))
+                                xprt_wake_pending_tasks(xprt, -ENOTCONN);
+                }
+                spin_unlock(&xprt->recv_lock);
         }
-        spin_unlock_bh(&xprt->transport_lock);
 }
 
 static void xprt_add_backlog(struct rpc_xprt *xprt, struct rpc_task *task)
diff --git a/net/sunrpc/xprtrdma/rpc_rdma.c b/net/sunrpc/xprtrdma/rpc_rdma.c
index ed34dc0f144c..a3f2ab283aeb 100644
--- a/net/sunrpc/xprtrdma/rpc_rdma.c
+++ b/net/sunrpc/xprtrdma/rpc_rdma.c
@@ -1408,11 +1408,7 @@ void rpcrdma_reply_handler(struct rpcrdma_rep *rep)
         dprintk("RPC:       %s: reply %p completes request %p (xid 0x%08x)\n",
                 __func__, rep, req, be32_to_cpu(rep->rr_xid));
 
-        if (list_empty(&req->rl_registered) &&
-            !test_bit(RPCRDMA_REQ_F_TX_RESOURCES, &req->rl_flags))
-                rpcrdma_complete_rqst(rep);
-        else
-                queue_work(rpcrdma_receive_wq, &rep->rr_work);
+        queue_work_on(req->rl_cpu, rpcrdma_receive_wq, &rep->rr_work);
         return;
 
 out_badstatus:
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index 646c24494ea7..6ee1ad8978f3 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -52,6 +52,7 @@
 #include <linux/slab.h>
 #include <linux/seq_file.h>
 #include <linux/sunrpc/addr.h>
+#include <linux/smp.h>
 
 #include "xprt_rdma.h"
 
@@ -656,6 +657,7 @@ xprt_rdma_allocate(struct rpc_task *task)
                 task->tk_pid, __func__, rqst->rq_callsize,
                 rqst->rq_rcvsize, req);
 
+        req->rl_cpu = smp_processor_id();
         req->rl_connect_cookie = 0;     /* our reserved value */
         rpcrdma_set_xprtdata(rqst, req);
         rqst->rq_buffer = req->rl_sendbuf->rg_base;
diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 710b3f77db82..8607c029c0dd 100644
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -83,7 +83,7 @@ rpcrdma_alloc_wq(void)
         struct workqueue_struct *recv_wq;
 
         recv_wq = alloc_workqueue("xprtrdma_receive",
-                                  WQ_MEM_RECLAIM | WQ_UNBOUND | WQ_HIGHPRI,
+                                  WQ_MEM_RECLAIM | WQ_HIGHPRI,
                                   0);
         if (!recv_wq)
                 return -ENOMEM;
diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
index 51686d9eac5f..1342f743f1c4 100644
--- a/net/sunrpc/xprtrdma/xprt_rdma.h
+++ b/net/sunrpc/xprtrdma/xprt_rdma.h
@@ -342,6 +342,7 @@ enum {
 struct rpcrdma_buffer;
 struct rpcrdma_req {
         struct list_head        rl_list;
+        int                     rl_cpu;
         unsigned int            rl_connect_cookie;
         struct rpcrdma_buffer   *rl_buffer;
         struct rpcrdma_rep      *rl_reply;