2 files changed, 38 insertions, 19 deletions

diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index dab09dac8fc7..f09b7db2c492 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -750,14 +750,16 @@ EXPORT_SYMBOL_GPL(rpc_shutdown_client);
 /*
  * Free an RPC client
  */
-static void
+static struct rpc_clnt *
 rpc_free_client(struct rpc_clnt *clnt)
 {
+        struct rpc_clnt *parent = NULL;
+
         dprintk_rcu("RPC:       destroying %s client for %s\n",
                         clnt->cl_program->name,
                         rcu_dereference(clnt->cl_xprt)->servername);
         if (clnt->cl_parent != clnt)
-                rpc_release_client(clnt->cl_parent);
+                parent = clnt->cl_parent;
         rpc_clnt_remove_pipedir(clnt);
         rpc_unregister_client(clnt);
         rpc_free_iostats(clnt->cl_metrics);
@@ -766,18 +768,17 @@ rpc_free_client(struct rpc_clnt *clnt)
         rpciod_down();
         rpc_free_clid(clnt);
         kfree(clnt);
+        return parent;
 }
 
 /*
  * Free an RPC client
  */
-static void
+static struct rpc_clnt * 
 rpc_free_auth(struct rpc_clnt *clnt)
 {
-        if (clnt->cl_auth == NULL) {
-                rpc_free_client(clnt);
-                return;
-        }
+        if (clnt->cl_auth == NULL)
+                return rpc_free_client(clnt);
 
         /*
          * Note: RPCSEC_GSS may need to send NULL RPC calls in order to
@@ -788,7 +789,8 @@ rpc_free_auth(struct rpc_clnt *clnt)
         rpcauth_release(clnt->cl_auth);
         clnt->cl_auth = NULL;
         if (atomic_dec_and_test(&clnt->cl_count))
-                rpc_free_client(clnt);
+                return rpc_free_client(clnt);
+        return NULL;
 }
 
 /*
@@ -799,10 +801,13 @@ rpc_release_client(struct rpc_clnt *clnt)
 {
         dprintk("RPC:       rpc_release_client(%p)\n", clnt);
 
-        if (list_empty(&clnt->cl_tasks))
-                wake_up(&destroy_wait);
-        if (atomic_dec_and_test(&clnt->cl_count))
-                rpc_free_auth(clnt);
+        do {
+                if (list_empty(&clnt->cl_tasks))
+                        wake_up(&destroy_wait);
+                if (!atomic_dec_and_test(&clnt->cl_count))
+                        break;
+                clnt = rpc_free_auth(clnt);
+        } while (clnt != NULL);
 }
 EXPORT_SYMBOL_GPL(rpc_release_client);
 
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 17c88928b7db..dd9d295813cf 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -393,8 +393,10 @@ static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen,
         return kernel_sendmsg(sock, &msg, NULL, 0, 0);
 }
 
-static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
+static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more, bool zerocopy)
 {
+        ssize_t (*do_sendpage)(struct socket *sock, struct page *page,
+                        int offset, size_t size, int flags);
         struct page **ppage;
         unsigned int remainder;
         int err, sent = 0;
@@ -403,6 +405,9 @@ static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned i
         base += xdr->page_base;
         ppage = xdr->pages + (base >> PAGE_SHIFT);
         base &= ~PAGE_MASK;
+        do_sendpage = sock->ops->sendpage;
+        if (!zerocopy)
+                do_sendpage = sock_no_sendpage;
         for(;;) {
                 unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
                 int flags = XS_SENDMSG_FLAGS;
@@ -410,7 +415,7 @@ static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned i
                 remainder -= len;
                 if (remainder != 0 || more)
                         flags |= MSG_MORE;
-                err = sock->ops->sendpage(sock, *ppage, base, len, flags);
+                err = do_sendpage(sock, *ppage, base, len, flags);
                 if (remainder == 0 || err != len)
                         break;
                 sent += err;
@@ -431,9 +436,10 @@ static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned i
  * @addrlen: UDP only -- length of destination address
  * @xdr: buffer containing this request
  * @base: starting position in the buffer
+ * @zerocopy: true if it is safe to use sendpage()
  *
  */
-static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
+static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base, bool zerocopy)
 {
         unsigned int remainder = xdr->len - base;
         int err, sent = 0;
@@ -461,7 +467,7 @@ static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen,
         if (base < xdr->page_len) {
                 unsigned int len = xdr->page_len - base;
                 remainder -= len;
-                err = xs_send_pagedata(sock, xdr, base, remainder != 0);
+                err = xs_send_pagedata(sock, xdr, base, remainder != 0, zerocopy);
                 if (remainder == 0 || err != len)
                         goto out;
                 sent += err;
@@ -564,7 +570,7 @@ static int xs_local_send_request(struct rpc_task *task)
                         req->rq_svec->iov_base, req->rq_svec->iov_len);
 
         status = xs_sendpages(transport->sock, NULL, 0,
-                                                xdr, req->rq_bytes_sent);
+                                                xdr, req->rq_bytes_sent, true);
         dprintk("RPC:       %s(%u) = %d\n",
                         __func__, xdr->len - req->rq_bytes_sent, status);
         if (likely(status >= 0)) {
@@ -620,7 +626,7 @@ static int xs_udp_send_request(struct rpc_task *task)
         status = xs_sendpages(transport->sock,
                               xs_addr(xprt),
                               xprt->addrlen, xdr,
-                              req->rq_bytes_sent);
+                              req->rq_bytes_sent, true);
 
         dprintk("RPC:       xs_udp_send_request(%u) = %d\n",
                         xdr->len - req->rq_bytes_sent, status);
@@ -693,6 +699,7 @@ static int xs_tcp_send_request(struct rpc_task *task)
         struct rpc_xprt *xprt = req->rq_xprt;
         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
         struct xdr_buf *xdr = &req->rq_snd_buf;
+        bool zerocopy = true;
         int status;
 
         xs_encode_stream_record_marker(&req->rq_snd_buf);
@@ -700,13 +707,20 @@ static int xs_tcp_send_request(struct rpc_task *task)
         xs_pktdump("packet data:",
                                 req->rq_svec->iov_base,
                                 req->rq_svec->iov_len);
+        /* Don't use zero copy if this is a resend. If the RPC call
+         * completes while the socket holds a reference to the pages,
+         * then we may end up resending corrupted data.
+         */
+        if (task->tk_flags & RPC_TASK_SENT)
+                zerocopy = false;
 
         /* Continue transmitting the packet/record. We must be careful
          * to cope with writespace callbacks arriving _after_ we have
          * called sendmsg(). */
         while (1) {
                 status = xs_sendpages(transport->sock,
-                                        NULL, 0, xdr, req->rq_bytes_sent);
+                                        NULL, 0, xdr, req->rq_bytes_sent,
+                                        zerocopy);
 
                 dprintk("RPC:       xs_tcp_send_request(%u) = %d\n",
                                 xdr->len - req->rq_bytes_sent, status);