[PATCH RFC 01/10] nfsd41: sunrpc: svc_tcp_recv_record()

[PATCH RFC 01/10] nfsd41: sunrpc: svc_tcp_recv_record()

[Benny Halevy]

From: Alexandros Batsakis <batsakis@netapp.com>

Factor functionality out of svc_tcp_recvfrom() to simplify routine

Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 net/sunrpc/svcsock.c |   79 ++++++++++++++++++++++++++++++++-----------------
 1 files changed, 51 insertions(+), 28 deletions(-)

diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 23128ee..5a5bc8b 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -826,21 +826,15 @@ failed:
 }
 
 /*
- * Receive data from a TCP socket.
+ * Receive data.
+ * If we haven't gotten the record length yet, get the next four bytes.
+ * Otherwise try to gobble up as much as possible up to the complete
+ * record length.
  */
-static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
+static int svc_tcp_recv_record(struct svc_sock *svsk, struct svc_rqst *rqstp)
 {
-	struct svc_sock	*svsk =
-		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
 	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
-	int		len;
-	struct kvec *vec;
-	int pnum, vlen;
-
-	dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
-		svsk, test_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags),
-		test_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags),
-		test_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags));
+	int len;
 
 	if (test_and_clear_bit(XPT_CHNGBUF, &svsk->sk_xprt.xpt_flags))
 		/* sndbuf needs to have room for one request
@@ -861,10 +855,6 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 
 	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 
-	/* Receive data. If we haven't got the record length yet, get
-	 * the next four bytes. Otherwise try to gobble up as much as
-	 * possible up to the complete record length.
-	 */
 	if (svsk->sk_tcplen < sizeof(rpc_fraghdr)) {
 		int		want = sizeof(rpc_fraghdr) - svsk->sk_tcplen;
 		struct kvec	iov;
@@ -879,7 +869,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 			dprintk("svc: short recvfrom while reading record "
 				"length (%d of %d)\n", len, want);
 			svc_xprt_received(&svsk->sk_xprt);
-			return -EAGAIN; /* record header not complete */
+			goto err_again; /* record header not complete */
 		}
 
 		svsk->sk_reclen = ntohl(svsk->sk_reclen);
@@ -894,6 +884,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 					"per record not supported\n");
 			goto err_delete;
 		}
+
 		svsk->sk_reclen &= RPC_FRAGMENT_SIZE_MASK;
 		dprintk("svc: TCP record, %d bytes\n", svsk->sk_reclen);
 		if (svsk->sk_reclen > serv->sv_max_mesg) {
@@ -914,11 +905,45 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 		dprintk("svc: incomplete TCP record (%d of %d)\n",
 			len, svsk->sk_reclen);
 		svc_xprt_received(&svsk->sk_xprt);
-		return -EAGAIN;	/* record not complete */
+		goto err_again;	/* record not complete */
 	}
 	len = svsk->sk_reclen;
 	set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 
+	return len;
+ error:
+	if (len == -EAGAIN) {
+		dprintk("RPC: TCP recv_record got EAGAIN\n");
+		svc_xprt_received(&svsk->sk_xprt);
+	}
+	return len;
+ err_delete:
+	set_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags);
+ err_again:
+	return -EAGAIN;
+}
+
+/*
+ * Receive data from a TCP socket.
+ */
+static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
+{
+	struct svc_sock	*svsk =
+		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
+	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
+	int		len;
+	struct kvec *vec;
+	int pnum, vlen;
+
+	dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
+		svsk, test_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags),
+		test_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags),
+		test_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags));
+
+	len = svc_tcp_recv_record(svsk, rqstp);
+	if (len < 0)
+		goto error;
+
 	vec = rqstp->rq_vec;
 	vec[0] = rqstp->rq_arg.head[0];
 	vlen = PAGE_SIZE;
@@ -934,7 +959,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	/* Now receive data */
 	len = svc_recvfrom(rqstp, vec, pnum, len);
 	if (len < 0)
-		goto error;
+		goto err_again;
 
 	dprintk("svc: TCP complete record (%d bytes)\n", len);
 	rqstp->rq_arg.len = len;
@@ -960,21 +985,19 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 
 	return len;
 
- err_delete:
-	set_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags);
-	return -EAGAIN;
-
- error:
+err_again:
 	if (len == -EAGAIN) {
 		dprintk("RPC: TCP recvfrom got EAGAIN\n");
 		svc_xprt_received(&svsk->sk_xprt);
-	} else {
+		return len;
+	}
+error:
+	if (len != -EAGAIN) {
 		printk(KERN_NOTICE "%s: recvfrom returned errno %d\n",
 		       svsk->sk_xprt.xpt_server->sv_name, -len);
-		goto err_delete;
+		set_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags);
 	}
-
-	return len;
+	return -EAGAIN;
 }
 
 /*
-- 
1.6.4

Re: [PATCH RFC 01/10] nfsd41: sunrpc: svc_tcp_recv_record()

[J. Bruce Fields]

On Thu, Aug 20, 2009 at 03:34:19AM +0300, Benny Halevy wrote:
> From: Alexandros Batsakis <batsakis@netapp.com>
> 
> Factor functionality out of svc_tcp_recvfrom() to simplify routine

Thanks, applied.

--b.

> 
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> ---
>  net/sunrpc/svcsock.c |   79 ++++++++++++++++++++++++++++++++-----------------
>  1 files changed, 51 insertions(+), 28 deletions(-)
> 
> diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
> index 23128ee..5a5bc8b 100644
> --- a/net/sunrpc/svcsock.c
> +++ b/net/sunrpc/svcsock.c
> @@ -826,21 +826,15 @@ failed:
>  }
>  
>  /*
> - * Receive data from a TCP socket.
> + * Receive data.
> + * If we haven't gotten the record length yet, get the next four bytes.
> + * Otherwise try to gobble up as much as possible up to the complete
> + * record length.
>   */
> -static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
> +static int svc_tcp_recv_record(struct svc_sock *svsk, struct svc_rqst *rqstp)
>  {
> -	struct svc_sock	*svsk =
> -		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
>  	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
> -	int		len;
> -	struct kvec *vec;
> -	int pnum, vlen;
> -
> -	dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
> -		svsk, test_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags),
> -		test_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags),
> -		test_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags));
> +	int len;
>  
>  	if (test_and_clear_bit(XPT_CHNGBUF, &svsk->sk_xprt.xpt_flags))
>  		/* sndbuf needs to have room for one request
> @@ -861,10 +855,6 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  
>  	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
>  
> -	/* Receive data. If we haven't got the record length yet, get
> -	 * the next four bytes. Otherwise try to gobble up as much as
> -	 * possible up to the complete record length.
> -	 */
>  	if (svsk->sk_tcplen < sizeof(rpc_fraghdr)) {
>  		int		want = sizeof(rpc_fraghdr) - svsk->sk_tcplen;
>  		struct kvec	iov;
> @@ -879,7 +869,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  			dprintk("svc: short recvfrom while reading record "
>  				"length (%d of %d)\n", len, want);
>  			svc_xprt_received(&svsk->sk_xprt);
> -			return -EAGAIN; /* record header not complete */
> +			goto err_again; /* record header not complete */
>  		}
>  
>  		svsk->sk_reclen = ntohl(svsk->sk_reclen);
> @@ -894,6 +884,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  					"per record not supported\n");
>  			goto err_delete;
>  		}
> +
>  		svsk->sk_reclen &= RPC_FRAGMENT_SIZE_MASK;
>  		dprintk("svc: TCP record, %d bytes\n", svsk->sk_reclen);
>  		if (svsk->sk_reclen > serv->sv_max_mesg) {
> @@ -914,11 +905,45 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  		dprintk("svc: incomplete TCP record (%d of %d)\n",
>  			len, svsk->sk_reclen);
>  		svc_xprt_received(&svsk->sk_xprt);
> -		return -EAGAIN;	/* record not complete */
> +		goto err_again;	/* record not complete */
>  	}
>  	len = svsk->sk_reclen;
>  	set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
>  
> +	return len;
> + error:
> +	if (len == -EAGAIN) {
> +		dprintk("RPC: TCP recv_record got EAGAIN\n");
> +		svc_xprt_received(&svsk->sk_xprt);
> +	}
> +	return len;
> + err_delete:
> +	set_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags);
> + err_again:
> +	return -EAGAIN;
> +}
> +
> +/*
> + * Receive data from a TCP socket.
> + */
> +static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
> +{
> +	struct svc_sock	*svsk =
> +		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
> +	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
> +	int		len;
> +	struct kvec *vec;
> +	int pnum, vlen;
> +
> +	dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
> +		svsk, test_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags),
> +		test_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags),
> +		test_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags));
> +
> +	len = svc_tcp_recv_record(svsk, rqstp);
> +	if (len < 0)
> +		goto error;
> +
>  	vec = rqstp->rq_vec;
>  	vec[0] = rqstp->rq_arg.head[0];
>  	vlen = PAGE_SIZE;
> @@ -934,7 +959,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  	/* Now receive data */
>  	len = svc_recvfrom(rqstp, vec, pnum, len);
>  	if (len < 0)
> -		goto error;
> +		goto err_again;
>  
>  	dprintk("svc: TCP complete record (%d bytes)\n", len);
>  	rqstp->rq_arg.len = len;
> @@ -960,21 +985,19 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  
>  	return len;
>  
> - err_delete:
> -	set_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags);
> -	return -EAGAIN;
> -
> - error:
> +err_again:
>  	if (len == -EAGAIN) {
>  		dprintk("RPC: TCP recvfrom got EAGAIN\n");
>  		svc_xprt_received(&svsk->sk_xprt);
> -	} else {
> +		return len;
> +	}
> +error:
> +	if (len != -EAGAIN) {
>  		printk(KERN_NOTICE "%s: recvfrom returned errno %d\n",
>  		       svsk->sk_xprt.xpt_server->sv_name, -len);
> -		goto err_delete;
> +		set_bit(XPT_CLOSE, &svsk->sk_xprt.xpt_flags);
>  	}
> -
> -	return len;
> +	return -EAGAIN;
>  }
>  
>  /*
> -- 
> 1.6.4
> 

[PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-side backchannel handling

[Benny Halevy]

From: Rahul Iyer <iyer@netapp.com>

Signed-off-by: Rahul Iyer <iyer@netapp.com>
Signed-off-by: Mike Sager <sager@netapp.com>
Signed-off-by: Marc Eshel <eshel@almaden.ibm.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

When the call direction is a reply, copy the xid and call direction into the
req->rq_private_buf.head[0].iov_base otherwise rpc_verify_header returns
rpc_garbage.

Signed-off-by: Andy Adamson <andros@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[get rid of CONFIG_NFSD_V4_1]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[sunrpc: refactoring of svc_tcp_recvfrom]
Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: create common send routine for the fore and the back channels]
Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: Use free_page() to free server backchannel pages]
Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: Document server backchannel locking]
Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: remove bc_connect_worker()]
Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: Define xprt_server_backchannel()[
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: remove bc_close and bc_init_auto_disconnect dummy functions]
Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: eliminate unneeded switch statement in xs_setup_tcp()]
Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: Don't auto close the server backchannel connection]
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: sunrpc: Remove unused functions]
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: change bc_sock to bc_xprt]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 include/linux/sunrpc/clnt.h    |    1 +
 include/linux/sunrpc/svcsock.h |    1 +
 include/linux/sunrpc/xprt.h    |    7 ++
 net/sunrpc/clnt.c              |    1 +
 net/sunrpc/sunrpc.h            |    4 +
 net/sunrpc/svcsock.c           |  172 +++++++++++++++++++++++++++-------
 net/sunrpc/xprt.c              |   16 +++-
 net/sunrpc/xprtsock.c          |  203 ++++++++++++++++++++++++++++++++++++++--
 8 files changed, 361 insertions(+), 44 deletions(-)

diff --git a/include/linux/sunrpc/clnt.h b/include/linux/sunrpc/clnt.h
index 37881f1..d904889 100644
--- a/include/linux/sunrpc/clnt.h
+++ b/include/linux/sunrpc/clnt.h
@@ -110,6 +110,7 @@ struct rpc_create_args {
 	rpc_authflavor_t	authflavor;
 	unsigned long		flags;
 	char			*client_name;
+	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
 };
 
 /* Values for "flags" field */
diff --git a/include/linux/sunrpc/svcsock.h b/include/linux/sunrpc/svcsock.h
index 04dba23..4b854e2 100644
--- a/include/linux/sunrpc/svcsock.h
+++ b/include/linux/sunrpc/svcsock.h
@@ -28,6 +28,7 @@ struct svc_sock {
 	/* private TCP part */
 	u32			sk_reclen;	/* length of record */
 	u32			sk_tcplen;	/* current read length */
+	struct rpc_xprt	       *sk_bc_xprt;	/* NFSv4.1 backchannel xprt */
 };
 
 /*
diff --git a/include/linux/sunrpc/xprt.h b/include/linux/sunrpc/xprt.h
index 1175d58..75cb619 100644
--- a/include/linux/sunrpc/xprt.h
+++ b/include/linux/sunrpc/xprt.h
@@ -181,6 +181,7 @@ struct rpc_xprt {
 	spinlock_t		reserve_lock;	/* lock slot table */
 	u32			xid;		/* Next XID value to use */
 	struct rpc_task *	snd_task;	/* Task blocked in send */
+	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
 #if defined(CONFIG_NFS_V4_1)
 	struct svc_serv		*bc_serv;       /* The RPC service which will */
 						/* process the callback */
@@ -233,6 +234,7 @@ struct xprt_create {
 	struct sockaddr *	srcaddr;	/* optional local address */
 	struct sockaddr *	dstaddr;	/* remote peer address */
 	size_t			addrlen;
+	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
 };
 
 struct xprt_class {
@@ -368,6 +370,11 @@ static inline int xprt_test_and_set_binding(struct rpc_xprt *xprt)
 	return test_and_set_bit(XPRT_BINDING, &xprt->state);
 }
 
+static inline int xprt_server_backchannel(struct rpc_xprt *xprt)
+{
+	return xprt->bc_xprt != NULL;
+}
+
 #endif /* __KERNEL__*/
 
 #endif /* _LINUX_SUNRPC_XPRT_H */
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index ebfcf9b..f45d3bb 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -270,6 +270,7 @@ struct rpc_clnt *rpc_create(struct rpc_create_args *args)
 		.srcaddr = args->saddress,
 		.dstaddr = args->address,
 		.addrlen = args->addrsize,
+		.bc_xprt = args->bc_xprt,
 	};
 	char servername[48];
 
diff --git a/net/sunrpc/sunrpc.h b/net/sunrpc/sunrpc.h
index 5d9dd74..b1b2e64 100644
--- a/net/sunrpc/sunrpc.h
+++ b/net/sunrpc/sunrpc.h
@@ -33,5 +33,9 @@ static inline int rpc_reply_expected(struct rpc_task *task)
 		(task->tk_msg.rpc_proc->p_decode != NULL);
 }
 
+int svc_send_common(struct socket *sock, struct xdr_buf *xdr,
+		    struct page *headpage, unsigned long headoffset,
+		    struct page *tailpage, unsigned long tailoffset);
+
 #endif /* _NET_SUNRPC_SUNRPC_H */
 
diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 5a5bc8b..258306e 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -49,6 +49,7 @@
 #include <linux/sunrpc/msg_prot.h>
 #include <linux/sunrpc/svcsock.h>
 #include <linux/sunrpc/stats.h>
+#include <linux/sunrpc/xprt.h>
 
 #define RPCDBG_FACILITY	RPCDBG_SVCXPRT
 
@@ -153,49 +154,27 @@ static void svc_set_cmsg_data(struct svc_rqst *rqstp, struct cmsghdr *cmh)
 }
 
 /*
- * Generic sendto routine
+ * send routine intended to be shared by the fore- and back-channel
  */
