7 files changed, 70 insertions, 30 deletions

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 4e6d406264a0..004a2f9dc432 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -379,6 +379,7 @@ static void svc_sock_setbufsize(struct socket *sock, unsigned int snd,
         lock_sock(sock->sk);
         sock->sk->sk_sndbuf = snd * 2;
         sock->sk->sk_rcvbuf = rcv * 2;
+        sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK|SOCK_RCVBUF_LOCK;
         release_sock(sock->sk);
 #endif
 }
@@ -831,6 +832,23 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
                 test_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags),
                 test_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags));
 
+        if (test_and_clear_bit(XPT_CHNGBUF, &svsk->sk_xprt.xpt_flags))
+                /* sndbuf needs to have room for one request
+                 * per thread, otherwise we can stall even when the
+                 * network isn't a bottleneck.
+                 *
+                 * We count all threads rather than threads in a
+                 * particular pool, which provides an upper bound
+                 * on the number of threads which will access the socket.
+                 *
+                 * rcvbuf just needs to be able to hold a few requests.
+                 * Normally they will be removed from the queue
+                 * as soon a a complete request arrives.
+                 */
+                svc_sock_setbufsize(svsk->sk_sock,
+                                    (serv->sv_nrthreads+3) * serv->sv_max_mesg,
+                                    3 * serv->sv_max_mesg);
+
         clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 
         /* Receive data. If we haven't got the record length yet, get
@@ -1078,6 +1096,15 @@ static void svc_tcp_init(struct svc_sock *svsk, struct svc_serv *serv)
 
                 tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
 
+                /* initialise setting must have enough space to
+                 * receive and respond to one request.
+                 * svc_tcp_recvfrom will re-adjust if necessary
+                 */
+                svc_sock_setbufsize(svsk->sk_sock,
+                                    3 * svsk->sk_xprt.xpt_server->sv_max_mesg,
+                                    3 * svsk->sk_xprt.xpt_server->sv_max_mesg);
+
+                set_bit(XPT_CHNGBUF, &svsk->sk_xprt.xpt_flags);
                 set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
                 if (sk->sk_state != TCP_ESTABLISHED)
                         set_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags);
