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

[Benny Halevy]

On Nov. 28, 2007, 6:25 +0200, ricardo.labiaga at netapp.com wrote:
> 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?

you must disconnecting if you drop the request
yet I think we can so a better job with preallocating a request.
I'll send a proposal for that soonish...

Benny

> + 			 */
> +			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);

hrm, the lock isn't taken in the alldir != RPC_REPLY path

>  	xs_tcp_check_fraghdr(transport);
> +	return;
> +
> +error:
> +	transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
> +	spin_unlock(&xprt->transport_lock);

ditto

I'll send a patch to fix that

> +	return;
>  }
>  
>  static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)