-static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
+int svc_send_common(struct socket *sock, struct xdr_buf *xdr,
+		    struct page *headpage, unsigned long headoffset,
+		    struct page *tailpage, unsigned long tailoffset)
 {
-	struct svc_sock	*svsk =
-		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
-	struct socket	*sock = svsk->sk_sock;
-	int		slen;
-	union {
-		struct cmsghdr	hdr;
-		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
-	} buffer;
-	struct cmsghdr *cmh = &buffer.hdr;
-	int		len = 0;
 	int		result;
 	int		size;
 	struct page	**ppage = xdr->pages;
 	size_t		base = xdr->page_base;
 	unsigned int	pglen = xdr->page_len;
 	unsigned int	flags = MSG_MORE;
-	RPC_IFDEBUG(char buf[RPC_MAX_ADDRBUFLEN]);
+	int		slen;
+	int		len = 0;
 
 	slen = xdr->len;
 
-	if (rqstp->rq_prot == IPPROTO_UDP) {
-		struct msghdr msg = {
-			.msg_name	= &rqstp->rq_addr,
-			.msg_namelen	= rqstp->rq_addrlen,
-			.msg_control	= cmh,
-			.msg_controllen	= sizeof(buffer),
-			.msg_flags	= MSG_MORE,
-		};
-
-		svc_set_cmsg_data(rqstp, cmh);
-
-		if (sock_sendmsg(sock, &msg, 0) < 0)
-			goto out;
-	}
-
 	/* send head */
 	if (slen == xdr->head[0].iov_len)
 		flags = 0;
-	len = kernel_sendpage(sock, rqstp->rq_respages[0], 0,
+	len = kernel_sendpage(sock, headpage, headoffset,
 				  xdr->head[0].iov_len, flags);
 	if (len != xdr->head[0].iov_len)
 		goto out;
@@ -219,16 +198,58 @@ static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
 		base = 0;
 		ppage++;
 	}
+
 	/* send tail */
 	if (xdr->tail[0].iov_len) {
-		result = kernel_sendpage(sock, rqstp->rq_respages[0],
-					     ((unsigned long)xdr->tail[0].iov_base)
-						& (PAGE_SIZE-1),
-					     xdr->tail[0].iov_len, 0);
-
+		result = kernel_sendpage(sock, tailpage, tailoffset,
+				   xdr->tail[0].iov_len, 0);
 		if (result > 0)
 			len += result;
 	}
+
+out:
+	return len;
+}
+
+
+/*
+ * Generic sendto routine
+ */
+static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
+{
+	struct svc_sock	*svsk =
+		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
+	struct socket	*sock = svsk->sk_sock;
+	union {
+		struct cmsghdr	hdr;
+		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
+	} buffer;
+	struct cmsghdr *cmh = &buffer.hdr;
+	int		len = 0;
+	unsigned long tailoff;
+	unsigned long headoff;
+	RPC_IFDEBUG(char buf[RPC_MAX_ADDRBUFLEN]);
+
+	if (rqstp->rq_prot == IPPROTO_UDP) {
+		struct msghdr msg = {
+			.msg_name	= &rqstp->rq_addr,
+			.msg_namelen	= rqstp->rq_addrlen,
+			.msg_control	= cmh,
+			.msg_controllen	= sizeof(buffer),
+			.msg_flags	= MSG_MORE,
+		};
+
+		svc_set_cmsg_data(rqstp, cmh);
+
+		if (sock_sendmsg(sock, &msg, 0) < 0)
+			goto out;
+	}
+
+	tailoff = ((unsigned long)xdr->tail[0].iov_base) & (PAGE_SIZE-1);
+	headoff = 0;
+	len = svc_send_common(sock, xdr, rqstp->rq_respages[0], headoff,
+			       rqstp->rq_respages[0], tailoff);
+
 out:
 	dprintk("svc: socket %p sendto([%p %Zu... ], %d) = %d (addr %s)\n",
 		svsk, xdr->head[0].iov_base, xdr->head[0].iov_len,
@@ -923,6 +944,57 @@ static int svc_tcp_recv_record(struct svc_sock *svsk, struct svc_rqst *rqstp)
 	return -EAGAIN;
 }
 
+static int svc_process_calldir(struct svc_sock *svsk, struct svc_rqst *rqstp,
+			       struct rpc_rqst **reqpp, struct kvec *vec)
+{
+	struct rpc_rqst *req = NULL;
+	u32 *p;
+	u32 xid;
+	u32 calldir;
+	int len;
+
+	len = svc_recvfrom(rqstp, vec, 1, 8);
+	if (len < 0)
+		goto error;
+
+	p = (u32 *)rqstp->rq_arg.head[0].iov_base;
+	xid = *p++;
+	calldir = *p;
+
+	if (calldir == 0) {
+		/* REQUEST is the most common case */
+		vec[0] = rqstp->rq_arg.head[0];
+	} else {
+		/* REPLY */
+		if (svsk->sk_bc_xprt)
+			req = xprt_lookup_rqst(svsk->sk_bc_xprt, xid);
+
+		if (!req) {
+			printk(KERN_NOTICE
+				"%s: Got unrecognized reply: "
+				"calldir 0x%x sk_bc_xprt %p xid %08x\n",
+				__func__, ntohl(calldir),
+				svsk->sk_bc_xprt, xid);
+			vec[0] = rqstp->rq_arg.head[0];
+			goto out;
+		}
+
+		memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
+		       sizeof(struct xdr_buf));
+		/* copy the xid and call direction */
+		memcpy(req->rq_private_buf.head[0].iov_base,
+		       rqstp->rq_arg.head[0].iov_base, 8);
+		vec[0] = req->rq_private_buf.head[0];
+	}
+ out:
+	vec[0].iov_base += 8;
+	vec[0].iov_len -= 8;
+	len = svsk->sk_reclen - 8;
+ error:
+	*reqpp = req;
+	return len;
+}
+
 /*
  * Receive data from a TCP socket.
  */
@@ -934,6 +1006,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	int		len;
 	struct kvec *vec;
 	int pnum, vlen;
+	struct rpc_rqst *req = NULL;
 
 	dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
 		svsk, test_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags),
@@ -947,9 +1020,27 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	vec = rqstp->rq_vec;
 	vec[0] = rqstp->rq_arg.head[0];
 	vlen = PAGE_SIZE;
+
+	/*
+	 * We have enough data for the whole tcp record. Let's try and read the
+	 * first 8 bytes to get the xid and the call direction. We can use this
+	 * to figure out if this is a call or a reply to a callback. If
+	 * sk_reclen is < 8 (xid and calldir), then this is a malformed packet.
+	 * In that case, don't bother with the calldir and just read the data.
+	 * It will be rejected in svc_process.
+	 */
+	if (len >= 8) {
+		len = svc_process_calldir(svsk, rqstp, &req, vec);
+		if (len < 0)
+			goto err_again;
+		vlen -= 8;
+	}
+
 	pnum = 1;
 	while (vlen < len) {
-		vec[pnum].iov_base = page_address(rqstp->rq_pages[pnum]);
+		vec[pnum].iov_base = (req) ?
+			page_address(req->rq_private_buf.pages[pnum - 1]) :
+			page_address(rqstp->rq_pages[pnum]);
 		vec[pnum].iov_len = PAGE_SIZE;
 		pnum++;
 		vlen += PAGE_SIZE;
@@ -961,6 +1052,16 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	if (len < 0)
 		goto err_again;
 
+	/*
+	 * Account for the 8 bytes we read earlier
+	 */
+	len += 8;
+
+	if (req) {
+		xprt_complete_rqst(req->rq_task, len);
+		len = 0;
+		goto out;
+	}
 	dprintk("svc: TCP complete record (%d bytes)\n", len);
 	rqstp->rq_arg.len = len;
 	rqstp->rq_arg.page_base = 0;
@@ -974,6 +1075,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
 	rqstp->rq_xprt_ctxt   = NULL;
 	rqstp->rq_prot	      = IPPROTO_TCP;
 
+out:
 	/* Reset TCP read info */
 	svsk->sk_reclen = 0;
 	svsk->sk_tcplen = 0;
diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index f412a85..b6d4d0d 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -599,6 +599,9 @@ static void xprt_autoclose(struct work_struct *work)
 	struct rpc_xprt *xprt =
 		container_of(work, struct rpc_xprt, task_cleanup);
 
+	if (xprt_server_backchannel(xprt))
+		return;
+
 	xprt->ops->close(xprt);
 	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
 	xprt_release_write(xprt, NULL);
@@ -669,6 +672,9 @@ xprt_init_autodisconnect(unsigned long data)
 {
 	struct rpc_xprt *xprt = (struct rpc_xprt *)data;
 
+	if (xprt_server_backchannel(xprt))
+		return;
+
 	spin_lock(&xprt->transport_lock);
 	if (!list_empty(&xprt->recv) || xprt->shutdown)
 		goto out_abort;
@@ -1083,7 +1089,8 @@ found:
 
 	INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
 	setup_timer(&xprt->timer, xprt_init_autodisconnect,
-			(unsigned long)xprt);
+		    (unsigned long)xprt);
+
 	xprt->last_used = jiffies;
 	xprt->cwnd = RPC_INITCWND;
 	xprt->bind_index = 0;
@@ -1103,6 +1110,13 @@ found:
 	dprintk("RPC:       created transport %p with %u slots\n", xprt,
 			xprt->max_reqs);
 
+	/*
+	 * Since we don't want connections for the backchannel, we set
+	 * the xprt status to connected
+	 */
+	if (args->bc_xprt)
+		xprt_set_connected(xprt);
+
 	return xprt;
 }
 
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 83c73c4..6e6f939 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -32,6 +32,7 @@
 #include <linux/tcp.h>
 #include <linux/sunrpc/clnt.h>
 #include <linux/sunrpc/sched.h>
+#include <linux/sunrpc/svcsock.h>
 #include <linux/sunrpc/xprtsock.h>
 #include <linux/file.h>
 #ifdef CONFIG_NFS_V4_1
@@ -43,6 +44,7 @@
 #include <net/udp.h>
 #include <net/tcp.h>
 
+#include "sunrpc.h"
 /*
  * xprtsock tunables
  */
@@ -2156,6 +2158,133 @@ static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
 			xprt->stat.bklog_u);
 }
 
+struct rpc_buffer {
+	size_t	len;
+	char	data[];
+};
+/*
+ * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
+ * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
+ * to use the server side send routines.
+ */
+void *bc_malloc(struct rpc_task *task, size_t size)
+{
+	struct page *page;
+	struct rpc_buffer *buf;
+
+	BUG_ON(size > PAGE_SIZE - sizeof(struct rpc_buffer));
+	page = alloc_page(GFP_KERNEL);
+
+	if (!page)
+		return NULL;
+
+	buf = page_address(page);
+	buf->len = PAGE_SIZE;
+
+	return buf->data;
+}
+
+/*
+ * Free the space allocated in the bc_alloc routine
+ */
+void bc_free(void *buffer)
+{
+	struct rpc_buffer *buf;
+
+	if (!buffer)
+		return;
+
+	buf = container_of(buffer, struct rpc_buffer, data);
+	free_page((unsigned long)buf);
+}
+
+/*
+ * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
+ * held. Borrows heavily from svc_tcp_sendto and xs_tcp_semd_request.
+ */
+static int bc_sendto(struct rpc_rqst *req)
+{
+	int len;
+	struct xdr_buf *xbufp = &req->rq_snd_buf;
+	struct rpc_xprt *xprt = req->rq_xprt;
+	struct sock_xprt *transport =
+				container_of(xprt, struct sock_xprt, xprt);
+	struct socket *sock = transport->sock;
+	unsigned long headoff;
+	unsigned long tailoff;
+
+	/*
+	 * Set up the rpc header and record marker stuff
+	 */
+	xs_encode_tcp_record_marker(xbufp);
+
+	tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
+	headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
+	len = svc_send_common(sock, xbufp,
+			      virt_to_page(xbufp->head[0].iov_base), headoff,
+			      xbufp->tail[0].iov_base, tailoff);
+
+	if (len != xbufp->len) {
+		printk(KERN_NOTICE "Error sending entire callback!\n");
+		len = -EAGAIN;
+	}
+
+	return len;
+}
+
+/*
+ * The send routine. Borrows from svc_send
+ */
+static int bc_send_request(struct rpc_task *task)
+{
+	struct rpc_rqst *req = task->tk_rqstp;
+	struct svc_xprt	*xprt;
+	struct svc_sock         *svsk;
+	u32                     len;
+
+	dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
+	/*
+	 * Get the server socket associated with this callback xprt
+	 */
+	xprt = req->rq_xprt->bc_xprt;
+	svsk = container_of(xprt, struct svc_sock, sk_xprt);
+
+	/*
+	 * Grab the mutex to serialize data as the connection is shared
+	 * with the fore channel
+	 */
+	mutex_lock(&xprt->xpt_mutex);
+	if (test_bit(XPT_DEAD, &xprt->xpt_flags))
+		len = -ENOTCONN;
+	else
+		len = bc_sendto(req);
+	mutex_unlock(&xprt->xpt_mutex);
+
+	if (len > 0)
+		len = 0;
+
+	return len;
+}
+
+/*
+ * The close routine. Since this is client initiated, we do nothing
+ */
+
+static void bc_close(struct rpc_xprt *xprt)
+{
+	return;
+}
+
+/*
+ * The xprt destroy routine. Again, because this connection is client
+ * initiated, we do nothing
+ */
+
+static void bc_destroy(struct rpc_xprt *xprt)
+{
+	return;
+}
+
 static struct rpc_xprt_ops xs_udp_ops = {
 	.set_buffer_size	= xs_udp_set_buffer_size,
 	.reserve_xprt		= xprt_reserve_xprt_cong,
@@ -2192,6 +2321,22 @@ static struct rpc_xprt_ops xs_tcp_ops = {
 	.print_stats		= xs_tcp_print_stats,
 };
 
+/*
+ * The rpc_xprt_ops for the server backchannel
+ */
+
+static struct rpc_xprt_ops bc_tcp_ops = {
+	.reserve_xprt		= xprt_reserve_xprt,
+	.release_xprt		= xprt_release_xprt,
+	.buf_alloc		= bc_malloc,
+	.buf_free		= bc_free,
+	.send_request		= bc_send_request,
+	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
+	.close			= bc_close,
+	.destroy		= bc_destroy,
+	.print_stats		= xs_tcp_print_stats,
+};
+
 static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
 				      unsigned int slot_table_size)
 {
@@ -2323,14 +2468,46 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
 	xprt->prot = IPPROTO_TCP;
 	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
 	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
+	xprt->timeout = &xs_tcp_default_timeout;
 
-	xprt->bind_timeout = XS_BIND_TO;
-	xprt->connect_timeout = XS_TCP_CONN_TO;
-	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
-	xprt->idle_timeout = XS_IDLE_DISC_TO;
+	if (args->bc_xprt) {
+		struct svc_sock *bc_sock;
 
-	xprt->ops = &xs_tcp_ops;
-	xprt->timeout = &xs_tcp_default_timeout;
+		/* backchannel */
+		xprt_set_bound(xprt);
+		xprt->bind_timeout = 0;
+		xprt->connect_timeout = 0;
+		xprt->reestablish_timeout = 0;
+		xprt->idle_timeout = (~0);
+
+		/*
+		 * The backchannel uses the same socket connection as the
+		 * forechannel
+		 */
+		xprt->bc_xprt = args->bc_xprt;
+		bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
+		bc_sock->sk_bc_xprt = xprt;
+		transport->sock = bc_sock->sk_sock;
+		transport->inet = bc_sock->sk_sk;
+
+		xprt->ops = &bc_tcp_ops;
+
+		switch (addr->sa_family) {
+		case AF_INET:
+			xs_format_ipv4_peer_addresses(xprt, "tcp",
+						      RPCBIND_NETID_TCP);
+			break;
+		case AF_INET6:
+			xs_format_ipv6_peer_addresses(xprt, "tcp",
+						      RPCBIND_NETID_TCP6);
+			break;
+		default:
+			kfree(xprt);
+			return ERR_PTR(-EAFNOSUPPORT);
+		}
+
+		goto out;
+	}
 
 	switch (addr->sa_family) {
 	case AF_INET:
@@ -2338,20 +2515,30 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
 			xprt_set_bound(xprt);
 
 		INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker4);
-		xs_format_ipv4_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
+		xs_format_ipv4_peer_addresses(xprt, "tcp",
+					      RPCBIND_NETID_TCP);
 		break;
 	case AF_INET6:
 		if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
 			xprt_set_bound(xprt);
 
 		INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker6);
-		xs_format_ipv6_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
+		xs_format_ipv6_peer_addresses(xprt, "tcp",
+					      RPCBIND_NETID_TCP);
 		break;
 	default:
 		kfree(xprt);
 		return ERR_PTR(-EAFNOSUPPORT);
 	}
 
+	xprt->bind_timeout = XS_BIND_TO;
+	xprt->connect_timeout = XS_TCP_CONN_TO;
+	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
+	xprt->idle_timeout = XS_IDLE_DISC_TO;
+
+	xprt->ops = &xs_tcp_ops;
+
+out:
 	dprintk("RPC:       set up transport to address %s\n",
 			xprt->address_strings[RPC_DISPLAY_ALL]);
 
-- 
1.6.4