@@ -1147,14 +1174,8 @@ static struct svc_sock *svc_setup_socket(struct svc_serv *serv,
         /* Initialize the socket */
         if (sock->type == SOCK_DGRAM)
                 svc_udp_init(svsk, serv);
-        else {
-                /* initialise setting must have enough space to
-                 * receive and respond to one request.
-                 */
-                svc_sock_setbufsize(svsk->sk_sock, 4 * serv->sv_max_mesg,
-                                        4 * serv->sv_max_mesg);
+        else
                 svc_tcp_init(svsk, serv);
-        }
 
         dprintk("svc: svc_setup_socket created %p (inet %p)\n",
                                 svsk, svsk->sk_sk);
diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index a0bfe53f1621..06ca058572f2 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -672,10 +672,8 @@ xprt_init_autodisconnect(unsigned long data)
         if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
                 goto out_abort;
         spin_unlock(&xprt->transport_lock);
-        if (xprt_connecting(xprt))
-                xprt_release_write(xprt, NULL);
-        else
-                queue_work(rpciod_workqueue, &xprt->task_cleanup);
+        set_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
+        queue_work(rpciod_workqueue, &xprt->task_cleanup);
         return;
 out_abort:
         spin_unlock(&xprt->transport_lock);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 629a28764da9..42a6f9f20285 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -265,7 +265,7 @@ static int fast_reg_read_chunks(struct svcxprt_rdma *xprt,
                 frmr->page_list->page_list[page_no] =
                         ib_dma_map_single(xprt->sc_cm_id->device,
                                           page_address(rqstp->rq_arg.pages[page_no]),
-                                          PAGE_SIZE, DMA_TO_DEVICE);
+                                          PAGE_SIZE, DMA_FROM_DEVICE);
                 if (ib_dma_mapping_error(xprt->sc_cm_id->device,
                                          frmr->page_list->page_list[page_no]))
                         goto fatal_err;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index 6c26a675435a..f11be72a1a80 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -128,7 +128,8 @@ static int fast_reg_xdr(struct svcxprt_rdma *xprt,
                 page_bytes -= sge_bytes;
 
                 frmr->page_list->page_list[page_no] =
-                        ib_dma_map_page(xprt->sc_cm_id->device, page, 0,
+                        ib_dma_map_single(xprt->sc_cm_id->device,
+                                          page_address(page),
                                           PAGE_SIZE, DMA_TO_DEVICE);
                 if (ib_dma_mapping_error(xprt->sc_cm_id->device,
                                          frmr->page_list->page_list[page_no]))
@@ -183,6 +184,7 @@ static int fast_reg_xdr(struct svcxprt_rdma *xprt,
 
  fatal_err:
         printk("svcrdma: Error fast registering memory for xprt %p\n", xprt);
+        vec->frmr = NULL;
         svc_rdma_put_frmr(xprt, frmr);
         return -EIO;
 }
@@ -516,6 +518,7 @@ static int send_reply(struct svcxprt_rdma *rdma,
                        "svcrdma: could not post a receive buffer, err=%d."
                        "Closing transport %p.\n", ret, rdma);
                 set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
+                svc_rdma_put_frmr(rdma, vec->frmr);
                 svc_rdma_put_context(ctxt, 0);
                 return -ENOTCONN;
         }
@@ -530,18 +533,17 @@ static int send_reply(struct svcxprt_rdma *rdma,
                 clear_bit(RDMACTXT_F_FAST_UNREG, &ctxt->flags);
 
         /* Prepare the SGE for the RPCRDMA Header */
+        ctxt->sge[0].lkey = rdma->sc_dma_lkey;
+        ctxt->sge[0].length = svc_rdma_xdr_get_reply_hdr_len(rdma_resp);
         ctxt->sge[0].addr =
-                ib_dma_map_page(rdma->sc_cm_id->device,
-                                page, 0, PAGE_SIZE, DMA_TO_DEVICE);
+                ib_dma_map_single(rdma->sc_cm_id->device, page_address(page),
+                                  ctxt->sge[0].length, DMA_TO_DEVICE);
         if (ib_dma_mapping_error(rdma->sc_cm_id->device, ctxt->sge[0].addr))
                 goto err;
         atomic_inc(&rdma->sc_dma_used);
 
         ctxt->direction = DMA_TO_DEVICE;
 
-        ctxt->sge[0].length = svc_rdma_xdr_get_reply_hdr_len(rdma_resp);
-        ctxt->sge[0].lkey = rdma->sc_dma_lkey;
-
         /* Determine how many of our SGE are to be transmitted */
         for (sge_no = 1; byte_count && sge_no < vec->count; sge_no++) {
                 sge_bytes = min_t(size_t, vec->sge[sge_no].iov_len, byte_count);
@@ -606,6 +608,7 @@ static int send_reply(struct svcxprt_rdma *rdma,
         return 0;
 
  err:
+        svc_rdma_unmap_dma(ctxt);
         svc_rdma_put_frmr(rdma, vec->frmr);
         svc_rdma_put_context(ctxt, 1);
         return -EIO;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 3d810e7df3fb..5151f9f6c573 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -500,8 +500,8 @@ int svc_rdma_post_recv(struct svcxprt_rdma *xprt)
                 BUG_ON(sge_no >= xprt->sc_max_sge);
                 page = svc_rdma_get_page();
                 ctxt->pages[sge_no] = page;
-                pa = ib_dma_map_page(xprt->sc_cm_id->device,
-                                     page, 0, PAGE_SIZE,
+                pa = ib_dma_map_single(xprt->sc_cm_id->device,
+                                     page_address(page), PAGE_SIZE,
                                      DMA_FROM_DEVICE);
                 if (ib_dma_mapping_error(xprt->sc_cm_id->device, pa))
                         goto err_put_ctxt;
@@ -520,8 +520,9 @@ int svc_rdma_post_recv(struct svcxprt_rdma *xprt)
         svc_xprt_get(&xprt->sc_xprt);
         ret = ib_post_recv(xprt->sc_qp, &recv_wr, &bad_recv_wr);
         if (ret) {
-                svc_xprt_put(&xprt->sc_xprt);
+                svc_rdma_unmap_dma(ctxt);
                 svc_rdma_put_context(ctxt, 1);
+                svc_xprt_put(&xprt->sc_xprt);
         }
         return ret;
 
@@ -1314,8 +1315,8 @@ void svc_rdma_send_error(struct svcxprt_rdma *xprt, struct rpcrdma_msg *rmsgp,
         length = svc_rdma_xdr_encode_error(xprt, rmsgp, err, va);
 
         /* Prepare SGE for local address */
-        sge.addr = ib_dma_map_page(xprt->sc_cm_id->device,
-                                   p, 0, PAGE_SIZE, DMA_FROM_DEVICE);
+        sge.addr = ib_dma_map_single(xprt->sc_cm_id->device,
+                                   page_address(p), PAGE_SIZE, DMA_FROM_DEVICE);
         if (ib_dma_mapping_error(xprt->sc_cm_id->device, sge.addr)) {
                 put_page(p);
                 return;
@@ -1342,7 +1343,7 @@ void svc_rdma_send_error(struct svcxprt_rdma *xprt, struct rpcrdma_msg *rmsgp,
         if (ret) {
                 dprintk("svcrdma: Error %d posting send for protocol error\n",
                         ret);
-                ib_dma_unmap_page(xprt->sc_cm_id->device,
+                ib_dma_unmap_single(xprt->sc_cm_id->device,
                                   sge.addr, PAGE_SIZE,
                                   DMA_FROM_DEVICE);
                 svc_rdma_put_context(ctxt, 1);
diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 3b21e0cc5e69..465aafc2007f 100644
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -1495,7 +1495,8 @@ rpcrdma_register_frmr_external(struct rpcrdma_mr_seg *seg,
         frmr_wr.wr.fast_reg.page_shift = PAGE_SHIFT;
         frmr_wr.wr.fast_reg.length = i << PAGE_SHIFT;
         frmr_wr.wr.fast_reg.access_flags = (writing ?
-                                IB_ACCESS_REMOTE_WRITE : IB_ACCESS_REMOTE_READ);
+                                IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE :
+                                IB_ACCESS_REMOTE_READ);
         frmr_wr.wr.fast_reg.rkey = seg1->mr_chunk.rl_mw->r.frmr.fr_mr->rkey;
         DECR_CQCOUNT(&r_xprt->rx_ep);
 
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index d40ff50887aa..e18596146013 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -807,6 +807,9 @@ static void xs_reset_transport(struct sock_xprt *transport)
  *
  * This is used when all requests are complete; ie, no DRC state remains
  * on the server we want to save.
+ *
+ * The caller _must_ be holding XPRT_LOCKED in order to avoid issues with
+ * xs_reset_transport() zeroing the socket from underneath a writer.
  */
 static void xs_close(struct rpc_xprt *xprt)
 {
@@ -824,6 +827,14 @@ static void xs_close(struct rpc_xprt *xprt)
         xprt_disconnect_done(xprt);
 }
 
+static void xs_tcp_close(struct rpc_xprt *xprt)
+{
+        if (test_and_clear_bit(XPRT_CONNECTION_CLOSE, &xprt->state))
+                xs_close(xprt);
+        else
+                xs_tcp_shutdown(xprt);
+}
+
 /**
  * xs_destroy - prepare to shutdown a transport
  * @xprt: doomed transport
@@ -1772,6 +1783,15 @@ static void xs_tcp_setup_socket(struct rpc_xprt *xprt,
                         xprt, -status, xprt_connected(xprt),
                         sock->sk->sk_state);
         switch (status) {
+        default:
+                printk("%s: connect returned unhandled error %d\n",
+                        __func__, status);
+        case -EADDRNOTAVAIL:
+                /* We're probably in TIME_WAIT. Get rid of existing socket,
+                 * and retry
+                 */
+                set_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
+                xprt_force_disconnect(xprt);
         case -ECONNREFUSED:
         case -ECONNRESET:
         case -ENETUNREACH:
@@ -1782,10 +1802,6 @@ static void xs_tcp_setup_socket(struct rpc_xprt *xprt,
                 xprt_clear_connecting(xprt);
                 return;
         }
-        /* get rid of existing socket, and retry */
-        xs_tcp_shutdown(xprt);
-        printk("%s: connect returned unhandled error %d\n",
-                        __func__, status);
 out_eagain:
         status = -EAGAIN;
 out:
@@ -1994,7 +2010,7 @@ static struct rpc_xprt_ops xs_tcp_ops = {
         .buf_free               = rpc_free,
         .send_request           = xs_tcp_send_request,
         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
-        .close                  = xs_tcp_shutdown,
+        .close                  = xs_tcp_close,
         .destroy                = xs_destroy,
         .print_stats            = xs_tcp_print_stats,
 };