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

diff --combined net/sunrpc/xprt.c
index 333b9d697ae5373d00c6001b9c7f75f3d6c0ed91,5b06f6906a27ea3b2ae0683a167dbde6fe1820ec..33b74fd8405185d906d07e315c9b5a83775e747d
--- 1/net/sunrpc/xprt.c
--- 2/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@@ -696,9 -696,9 +696,9 @@@ xprt_schedule_autodisconnect(struct rpc
  }
  
  static void
 -xprt_init_autodisconnect(unsigned long data)
 +xprt_init_autodisconnect(struct timer_list *t)
  {
 -      struct rpc_xprt *xprt = (struct rpc_xprt *)data;
 +      struct rpc_xprt *xprt = from_timer(xprt, t, timer);
  
        spin_lock(&xprt->transport_lock);
        if (!list_empty(&xprt->recv))
@@@ -1001,6 -1001,7 +1001,7 @@@ void xprt_transmit(struct rpc_task *tas
  {
        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 +1025,7 @@@
        } 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 +1049,28 @@@
        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)
@@@ -1334,7 -1344,7 +1344,7 @@@ void xprt_release(struct rpc_task *task
                rpc_count_iostats(task, task->tk_client->cl_metrics);
        spin_lock(&xprt->recv_lock);
        if (!list_empty(&req->rq_list)) {
 -              list_del(&req->rq_list);
 +              list_del_init(&req->rq_list);
                xprt_wait_on_pinned_rqst(req);
        }
        spin_unlock(&xprt->recv_lock);
@@@ -1423,9 -1433,10 +1433,9 @@@ found
                xprt->idle_timeout = 0;
        INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
        if (xprt_has_timer(xprt))
 -              setup_timer(&xprt->timer, xprt_init_autodisconnect,
 -                          (unsigned long)xprt);
 +              timer_setup(&xprt->timer, xprt_init_autodisconnect, 0);
        else
 -              init_timer(&xprt->timer);
 +              timer_setup(&xprt->timer, NULL, 0);
  
        if (strlen(args->servername) > RPC_MAXNETNAMELEN) {
                xprt_destroy(xprt);
@@@ -1445,23 -1456,6 +1455,23 @@@ out
        return xprt;
  }
  
 +static void xprt_destroy_cb(struct work_struct *work)
 +{
 +      struct rpc_xprt *xprt =
 +              container_of(work, struct rpc_xprt, task_cleanup);
 +
 +      rpc_xprt_debugfs_unregister(xprt);
 +      rpc_destroy_wait_queue(&xprt->binding);
 +      rpc_destroy_wait_queue(&xprt->pending);
 +      rpc_destroy_wait_queue(&xprt->sending);
 +      rpc_destroy_wait_queue(&xprt->backlog);
 +      kfree(xprt->servername);
 +      /*
 +       * Tear down transport state and free the rpc_xprt
 +       */
 +      xprt->ops->destroy(xprt);
 +}
 +
  /**
   * xprt_destroy - destroy an RPC transport, killing off all requests.
   * @xprt: transport to destroy
@@@ -1471,19 -1465,22 +1481,19 @@@ static void xprt_destroy(struct rpc_xpr
  {
        dprintk("RPC:       destroying transport %p\n", xprt);
  
 -      /* Exclude transport connect/disconnect handlers */
 +      /*
 +       * Exclude transport connect/disconnect handlers and autoclose
 +       */
        wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_UNINTERRUPTIBLE);
  
        del_timer_sync(&xprt->timer);
  
 -      rpc_xprt_debugfs_unregister(xprt);
 -      rpc_destroy_wait_queue(&xprt->binding);
 -      rpc_destroy_wait_queue(&xprt->pending);
 -      rpc_destroy_wait_queue(&xprt->sending);
 -      rpc_destroy_wait_queue(&xprt->backlog);
 -      cancel_work_sync(&xprt->task_cleanup);
 -      kfree(xprt->servername);
        /*
 -       * Tear down transport state and free the rpc_xprt
 +       * Destroy sockets etc from the system workqueue so they can
 +       * safely flush receive work running on rpciod.
         */
 -      xprt->ops->destroy(xprt);
 +      INIT_WORK(&xprt->task_cleanup, xprt_destroy_cb);
 +      schedule_work(&xprt->task_cleanup);
  }
  
  static void xprt_destroy_kref(struct kref *kref)