Re: [PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-side backchannel handling

[J. Bruce Fields]

On Thu, Aug 20, 2009 at 03:34:23AM +0300, Benny Halevy wrote:
> From: Rahul Iyer <iyer@netapp.com>
> 
> Signed-off-by: Rahul Iyer <iyer@netapp.com>
> Signed-off-by: Mike Sager <sager@netapp.com>
> Signed-off-by: Marc Eshel <eshel@almaden.ibm.com>
> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

This patch needs an ACK from Trond.

> 
> When the call direction is a reply, copy the xid and call direction into the
> req->rq_private_buf.head[0].iov_base otherwise rpc_verify_header returns
> rpc_garbage.

Looks mostly OK, though blocking the client rpciod on the
bc_send_request method may be a problem--rpciod normally tries not to
sleep, and the other send_request methods look like they avoid it.

Other minor comments follow.

What are you using to test the backchannel?

> 
> Signed-off-by: Andy Adamson <andros@netapp.com>
> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> [get rid of CONFIG_NFSD_V4_1]
> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> [sunrpc: refactoring of svc_tcp_recvfrom]
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: create common send routine for the fore and the back channels]
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: Use free_page() to free server backchannel pages]
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: Document server backchannel locking]
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: remove bc_connect_worker()]
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: Define xprt_server_backchannel()[
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: remove bc_close and bc_init_auto_disconnect dummy functions]
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: eliminate unneeded switch statement in xs_setup_tcp()]
> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: Don't auto close the server backchannel connection]
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> [nfsd41: sunrpc: Remove unused functions]
> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> [nfsd41: change bc_sock to bc_xprt]
> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> ---
>  include/linux/sunrpc/clnt.h    |    1 +
>  include/linux/sunrpc/svcsock.h |    1 +
>  include/linux/sunrpc/xprt.h    |    7 ++
>  net/sunrpc/clnt.c              |    1 +
>  net/sunrpc/sunrpc.h            |    4 +
>  net/sunrpc/svcsock.c           |  172 +++++++++++++++++++++++++++-------
>  net/sunrpc/xprt.c              |   16 +++-
>  net/sunrpc/xprtsock.c          |  203 ++++++++++++++++++++++++++++++++++++++--
>  8 files changed, 361 insertions(+), 44 deletions(-)
> 
> diff --git a/include/linux/sunrpc/clnt.h b/include/linux/sunrpc/clnt.h
> index 37881f1..d904889 100644
> --- a/include/linux/sunrpc/clnt.h
> +++ b/include/linux/sunrpc/clnt.h
> @@ -110,6 +110,7 @@ struct rpc_create_args {
>  	rpc_authflavor_t	authflavor;
>  	unsigned long		flags;
>  	char			*client_name;
> +	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
>  };
>  
>  /* Values for "flags" field */
> diff --git a/include/linux/sunrpc/svcsock.h b/include/linux/sunrpc/svcsock.h
> index 04dba23..4b854e2 100644
> --- a/include/linux/sunrpc/svcsock.h
> +++ b/include/linux/sunrpc/svcsock.h
> @@ -28,6 +28,7 @@ struct svc_sock {
>  	/* private TCP part */
>  	u32			sk_reclen;	/* length of record */
>  	u32			sk_tcplen;	/* current read length */
> +	struct rpc_xprt	       *sk_bc_xprt;	/* NFSv4.1 backchannel xprt */
>  };
>  
>  /*
> diff --git a/include/linux/sunrpc/xprt.h b/include/linux/sunrpc/xprt.h
> index 1175d58..75cb619 100644
> --- a/include/linux/sunrpc/xprt.h
> +++ b/include/linux/sunrpc/xprt.h
> @@ -181,6 +181,7 @@ struct rpc_xprt {
>  	spinlock_t		reserve_lock;	/* lock slot table */
>  	u32			xid;		/* Next XID value to use */
>  	struct rpc_task *	snd_task;	/* Task blocked in send */
> +	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
>  #if defined(CONFIG_NFS_V4_1)
>  	struct svc_serv		*bc_serv;       /* The RPC service which will */
>  						/* process the callback */
> @@ -233,6 +234,7 @@ struct xprt_create {
>  	struct sockaddr *	srcaddr;	/* optional local address */
>  	struct sockaddr *	dstaddr;	/* remote peer address */
>  	size_t			addrlen;
> +	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
>  };
>  
>  struct xprt_class {
> @@ -368,6 +370,11 @@ static inline int xprt_test_and_set_binding(struct rpc_xprt *xprt)
>  	return test_and_set_bit(XPRT_BINDING, &xprt->state);
>  }
>  
> +static inline int xprt_server_backchannel(struct rpc_xprt *xprt)
> +{
> +	return xprt->bc_xprt != NULL;
> +}
> +
>  #endif /* __KERNEL__*/
>  
>  #endif /* _LINUX_SUNRPC_XPRT_H */
> diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
> index ebfcf9b..f45d3bb 100644
> --- a/net/sunrpc/clnt.c
> +++ b/net/sunrpc/clnt.c
> @@ -270,6 +270,7 @@ struct rpc_clnt *rpc_create(struct rpc_create_args *args)
>  		.srcaddr = args->saddress,
>  		.dstaddr = args->address,
>  		.addrlen = args->addrsize,
> +		.bc_xprt = args->bc_xprt,
>  	};
>  	char servername[48];
>  
> diff --git a/net/sunrpc/sunrpc.h b/net/sunrpc/sunrpc.h
> index 5d9dd74..b1b2e64 100644
> --- a/net/sunrpc/sunrpc.h
> +++ b/net/sunrpc/sunrpc.h
> @@ -33,5 +33,9 @@ static inline int rpc_reply_expected(struct rpc_task *task)
>  		(task->tk_msg.rpc_proc->p_decode != NULL);
>  }
>  
> +int svc_send_common(struct socket *sock, struct xdr_buf *xdr,
> +		    struct page *headpage, unsigned long headoffset,
> +		    struct page *tailpage, unsigned long tailoffset);
> +
>  #endif /* _NET_SUNRPC_SUNRPC_H */
>  
> diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
> index 5a5bc8b..258306e 100644
> --- a/net/sunrpc/svcsock.c
> +++ b/net/sunrpc/svcsock.c
> @@ -49,6 +49,7 @@
>  #include <linux/sunrpc/msg_prot.h>
>  #include <linux/sunrpc/svcsock.h>
>  #include <linux/sunrpc/stats.h>
> +#include <linux/sunrpc/xprt.h>
>  
>  #define RPCDBG_FACILITY	RPCDBG_SVCXPRT
>  
> @@ -153,49 +154,27 @@ static void svc_set_cmsg_data(struct svc_rqst *rqstp, struct cmsghdr *cmh)
>  }
>  
>  /*
> - * Generic sendto routine
> + * send routine intended to be shared by the fore- and back-channel
>   */
> -static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
> +int svc_send_common(struct socket *sock, struct xdr_buf *xdr,
> +		    struct page *headpage, unsigned long headoffset,
> +		    struct page *tailpage, unsigned long tailoffset)
>  {
> -	struct svc_sock	*svsk =
> -		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
> -	struct socket	*sock = svsk->sk_sock;
> -	int		slen;
> -	union {
> -		struct cmsghdr	hdr;
> -		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
> -	} buffer;
> -	struct cmsghdr *cmh = &buffer.hdr;
> -	int		len = 0;
>  	int		result;
>  	int		size;
>  	struct page	**ppage = xdr->pages;
>  	size_t		base = xdr->page_base;
>  	unsigned int	pglen = xdr->page_len;
>  	unsigned int	flags = MSG_MORE;
> -	RPC_IFDEBUG(char buf[RPC_MAX_ADDRBUFLEN]);
> +	int		slen;
> +	int		len = 0;
>  
>  	slen = xdr->len;
>  
> -	if (rqstp->rq_prot == IPPROTO_UDP) {
> -		struct msghdr msg = {
> -			.msg_name	= &rqstp->rq_addr,
> -			.msg_namelen	= rqstp->rq_addrlen,
> -			.msg_control	= cmh,
> -			.msg_controllen	= sizeof(buffer),
> -			.msg_flags	= MSG_MORE,
> -		};
> -
> -		svc_set_cmsg_data(rqstp, cmh);
> -
> -		if (sock_sendmsg(sock, &msg, 0) < 0)
> -			goto out;
> -	}
> -
>  	/* send head */
>  	if (slen == xdr->head[0].iov_len)
>  		flags = 0;
> -	len = kernel_sendpage(sock, rqstp->rq_respages[0], 0,
> +	len = kernel_sendpage(sock, headpage, headoffset,
>  				  xdr->head[0].iov_len, flags);
>  	if (len != xdr->head[0].iov_len)
>  		goto out;
> @@ -219,16 +198,58 @@ static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
>  		base = 0;
>  		ppage++;
>  	}
> +
>  	/* send tail */
>  	if (xdr->tail[0].iov_len) {
> -		result = kernel_sendpage(sock, rqstp->rq_respages[0],
> -					     ((unsigned long)xdr->tail[0].iov_base)
> -						& (PAGE_SIZE-1),
> -					     xdr->tail[0].iov_len, 0);
> -
> +		result = kernel_sendpage(sock, tailpage, tailoffset,
> +				   xdr->tail[0].iov_len, 0);
>  		if (result > 0)
>  			len += result;
>  	}
> +
> +out:
> +	return len;
> +}
> +
> +
> +/*
> + * Generic sendto routine
> + */
> +static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
> +{
> +	struct svc_sock	*svsk =
> +		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
> +	struct socket	*sock = svsk->sk_sock;
> +	union {
> +		struct cmsghdr	hdr;
> +		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
> +	} buffer;
> +	struct cmsghdr *cmh = &buffer.hdr;
> +	int		len = 0;
> +	unsigned long tailoff;
> +	unsigned long headoff;
> +	RPC_IFDEBUG(char buf[RPC_MAX_ADDRBUFLEN]);
> +
> +	if (rqstp->rq_prot == IPPROTO_UDP) {
> +		struct msghdr msg = {
> +			.msg_name	= &rqstp->rq_addr,
> +			.msg_namelen	= rqstp->rq_addrlen,
> +			.msg_control	= cmh,
> +			.msg_controllen	= sizeof(buffer),
> +			.msg_flags	= MSG_MORE,
> +		};
> +
> +		svc_set_cmsg_data(rqstp, cmh);
> +
> +		if (sock_sendmsg(sock, &msg, 0) < 0)
> +			goto out;
> +	}
> +
> +	tailoff = ((unsigned long)xdr->tail[0].iov_base) & (PAGE_SIZE-1);
> +	headoff = 0;
> +	len = svc_send_common(sock, xdr, rqstp->rq_respages[0], headoff,
> +			       rqstp->rq_respages[0], tailoff);
> +
>  out:
>  	dprintk("svc: socket %p sendto([%p %Zu... ], %d) = %d (addr %s)\n",
>  		svsk, xdr->head[0].iov_base, xdr->head[0].iov_len,
> @@ -923,6 +944,57 @@ static int svc_tcp_recv_record(struct svc_sock *svsk, struct svc_rqst *rqstp)
>  	return -EAGAIN;
>  }
>  
> +static int svc_process_calldir(struct svc_sock *svsk, struct svc_rqst *rqstp,
> +			       struct rpc_rqst **reqpp, struct kvec *vec)
> +{
> +	struct rpc_rqst *req = NULL;
> +	u32 *p;
> +	u32 xid;
> +	u32 calldir;
> +	int len;
> +
> +	len = svc_recvfrom(rqstp, vec, 1, 8);
> +	if (len < 0)
> +		goto error;
> +
> +	p = (u32 *)rqstp->rq_arg.head[0].iov_base;
> +	xid = *p++;
> +	calldir = *p;
> +
> +	if (calldir == 0) {
> +		/* REQUEST is the most common case */
> +		vec[0] = rqstp->rq_arg.head[0];
> +	} else {
> +		/* REPLY */
> +		if (svsk->sk_bc_xprt)
> +			req = xprt_lookup_rqst(svsk->sk_bc_xprt, xid);
> +
> +		if (!req) {
> +			printk(KERN_NOTICE
> +				"%s: Got unrecognized reply: "
> +				"calldir 0x%x sk_bc_xprt %p xid %08x\n",
> +				__func__, ntohl(calldir),
> +				svsk->sk_bc_xprt, xid);
> +			vec[0] = rqstp->rq_arg.head[0];
> +			goto out;
> +		}
> +
> +		memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
> +		       sizeof(struct xdr_buf));
> +		/* copy the xid and call direction */
> +		memcpy(req->rq_private_buf.head[0].iov_base,
> +		       rqstp->rq_arg.head[0].iov_base, 8);
> +		vec[0] = req->rq_private_buf.head[0];
> +	}
> + out:
> +	vec[0].iov_base += 8;
> +	vec[0].iov_len -= 8;
> +	len = svsk->sk_reclen - 8;
> + error:
> +	*reqpp = req;
> +	return len;
> +}
> +
>  /*
>   * Receive data from a TCP socket.
>   */
> @@ -934,6 +1006,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  	int		len;
>  	struct kvec *vec;
>  	int pnum, vlen;
> +	struct rpc_rqst *req = NULL;
>  
>  	dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
>  		svsk, test_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags),
> @@ -947,9 +1020,27 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  	vec = rqstp->rq_vec;
>  	vec[0] = rqstp->rq_arg.head[0];
>  	vlen = PAGE_SIZE;
> +
> +	/*
> +	 * We have enough data for the whole tcp record. Let's try and read the
> +	 * first 8 bytes to get the xid and the call direction. We can use this
> +	 * to figure out if this is a call or a reply to a callback. If
> +	 * sk_reclen is < 8 (xid and calldir), then this is a malformed packet.
> +	 * In that case, don't bother with the calldir and just read the data.
> +	 * It will be rejected in svc_process.
> +	 */
> +	if (len >= 8) {
> +		len = svc_process_calldir(svsk, rqstp, &req, vec);
> +		if (len < 0)
> +			goto err_again;
> +		vlen -= 8;
> +	}
> +
>  	pnum = 1;
>  	while (vlen < len) {
> -		vec[pnum].iov_base = page_address(rqstp->rq_pages[pnum]);
> +		vec[pnum].iov_base = (req) ?
> +			page_address(req->rq_private_buf.pages[pnum - 1]) :
> +			page_address(rqstp->rq_pages[pnum]);
>  		vec[pnum].iov_len = PAGE_SIZE;
>  		pnum++;
>  		vlen += PAGE_SIZE;
> @@ -961,6 +1052,16 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  	if (len < 0)
>  		goto err_again;
>  
> +	/*
> +	 * Account for the 8 bytes we read earlier
> +	 */
> +	len += 8;
> +
> +	if (req) {
> +		xprt_complete_rqst(req->rq_task, len);
> +		len = 0;
> +		goto out;
> +	}
>  	dprintk("svc: TCP complete record (%d bytes)\n", len);
>  	rqstp->rq_arg.len = len;
>  	rqstp->rq_arg.page_base = 0;
> @@ -974,6 +1075,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>  	rqstp->rq_xprt_ctxt   = NULL;
>  	rqstp->rq_prot	      = IPPROTO_TCP;
>  
> +out:
>  	/* Reset TCP read info */
>  	svsk->sk_reclen = 0;
>  	svsk->sk_tcplen = 0;
> diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
> index f412a85..b6d4d0d 100644
> --- a/net/sunrpc/xprt.c
> +++ b/net/sunrpc/xprt.c
> @@ -599,6 +599,9 @@ static void xprt_autoclose(struct work_struct *work)
>  	struct rpc_xprt *xprt =
>  		container_of(work, struct rpc_xprt, task_cleanup);
>  
> +	if (xprt_server_backchannel(xprt))
> +		return;
> +
>  	xprt->ops->close(xprt);
>  	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
>  	xprt_release_write(xprt, NULL);
> @@ -669,6 +672,9 @@ xprt_init_autodisconnect(unsigned long data)
>  {
>  	struct rpc_xprt *xprt = (struct rpc_xprt *)data;
>  
> +	if (xprt_server_backchannel(xprt))
> +		return;
> +
>  	spin_lock(&xprt->transport_lock);
>  	if (!list_empty(&xprt->recv) || xprt->shutdown)
>  		goto out_abort;
> @@ -1083,7 +1089,8 @@ found:
>  
>  	INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
>  	setup_timer(&xprt->timer, xprt_init_autodisconnect,
> -			(unsigned long)xprt);
> +		    (unsigned long)xprt);
> +

If you must do miscellaneous reformatting, please do it in a separate
patch, so it's easier to toss if there's a conflict....

>  	xprt->last_used = jiffies;
>  	xprt->cwnd = RPC_INITCWND;
>  	xprt->bind_index = 0;
> @@ -1103,6 +1110,13 @@ found:
>  	dprintk("RPC:       created transport %p with %u slots\n", xprt,
>  			xprt->max_reqs);
>  
> +	/*
> +	 * Since we don't want connections for the backchannel, we set
> +	 * the xprt status to connected
> +	 */
> +	if (args->bc_xprt)
> +		xprt_set_connected(xprt);
> +
>  	return xprt;
>  }
>  
> diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
> index 83c73c4..6e6f939 100644
> --- a/net/sunrpc/xprtsock.c
> +++ b/net/sunrpc/xprtsock.c
> @@ -32,6 +32,7 @@
>  #include <linux/tcp.h>
>  #include <linux/sunrpc/clnt.h>
>  #include <linux/sunrpc/sched.h>
> +#include <linux/sunrpc/svcsock.h>
>  #include <linux/sunrpc/xprtsock.h>
>  #include <linux/file.h>
>  #ifdef CONFIG_NFS_V4_1
> @@ -43,6 +44,7 @@
>  #include <net/udp.h>
>  #include <net/tcp.h>
>  
> +#include "sunrpc.h"
>  /*
>   * xprtsock tunables
>   */
> @@ -2156,6 +2158,133 @@ static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
>  			xprt->stat.bklog_u);
>  }
>  
> +struct rpc_buffer {
> +	size_t	len;
> +	char	data[];
> +};
> +/*
> + * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
> + * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
> + * to use the server side send routines.
> + */
> +void *bc_malloc(struct rpc_task *task, size_t size)
> +{
> +	struct page *page;
> +	struct rpc_buffer *buf;
> +
> +	BUG_ON(size > PAGE_SIZE - sizeof(struct rpc_buffer));
> +	page = alloc_page(GFP_KERNEL);
> +
> +	if (!page)
> +		return NULL;
> +
> +	buf = page_address(page);
> +	buf->len = PAGE_SIZE;
> +
> +	return buf->data;
> +}
> +
> +/*
> + * Free the space allocated in the bc_alloc routine
> + */
> +void bc_free(void *buffer)
> +{
> +	struct rpc_buffer *buf;
> +
> +	if (!buffer)
> +		return;
> +
> +	buf = container_of(buffer, struct rpc_buffer, data);
> +	free_page((unsigned long)buf);
> +}
> +
> +/*
> + * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
> + * held. Borrows heavily from svc_tcp_sendto and xs_tcp_semd_request.

s/tcp_semd/tcp_send/.

> + */
> +static int bc_sendto(struct rpc_rqst *req)
> +{
> +	int len;
> +	struct xdr_buf *xbufp = &req->rq_snd_buf;
> +	struct rpc_xprt *xprt = req->rq_xprt;
> +	struct sock_xprt *transport =
> +				container_of(xprt, struct sock_xprt, xprt);
> +	struct socket *sock = transport->sock;
> +	unsigned long headoff;
> +	unsigned long tailoff;
> +
> +	/*
> +	 * Set up the rpc header and record marker stuff
> +	 */
> +	xs_encode_tcp_record_marker(xbufp);
> +
> +	tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
> +	headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
> +	len = svc_send_common(sock, xbufp,
> +			      virt_to_page(xbufp->head[0].iov_base), headoff,
> +			      xbufp->tail[0].iov_base, tailoff);
> +
> +	if (len != xbufp->len) {
> +		printk(KERN_NOTICE "Error sending entire callback!\n");
> +		len = -EAGAIN;
> +	}
> +
> +	return len;
> +}
> +
> +/*
> + * The send routine. Borrows from svc_send
> + */
> +static int bc_send_request(struct rpc_task *task)
> +{
> +	struct rpc_rqst *req = task->tk_rqstp;
> +	struct svc_xprt	*xprt;
> +	struct svc_sock         *svsk;
> +	u32                     len;
> +
> +	dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
> +	/*
> +	 * Get the server socket associated with this callback xprt
> +	 */
> +	xprt = req->rq_xprt->bc_xprt;
> +	svsk = container_of(xprt, struct svc_sock, sk_xprt);
> +
> +	/*
> +	 * Grab the mutex to serialize data as the connection is shared
> +	 * with the fore channel
> +	 */
> +	mutex_lock(&xprt->xpt_mutex);
> +	if (test_bit(XPT_DEAD, &xprt->xpt_flags))
> +		len = -ENOTCONN;
> +	else
> +		len = bc_sendto(req);
> +	mutex_unlock(&xprt->xpt_mutex);

Is it OK to block rpciod here?

> +
> +	if (len > 0)
> +		len = 0;
> +
> +	return len;
> +}
> +
> +/*
> + * The close routine. Since this is client initiated, we do nothing
> + */
> +
> +static void bc_close(struct rpc_xprt *xprt)
> +{
> +	return;
> +}
> +
> +/*
> + * The xprt destroy routine. Again, because this connection is client
> + * initiated, we do nothing
> + */
> +
> +static void bc_destroy(struct rpc_xprt *xprt)
> +{
> +	return;
> +}
> +
>  static struct rpc_xprt_ops xs_udp_ops = {
>  	.set_buffer_size	= xs_udp_set_buffer_size,
>  	.reserve_xprt		= xprt_reserve_xprt_cong,
> @@ -2192,6 +2321,22 @@ static struct rpc_xprt_ops xs_tcp_ops = {
>  	.print_stats		= xs_tcp_print_stats,
>  };
>  
> +/*
> + * The rpc_xprt_ops for the server backchannel
> + */
> +
> +static struct rpc_xprt_ops bc_tcp_ops = {
> +	.reserve_xprt		= xprt_reserve_xprt,
> +	.release_xprt		= xprt_release_xprt,
> +	.buf_alloc		= bc_malloc,
> +	.buf_free		= bc_free,
> +	.send_request		= bc_send_request,
> +	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
> +	.close			= bc_close,
> +	.destroy		= bc_destroy,
> +	.print_stats		= xs_tcp_print_stats,
> +};
> +
>  static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
>  				      unsigned int slot_table_size)
>  {
> @@ -2323,14 +2468,46 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
>  	xprt->prot = IPPROTO_TCP;
>  	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
>  	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
> +	xprt->timeout = &xs_tcp_default_timeout;
>  
> -	xprt->bind_timeout = XS_BIND_TO;
> -	xprt->connect_timeout = XS_TCP_CONN_TO;
> -	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
> -	xprt->idle_timeout = XS_IDLE_DISC_TO;
> +	if (args->bc_xprt) {
> +		struct svc_sock *bc_sock;
>  
> -	xprt->ops = &xs_tcp_ops;
> -	xprt->timeout = &xs_tcp_default_timeout;
> +		/* backchannel */
> +		xprt_set_bound(xprt);
> +		xprt->bind_timeout = 0;
> +		xprt->connect_timeout = 0;
> +		xprt->reestablish_timeout = 0;
> +		xprt->idle_timeout = (~0);
> +
> +		/*
> +		 * The backchannel uses the same socket connection as the
> +		 * forechannel
> +		 */
> +		xprt->bc_xprt = args->bc_xprt;
> +		bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
> +		bc_sock->sk_bc_xprt = xprt;
> +		transport->sock = bc_sock->sk_sock;
> +		transport->inet = bc_sock->sk_sk;
> +
> +		xprt->ops = &bc_tcp_ops;
> +
> +		switch (addr->sa_family) {
> +		case AF_INET:
> +			xs_format_ipv4_peer_addresses(xprt, "tcp",
> +						      RPCBIND_NETID_TCP);
> +			break;
> +		case AF_INET6:
> +			xs_format_ipv6_peer_addresses(xprt, "tcp",
> +						      RPCBIND_NETID_TCP6);
> +			break;
> +		default:
> +			kfree(xprt);
> +			return ERR_PTR(-EAFNOSUPPORT);
> +		}
> +
> +		goto out;
> +	}
>  
>  	switch (addr->sa_family) {
>  	case AF_INET:
> @@ -2338,20 +2515,30 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
>  			xprt_set_bound(xprt);
>  
>  		INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker4);
> -		xs_format_ipv4_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
> +		xs_format_ipv4_peer_addresses(xprt, "tcp",
> +					      RPCBIND_NETID_TCP);

Again, try to avoid mixing in this kind of reformatting.

>  		break;
>  	case AF_INET6:
>  		if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
>  			xprt_set_bound(xprt);
>  
>  		INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker6);
> -		xs_format_ipv6_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
> +		xs_format_ipv6_peer_addresses(xprt, "tcp",
> +					      RPCBIND_NETID_TCP);

