[pnfs] [PATCH 09/11] Update xs_tcp_read_request() to handle back channel callbacks.

[ricardo.labiaga at netapp.com]

Traditionally NFS clients only expect RPC replies on the open connections.  With
NFSv4.1, callbacks can arrive over an existing open connections.
Update xs_tcp_read_request() to queue callback requests onto a queue where
the callback service (a separate thread) is listening for the processing.
Original code by Rahul Iyer (iyer at netapp.com).

Signed-off-by: Ricardo Labiaga <ricardo.labiaga at netapp.com>
---
 net/sunrpc/xprtsock.c |   65 ++++++++++++++++++++++++++++++++++++++++---------
 1 files changed, 53 insertions(+), 12 deletions(-)

diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 8347aac..0eda706 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -35,6 +35,9 @@
 #include <linux/sunrpc/svcsock.h>
 #include <linux/sunrpc/xprtsock.h>
 #include <linux/file.h>
+#ifdef CONFIG_NFS_V4_1
+#include <linux/sunrpc/bc_xprt.h>
+#endif
 
 #include <net/sock.h>
 #include <net/checksum.h>
@@ -960,16 +963,34 @@ static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_rea
 	struct xdr_buf *rcvbuf;
 	size_t len;
 	ssize_t r;
-
-	/* Find and lock the request corresponding to this xid */
-	spin_lock(&xprt->transport_lock);
-	req = xprt_lookup_rqst(xprt, transport->tcp_xid);
-	if (!req) {
-		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
-		dprintk("RPC:       XID %08x request not found!\n",
-				ntohl(transport->tcp_xid));
-		spin_unlock(&xprt->transport_lock);
-		return;
+	u32 calldir;
+
+	calldir = ntohl(transport->tcp_calldir);
+	if (calldir == RPC_REPLY) {
+		/* Find and lock the request corresponding to this xid */
+		spin_lock(&xprt->transport_lock);
+		req = xprt_lookup_rqst(xprt, transport->tcp_xid);
+		if (!req) {
+			dprintk("RPC:       XID %08x request not found!\n",
+					ntohl(transport->tcp_xid));
+			goto error;
+		}
+	} else {
+		/* RPC_CALL */
+		if (xprt->bc_mempool) {
+			req = xprt_alloc_bc_request(xprt);
+		} else {
+			dprintk("RPC:       Unexpected callback dropped\n");
+			goto error;
+		}
+		if (req == NULL) {
+			/*
+			 * Drop the callback.
+			 * XXX Should we instead terminate the connection?
+ 			 */
+			dprintk("RPC:       Couldn't get rpc_rqst for the callback! Dropping callback...\n");
+			goto error;
+		}
 	}
 
 	rcvbuf = &req->rq_private_buf;
@@ -1026,10 +1047,30 @@ static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_rea
 	}
 
 out:
-	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
-		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
+	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) {
+		if (calldir == RPC_REPLY) {
+			xprt_complete_rqst(req->rq_task, transport->tcp_copied);
+		} else {
+			/*
+ 			 * 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.
+ 			 */
+			spin_lock(&xprt->serv->sv_cb_lock);
+			list_add(&req->rq_list, &xprt->serv->sv_cb_list);
+			spin_unlock(&xprt->serv->sv_cb_lock);
+			wake_up(&xprt->serv->sv_cb_waitq);
+		}
+	}
 	spin_unlock(&xprt->transport_lock);
 	xs_tcp_check_fraghdr(transport);
+	return;
+
+error:
+	transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
+	spin_unlock(&xprt->transport_lock);
+	return;
 }
 
 static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
-- 
1.5.2