Is the TCP->TCP6 change a typo?

--b.

>  		break;
>  	default:
>  		kfree(xprt);
>  		return ERR_PTR(-EAFNOSUPPORT);
>  	}
>  
> +	xprt->bind_timeout = XS_BIND_TO;
> +	xprt->connect_timeout = XS_TCP_CONN_TO;
> +	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
> +	xprt->idle_timeout = XS_IDLE_DISC_TO;
> +
> +	xprt->ops = &xs_tcp_ops;
> +
> +out:
>  	dprintk("RPC:       set up transport to address %s\n",
>  			xprt->address_strings[RPC_DISPLAY_ALL]);
>  
> -- 
> 1.6.4
> 

Re: [PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-side backchannel handling

[Trond Myklebust]

On Mon, 2009-08-24 at 19:34 -0400, J. Bruce Fields wrote:
> On Thu, Aug 20, 2009 at 03:34:23AM +0300, Benny Halevy wrote:
> > From: Rahul Iyer <iyer@netapp.com>
> > 
> > Signed-off-by: Rahul Iyer <iyer@netapp.com>
> > Signed-off-by: Mike Sager <sager@netapp.com>
> > Signed-off-by: Marc Eshel <eshel@almaden.ibm.com>
> > Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> > Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> 
> This patch needs an ACK from Trond.
> 
> > 
> > When the call direction is a reply, copy the xid and call direction into the
> > req->rq_private_buf.head[0].iov_base otherwise rpc_verify_header returns
> > rpc_garbage.
> 
> Looks mostly OK, though blocking the client rpciod on the
> bc_send_request method may be a problem--rpciod normally tries not to
> sleep, and the other send_request methods look like they avoid it.

Agreed. Blocking on sending is unacceptable inside rpciod. Please either
use non-blocking I/O, or use a different thread context for this.


-- 
Trond Myklebust
Linux NFS client maintainer

NetApp
Trond.Myklebust@netapp.com
www.netapp.com

RE: [PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-side backchannel handling

[Batsakis, Alexandros]

> -----Original Message-----
> From: Myklebust, Trond
> Sent: Monday, August 24, 2009 4:42 PM
> To: J. Bruce Fields
> Cc: Benny Halevy; linux-nfs@vger.kernel.org; pnfs@linux-nfs.org; Iyer,
> Rahul; Sager, Mike; Marc Eshel; Labiaga, Ricardo; Adamson, Andy;
> Batsakis, Alexandros; Myklebust, Trond
> Subject: Re: [PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-side
> backchannel handling
> 
> > >
> > > When the call direction is a reply, copy the xid and call
direction
> into the
> > > req->rq_private_buf.head[0].iov_base otherwise rpc_verify_header
> returns
> > > rpc_garbage.
> >
> > Looks mostly OK, though blocking the client rpciod on the
> > bc_send_request method may be a problem--rpciod normally tries not
to
> > sleep, and the other send_request methods look like they avoid it.
> 
> Agreed. Blocking on sending is unacceptable inside rpciod. Please
> either
> use non-blocking I/O, or use a different thread context for this.
> 

OK thanks, it makes sense. I ll take care of it

-alexandros

> 
> --
> Trond Myklebust
> Linux NFS client maintainer
> 
> NetApp
> Trond.Myklebust@netapp.com
> www.netapp.com

Re: [PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-side backchannel handling

[J. Bruce Fields]

On Mon, Aug 24, 2009 at 07:42:27PM -0400, Trond Myklebust wrote:
> On Mon, 2009-08-24 at 19:34 -0400, J. Bruce Fields wrote:
> > On Thu, Aug 20, 2009 at 03:34:23AM +0300, Benny Halevy wrote:
> > > From: Rahul Iyer <iyer@netapp.com>
> > > 
> > > Signed-off-by: Rahul Iyer <iyer@netapp.com>
> > > Signed-off-by: Mike Sager <sager@netapp.com>
> > > Signed-off-by: Marc Eshel <eshel@almaden.ibm.com>
> > > Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> > > Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> > 
> > This patch needs an ACK from Trond.
> > 
> > > 
> > > When the call direction is a reply, copy the xid and call direction into the
> > > req->rq_private_buf.head[0].iov_base otherwise rpc_verify_header returns
> > > rpc_garbage.
> > 
> > Looks mostly OK, though blocking the client rpciod on the
> > bc_send_request method may be a problem--rpciod normally tries not to
> > sleep, and the other send_request methods look like they avoid it.
> 
> Agreed. Blocking on sending is unacceptable inside rpciod. Please either
> use non-blocking I/O, or use a different thread context for this.

We did some work to avoid having to spawn a thread on the server for
every recall, and I'd still prefer to avoid that.

But I'm not sure what's required to make the server send routine
non-blocking.

--b.

RE: [PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-sidebackchannel handling

[Labiaga, Ricardo]

> -----Original Message-----
> From: J. Bruce Fields [mailto:bfields@fieldses.org]
> Sent: Monday, August 24, 2009 4:34 PM
> 
> What are you using to test the backchannel?
> 

We use delegation recalls to test the backchannel.

> >  		break;
> >  	case AF_INET6:
> >  		if (((struct sockaddr_in6 *)addr)->sin6_port !=
htons(0))
> >  			xprt_set_bound(xprt);
> >
> >  		INIT_DELAYED_WORK(&transport->connect_worker,
> xs_tcp_connect_worker6);
> > -		xs_format_ipv6_peer_addresses(xprt, "tcp",
RPCBIND_NETID_TCP6);
> > +		xs_format_ipv6_peer_addresses(xprt, "tcp",
> > +					      RPCBIND_NETID_TCP);
> 
> Is the TCP->TCP6 change a typo?
> 

Yes, thanks for catching this.  Will be fixed when the synchronization
problem is addressed.

- ricardo

Re: [PATCH RFC 02/10] nfsd41: sunrpc: Added rpc server-side backchannel handling

[Benny Halevy]

On Aug. 25, 2009, 2:34 +0300, "J. Bruce Fields" <bfields@fieldses.org> wrote:
> On Thu, Aug 20, 2009 at 03:34:23AM +0300, Benny Halevy wrote:
>> From: Rahul Iyer <iyer@netapp.com>
>>
>> Signed-off-by: Rahul Iyer <iyer@netapp.com>
>> Signed-off-by: Mike Sager <sager@netapp.com>
>> Signed-off-by: Marc Eshel <eshel@almaden.ibm.com>
>> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
> 
> This patch needs an ACK from Trond.
> 
>> When the call direction is a reply, copy the xid and call direction into the
>> req->rq_private_buf.head[0].iov_base otherwise rpc_verify_header returns
>> rpc_garbage.
> 
> Looks mostly OK, though blocking the client rpciod on the
> bc_send_request method may be a problem--rpciod normally tries not to
> sleep, and the other send_request methods look like they avoid it.
> 
> Other minor comments follow.
> 
> What are you using to test the backchannel?
> 

I use the connectathon tests against the Panasas server
with the "return on close" LAYOUTGET reply flag set to false.
This causes the server to recall the layout on close.
(I do see layout recalls also with roc==true but roc==false
is more targeted for testing the callback path)

With the files layout, using pnfsd-lexp, I run
$ cat /mnt/.../foo
$ tail -f /mnt/.../foo &

The nfs client should get a read delegation at this point

and from another nfs client or on the server:
$ date > /.../foo

To recall the delegation.

Benny

>> Signed-off-by: Andy Adamson <andros@netapp.com>
>> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
>> [get rid of CONFIG_NFSD_V4_1]
>> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
>> [sunrpc: refactoring of svc_tcp_recvfrom]
>> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: create common send routine for the fore and the back channels]
>> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: Use free_page() to free server backchannel pages]
>> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: Document server backchannel locking]
>> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: remove bc_connect_worker()]
>> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: Define xprt_server_backchannel()[
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: remove bc_close and bc_init_auto_disconnect dummy functions]
>> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: eliminate unneeded switch statement in xs_setup_tcp()]
>> Signed-off-by: Alexandros Batsakis <batsakis@netapp.com>
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: Don't auto close the server backchannel connection]
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> [nfsd41: sunrpc: Remove unused functions]
>> Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
>> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
>> [nfsd41: change bc_sock to bc_xprt]
>> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
>> ---
>>  include/linux/sunrpc/clnt.h    |    1 +
>>  include/linux/sunrpc/svcsock.h |    1 +
>>  include/linux/sunrpc/xprt.h    |    7 ++
>>  net/sunrpc/clnt.c              |    1 +
>>  net/sunrpc/sunrpc.h            |    4 +
>>  net/sunrpc/svcsock.c           |  172 +++++++++++++++++++++++++++-------
>>  net/sunrpc/xprt.c              |   16 +++-
>>  net/sunrpc/xprtsock.c          |  203 ++++++++++++++++++++++++++++++++++++++--
>>  8 files changed, 361 insertions(+), 44 deletions(-)
>>
>> diff --git a/include/linux/sunrpc/clnt.h b/include/linux/sunrpc/clnt.h
>> index 37881f1..d904889 100644
>> --- a/include/linux/sunrpc/clnt.h
>> +++ b/include/linux/sunrpc/clnt.h
>> @@ -110,6 +110,7 @@ struct rpc_create_args {
>>  	rpc_authflavor_t	authflavor;
>>  	unsigned long		flags;
>>  	char			*client_name;
>> +	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
>>  };
>>  
>>  /* Values for "flags" field */
>> diff --git a/include/linux/sunrpc/svcsock.h b/include/linux/sunrpc/svcsock.h
>> index 04dba23..4b854e2 100644
>> --- a/include/linux/sunrpc/svcsock.h
>> +++ b/include/linux/sunrpc/svcsock.h
>> @@ -28,6 +28,7 @@ struct svc_sock {
>>  	/* private TCP part */
>>  	u32			sk_reclen;	/* length of record */
>>  	u32			sk_tcplen;	/* current read length */
>> +	struct rpc_xprt	       *sk_bc_xprt;	/* NFSv4.1 backchannel xprt */
>>  };
>>  
>>  /*
>> diff --git a/include/linux/sunrpc/xprt.h b/include/linux/sunrpc/xprt.h
>> index 1175d58..75cb619 100644
>> --- a/include/linux/sunrpc/xprt.h
>> +++ b/include/linux/sunrpc/xprt.h
>> @@ -181,6 +181,7 @@ struct rpc_xprt {
>>  	spinlock_t		reserve_lock;	/* lock slot table */
>>  	u32			xid;		/* Next XID value to use */
>>  	struct rpc_task *	snd_task;	/* Task blocked in send */
>> +	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
>>  #if defined(CONFIG_NFS_V4_1)
>>  	struct svc_serv		*bc_serv;       /* The RPC service which will */
>>  						/* process the callback */
>> @@ -233,6 +234,7 @@ struct xprt_create {
>>  	struct sockaddr *	srcaddr;	/* optional local address */
>>  	struct sockaddr *	dstaddr;	/* remote peer address */
>>  	size_t			addrlen;
>> +	struct svc_xprt		*bc_xprt;	/* NFSv4.1 backchannel */
>>  };
>>  
>>  struct xprt_class {
>> @@ -368,6 +370,11 @@ static inline int xprt_test_and_set_binding(struct rpc_xprt *xprt)
>>  	return test_and_set_bit(XPRT_BINDING, &xprt->state);
>>  }
>>  
>> +static inline int xprt_server_backchannel(struct rpc_xprt *xprt)
>> +{
>> +	return xprt->bc_xprt != NULL;
>> +}
>> +
>>  #endif /* __KERNEL__*/
>>  
>>  #endif /* _LINUX_SUNRPC_XPRT_H */
>> diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
>> index ebfcf9b..f45d3bb 100644
>> --- a/net/sunrpc/clnt.c
>> +++ b/net/sunrpc/clnt.c
>> @@ -270,6 +270,7 @@ struct rpc_clnt *rpc_create(struct rpc_create_args *args)
>>  		.srcaddr = args->saddress,
>>  		.dstaddr = args->address,
>>  		.addrlen = args->addrsize,
>> +		.bc_xprt = args->bc_xprt,
>>  	};
>>  	char servername[48];
>>  
>> diff --git a/net/sunrpc/sunrpc.h b/net/sunrpc/sunrpc.h
>> index 5d9dd74..b1b2e64 100644
>> --- a/net/sunrpc/sunrpc.h
>> +++ b/net/sunrpc/sunrpc.h
>> @@ -33,5 +33,9 @@ static inline int rpc_reply_expected(struct rpc_task *task)
>>  		(task->tk_msg.rpc_proc->p_decode != NULL);
>>  }
>>  
>> +int svc_send_common(struct socket *sock, struct xdr_buf *xdr,
>> +		    struct page *headpage, unsigned long headoffset,
>> +		    struct page *tailpage, unsigned long tailoffset);
>> +
>>  #endif /* _NET_SUNRPC_SUNRPC_H */
>>  
>> diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
>> index 5a5bc8b..258306e 100644
>> --- a/net/sunrpc/svcsock.c
>> +++ b/net/sunrpc/svcsock.c
>> @@ -49,6 +49,7 @@
>>  #include <linux/sunrpc/msg_prot.h>
>>  #include <linux/sunrpc/svcsock.h>
>>  #include <linux/sunrpc/stats.h>
>> +#include <linux/sunrpc/xprt.h>
>>  
>>  #define RPCDBG_FACILITY	RPCDBG_SVCXPRT
>>  
>> @@ -153,49 +154,27 @@ static void svc_set_cmsg_data(struct svc_rqst *rqstp, struct cmsghdr *cmh)
>>  }
>>  
>>  /*
>> - * Generic sendto routine
>> + * send routine intended to be shared by the fore- and back-channel
>>   */
>> -static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
>> +int svc_send_common(struct socket *sock, struct xdr_buf *xdr,
>> +		    struct page *headpage, unsigned long headoffset,
>> +		    struct page *tailpage, unsigned long tailoffset)
>>  {
>> -	struct svc_sock	*svsk =
>> -		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
>> -	struct socket	*sock = svsk->sk_sock;
>> -	int		slen;
>> -	union {
>> -		struct cmsghdr	hdr;
>> -		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
>> -	} buffer;
>> -	struct cmsghdr *cmh = &buffer.hdr;
>> -	int		len = 0;
>>  	int		result;
>>  	int		size;
>>  	struct page	**ppage = xdr->pages;
>>  	size_t		base = xdr->page_base;
>>  	unsigned int	pglen = xdr->page_len;
>>  	unsigned int	flags = MSG_MORE;
>> -	RPC_IFDEBUG(char buf[RPC_MAX_ADDRBUFLEN]);
>> +	int		slen;
>> +	int		len = 0;
>>  
>>  	slen = xdr->len;
>>  
>> -	if (rqstp->rq_prot == IPPROTO_UDP) {
>> -		struct msghdr msg = {
>> -			.msg_name	= &rqstp->rq_addr,
>> -			.msg_namelen	= rqstp->rq_addrlen,
>> -			.msg_control	= cmh,
>> -			.msg_controllen	= sizeof(buffer),
>> -			.msg_flags	= MSG_MORE,
>> -		};
>> -
>> -		svc_set_cmsg_data(rqstp, cmh);
>> -
>> -		if (sock_sendmsg(sock, &msg, 0) < 0)
>> -			goto out;
>> -	}
>> -
>>  	/* send head */
>>  	if (slen == xdr->head[0].iov_len)
>>  		flags = 0;
>> -	len = kernel_sendpage(sock, rqstp->rq_respages[0], 0,
>> +	len = kernel_sendpage(sock, headpage, headoffset,
>>  				  xdr->head[0].iov_len, flags);
>>  	if (len != xdr->head[0].iov_len)
>>  		goto out;
>> @@ -219,16 +198,58 @@ static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
>>  		base = 0;
>>  		ppage++;
>>  	}
>> +
>>  	/* send tail */
>>  	if (xdr->tail[0].iov_len) {
>> -		result = kernel_sendpage(sock, rqstp->rq_respages[0],
>> -					     ((unsigned long)xdr->tail[0].iov_base)
>> -						& (PAGE_SIZE-1),
>> -					     xdr->tail[0].iov_len, 0);
>> -
>> +		result = kernel_sendpage(sock, tailpage, tailoffset,
>> +				   xdr->tail[0].iov_len, 0);
>>  		if (result > 0)
>>  			len += result;
>>  	}
>> +
>> +out:
>> +	return len;
>> +}
>> +
>> +
>> +/*
>> + * Generic sendto routine
>> + */
>> +static int svc_sendto(struct svc_rqst *rqstp, struct xdr_buf *xdr)
>> +{
>> +	struct svc_sock	*svsk =
>> +		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
>> +	struct socket	*sock = svsk->sk_sock;
>> +	union {
>> +		struct cmsghdr	hdr;
>> +		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
>> +	} buffer;
>> +	struct cmsghdr *cmh = &buffer.hdr;
>> +	int		len = 0;
>> +	unsigned long tailoff;
>> +	unsigned long headoff;
>> +	RPC_IFDEBUG(char buf[RPC_MAX_ADDRBUFLEN]);
>> +
>> +	if (rqstp->rq_prot == IPPROTO_UDP) {
>> +		struct msghdr msg = {
>> +			.msg_name	= &rqstp->rq_addr,
>> +			.msg_namelen	= rqstp->rq_addrlen,
>> +			.msg_control	= cmh,
>> +			.msg_controllen	= sizeof(buffer),
>> +			.msg_flags	= MSG_MORE,
>> +		};
>> +
>> +		svc_set_cmsg_data(rqstp, cmh);
>> +
>> +		if (sock_sendmsg(sock, &msg, 0) < 0)
>> +			goto out;
>> +	}
>> +
>> +	tailoff = ((unsigned long)xdr->tail[0].iov_base) & (PAGE_SIZE-1);
>> +	headoff = 0;
>> +	len = svc_send_common(sock, xdr, rqstp->rq_respages[0], headoff,
>> +			       rqstp->rq_respages[0], tailoff);
>> +
>>  out:
>>  	dprintk("svc: socket %p sendto([%p %Zu... ], %d) = %d (addr %s)\n",
>>  		svsk, xdr->head[0].iov_base, xdr->head[0].iov_len,
>> @@ -923,6 +944,57 @@ static int svc_tcp_recv_record(struct svc_sock *svsk, struct svc_rqst *rqstp)
>>  	return -EAGAIN;
>>  }
>>  
>> +static int svc_process_calldir(struct svc_sock *svsk, struct svc_rqst *rqstp,
>> +			       struct rpc_rqst **reqpp, struct kvec *vec)
>> +{
>> +	struct rpc_rqst *req = NULL;
>> +	u32 *p;
>> +	u32 xid;
>> +	u32 calldir;
>> +	int len;
>> +
>> +	len = svc_recvfrom(rqstp, vec, 1, 8);
>> +	if (len < 0)
>> +		goto error;
>> +
>> +	p = (u32 *)rqstp->rq_arg.head[0].iov_base;
>> +	xid = *p++;
>> +	calldir = *p;
>> +
>> +	if (calldir == 0) {
>> +		/* REQUEST is the most common case */
>> +		vec[0] = rqstp->rq_arg.head[0];
>> +	} else {
>> +		/* REPLY */
>> +		if (svsk->sk_bc_xprt)
>> +			req = xprt_lookup_rqst(svsk->sk_bc_xprt, xid);
>> +
>> +		if (!req) {
>> +			printk(KERN_NOTICE
>> +				"%s: Got unrecognized reply: "
>> +				"calldir 0x%x sk_bc_xprt %p xid %08x\n",
>> +				__func__, ntohl(calldir),
>> +				svsk->sk_bc_xprt, xid);
>> +			vec[0] = rqstp->rq_arg.head[0];
>> +			goto out;
>> +		}
>> +
>> +		memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
>> +		       sizeof(struct xdr_buf));
>> +		/* copy the xid and call direction */
>> +		memcpy(req->rq_private_buf.head[0].iov_base,
>> +		       rqstp->rq_arg.head[0].iov_base, 8);
>> +		vec[0] = req->rq_private_buf.head[0];
>> +	}
>> + out:
>> +	vec[0].iov_base += 8;
>> +	vec[0].iov_len -= 8;
>> +	len = svsk->sk_reclen - 8;
>> + error:
>> +	*reqpp = req;
>> +	return len;
>> +}
>> +
>>  /*
>>   * Receive data from a TCP socket.
>>   */
>> @@ -934,6 +1006,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>>  	int		len;
>>  	struct kvec *vec;
>>  	int pnum, vlen;
>> +	struct rpc_rqst *req = NULL;
>>  
>>  	dprintk("svc: tcp_recv %p data %d conn %d close %d\n",
>>  		svsk, test_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags),
>> @@ -947,9 +1020,27 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>>  	vec = rqstp->rq_vec;
>>  	vec[0] = rqstp->rq_arg.head[0];
>>  	vlen = PAGE_SIZE;
>> +
>> +	/*
>> +	 * We have enough data for the whole tcp record. Let's try and read the
>> +	 * first 8 bytes to get the xid and the call direction. We can use this
>> +	 * to figure out if this is a call or a reply to a callback. If
>> +	 * sk_reclen is < 8 (xid and calldir), then this is a malformed packet.
>> +	 * In that case, don't bother with the calldir and just read the data.
>> +	 * It will be rejected in svc_process.
>> +	 */
>> +	if (len >= 8) {
>> +		len = svc_process_calldir(svsk, rqstp, &req, vec);
>> +		if (len < 0)
>> +			goto err_again;
>> +		vlen -= 8;
>> +	}
>> +
>>  	pnum = 1;
>>  	while (vlen < len) {
>> -		vec[pnum].iov_base = page_address(rqstp->rq_pages[pnum]);
>> +		vec[pnum].iov_base = (req) ?
>> +			page_address(req->rq_private_buf.pages[pnum - 1]) :
>> +			page_address(rqstp->rq_pages[pnum]);
>>  		vec[pnum].iov_len = PAGE_SIZE;
>>  		pnum++;
>>  		vlen += PAGE_SIZE;
>> @@ -961,6 +1052,16 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>>  	if (len < 0)
>>  		goto err_again;
>>  
>> +	/*
>> +	 * Account for the 8 bytes we read earlier
>> +	 */
>> +	len += 8;
>> +
>> +	if (req) {
>> +		xprt_complete_rqst(req->rq_task, len);
>> +		len = 0;
>> +		goto out;
>> +	}
>>  	dprintk("svc: TCP complete record (%d bytes)\n", len);
>>  	rqstp->rq_arg.len = len;
>>  	rqstp->rq_arg.page_base = 0;
>> @@ -974,6 +1075,7 @@ static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
>>  	rqstp->rq_xprt_ctxt   = NULL;
>>  	rqstp->rq_prot	      = IPPROTO_TCP;
>>  
>> +out:
>>  	/* Reset TCP read info */
>>  	svsk->sk_reclen = 0;
>>  	svsk->sk_tcplen = 0;
>> diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
>> index f412a85..b6d4d0d 100644
>> --- a/net/sunrpc/xprt.c
>> +++ b/net/sunrpc/xprt.c
>> @@ -599,6 +599,9 @@ static void xprt_autoclose(struct work_struct *work)
>>  	struct rpc_xprt *xprt =
>>  		container_of(work, struct rpc_xprt, task_cleanup);
>>  
>> +	if (xprt_server_backchannel(xprt))
>> +		return;
>> +
>>  	xprt->ops->close(xprt);
>>  	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
>>  	xprt_release_write(xprt, NULL);
>> @@ -669,6 +672,9 @@ xprt_init_autodisconnect(unsigned long data)
>>  {
>>  	struct rpc_xprt *xprt = (struct rpc_xprt *)data;
>>  
>> +	if (xprt_server_backchannel(xprt))
>> +		return;
>> +
>>  	spin_lock(&xprt->transport_lock);
>>  	if (!list_empty(&xprt->recv) || xprt->shutdown)
>>  		goto out_abort;
>> @@ -1083,7 +1089,8 @@ found:
>>  
>>  	INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
>>  	setup_timer(&xprt->timer, xprt_init_autodisconnect,
>> -			(unsigned long)xprt);
>> +		    (unsigned long)xprt);
>> +
> 
> If you must do miscellaneous reformatting, please do it in a separate
> patch, so it's easier to toss if there's a conflict....
> 
>>  	xprt->last_used = jiffies;
>>  	xprt->cwnd = RPC_INITCWND;
>>  	xprt->bind_index = 0;
>> @@ -1103,6 +1110,13 @@ found:
>>  	dprintk("RPC:       created transport %p with %u slots\n", xprt,
>>  			xprt->max_reqs);
>>  
>> +	/*
>> +	 * Since we don't want connections for the backchannel, we set
>> +	 * the xprt status to connected
>> +	 */
>> +	if (args->bc_xprt)
>> +		xprt_set_connected(xprt);
>> +
>>  	return xprt;
>>  }
>>  
>> diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
>> index 83c73c4..6e6f939 100644
>> --- a/net/sunrpc/xprtsock.c
>> +++ b/net/sunrpc/xprtsock.c
>> @@ -32,6 +32,7 @@
>>  #include <linux/tcp.h>
>>  #include <linux/sunrpc/clnt.h>
>>  #include <linux/sunrpc/sched.h>
>> +#include <linux/sunrpc/svcsock.h>
>>  #include <linux/sunrpc/xprtsock.h>
>>  #include <linux/file.h>
>>  #ifdef CONFIG_NFS_V4_1
>> @@ -43,6 +44,7 @@
>>  #include <net/udp.h>
>>  #include <net/tcp.h>
>>  
>> +#include "sunrpc.h"
>>  /*
>>   * xprtsock tunables
>>   */
>> @@ -2156,6 +2158,133 @@ static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
>>  			xprt->stat.bklog_u);
>>  }
>>  
>> +struct rpc_buffer {
>> +	size_t	len;
>> +	char	data[];
>> +};
>> +/*
>> + * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
>> + * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
>> + * to use the server side send routines.
>> + */
>> +void *bc_malloc(struct rpc_task *task, size_t size)
>> +{
>> +	struct page *page;
>> +	struct rpc_buffer *buf;
>> +
>> +	BUG_ON(size > PAGE_SIZE - sizeof(struct rpc_buffer));
>> +	page = alloc_page(GFP_KERNEL);
>> +
>> +	if (!page)
>> +		return NULL;
>> +
>> +	buf = page_address(page);
>> +	buf->len = PAGE_SIZE;
>> +
>> +	return buf->data;
>> +}
>> +
>> +/*
>> + * Free the space allocated in the bc_alloc routine
>> + */
>> +void bc_free(void *buffer)
>> +{
>> +	struct rpc_buffer *buf;
>> +
>> +	if (!buffer)
>> +		return;
>> +
>> +	buf = container_of(buffer, struct rpc_buffer, data);
>> +	free_page((unsigned long)buf);
>> +}
>> +
>> +/*
>> + * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
>> + * held. Borrows heavily from svc_tcp_sendto and xs_tcp_semd_request.
> 
> s/tcp_semd/tcp_send/.
> 
>> + */
>> +static int bc_sendto(struct rpc_rqst *req)
>> +{
>> +	int len;
>> +	struct xdr_buf *xbufp = &req->rq_snd_buf;
>> +	struct rpc_xprt *xprt = req->rq_xprt;
>> +	struct sock_xprt *transport =
>> +				container_of(xprt, struct sock_xprt, xprt);
>> +	struct socket *sock = transport->sock;
>> +	unsigned long headoff;
>> +	unsigned long tailoff;
>> +
>> +	/*
>> +	 * Set up the rpc header and record marker stuff
>> +	 */
>> +	xs_encode_tcp_record_marker(xbufp);
>> +
>> +	tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
>> +	headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
>> +	len = svc_send_common(sock, xbufp,
>> +			      virt_to_page(xbufp->head[0].iov_base), headoff,
>> +			      xbufp->tail[0].iov_base, tailoff);
>> +
>> +	if (len != xbufp->len) {
>> +		printk(KERN_NOTICE "Error sending entire callback!\n");
>> +		len = -EAGAIN;
>> +	}
>> +
>> +	return len;
>> +}
>> +
>> +/*
>> + * The send routine. Borrows from svc_send
>> + */
>> +static int bc_send_request(struct rpc_task *task)
>> +{
>> +	struct rpc_rqst *req = task->tk_rqstp;
>> +	struct svc_xprt	*xprt;
>> +	struct svc_sock         *svsk;
>> +	u32                     len;
>> +
>> +	dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
>> +	/*
>> +	 * Get the server socket associated with this callback xprt
>> +	 */
>> +	xprt = req->rq_xprt->bc_xprt;
>> +	svsk = container_of(xprt, struct svc_sock, sk_xprt);
>> +
>> +	/*
>> +	 * Grab the mutex to serialize data as the connection is shared
>> +	 * with the fore channel
>> +	 */
>> +	mutex_lock(&xprt->xpt_mutex);
>> +	if (test_bit(XPT_DEAD, &xprt->xpt_flags))
>> +		len = -ENOTCONN;
>> +	else
>> +		len = bc_sendto(req);
>> +	mutex_unlock(&xprt->xpt_mutex);
> 
> Is it OK to block rpciod here?
> 
>> +
>> +	if (len > 0)
>> +		len = 0;
>> +
>> +	return len;
>> +}
>> +
>> +/*
>> + * The close routine. Since this is client initiated, we do nothing
>> + */
>> +
>> +static void bc_close(struct rpc_xprt *xprt)
>> +{
>> +	return;
>> +}
>> +
>> +/*
>> + * The xprt destroy routine. Again, because this connection is client
>> + * initiated, we do nothing
>> + */
>> +
>> +static void bc_destroy(struct rpc_xprt *xprt)
>> +{
>> +	return;
>> +}
>> +
>>  static struct rpc_xprt_ops xs_udp_ops = {
>>  	.set_buffer_size	= xs_udp_set_buffer_size,
>>  	.reserve_xprt		= xprt_reserve_xprt_cong,
>> @@ -2192,6 +2321,22 @@ static struct rpc_xprt_ops xs_tcp_ops = {
>>  	.print_stats		= xs_tcp_print_stats,
>>  };
>>  
>> +/*
>> + * The rpc_xprt_ops for the server backchannel
>> + */
>> +
>> +static struct rpc_xprt_ops bc_tcp_ops = {
>> +	.reserve_xprt		= xprt_reserve_xprt,
>> +	.release_xprt		= xprt_release_xprt,
>> +	.buf_alloc		= bc_malloc,
>> +	.buf_free		= bc_free,
>> +	.send_request		= bc_send_request,
>> +	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
>> +	.close			= bc_close,
>> +	.destroy		= bc_destroy,
>> +	.print_stats		= xs_tcp_print_stats,
>> +};
>> +
>>  static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
>>  				      unsigned int slot_table_size)
>>  {
>> @@ -2323,14 +2468,46 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
>>  	xprt->prot = IPPROTO_TCP;
>>  	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
>>  	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
>> +	xprt->timeout = &xs_tcp_default_timeout;
>>  
>> -	xprt->bind_timeout = XS_BIND_TO;
>> -	xprt->connect_timeout = XS_TCP_CONN_TO;
>> -	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
>> -	xprt->idle_timeout = XS_IDLE_DISC_TO;
>> +	if (args->bc_xprt) {
>> +		struct svc_sock *bc_sock;
>>  
>> -	xprt->ops = &xs_tcp_ops;
>> -	xprt->timeout = &xs_tcp_default_timeout;
>> +		/* backchannel */
>> +		xprt_set_bound(xprt);
>> +		xprt->bind_timeout = 0;
>> +		xprt->connect_timeout = 0;
>> +		xprt->reestablish_timeout = 0;
>> +		xprt->idle_timeout = (~0);
>> +
>> +		/*
>> +		 * The backchannel uses the same socket connection as the
>> +		 * forechannel
>> +		 */
>> +		xprt->bc_xprt = args->bc_xprt;
>> +		bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
>> +		bc_sock->sk_bc_xprt = xprt;
>> +		transport->sock = bc_sock->sk_sock;
>> +		transport->inet = bc_sock->sk_sk;
>> +
>> +		xprt->ops = &bc_tcp_ops;
>> +
>> +		switch (addr->sa_family) {
>> +		case AF_INET:
>> +			xs_format_ipv4_peer_addresses(xprt, "tcp",
>> +						      RPCBIND_NETID_TCP);
>> +			break;
>> +		case AF_INET6:
>> +			xs_format_ipv6_peer_addresses(xprt, "tcp",
>> +						      RPCBIND_NETID_TCP6);
>> +			break;
>> +		default:
>> +			kfree(xprt);
>> +			return ERR_PTR(-EAFNOSUPPORT);
>> +		}
>> +
>> +		goto out;
>> +	}
>>  
>>  	switch (addr->sa_family) {
>>  	case AF_INET:
>> @@ -2338,20 +2515,30 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
>>  			xprt_set_bound(xprt);
>>  
>>  		INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker4);
>> -		xs_format_ipv4_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
>> +		xs_format_ipv4_peer_addresses(xprt, "tcp",
>> +					      RPCBIND_NETID_TCP);
> 
> Again, try to avoid mixing in this kind of reformatting.
> 
>>  		break;
>>  	case AF_INET6:
>>  		if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
>>  			xprt_set_bound(xprt);
>>  
>>  		INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker6);
>> -		xs_format_ipv6_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
>> +		xs_format_ipv6_peer_addresses(xprt, "tcp",
>> +					      RPCBIND_NETID_TCP);
> 
> Is the TCP->TCP6 change a typo?
> 
> --b.
> 
>>  		break;
>>  	default:
>>  		kfree(xprt);
>>  		return ERR_PTR(-EAFNOSUPPORT);
>>  	}
>>  
>> +	xprt->bind_timeout = XS_BIND_TO;
>> +	xprt->connect_timeout = XS_TCP_CONN_TO;
>> +	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
>> +	xprt->idle_timeout = XS_IDLE_DISC_TO;
>> +
>> +	xprt->ops = &xs_tcp_ops;
>> +
>> +out:
>>  	dprintk("RPC:       set up transport to address %s\n",
>>  			xprt->address_strings[RPC_DISPLAY_ALL]);
>>  
>> -- 
>> 1.6.4
>>

[PATCH RFC 03/10] nfsd41: sunrpc: move struct rpc_buffer def into a common header file

[Benny Halevy]

struct rpc_buffer is currently defined twice, the same way, in sched.c
and xprtsock.c.  Move its definition into a sunrpc.h, a common, internal
header file.

Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 net/sunrpc/sched.c    |    7 ++-----
 net/sunrpc/sunrpc.h   |    8 ++++++++
 net/sunrpc/xprtsock.c |    4 ----
 3 files changed, 10 insertions(+), 9 deletions(-)

diff --git a/net/sunrpc/sched.c b/net/sunrpc/sched.c
index 8f459ab..cef74ba 100644
--- a/net/sunrpc/sched.c
+++ b/net/sunrpc/sched.c
@@ -21,6 +21,8 @@
 
 #include <linux/sunrpc/clnt.h>
 
+#include "sunrpc.h"
+
 #ifdef RPC_DEBUG
 #define RPCDBG_FACILITY		RPCDBG_SCHED
 #define RPC_TASK_MAGIC_ID	0xf00baa
@@ -711,11 +713,6 @@ static void rpc_async_schedule(struct work_struct *work)
 	__rpc_execute(container_of(work, struct rpc_task, u.tk_work));
 }
 
-struct rpc_buffer {
-	size_t	len;
-	char	data[];
-};
-
 /**
  * rpc_malloc - allocate an RPC buffer
  * @task: RPC task that will use this buffer
diff --git a/net/sunrpc/sunrpc.h b/net/sunrpc/sunrpc.h
index b1b2e64..7b68daf 100644
--- a/net/sunrpc/sunrpc.h
+++ b/net/sunrpc/sunrpc.h
@@ -27,6 +27,14 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 #ifndef _NET_SUNRPC_SUNRPC_H
 #define _NET_SUNRPC_SUNRPC_H
 
+/*
+ * Header for dynamically allocated rpc buffers.
+ */
+struct rpc_buffer {
+	size_t	len;
+	char	data[];
+};
+
 static inline int rpc_reply_expected(struct rpc_task *task)
 {
 	return (task->tk_msg.rpc_proc != NULL) &&
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 6e6f939..4098a92 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -2158,10 +2158,6 @@ static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
 			xprt->stat.bklog_u);
 }
 
-struct rpc_buffer {
-	size_t	len;
-	char	data[];
-};
 /*
  * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
  * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
-- 
1.6.4

Re: [PATCH RFC 03/10] nfsd41: sunrpc: move struct rpc_buffer def into a common header file

[J. Bruce Fields]

On Thu, Aug 20, 2009 at 03:34:28AM +0300, Benny Halevy wrote:
> struct rpc_buffer is currently defined twice, the same way, in sched.c
> and xprtsock.c.  Move its definition into a sunrpc.h, a common, internal
> header file.

Not a huge deal, but: ideally this move would be done before the
previous patch, which would use it.  (Under the general principal when
we have the chance, we we anticipate problems rather than introduce them
and then solve them in a later patch.)

--b.

> 
> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
> ---
>  net/sunrpc/sched.c    |    7 ++-----
>  net/sunrpc/sunrpc.h   |    8 ++++++++
>  net/sunrpc/xprtsock.c |    4 ----
>  3 files changed, 10 insertions(+), 9 deletions(-)
> 
> diff --git a/net/sunrpc/sched.c b/net/sunrpc/sched.c
> index 8f459ab..cef74ba 100644
> --- a/net/sunrpc/sched.c
> +++ b/net/sunrpc/sched.c
> @@ -21,6 +21,8 @@
>  
>  #include <linux/sunrpc/clnt.h>
>  
> +#include "sunrpc.h"
> +
>  #ifdef RPC_DEBUG
>  #define RPCDBG_FACILITY		RPCDBG_SCHED
>  #define RPC_TASK_MAGIC_ID	0xf00baa
> @@ -711,11 +713,6 @@ static void rpc_async_schedule(struct work_struct *work)
>  	__rpc_execute(container_of(work, struct rpc_task, u.tk_work));
>  }
>  
> -struct rpc_buffer {
> -	size_t	len;
> -	char	data[];
> -};
> -
>  /**
>   * rpc_malloc - allocate an RPC buffer
>   * @task: RPC task that will use this buffer
> diff --git a/net/sunrpc/sunrpc.h b/net/sunrpc/sunrpc.h
> index b1b2e64..7b68daf 100644
> --- a/net/sunrpc/sunrpc.h
> +++ b/net/sunrpc/sunrpc.h
> @@ -27,6 +27,14 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
>  #ifndef _NET_SUNRPC_SUNRPC_H
>  #define _NET_SUNRPC_SUNRPC_H
>  
> +/*
> + * Header for dynamically allocated rpc buffers.
> + */
> +struct rpc_buffer {
> +	size_t	len;
> +	char	data[];
> +};
> +
>  static inline int rpc_reply_expected(struct rpc_task *task)
>  {
>  	return (task->tk_msg.rpc_proc != NULL) &&
> diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
> index 6e6f939..4098a92 100644
> --- a/net/sunrpc/xprtsock.c
> +++ b/net/sunrpc/xprtsock.c
> @@ -2158,10 +2158,6 @@ static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
>  			xprt->stat.bklog_u);
>  }
>  
> -struct rpc_buffer {
> -	size_t	len;
> -	char	data[];
> -};
>  /*
>   * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
>   * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
> -- 
> 1.6.4
> 

Re: [PATCH RFC 03/10] nfsd41: sunrpc: move struct rpc_buffer def into a common header file

[Benny Halevy]

On Aug. 25, 2009, 2:35 +0300, "J. Bruce Fields" <bfields@fieldses.org> wrote:
> On Thu, Aug 20, 2009 at 03:34:28AM +0300, Benny Halevy wrote:
>> struct rpc_buffer is currently defined twice, the same way, in sched.c
>> and xprtsock.c.  Move its definition into a sunrpc.h, a common, internal
>> header file.
> 
> Not a huge deal, but: ideally this move would be done before the
> previous patch, which would use it.  (Under the general principal when
> we have the chance, we we anticipate problems rather than introduce them
> and then solve them in a later patch.)

Absolutely. Will do.

Benny

> 
> --b.
> 
>> Signed-off-by: Benny Halevy <bhalevy@panasas.com>
>> ---
>>  net/sunrpc/sched.c    |    7 ++-----
>>  net/sunrpc/sunrpc.h   |    8 ++++++++
>>  net/sunrpc/xprtsock.c |    4 ----
>>  3 files changed, 10 insertions(+), 9 deletions(-)
>>
>> diff --git a/net/sunrpc/sched.c b/net/sunrpc/sched.c
>> index 8f459ab..cef74ba 100644
>> --- a/net/sunrpc/sched.c
>> +++ b/net/sunrpc/sched.c
>> @@ -21,6 +21,8 @@
>>  
>>  #include <linux/sunrpc/clnt.h>
>>  
>> +#include "sunrpc.h"
>> +
>>  #ifdef RPC_DEBUG
>>  #define RPCDBG_FACILITY		RPCDBG_SCHED
>>  #define RPC_TASK_MAGIC_ID	0xf00baa
>> @@ -711,11 +713,6 @@ static void rpc_async_schedule(struct work_struct *work)
>>  	__rpc_execute(container_of(work, struct rpc_task, u.tk_work));
>>  }
>>  
>> -struct rpc_buffer {
>> -	size_t	len;
>> -	char	data[];
>> -};
>> -
>>  /**
>>   * rpc_malloc - allocate an RPC buffer
>>   * @task: RPC task that will use this buffer
>> diff --git a/net/sunrpc/sunrpc.h b/net/sunrpc/sunrpc.h
>> index b1b2e64..7b68daf 100644
>> --- a/net/sunrpc/sunrpc.h
>> +++ b/net/sunrpc/sunrpc.h
>> @@ -27,6 +27,14 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
>>  #ifndef _NET_SUNRPC_SUNRPC_H
>>  #define _NET_SUNRPC_SUNRPC_H
>>  
>> +/*
>> + * Header for dynamically allocated rpc buffers.
>> + */
>> +struct rpc_buffer {
>> +	size_t	len;
>> +	char	data[];
>> +};
>> +
>>  static inline int rpc_reply_expected(struct rpc_task *task)
>>  {
>>  	return (task->tk_msg.rpc_proc != NULL) &&
>> diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
>> index 6e6f939..4098a92 100644
>> --- a/net/sunrpc/xprtsock.c
>> +++ b/net/sunrpc/xprtsock.c
>> @@ -2158,10 +2158,6 @@ static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
>>  			xprt->stat.bklog_u);
>>  }
>>  
>> -struct rpc_buffer {
>> -	size_t	len;
>> -	char	data[];
>> -};
>>  /*
>>   * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
>>   * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
>> -- 
>> 1.6.4
>>

[PATCH RFC 04/10] nfsd41: Backchannel: callback infrastructure

[Benny Halevy]

From: Andy Adamson <andros@netapp.com>

Keep the xprt used for create_session in cl_cb_xprt.
Mark cl_callback.cb_minorversion = 1 and remember
the client provided cl_callback.cb_prog rpc program number.
Use it to probe the callback path.

Define xdr sizes and code nfs4_cb_compound header to be able
to send a null callback rpc.

Signed-off-by: Andy Adamson<andros@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[get callback minorversion from fore channel's]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: change bc_sock to bc_xprt]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 fs/nfsd/nfs4callback.c |   25 ++++++++++++++++++++++---
 fs/nfsd/nfs4state.c    |   11 +++++++++++
 2 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/fs/nfsd/nfs4callback.c b/fs/nfsd/nfs4callback.c
index 3fd23f7..a39548e 100644
--- a/fs/nfsd/nfs4callback.c
+++ b/fs/nfsd/nfs4callback.c
@@ -43,6 +43,7 @@
 #include <linux/sunrpc/xdr.h>
 #include <linux/sunrpc/svc.h>
 #include <linux/sunrpc/clnt.h>
+#include <linux/sunrpc/svcsock.h>
 #include <linux/nfsd/nfsd.h>
 #include <linux/nfsd/state.h>
 #include <linux/sunrpc/sched.h>
@@ -52,16 +53,19 @@
 
 #define NFSPROC4_CB_NULL 0
 #define NFSPROC4_CB_COMPOUND 1
+#define NFS4_STATEID_SIZE 16
 
 /* Index of predefined Linux callback client operations */
 
 enum {
-        NFSPROC4_CLNT_CB_NULL = 0,
+	NFSPROC4_CLNT_CB_NULL = 0,
 	NFSPROC4_CLNT_CB_RECALL,
+	NFSPROC4_CLNT_CB_SEQUENCE,
 };
 
 enum nfs_cb_opnum4 {
 	OP_CB_RECALL            = 4,
+	OP_CB_SEQUENCE          = 11,
 };
 
 #define NFS4_MAXTAGLEN		20
@@ -70,15 +74,22 @@ enum nfs_cb_opnum4 {
 #define NFS4_dec_cb_null_sz		0
 #define cb_compound_enc_hdr_sz		4
 #define cb_compound_dec_hdr_sz		(3 + (NFS4_MAXTAGLEN >> 2))
+#define sessionid_sz			(NFS4_MAX_SESSIONID_LEN >> 2)
+#define cb_sequence_enc_sz		(sessionid_sz + 4 +             \
+					1 /* no referring calls list yet */)
+#define cb_sequence_dec_sz		(op_dec_sz + sessionid_sz + 4)
+
 #define op_enc_sz			1
 #define op_dec_sz			2
 #define enc_nfs4_fh_sz			(1 + (NFS4_FHSIZE >> 2))
 #define enc_stateid_sz			(NFS4_STATEID_SIZE >> 2)
 #define NFS4_enc_cb_recall_sz		(cb_compound_enc_hdr_sz +       \
+					cb_sequence_enc_sz +            \
 					1 + enc_stateid_sz +            \
 					enc_nfs4_fh_sz)
 
 #define NFS4_dec_cb_recall_sz		(cb_compound_dec_hdr_sz  +      \
+					cb_sequence_dec_sz +            \
 					op_dec_sz)
 
 /*
@@ -137,11 +148,13 @@ xdr_error:                                      \
 } while (0)
 
 struct nfs4_cb_compound_hdr {
-	int		status;
-	u32		ident;
+	/* args */
+	u32		ident;	/* minorversion 0 only */
 	u32		nops;
 	__be32		*nops_p;
 	u32		minorversion;
+	/* res */
+	int		status;
 	u32		taglen;
 	char		*tag;
 };
@@ -405,6 +418,12 @@ int setup_callback_client(struct nfs4_client *clp)
 	addr.sin_family = AF_INET;
 	addr.sin_port = htons(cb->cb_port);
 	addr.sin_addr.s_addr = htonl(cb->cb_addr);
+	if (cb->cb_minorversion)
+		args.bc_xprt = clp->cl_cb_xprt;
+
+	dprintk("%s: program %s 0x%x nrvers %u version %u minorversion %u\n",
+		__func__, args.program->name, args.prognumber,
+		args.program->nrvers, args.version, cb->cb_minorversion);
 
 	/* Create RPC client */
 	client = rpc_create(&args);
diff --git a/fs/nfsd/nfs4state.c b/fs/nfsd/nfs4state.c
index 8b4d1ad..a420197 100644
--- a/fs/nfsd/nfs4state.c
+++ b/fs/nfsd/nfs4state.c
@@ -653,6 +653,8 @@ static inline void
 free_client(struct nfs4_client *clp)
 {
 	shutdown_callback_client(clp);
+	if (clp->cl_cb_xprt)
+		svc_xprt_put(clp->cl_cb_xprt);
 	if (clp->cl_cred.cr_group_info)
 		put_group_info(clp->cl_cred.cr_group_info);
 	kfree(clp->cl_principal);
@@ -1334,6 +1336,15 @@ nfsd4_create_session(struct svc_rqst *rqstp,
 		cr_ses->flags &= ~SESSION4_PERSIST;
 		cr_ses->flags &= ~SESSION4_RDMA;
 
+		if (cr_ses->flags & SESSION4_BACK_CHAN) {
+			unconf->cl_cb_xprt = rqstp->rq_xprt;
+			svc_xprt_get(unconf->cl_cb_xprt);
+			unconf->cl_cb_conn.cb_minorversion =
+				cstate->minorversion;
+			unconf->cl_cb_conn.cb_prog = cr_ses->callback_prog;
+			unconf->cl_cb_seq_nr = 1;
+			nfsd4_probe_callback(unconf);
+		}
 		conf = unconf;
 	} else {
 		status = nfserr_stale_clientid;
-- 
1.6.4

[PATCH RFC 05/10] nfsd41: Backchannel: Add sequence arguments to callback RPC arguments

[Benny Halevy]

From: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

Follow the model we use in the client. Make the sequence arguments
part of the regular RPC arguments.  None of the callbacks that are
soon to be implemented expect results that need to be passed back
to the caller, so we don't define a separate RPC results structure.
For session validation, the cb_sequence decoding will use a pointer
to the sequence arguments that are part of the RPC argument.

Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[define struct nfsd4_cb_sequence here]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 fs/nfsd/nfs4callback.c     |    5 +++++
 include/linux/nfsd/state.h |    9 +++++++++
 2 files changed, 14 insertions(+), 0 deletions(-)

diff --git a/fs/nfsd/nfs4callback.c b/fs/nfsd/nfs4callback.c
index a39548e..a301a14 100644
--- a/fs/nfsd/nfs4callback.c
+++ b/fs/nfsd/nfs4callback.c
@@ -92,6 +92,11 @@ enum nfs_cb_opnum4 {
 					cb_sequence_dec_sz +            \
 					op_dec_sz)
 
+struct nfs4_rpc_args {
+	void				*args_op;
+	struct nfsd4_cb_sequence	args_seq;
+};
+
 /*
 * Generic encode routines from fs/nfs/nfs4xdr.c
 */
diff --git a/include/linux/nfsd/state.h b/include/linux/nfsd/state.h
index bccda62..50c515f 100644
--- a/include/linux/nfsd/state.h
+++ b/include/linux/nfsd/state.h
@@ -60,6 +60,12 @@ typedef struct {
 #define si_stateownerid   si_opaque.so_stateownerid
 #define si_fileid         si_opaque.so_fileid
 
+struct nfsd4_cb_sequence {
+	/* args/res */
+	u32			cbs_minorversion;
+	struct nfs4_client	*cbs_clp;
+};
+
 struct nfs4_delegation {
 	struct list_head	dl_perfile;
 	struct list_head	dl_perclnt;
@@ -205,6 +211,9 @@ struct nfs4_client {
 	struct nfsd4_clid_slot	cl_cs_slot;	/* create_session slot */
 	u32			cl_exchange_flags;
 	struct nfs4_sessionid	cl_sessionid;
+
+	/* for nfs41 callbacks */
+	struct svc_xprt		*cl_cb_xprt;	/* 4.1 callback transport */
 };
 
 /* struct nfs4_client_reset
-- 
1.6.4

[PATCH RFC 06/10] nfsd41: Backchannel: Server backchannel RPC wait queue

[Benny Halevy]

From: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

RPC callback requests will wait on this wait queue if the backchannel
is out of slots.

Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 fs/nfsd/nfs4state.c        |    2 ++
 include/linux/nfsd/state.h |    4 ++++
 2 files changed, 6 insertions(+), 0 deletions(-)

diff --git a/fs/nfsd/nfs4state.c b/fs/nfsd/nfs4state.c
index a420197..ffedc50 100644
--- a/fs/nfsd/nfs4state.c
+++ b/fs/nfsd/nfs4state.c
@@ -726,6 +726,8 @@ static struct nfs4_client *create_client(struct xdr_netobj name, char *recdir)
 	INIT_LIST_HEAD(&clp->cl_delegations);
 	INIT_LIST_HEAD(&clp->cl_sessions);
 	INIT_LIST_HEAD(&clp->cl_lru);
+	clear_bit(0, &clp->cl_cb_slot_busy);
+	rpc_init_wait_queue(&clp->cl_cb_waitq, "Backchannel slot table");
 	return clp;
 }
 
diff --git a/include/linux/nfsd/state.h b/include/linux/nfsd/state.h
index 50c515f..637f67c 100644
--- a/include/linux/nfsd/state.h
+++ b/include/linux/nfsd/state.h
@@ -213,7 +213,11 @@ struct nfs4_client {
 	struct nfs4_sessionid	cl_sessionid;
 
 	/* for nfs41 callbacks */
+	/* We currently support a single back channel with a single slot */
+	unsigned long		cl_cb_slot_busy;
 	struct svc_xprt		*cl_cb_xprt;	/* 4.1 callback transport */
+	struct rpc_wait_queue	cl_cb_waitq;	/* backchannel callers may */
+						/* wait here for slots */
 };
 
 /* struct nfs4_client_reset
-- 
1.6.4

[PATCH RFC 07/10] nfsd41: Backchannel: Setup sequence information

[Benny Halevy]

From: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

Follows the model used by the NFS client.  Setup the RPC prepare and done
function pointers so that we can populate the sequence information if
minorversion == 1.  rpc_run_task() is then invoked directly just like
existing NFS client operations do.

nfsd4_cb_prepare() determines if the sequence information needs to be setup.
If the slot is in use, it adds itself to the wait queue.

nfsd4_cb_done() wakes anyone sleeping on the callback channel wait queue
after our RPC reply has been received.  It also sets the task message
result pointer to NULL to clearly indicate we're done using it.

Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[define cl_cb_seq_nr here]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 fs/nfsd/nfs4callback.c     |   83 ++++++++++++++++++++++++++++++++++++++++++++
 include/linux/nfsd/state.h |    1 +
 2 files changed, 84 insertions(+), 0 deletions(-)

diff --git a/fs/nfsd/nfs4callback.c b/fs/nfsd/nfs4callback.c
index a301a14..896f187 100644
--- a/fs/nfsd/nfs4callback.c
+++ b/fs/nfsd/nfs4callback.c
@@ -527,6 +527,88 @@ nfsd4_probe_callback(struct nfs4_client *clp)
 	do_probe_callback(clp);
 }
 
+/*
+ * There's currently a single callback channel slot.
+ * If the slot is available, then mark it busy.  Otherwise, set the
+ * thread for sleeping on the callback RPC wait queue.
+ */
+static int nfsd41_cb_setup_sequence(struct nfs4_client *clp,
+		struct rpc_task *task)
+{
+	struct nfs4_rpc_args *args = task->tk_msg.rpc_argp;
+	u32 *ptr = (u32 *)clp->cl_sessionid.data;
+	int status = 0;
+
+	dprintk("%s: %u:%u:%u:%u\n", __func__,
+		ptr[0], ptr[1], ptr[2], ptr[3]);
+
+	if (test_and_set_bit(0, &clp->cl_cb_slot_busy) != 0) {
+		rpc_sleep_on(&clp->cl_cb_waitq, task, NULL);
+		dprintk("%s slot is busy\n", __func__);
+		status = -EAGAIN;
+		goto out;
+	}
+
+	/*
+	 * We'll need the clp during XDR encoding and decoding,
+	 * and the sequence during decoding to verify the reply
+	 */
+	args->args_seq.cbs_clp = clp;
+	task->tk_msg.rpc_resp = &args->args_seq;
+
+out:
+	dprintk("%s status=%d\n", __func__, status);
+	return status;
+}
+
+/*
+ * TODO: cb_sequence should support referring call lists, cachethis, multiple
+ * slots, and mark callback channel down on communication errors.
+ */
+static void nfsd4_cb_prepare(struct rpc_task *task, void *calldata)
+{
+	struct nfs4_delegation *dp = calldata;
+	struct nfs4_client *clp = dp->dl_client;
+	struct nfs4_rpc_args *args = task->tk_msg.rpc_argp;
+	u32 minorversion = clp->cl_cb_conn.cb_minorversion;
+	int status = 0;
+
+	args->args_seq.cbs_minorversion = minorversion;
+	if (minorversion) {
+		status = nfsd41_cb_setup_sequence(clp, task);
+		if (status) {
+			if (status != -EAGAIN) {
+				/* terminate rpc task */
+				task->tk_status = status;
+				task->tk_action = NULL;
+			}
+			return;
+		}
+	}
+	rpc_call_start(task);
+}
+
+static void nfsd4_cb_done(struct rpc_task *task, void *calldata)
+{
+	struct nfs4_delegation *dp = calldata;
+	struct nfs4_client *clp = dp->dl_client;
+
+	dprintk("%s: minorversion=%d\n", __func__,
+		clp->cl_cb_conn.cb_minorversion);
+
+	if (clp->cl_cb_conn.cb_minorversion) {
+		/* No need for lock, access serialized in nfsd4_cb_prepare */
+		++clp->cl_cb_seq_nr;
+		clear_bit(0, &clp->cl_cb_slot_busy);
+		rpc_wake_up_next(&clp->cl_cb_waitq);
+		dprintk("%s: freed slot, new seqid=%d\n", __func__,
+			clp->cl_cb_seq_nr);
+
+		/* We're done looking into the sequence information */
+		task->tk_msg.rpc_resp = NULL;
+	}
+}
+
 static void nfsd4_cb_recall_done(struct rpc_task *task, void *calldata)
 {
 	struct nfs4_delegation *dp = calldata;
@@ -566,6 +648,7 @@ static void nfsd4_cb_recall_release(void *calldata)
 }
 
 static const struct rpc_call_ops nfsd4_cb_recall_ops = {
+	.rpc_call_prepare = nfsd4_cb_prepare,
 	.rpc_call_done = nfsd4_cb_recall_done,
 	.rpc_release = nfsd4_cb_recall_release,
 };
diff --git a/include/linux/nfsd/state.h b/include/linux/nfsd/state.h
index 637f67c..295f64a 100644
--- a/include/linux/nfsd/state.h
+++ b/include/linux/nfsd/state.h
@@ -215,6 +215,7 @@ struct nfs4_client {
 	/* for nfs41 callbacks */
 	/* We currently support a single back channel with a single slot */
 	unsigned long		cl_cb_slot_busy;
+	u32			cl_cb_seq_nr;
 	struct svc_xprt		*cl_cb_xprt;	/* 4.1 callback transport */
 	struct rpc_wait_queue	cl_cb_waitq;	/* backchannel callers may */
 						/* wait here for slots */
-- 
1.6.4

[PATCH RFC 08/10] nfsd41: Backchannel: cb_sequence callback

[Benny Halevy]

Implement the cb_sequence callback conforming to draft-ietf-nfsv4-minorversion1

Note: highest slot id and target highest slot id do not have to be 0
as was previously implemented.  They can be greater than what the
nfs server sent if the client supports a larger slot table on the
backchannel.  At this point we just ignore that.

Signed-off-by: Benny Halevy <bhalevy@panasas.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[Rework the back channel xdr using the shared v4.0 and v4.1 framework.]
Signed-off-by: Andy Adamson <andros@netapp.com>
[fixed indentation]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: use nfsd4_cb_sequence for callback minorversion]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: fix verification of CB_SEQUENCE highest slot id[
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: Backchannel: Remove old backchannel serialization]
[nfsd41: Backchannel: First callback sequence ID should be 1]
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: decode_cb_sequence does not need to actually decode ignored fields]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 fs/nfsd/nfs4callback.c |   72 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 72 insertions(+), 0 deletions(-)

diff --git a/fs/nfsd/nfs4callback.c b/fs/nfsd/nfs4callback.c
index 896f187..dc848d1 100644
--- a/fs/nfsd/nfs4callback.c
+++ b/fs/nfsd/nfs4callback.c
@@ -256,6 +256,27 @@ encode_cb_recall(struct xdr_stream *xdr, struct nfs4_delegation *dp,
 	hdr->nops++;
 }
 
+static void
+encode_cb_sequence(struct xdr_stream *xdr, struct nfsd4_cb_sequence *args,
+		   struct nfs4_cb_compound_hdr *hdr)
+{
+	__be32 *p;
+
+	if (hdr->minorversion == 0)
+		return;
+
+	RESERVE_SPACE(1 + NFS4_MAX_SESSIONID_LEN + 20);
+
+	WRITE32(OP_CB_SEQUENCE);
+	WRITEMEM(args->cbs_clp->cl_sessionid.data, NFS4_MAX_SESSIONID_LEN);
+	WRITE32(args->cbs_clp->cl_cb_seq_nr);
+	WRITE32(0);		/* slotid, always 0 */
+	WRITE32(0);		/* highest slotid always 0 */
+	WRITE32(0);		/* cachethis always 0 */
+	WRITE32(0); /* FIXME: support referring_call_lists */
+	hdr->nops++;
+}
+
 static int
 nfs4_xdr_enc_cb_null(struct rpc_rqst *req, __be32 *p)
 {
@@ -317,6 +338,57 @@ decode_cb_op_hdr(struct xdr_stream *xdr, enum nfs_opnum4 expected)
 	return 0;
 }
 
+/*
+ * Our current back channel implmentation supports a single backchannel
+ * with a single slot.
+ */
+static int
+decode_cb_sequence(struct xdr_stream *xdr, struct nfsd4_cb_sequence *res,
+		   struct rpc_rqst *rqstp)
+{
+	struct nfs4_sessionid id;
+	int status;
+	u32 dummy;
+	__be32 *p;
+
+	if (res->cbs_minorversion == 0)
+		return 0;
+
+	status = decode_cb_op_hdr(xdr, OP_CB_SEQUENCE);
+	if (status)
+		return status;
+
+	/*
+	 * If the server returns different values for sessionID, slotID or
+	 * sequence number, the server is looney tunes.
+	 */
+	status = -ESERVERFAULT;
+
+	READ_BUF(NFS4_MAX_SESSIONID_LEN + 16);
+	memcpy(id.data, p, NFS4_MAX_SESSIONID_LEN);
+	p += XDR_QUADLEN(NFS4_MAX_SESSIONID_LEN);
+	if (memcmp(id.data, res->cbs_clp->cl_sessionid.data,
+		   NFS4_MAX_SESSIONID_LEN)) {
+		dprintk("%s Invalid session id\n", __func__);
+		goto out;
+	}
+	READ32(dummy);
+	if (dummy != res->cbs_clp->cl_cb_seq_nr) {
+		dprintk("%s Invalid sequence number\n", __func__);
+		goto out;
+	}
+	READ32(dummy); 	/* slotid must be 0 */
+	if (dummy != 0) {
+		dprintk("%s Invalid slotid\n", __func__);
+		goto out;
+	}
+	/* FIXME: process highest slotid and target highest slotid */
+	status = 0;
+out:
+	return status;
+}
+
+
 static int
 nfs4_xdr_dec_cb_null(struct rpc_rqst *req, __be32 *p)
 {
-- 
1.6.4

[PATCH RFC 09/10] nfsd41: Backchannel: Implement cb_recall over NFSv4.1

[Benny Halevy]

From: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
[nfsd41: cb_recall callback]
[Share v4.0 and v4.1 back channel xdr]
Signed-off-by: Andy Adamson <andros@netapp.com>
Signed-off-by: Ricardo Labiaga <ricardo.labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[Share v4.0 and v4.1 back channel xdr]
Signed-off-by: Andy Adamson <andros@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: use nfsd4_cb_sequence for callback minorversion]
[nfsd41: conditionally decode_sequence in nfs4_xdr_dec_cb_recall]
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
[nfsd41: Backchannel: Add sequence arguments to callback RPC arguments]
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 fs/nfsd/nfs4callback.c |   31 +++++++++++++++++++++++++++----
 1 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/fs/nfsd/nfs4callback.c b/fs/nfsd/nfs4callback.c
index dc848d1..6f1c046 100644
--- a/fs/nfsd/nfs4callback.c
+++ b/fs/nfsd/nfs4callback.c
@@ -288,15 +288,19 @@ nfs4_xdr_enc_cb_null(struct rpc_rqst *req, __be32 *p)
 }
 
 static int
-nfs4_xdr_enc_cb_recall(struct rpc_rqst *req, __be32 *p, struct nfs4_delegation *args)
+nfs4_xdr_enc_cb_recall(struct rpc_rqst *req, __be32 *p,
+		struct nfs4_rpc_args *rpc_args)
 {
 	struct xdr_stream xdr;
+	struct nfs4_delegation *args = rpc_args->args_op;
 	struct nfs4_cb_compound_hdr hdr = {
 		.ident = args->dl_ident,
+		.minorversion = rpc_args->args_seq.cbs_minorversion,
 	};
 
 	xdr_init_encode(&xdr, &req->rq_snd_buf, p);
 	encode_cb_compound_hdr(&xdr, &hdr);
+	encode_cb_sequence(&xdr, &rpc_args->args_seq, &hdr);
 	encode_cb_recall(&xdr, args, &hdr);
 	encode_cb_nops(&hdr);
 	return 0;
@@ -396,7 +400,8 @@ nfs4_xdr_dec_cb_null(struct rpc_rqst *req, __be32 *p)
 }
 
 static int
-nfs4_xdr_dec_cb_recall(struct rpc_rqst *rqstp, __be32 *p)
+nfs4_xdr_dec_cb_recall(struct rpc_rqst *rqstp, __be32 *p,
+		struct nfsd4_cb_sequence *seq)
 {
 	struct xdr_stream xdr;
 	struct nfs4_cb_compound_hdr hdr;
@@ -406,6 +411,11 @@ nfs4_xdr_dec_cb_recall(struct rpc_rqst *rqstp, __be32 *p)
 	status = decode_cb_compound_hdr(&xdr, &hdr);
 	if (status)
 		goto out;
+	if (seq) {
+		status = decode_cb_sequence(&xdr, seq, rqstp);
+		if (status)
+			goto out;
+	}
 	status = decode_cb_op_hdr(&xdr, OP_CB_RECALL);
 out:
 	return status;
@@ -686,6 +696,8 @@ static void nfsd4_cb_recall_done(struct rpc_task *task, void *calldata)
 	struct nfs4_delegation *dp = calldata;
 	struct nfs4_client *clp = dp->dl_client;
 
+	nfsd4_cb_done(task, calldata);
+
 	switch (task->tk_status) {
 	case -EIO:
 		/* Network partition? */
@@ -698,16 +710,19 @@ static void nfsd4_cb_recall_done(struct rpc_task *task, void *calldata)
 		break;
 	default:
 		/* success, or error we can't handle */
-		return;
+		goto done;
 	}
 	if (dp->dl_retries--) {
 		rpc_delay(task, 2*HZ);
 		task->tk_status = 0;
 		rpc_restart_call(task);
+		return;
 	} else {
 		atomic_set(&clp->cl_cb_conn.cb_set, 0);
 		warn_no_callback_path(clp, task->tk_status);
 	}
+done:
+	kfree(task->tk_msg.rpc_argp);
 }
 
 static void nfsd4_cb_recall_release(void *calldata)
@@ -733,16 +748,24 @@ nfsd4_cb_recall(struct nfs4_delegation *dp)
 {
 	struct nfs4_client *clp = dp->dl_client;
 	struct rpc_clnt *clnt = clp->cl_cb_conn.cb_client;
+	struct nfs4_rpc_args *args;
 	struct rpc_message msg = {
 		.rpc_proc = &nfs4_cb_procedures[NFSPROC4_CLNT_CB_RECALL],
-		.rpc_argp = dp,
 		.rpc_cred = clp->cl_cb_conn.cb_cred
 	};
 	int status;
 
+	args = kzalloc(sizeof(*args), GFP_KERNEL);
+	if (!args) {
+		status = -ENOMEM;
+		goto out;
+	}
+	args->args_op = dp;
+	msg.rpc_argp = args;
 	dp->dl_retries = 1;
 	status = rpc_call_async(clnt, &msg, RPC_TASK_SOFT,
 				&nfsd4_cb_recall_ops, dp);
+out:
 	if (status) {
 		put_nfs4_client(clp);
 		nfs4_put_delegation(dp);
-- 
1.6.4

[PATCH RFC 10/10] nfsd41: Refactor create_client()

[Benny Halevy]

From: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

Move common initialization of 'struct nfs4_client' inside create_client().

Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>

[nfsd41: Remember the auth flavor to use for callbacks]
Signed-off-by: Ricardo Labiaga <Ricardo.Labiaga@netapp.com>
Signed-off-by: Benny Halevy <bhalevy@panasas.com>
---
 fs/nfsd/nfs4state.c |   90 +++++++++++++++++++++++++-------------------------
 1 files changed, 45 insertions(+), 45 deletions(-)

diff --git a/fs/nfsd/nfs4state.c b/fs/nfsd/nfs4state.c
index ffedc50..1dedde9 100644
--- a/fs/nfsd/nfs4state.c
+++ b/fs/nfsd/nfs4state.c
@@ -710,27 +710,6 @@ expire_client(struct nfs4_client *clp)
 	put_nfs4_client(clp);
 }
 
-static struct nfs4_client *create_client(struct xdr_netobj name, char *recdir)
-{
-	struct nfs4_client *clp;
-
-	clp = alloc_client(name);
-	if (clp == NULL)
-		return NULL;
-	memcpy(clp->cl_recdir, recdir, HEXDIR_LEN);
-	atomic_set(&clp->cl_count, 1);
-	atomic_set(&clp->cl_cb_conn.cb_set, 0);
-	INIT_LIST_HEAD(&clp->cl_idhash);
-	INIT_LIST_HEAD(&clp->cl_strhash);
-	INIT_LIST_HEAD(&clp->cl_openowners);
-	INIT_LIST_HEAD(&clp->cl_delegations);
-	INIT_LIST_HEAD(&clp->cl_sessions);
-	INIT_LIST_HEAD(&clp->cl_lru);
-	clear_bit(0, &clp->cl_cb_slot_busy);
-	rpc_init_wait_queue(&clp->cl_cb_waitq, "Backchannel slot table");
-	return clp;
-}
-
 static void copy_verf(struct nfs4_client *target, nfs4_verifier *source)
 {
 	memcpy(target->cl_verifier.data, source->data,
@@ -793,6 +772,46 @@ static void gen_confirm(struct nfs4_client *clp)
 	*p++ = i++;
 }
 
+static struct nfs4_client *create_client(struct xdr_netobj name, char *recdir,
+		struct svc_rqst *rqstp, nfs4_verifier *verf)
+{
+	struct nfs4_client *clp;
+	u32 ip_addr = svc_addr_in(rqstp)->sin_addr.s_addr;
+	char *princ;
+
+	clp = alloc_client(name);
+	if (clp == NULL)
+		return NULL;
+
+	princ = svc_gss_principal(rqstp);
+	if (princ) {
+		clp->cl_principal = kstrdup(princ, GFP_KERNEL);
+		if (clp->cl_principal == NULL) {
+			free_client(clp);
+			return NULL;
+		}
+	}
+
+	memcpy(clp->cl_recdir, recdir, HEXDIR_LEN);
+	atomic_set(&clp->cl_count, 1);
+	atomic_set(&clp->cl_cb_conn.cb_set, 0);
+	INIT_LIST_HEAD(&clp->cl_idhash);
+	INIT_LIST_HEAD(&clp->cl_strhash);
+	INIT_LIST_HEAD(&clp->cl_openowners);
+	INIT_LIST_HEAD(&clp->cl_delegations);
+	INIT_LIST_HEAD(&clp->cl_sessions);
+	INIT_LIST_HEAD(&clp->cl_lru);
+	clear_bit(0, &clp->cl_cb_slot_busy);
+	rpc_init_wait_queue(&clp->cl_cb_waitq, "Backchannel slot table");
+	copy_verf(clp, verf);
+	clp->cl_addr = ip_addr;
+	clp->cl_flavor = rqstp->rq_flavor;
+	copy_cred(&clp->cl_cred, &rqstp->rq_cred);
+	gen_confirm(clp);
+
+	return clp;
+}
+
 static int check_name(struct xdr_netobj name)
 {
 	if (name.len == 0) 
@@ -1206,17 +1225,13 @@ nfsd4_exchange_id(struct svc_rqst *rqstp,
 
 out_new:
 	/* Normal case */
-	new = create_client(exid->clname, dname);
+	new = create_client(exid->clname, dname, rqstp, &verf);
 	if (new == NULL) {
 		status = nfserr_resource;
 		goto out;
 	}
 
-	copy_verf(new, &verf);
-	copy_cred(&new->cl_cred, &rqstp->rq_cred);
-	new->cl_addr = ip_addr;
 	gen_clid(new);
-	gen_confirm(new);
 	add_to_unconfirmed(new, strhashval);
 out_copy:
 	exid->clientid.cl_boot = new->cl_clientid.cl_boot;
@@ -1472,7 +1487,6 @@ __be32
 nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
 		  struct nfsd4_setclientid *setclid)
 {
-	struct sockaddr_in	*sin = svc_addr_in(rqstp);
 	struct xdr_netobj 	clname = { 
 		.len = setclid->se_namelen,
 		.data = setclid->se_name,
@@ -1481,7 +1495,6 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
 	unsigned int 		strhashval;
 	struct nfs4_client	*conf, *unconf, *new;
 	__be32 			status;
-	char			*princ;
 	char                    dname[HEXDIR_LEN];
 	
 	if (!check_name(clname))
@@ -1523,7 +1536,7 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
 		 */
 		if (unconf)
 			expire_client(unconf);
-		new = create_client(clname, dname);
+		new = create_client(clname, dname, rqstp, &clverifier);
 		if (new == NULL)
 			goto out;
 		gen_clid(new);
@@ -1540,7 +1553,7 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
 			 */
 			expire_client(unconf);
 		}
-		new = create_client(clname, dname);
+		new = create_client(clname, dname, rqstp, &clverifier);
 		if (new == NULL)
 			goto out;
 		copy_clid(new, conf);
@@ -1550,7 +1563,7 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
 		 * probable client reboot; state will be removed if
 		 * confirmed.
 		 */
-		new = create_client(clname, dname);
+		new = create_client(clname, dname, rqstp, &clverifier);
 		if (new == NULL)
 			goto out;
 		gen_clid(new);
@@ -1561,24 +1574,11 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
 		 * confirmed.
 		 */
 		expire_client(unconf);
-		new = create_client(clname, dname);
+		new = create_client(clname, dname, rqstp, &clverifier);
 		if (new == NULL)
 			goto out;
 		gen_clid(new);
 	}
-	copy_verf(new, &clverifier);
-	new->cl_addr = sin->sin_addr.s_addr;
-	new->cl_flavor = rqstp->rq_flavor;
-	princ = svc_gss_principal(rqstp);
-	if (princ) {
-		new->cl_principal = kstrdup(princ, GFP_KERNEL);
-		if (new->cl_principal == NULL) {
-			free_client(new);
-			goto out;
-		}
-	}
-	copy_cred(&new->cl_cred, &rqstp->rq_cred);
-	gen_confirm(new);
 	gen_callback(new, setclid);
 	add_to_unconfirmed(new, strhashval);
 	setclid->se_clientid.cl_boot = new->cl_clientid.cl_boot;
-- 
1.6.4