aboutsummaryrefslogtreecommitdiffstats
path: root/net
diff options
context:
space:
mode:
Diffstat (limited to 'net')
-rw-r--r--net/sunrpc/Makefile2
-rw-r--r--net/sunrpc/auth.c1
-rw-r--r--net/sunrpc/auth_gss/auth_gss.c8
-rw-r--r--net/sunrpc/auth_gss/gss_krb5_mech.c1
-rw-r--r--net/sunrpc/auth_gss/gss_mech_switch.c1
-rw-r--r--net/sunrpc/auth_null.c2
-rw-r--r--net/sunrpc/auth_unix.c2
-rw-r--r--net/sunrpc/clnt.c118
-rw-r--r--net/sunrpc/pmap_clnt.c12
-rw-r--r--net/sunrpc/socklib.c175
-rw-r--r--net/sunrpc/sunrpc_syms.c1
-rw-r--r--net/sunrpc/svcsock.c3
-rw-r--r--net/sunrpc/sysctl.c32
-rw-r--r--net/sunrpc/xdr.c177
-rw-r--r--net/sunrpc/xprt.c1604
-rw-r--r--net/sunrpc/xprtsock.c1252
16 files changed, 1990 insertions, 1401 deletions
diff --git a/net/sunrpc/Makefile b/net/sunrpc/Makefile
index 46a2ce00a29b..cdcab9ca4c60 100644
--- a/net/sunrpc/Makefile
+++ b/net/sunrpc/Makefile
@@ -6,7 +6,7 @@
6obj-$(CONFIG_SUNRPC) += sunrpc.o 6obj-$(CONFIG_SUNRPC) += sunrpc.o
7obj-$(CONFIG_SUNRPC_GSS) += auth_gss/ 7obj-$(CONFIG_SUNRPC_GSS) += auth_gss/
8 8
9sunrpc-y := clnt.o xprt.o sched.o \ 9sunrpc-y := clnt.o xprt.o socklib.o xprtsock.o sched.o \
10 auth.o auth_null.o auth_unix.o \ 10 auth.o auth_null.o auth_unix.o \
11 svc.o svcsock.o svcauth.o svcauth_unix.o \ 11 svc.o svcsock.o svcauth.o svcauth_unix.o \
12 pmap_clnt.o timer.o xdr.o \ 12 pmap_clnt.o timer.o xdr.o \
diff --git a/net/sunrpc/auth.c b/net/sunrpc/auth.c
index 505e2d4b3d62..a415d99c394d 100644
--- a/net/sunrpc/auth.c
+++ b/net/sunrpc/auth.c
@@ -11,7 +11,6 @@
11#include <linux/module.h> 11#include <linux/module.h>
12#include <linux/slab.h> 12#include <linux/slab.h>
13#include <linux/errno.h> 13#include <linux/errno.h>
14#include <linux/socket.h>
15#include <linux/sunrpc/clnt.h> 14#include <linux/sunrpc/clnt.h>
16#include <linux/spinlock.h> 15#include <linux/spinlock.h>
17 16
diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index 2f7b867161d2..d2b08f16c257 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -42,8 +42,6 @@
42#include <linux/init.h> 42#include <linux/init.h>
43#include <linux/types.h> 43#include <linux/types.h>
44#include <linux/slab.h> 44#include <linux/slab.h>
45#include <linux/socket.h>
46#include <linux/in.h>
47#include <linux/sched.h> 45#include <linux/sched.h>
48#include <linux/sunrpc/clnt.h> 46#include <linux/sunrpc/clnt.h>
49#include <linux/sunrpc/auth.h> 47#include <linux/sunrpc/auth.h>
@@ -846,10 +844,8 @@ gss_marshal(struct rpc_task *task, u32 *p)
846 844
847 /* We compute the checksum for the verifier over the xdr-encoded bytes 845 /* We compute the checksum for the verifier over the xdr-encoded bytes
848 * starting with the xid and ending at the end of the credential: */ 846 * starting with the xid and ending at the end of the credential: */
849 iov.iov_base = req->rq_snd_buf.head[0].iov_base; 847 iov.iov_base = xprt_skip_transport_header(task->tk_xprt,
850 if (task->tk_client->cl_xprt->stream) 848 req->rq_snd_buf.head[0].iov_base);
851 /* See clnt.c:call_header() */
852 iov.iov_base += 4;
853 iov.iov_len = (u8 *)p - (u8 *)iov.iov_base; 849 iov.iov_len = (u8 *)p - (u8 *)iov.iov_base;
854 xdr_buf_from_iov(&iov, &verf_buf); 850 xdr_buf_from_iov(&iov, &verf_buf);
855 851
diff --git a/net/sunrpc/auth_gss/gss_krb5_mech.c b/net/sunrpc/auth_gss/gss_krb5_mech.c
index 606a8a82cafb..462c5b86b073 100644
--- a/net/sunrpc/auth_gss/gss_krb5_mech.c
+++ b/net/sunrpc/auth_gss/gss_krb5_mech.c
@@ -39,7 +39,6 @@
39#include <linux/types.h> 39#include <linux/types.h>
40#include <linux/slab.h> 40#include <linux/slab.h>
41#include <linux/sunrpc/auth.h> 41#include <linux/sunrpc/auth.h>
42#include <linux/in.h>
43#include <linux/sunrpc/gss_krb5.h> 42#include <linux/sunrpc/gss_krb5.h>
44#include <linux/sunrpc/xdr.h> 43#include <linux/sunrpc/xdr.h>
45#include <linux/crypto.h> 44#include <linux/crypto.h>
diff --git a/net/sunrpc/auth_gss/gss_mech_switch.c b/net/sunrpc/auth_gss/gss_mech_switch.c
index 9dfb68377d69..58aeaddd8c79 100644
--- a/net/sunrpc/auth_gss/gss_mech_switch.c
+++ b/net/sunrpc/auth_gss/gss_mech_switch.c
@@ -35,7 +35,6 @@
35 35
36#include <linux/types.h> 36#include <linux/types.h>
37#include <linux/slab.h> 37#include <linux/slab.h>
38#include <linux/socket.h>
39#include <linux/module.h> 38#include <linux/module.h>
40#include <linux/sunrpc/msg_prot.h> 39#include <linux/sunrpc/msg_prot.h>
41#include <linux/sunrpc/gss_asn1.h> 40#include <linux/sunrpc/gss_asn1.h>
diff --git a/net/sunrpc/auth_null.c b/net/sunrpc/auth_null.c
index 9b72d3abf823..f56767aaa927 100644
--- a/net/sunrpc/auth_null.c
+++ b/net/sunrpc/auth_null.c
@@ -7,9 +7,7 @@
7 */ 7 */
8 8
9#include <linux/types.h> 9#include <linux/types.h>
10#include <linux/socket.h>
11#include <linux/module.h> 10#include <linux/module.h>
12#include <linux/in.h>
13#include <linux/utsname.h> 11#include <linux/utsname.h>
14#include <linux/sunrpc/clnt.h> 12#include <linux/sunrpc/clnt.h>
15#include <linux/sched.h> 13#include <linux/sched.h>
diff --git a/net/sunrpc/auth_unix.c b/net/sunrpc/auth_unix.c
index 4ff297a9b15b..890fb5ea0dcb 100644
--- a/net/sunrpc/auth_unix.c
+++ b/net/sunrpc/auth_unix.c
@@ -9,8 +9,6 @@
9#include <linux/types.h> 9#include <linux/types.h>
10#include <linux/sched.h> 10#include <linux/sched.h>
11#include <linux/module.h> 11#include <linux/module.h>
12#include <linux/socket.h>
13#include <linux/in.h>
14#include <linux/sunrpc/clnt.h> 12#include <linux/sunrpc/clnt.h>
15#include <linux/sunrpc/auth.h> 13#include <linux/sunrpc/auth.h>
16 14
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index f17e6153b688..5a8f01d726e9 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -1,5 +1,5 @@
1/* 1/*
2 * linux/net/sunrpc/rpcclnt.c 2 * linux/net/sunrpc/clnt.c
3 * 3 *
4 * This file contains the high-level RPC interface. 4 * This file contains the high-level RPC interface.
5 * It is modeled as a finite state machine to support both synchronous 5 * It is modeled as a finite state machine to support both synchronous
@@ -27,7 +27,6 @@
27#include <linux/types.h> 27#include <linux/types.h>
28#include <linux/mm.h> 28#include <linux/mm.h>
29#include <linux/slab.h> 29#include <linux/slab.h>
30#include <linux/in.h>
31#include <linux/utsname.h> 30#include <linux/utsname.h>
32 31
33#include <linux/sunrpc/clnt.h> 32#include <linux/sunrpc/clnt.h>
@@ -53,6 +52,7 @@ static void call_allocate(struct rpc_task *task);
53static void call_encode(struct rpc_task *task); 52static void call_encode(struct rpc_task *task);
54static void call_decode(struct rpc_task *task); 53static void call_decode(struct rpc_task *task);
55static void call_bind(struct rpc_task *task); 54static void call_bind(struct rpc_task *task);
55static void call_bind_status(struct rpc_task *task);
56static void call_transmit(struct rpc_task *task); 56static void call_transmit(struct rpc_task *task);
57static void call_status(struct rpc_task *task); 57static void call_status(struct rpc_task *task);
58static void call_refresh(struct rpc_task *task); 58static void call_refresh(struct rpc_task *task);
@@ -517,15 +517,8 @@ void
517rpc_setbufsize(struct rpc_clnt *clnt, unsigned int sndsize, unsigned int rcvsize) 517rpc_setbufsize(struct rpc_clnt *clnt, unsigned int sndsize, unsigned int rcvsize)
518{ 518{
519 struct rpc_xprt *xprt = clnt->cl_xprt; 519 struct rpc_xprt *xprt = clnt->cl_xprt;
520 520 if (xprt->ops->set_buffer_size)
521 xprt->sndsize = 0; 521 xprt->ops->set_buffer_size(xprt, sndsize, rcvsize);
522 if (sndsize)
523 xprt->sndsize = sndsize + RPC_SLACK_SPACE;
524 xprt->rcvsize = 0;
525 if (rcvsize)
526 xprt->rcvsize = rcvsize + RPC_SLACK_SPACE;
527 if (xprt_connected(xprt))
528 xprt_sock_setbufsize(xprt);
529} 522}
530 523
531/* 524/*
@@ -734,43 +727,94 @@ static void
734call_bind(struct rpc_task *task) 727call_bind(struct rpc_task *task)
735{ 728{
736 struct rpc_clnt *clnt = task->tk_client; 729 struct rpc_clnt *clnt = task->tk_client;
737 struct rpc_xprt *xprt = clnt->cl_xprt;
738
739 dprintk("RPC: %4d call_bind xprt %p %s connected\n", task->tk_pid,
740 xprt, (xprt_connected(xprt) ? "is" : "is not"));
741 730
742 task->tk_action = (xprt_connected(xprt)) ? call_transmit : call_connect; 731 dprintk("RPC: %4d call_bind (status %d)\n",
732 task->tk_pid, task->tk_status);
743 733
734 task->tk_action = call_connect;
744 if (!clnt->cl_port) { 735 if (!clnt->cl_port) {
745 task->tk_action = call_connect; 736 task->tk_action = call_bind_status;
746 task->tk_timeout = RPC_CONNECT_TIMEOUT; 737 task->tk_timeout = task->tk_xprt->bind_timeout;
747 rpc_getport(task, clnt); 738 rpc_getport(task, clnt);
748 } 739 }
749} 740}
750 741
751/* 742/*
752 * 4a. Connect to the RPC server (TCP case) 743 * 4a. Sort out bind result
744 */
745static void
746call_bind_status(struct rpc_task *task)
747{
748 int status = -EACCES;
749
750 if (task->tk_status >= 0) {
751 dprintk("RPC: %4d call_bind_status (status %d)\n",
752 task->tk_pid, task->tk_status);
753 task->tk_status = 0;
754 task->tk_action = call_connect;
755 return;
756 }
757
758 switch (task->tk_status) {
759 case -EACCES:
760 dprintk("RPC: %4d remote rpcbind: RPC program/version unavailable\n",
761 task->tk_pid);
762 break;
763 case -ETIMEDOUT:
764 dprintk("RPC: %4d rpcbind request timed out\n",
765 task->tk_pid);
766 if (RPC_IS_SOFT(task)) {
767 status = -EIO;
768 break;
769 }
770 goto retry_bind;
771 case -EPFNOSUPPORT:
772 dprintk("RPC: %4d remote rpcbind service unavailable\n",
773 task->tk_pid);
774 break;
775 case -EPROTONOSUPPORT:
776 dprintk("RPC: %4d remote rpcbind version 2 unavailable\n",
777 task->tk_pid);
778 break;
779 default:
780 dprintk("RPC: %4d unrecognized rpcbind error (%d)\n",
781 task->tk_pid, -task->tk_status);
782 status = -EIO;
783 break;
784 }
785
786 rpc_exit(task, status);
787 return;
788
789retry_bind:
790 task->tk_status = 0;
791 task->tk_action = call_bind;
792 return;
793}
794
795/*
796 * 4b. Connect to the RPC server
753 */ 797 */
754static void 798static void
755call_connect(struct rpc_task *task) 799call_connect(struct rpc_task *task)
756{ 800{
757 struct rpc_clnt *clnt = task->tk_client; 801 struct rpc_xprt *xprt = task->tk_xprt;
758 802
759 dprintk("RPC: %4d call_connect status %d\n", 803 dprintk("RPC: %4d call_connect xprt %p %s connected\n",
760 task->tk_pid, task->tk_status); 804 task->tk_pid, xprt,
805 (xprt_connected(xprt) ? "is" : "is not"));
761 806
762 if (xprt_connected(clnt->cl_xprt)) { 807 task->tk_action = call_transmit;
763 task->tk_action = call_transmit; 808 if (!xprt_connected(xprt)) {
764 return; 809 task->tk_action = call_connect_status;
810 if (task->tk_status < 0)
811 return;
812 xprt_connect(task);
765 } 813 }
766 task->tk_action = call_connect_status;
767 if (task->tk_status < 0)
768 return;
769 xprt_connect(task);
770} 814}
771 815
772/* 816/*
773 * 4b. Sort out connect result 817 * 4c. Sort out connect result
774 */ 818 */
775static void 819static void
776call_connect_status(struct rpc_task *task) 820call_connect_status(struct rpc_task *task)
@@ -778,6 +822,9 @@ call_connect_status(struct rpc_task *task)
778 struct rpc_clnt *clnt = task->tk_client; 822 struct rpc_clnt *clnt = task->tk_client;
779 int status = task->tk_status; 823 int status = task->tk_status;
780 824
825 dprintk("RPC: %5u call_connect_status (status %d)\n",
826 task->tk_pid, task->tk_status);
827
781 task->tk_status = 0; 828 task->tk_status = 0;
782 if (status >= 0) { 829 if (status >= 0) {
783 clnt->cl_stats->netreconn++; 830 clnt->cl_stats->netreconn++;
@@ -785,17 +832,19 @@ call_connect_status(struct rpc_task *task)
785 return; 832 return;
786 } 833 }
787 834
788 /* Something failed: we may have to rebind */ 835 /* Something failed: remote service port may have changed */
789 if (clnt->cl_autobind) 836 if (clnt->cl_autobind)
790 clnt->cl_port = 0; 837 clnt->cl_port = 0;
838
791 switch (status) { 839 switch (status) {
792 case -ENOTCONN: 840 case -ENOTCONN:
793 case -ETIMEDOUT: 841 case -ETIMEDOUT:
794 case -EAGAIN: 842 case -EAGAIN:
795 task->tk_action = (clnt->cl_port == 0) ? call_bind : call_connect; 843 task->tk_action = call_bind;
796 break; 844 break;
797 default: 845 default:
798 rpc_exit(task, -EIO); 846 rpc_exit(task, -EIO);
847 break;
799 } 848 }
800} 849}
801 850
@@ -1020,13 +1069,12 @@ static u32 *
1020call_header(struct rpc_task *task) 1069call_header(struct rpc_task *task)
1021{ 1070{
1022 struct rpc_clnt *clnt = task->tk_client; 1071 struct rpc_clnt *clnt = task->tk_client;
1023 struct rpc_xprt *xprt = clnt->cl_xprt;
1024 struct rpc_rqst *req = task->tk_rqstp; 1072 struct rpc_rqst *req = task->tk_rqstp;
1025 u32 *p = req->rq_svec[0].iov_base; 1073 u32 *p = req->rq_svec[0].iov_base;
1026 1074
1027 /* FIXME: check buffer size? */ 1075 /* FIXME: check buffer size? */
1028 if (xprt->stream) 1076
1029 *p++ = 0; /* fill in later */ 1077 p = xprt_skip_transport_header(task->tk_xprt, p);
1030 *p++ = req->rq_xid; /* XID */ 1078 *p++ = req->rq_xid; /* XID */
1031 *p++ = htonl(RPC_CALL); /* CALL */ 1079 *p++ = htonl(RPC_CALL); /* CALL */
1032 *p++ = htonl(RPC_VERSION); /* RPC version */ 1080 *p++ = htonl(RPC_VERSION); /* RPC version */
diff --git a/net/sunrpc/pmap_clnt.c b/net/sunrpc/pmap_clnt.c
index 4e81f2766923..a398575f94b8 100644
--- a/net/sunrpc/pmap_clnt.c
+++ b/net/sunrpc/pmap_clnt.c
@@ -26,7 +26,7 @@
26#define PMAP_GETPORT 3 26#define PMAP_GETPORT 3
27 27
28static struct rpc_procinfo pmap_procedures[]; 28static struct rpc_procinfo pmap_procedures[];
29static struct rpc_clnt * pmap_create(char *, struct sockaddr_in *, int); 29static struct rpc_clnt * pmap_create(char *, struct sockaddr_in *, int, int);
30static void pmap_getport_done(struct rpc_task *); 30static void pmap_getport_done(struct rpc_task *);
31static struct rpc_program pmap_program; 31static struct rpc_program pmap_program;
32static DEFINE_SPINLOCK(pmap_lock); 32static DEFINE_SPINLOCK(pmap_lock);
@@ -65,7 +65,7 @@ rpc_getport(struct rpc_task *task, struct rpc_clnt *clnt)
65 map->pm_binding = 1; 65 map->pm_binding = 1;
66 spin_unlock(&pmap_lock); 66 spin_unlock(&pmap_lock);
67 67
68 pmap_clnt = pmap_create(clnt->cl_server, sap, map->pm_prot); 68 pmap_clnt = pmap_create(clnt->cl_server, sap, map->pm_prot, 0);
69 if (IS_ERR(pmap_clnt)) { 69 if (IS_ERR(pmap_clnt)) {
70 task->tk_status = PTR_ERR(pmap_clnt); 70 task->tk_status = PTR_ERR(pmap_clnt);
71 goto bailout; 71 goto bailout;
@@ -112,7 +112,7 @@ rpc_getport_external(struct sockaddr_in *sin, __u32 prog, __u32 vers, int prot)
112 NIPQUAD(sin->sin_addr.s_addr), prog, vers, prot); 112 NIPQUAD(sin->sin_addr.s_addr), prog, vers, prot);
113 113
114 sprintf(hostname, "%u.%u.%u.%u", NIPQUAD(sin->sin_addr.s_addr)); 114 sprintf(hostname, "%u.%u.%u.%u", NIPQUAD(sin->sin_addr.s_addr));
115 pmap_clnt = pmap_create(hostname, sin, prot); 115 pmap_clnt = pmap_create(hostname, sin, prot, 0);
116 if (IS_ERR(pmap_clnt)) 116 if (IS_ERR(pmap_clnt))
117 return PTR_ERR(pmap_clnt); 117 return PTR_ERR(pmap_clnt);
118 118
@@ -171,7 +171,7 @@ rpc_register(u32 prog, u32 vers, int prot, unsigned short port, int *okay)
171 171
172 sin.sin_family = AF_INET; 172 sin.sin_family = AF_INET;
173 sin.sin_addr.s_addr = htonl(INADDR_LOOPBACK); 173 sin.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
174 pmap_clnt = pmap_create("localhost", &sin, IPPROTO_UDP); 174 pmap_clnt = pmap_create("localhost", &sin, IPPROTO_UDP, 1);
175 if (IS_ERR(pmap_clnt)) { 175 if (IS_ERR(pmap_clnt)) {
176 error = PTR_ERR(pmap_clnt); 176 error = PTR_ERR(pmap_clnt);
177 dprintk("RPC: couldn't create pmap client. Error = %d\n", error); 177 dprintk("RPC: couldn't create pmap client. Error = %d\n", error);
@@ -198,7 +198,7 @@ rpc_register(u32 prog, u32 vers, int prot, unsigned short port, int *okay)
198} 198}
199 199
200static struct rpc_clnt * 200static struct rpc_clnt *
201pmap_create(char *hostname, struct sockaddr_in *srvaddr, int proto) 201pmap_create(char *hostname, struct sockaddr_in *srvaddr, int proto, int privileged)
202{ 202{
203 struct rpc_xprt *xprt; 203 struct rpc_xprt *xprt;
204 struct rpc_clnt *clnt; 204 struct rpc_clnt *clnt;
@@ -208,6 +208,8 @@ pmap_create(char *hostname, struct sockaddr_in *srvaddr, int proto)
208 if (IS_ERR(xprt)) 208 if (IS_ERR(xprt))
209 return (struct rpc_clnt *)xprt; 209 return (struct rpc_clnt *)xprt;
210 xprt->addr.sin_port = htons(RPC_PMAP_PORT); 210 xprt->addr.sin_port = htons(RPC_PMAP_PORT);
211 if (!privileged)
212 xprt->resvport = 0;
211 213
212 /* printk("pmap: create clnt\n"); */ 214 /* printk("pmap: create clnt\n"); */
213 clnt = rpc_new_client(xprt, hostname, 215 clnt = rpc_new_client(xprt, hostname,
diff --git a/net/sunrpc/socklib.c b/net/sunrpc/socklib.c
new file mode 100644
index 000000000000..8f97e90f36c8
--- /dev/null
+++ b/net/sunrpc/socklib.c
@@ -0,0 +1,175 @@
1/*
2 * linux/net/sunrpc/socklib.c
3 *
4 * Common socket helper routines for RPC client and server
5 *
6 * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
7 */
8
9#include <linux/types.h>
10#include <linux/pagemap.h>
11#include <linux/udp.h>
12#include <linux/sunrpc/xdr.h>
13
14
15/**
16 * skb_read_bits - copy some data bits from skb to internal buffer
17 * @desc: sk_buff copy helper
18 * @to: copy destination
19 * @len: number of bytes to copy
20 *
21 * Possibly called several times to iterate over an sk_buff and copy
22 * data out of it.
23 */
24static size_t skb_read_bits(skb_reader_t *desc, void *to, size_t len)
25{
26 if (len > desc->count)
27 len = desc->count;
28 if (skb_copy_bits(desc->skb, desc->offset, to, len))
29 return 0;
30 desc->count -= len;
31 desc->offset += len;
32 return len;
33}
34
35/**
36 * skb_read_and_csum_bits - copy and checksum from skb to buffer
37 * @desc: sk_buff copy helper
38 * @to: copy destination
39 * @len: number of bytes to copy
40 *
41 * Same as skb_read_bits, but calculate a checksum at the same time.
42 */
43static size_t skb_read_and_csum_bits(skb_reader_t *desc, void *to, size_t len)
44{
45 unsigned int csum2, pos;
46
47 if (len > desc->count)
48 len = desc->count;
49 pos = desc->offset;
50 csum2 = skb_copy_and_csum_bits(desc->skb, pos, to, len, 0);
51 desc->csum = csum_block_add(desc->csum, csum2, pos);
52 desc->count -= len;
53 desc->offset += len;
54 return len;
55}
56
57/**
58 * xdr_partial_copy_from_skb - copy data out of an skb
59 * @xdr: target XDR buffer
60 * @base: starting offset
61 * @desc: sk_buff copy helper
62 * @copy_actor: virtual method for copying data
63 *
64 */
65ssize_t xdr_partial_copy_from_skb(struct xdr_buf *xdr, unsigned int base, skb_reader_t *desc, skb_read_actor_t copy_actor)
66{
67 struct page **ppage = xdr->pages;
68 unsigned int len, pglen = xdr->page_len;
69 ssize_t copied = 0;
70 int ret;
71
72 len = xdr->head[0].iov_len;
73 if (base < len) {
74 len -= base;
75 ret = copy_actor(desc, (char *)xdr->head[0].iov_base + base, len);
76 copied += ret;
77 if (ret != len || !desc->count)
78 goto out;
79 base = 0;
80 } else
81 base -= len;
82
83 if (unlikely(pglen == 0))
84 goto copy_tail;
85 if (unlikely(base >= pglen)) {
86 base -= pglen;
87 goto copy_tail;
88 }
89 if (base || xdr->page_base) {
90 pglen -= base;
91 base += xdr->page_base;
92 ppage += base >> PAGE_CACHE_SHIFT;
93 base &= ~PAGE_CACHE_MASK;
94 }
95 do {
96 char *kaddr;
97
98 /* ACL likes to be lazy in allocating pages - ACLs
99 * are small by default but can get huge. */
100 if (unlikely(*ppage == NULL)) {
101 *ppage = alloc_page(GFP_ATOMIC);
102 if (unlikely(*ppage == NULL)) {
103 if (copied == 0)
104 copied = -ENOMEM;
105 goto out;
106 }
107 }
108
109 len = PAGE_CACHE_SIZE;
110 kaddr = kmap_atomic(*ppage, KM_SKB_SUNRPC_DATA);
111 if (base) {
112 len -= base;
113 if (pglen < len)
114 len = pglen;
115 ret = copy_actor(desc, kaddr + base, len);
116 base = 0;
117 } else {
118 if (pglen < len)
119 len = pglen;
120 ret = copy_actor(desc, kaddr, len);
121 }
122 flush_dcache_page(*ppage);
123 kunmap_atomic(kaddr, KM_SKB_SUNRPC_DATA);
124 copied += ret;
125 if (ret != len || !desc->count)
126 goto out;
127 ppage++;
128 } while ((pglen -= len) != 0);
129copy_tail:
130 len = xdr->tail[0].iov_len;
131 if (base < len)
132 copied += copy_actor(desc, (char *)xdr->tail[0].iov_base + base, len - base);
133out:
134 return copied;
135}
136
137/**
138 * csum_partial_copy_to_xdr - checksum and copy data
139 * @xdr: target XDR buffer
140 * @skb: source skb
141 *
142 * We have set things up such that we perform the checksum of the UDP
143 * packet in parallel with the copies into the RPC client iovec. -DaveM
144 */
145int csum_partial_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb)
146{
147 skb_reader_t desc;
148
149 desc.skb = skb;
150 desc.offset = sizeof(struct udphdr);
151 desc.count = skb->len - desc.offset;
152
153 if (skb->ip_summed == CHECKSUM_UNNECESSARY)
154 goto no_checksum;
155
156 desc.csum = csum_partial(skb->data, desc.offset, skb->csum);
157 if (xdr_partial_copy_from_skb(xdr, 0, &desc, skb_read_and_csum_bits) < 0)
158 return -1;
159 if (desc.offset != skb->len) {
160 unsigned int csum2;
161 csum2 = skb_checksum(skb, desc.offset, skb->len - desc.offset, 0);
162 desc.csum = csum_block_add(desc.csum, csum2, desc.offset);
163 }
164 if (desc.count)
165 return -1;
166 if ((unsigned short)csum_fold(desc.csum))
167 return -1;
168 return 0;
169no_checksum:
170 if (xdr_partial_copy_from_skb(xdr, 0, &desc, skb_read_bits) < 0)
171 return -1;
172 if (desc.count)
173 return -1;
174 return 0;
175}
diff --git a/net/sunrpc/sunrpc_syms.c b/net/sunrpc/sunrpc_syms.c
index ed48ff022d35..2387e7b823ff 100644
--- a/net/sunrpc/sunrpc_syms.c
+++ b/net/sunrpc/sunrpc_syms.c
@@ -10,7 +10,6 @@
10#include <linux/module.h> 10#include <linux/module.h>
11 11
12#include <linux/types.h> 12#include <linux/types.h>
13#include <linux/socket.h>
14#include <linux/sched.h> 13#include <linux/sched.h>
15#include <linux/uio.h> 14#include <linux/uio.h>
16#include <linux/unistd.h> 15#include <linux/unistd.h>
diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 30ec3efc48a6..130f2b5d93dd 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -548,9 +548,6 @@ svc_write_space(struct sock *sk)
548/* 548/*
549 * Receive a datagram from a UDP socket. 549 * Receive a datagram from a UDP socket.
550 */ 550 */
551extern int
552csum_partial_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb);
553
554static int 551static int
555svc_udp_recvfrom(struct svc_rqst *rqstp) 552svc_udp_recvfrom(struct svc_rqst *rqstp)
556{ 553{
diff --git a/net/sunrpc/sysctl.c b/net/sunrpc/sysctl.c
index 1b9616a12e24..d0c9f460e411 100644
--- a/net/sunrpc/sysctl.c
+++ b/net/sunrpc/sysctl.c
@@ -119,8 +119,18 @@ done:
119 return 0; 119 return 0;
120} 120}
121 121
122unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
123unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;
124unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
125EXPORT_SYMBOL(xprt_min_resvport);
126unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
127EXPORT_SYMBOL(xprt_max_resvport);
128
129
122static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE; 130static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
123static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE; 131static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
132static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
133static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
124 134
125static ctl_table debug_table[] = { 135static ctl_table debug_table[] = {
126 { 136 {
@@ -177,6 +187,28 @@ static ctl_table debug_table[] = {
177 .extra1 = &min_slot_table_size, 187 .extra1 = &min_slot_table_size,
178 .extra2 = &max_slot_table_size 188 .extra2 = &max_slot_table_size
179 }, 189 },
190 {
191 .ctl_name = CTL_MIN_RESVPORT,
192 .procname = "min_resvport",
193 .data = &xprt_min_resvport,
194 .maxlen = sizeof(unsigned int),
195 .mode = 0644,
196 .proc_handler = &proc_dointvec_minmax,
197 .strategy = &sysctl_intvec,
198 .extra1 = &xprt_min_resvport_limit,
199 .extra2 = &xprt_max_resvport_limit
200 },
201 {
202 .ctl_name = CTL_MAX_RESVPORT,
203 .procname = "max_resvport",
204 .data = &xprt_max_resvport,
205 .maxlen = sizeof(unsigned int),
206 .mode = 0644,
207 .proc_handler = &proc_dointvec_minmax,
208 .strategy = &sysctl_intvec,
209 .extra1 = &xprt_min_resvport_limit,
210 .extra2 = &xprt_max_resvport_limit
211 },
180 { .ctl_name = 0 } 212 { .ctl_name = 0 }
181}; 213};
182 214
diff --git a/net/sunrpc/xdr.c b/net/sunrpc/xdr.c
index fde16f40a581..32df43372ee9 100644
--- a/net/sunrpc/xdr.c
+++ b/net/sunrpc/xdr.c
@@ -6,15 +6,12 @@
6 * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de> 6 * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
7 */ 7 */
8 8
9#include <linux/module.h>
9#include <linux/types.h> 10#include <linux/types.h>
10#include <linux/socket.h>
11#include <linux/string.h> 11#include <linux/string.h>
12#include <linux/kernel.h> 12#include <linux/kernel.h>
13#include <linux/pagemap.h> 13#include <linux/pagemap.h>
14#include <linux/errno.h> 14#include <linux/errno.h>
15#include <linux/in.h>
16#include <linux/net.h>
17#include <net/sock.h>
18#include <linux/sunrpc/xdr.h> 15#include <linux/sunrpc/xdr.h>
19#include <linux/sunrpc/msg_prot.h> 16#include <linux/sunrpc/msg_prot.h>
20 17
@@ -176,178 +173,6 @@ xdr_inline_pages(struct xdr_buf *xdr, unsigned int offset,
176 xdr->buflen += len; 173 xdr->buflen += len;
177} 174}
178 175
179ssize_t
180xdr_partial_copy_from_skb(struct xdr_buf *xdr, unsigned int base,
181 skb_reader_t *desc,
182 skb_read_actor_t copy_actor)
183{
184 struct page **ppage = xdr->pages;
185 unsigned int len, pglen = xdr->page_len;
186 ssize_t copied = 0;
187 int ret;
188
189 len = xdr->head[0].iov_len;
190 if (base < len) {
191 len -= base;
192 ret = copy_actor(desc, (char *)xdr->head[0].iov_base + base, len);
193 copied += ret;
194 if (ret != len || !desc->count)
195 goto out;
196 base = 0;
197 } else
198 base -= len;
199
200 if (pglen == 0)
201 goto copy_tail;
202 if (base >= pglen) {
203 base -= pglen;
204 goto copy_tail;
205 }
206 if (base || xdr->page_base) {
207 pglen -= base;
208 base += xdr->page_base;
209 ppage += base >> PAGE_CACHE_SHIFT;
210 base &= ~PAGE_CACHE_MASK;
211 }
212 do {
213 char *kaddr;
214
215 /* ACL likes to be lazy in allocating pages - ACLs
216 * are small by default but can get huge. */
217 if (unlikely(*ppage == NULL)) {
218 *ppage = alloc_page(GFP_ATOMIC);
219 if (unlikely(*ppage == NULL)) {
220 if (copied == 0)
221 copied = -ENOMEM;
222 goto out;
223 }
224 }
225
226 len = PAGE_CACHE_SIZE;
227 kaddr = kmap_atomic(*ppage, KM_SKB_SUNRPC_DATA);
228 if (base) {
229 len -= base;
230 if (pglen < len)
231 len = pglen;
232 ret = copy_actor(desc, kaddr + base, len);
233 base = 0;
234 } else {
235 if (pglen < len)
236 len = pglen;
237 ret = copy_actor(desc, kaddr, len);
238 }
239 flush_dcache_page(*ppage);
240 kunmap_atomic(kaddr, KM_SKB_SUNRPC_DATA);
241 copied += ret;
242 if (ret != len || !desc->count)
243 goto out;
244 ppage++;
245 } while ((pglen -= len) != 0);
246copy_tail:
247 len = xdr->tail[0].iov_len;
248 if (base < len)
249 copied += copy_actor(desc, (char *)xdr->tail[0].iov_base + base, len - base);
250out:
251 return copied;
252}
253
254
255int
256xdr_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen,
257 struct xdr_buf *xdr, unsigned int base, int msgflags)
258{
259 struct page **ppage = xdr->pages;
260 unsigned int len, pglen = xdr->page_len;
261 int err, ret = 0;
262 ssize_t (*sendpage)(struct socket *, struct page *, int, size_t, int);
263
264 len = xdr->head[0].iov_len;
265 if (base < len || (addr != NULL && base == 0)) {
266 struct kvec iov = {
267 .iov_base = xdr->head[0].iov_base + base,
268 .iov_len = len - base,
269 };
270 struct msghdr msg = {
271 .msg_name = addr,
272 .msg_namelen = addrlen,
273 .msg_flags = msgflags,
274 };
275 if (xdr->len > len)
276 msg.msg_flags |= MSG_MORE;
277
278 if (iov.iov_len != 0)
279 err = kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
280 else
281 err = kernel_sendmsg(sock, &msg, NULL, 0, 0);
282 if (ret == 0)
283 ret = err;
284 else if (err > 0)
285 ret += err;
286 if (err != iov.iov_len)
287 goto out;
288 base = 0;
289 } else
290 base -= len;
291
292 if (pglen == 0)
293 goto copy_tail;
294 if (base >= pglen) {
295 base -= pglen;
296 goto copy_tail;
297 }
298 if (base || xdr->page_base) {
299 pglen -= base;
300 base += xdr->page_base;
301 ppage += base >> PAGE_CACHE_SHIFT;
302 base &= ~PAGE_CACHE_MASK;
303 }
304
305 sendpage = sock->ops->sendpage ? : sock_no_sendpage;
306 do {
307 int flags = msgflags;
308
309 len = PAGE_CACHE_SIZE;
310 if (base)
311 len -= base;
312 if (pglen < len)
313 len = pglen;
314
315 if (pglen != len || xdr->tail[0].iov_len != 0)
316 flags |= MSG_MORE;
317
318 /* Hmm... We might be dealing with highmem pages */
319 if (PageHighMem(*ppage))
320 sendpage = sock_no_sendpage;
321 err = sendpage(sock, *ppage, base, len, flags);
322 if (ret == 0)
323 ret = err;
324 else if (err > 0)
325 ret += err;
326 if (err != len)
327 goto out;
328 base = 0;
329 ppage++;
330 } while ((pglen -= len) != 0);
331copy_tail:
332 len = xdr->tail[0].iov_len;
333 if (base < len) {
334 struct kvec iov = {
335 .iov_base = xdr->tail[0].iov_base + base,
336 .iov_len = len - base,
337 };
338 struct msghdr msg = {
339 .msg_flags = msgflags,
340 };
341 err = kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
342 if (ret == 0)
343 ret = err;
344 else if (err > 0)
345 ret += err;
346 }
347out:
348 return ret;
349}
350
351 176
352/* 177/*
353 * Helper routines for doing 'memmove' like operations on a struct xdr_buf 178 * Helper routines for doing 'memmove' like operations on a struct xdr_buf
diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index 3c654e06b084..215be0d0ef6b 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -10,12 +10,12 @@
10 * one is available. Otherwise, it sleeps on the backlog queue 10 * one is available. Otherwise, it sleeps on the backlog queue
11 * (xprt_reserve). 11 * (xprt_reserve).
12 * - Next, the caller puts together the RPC message, stuffs it into 12 * - Next, the caller puts together the RPC message, stuffs it into
13 * the request struct, and calls xprt_call(). 13 * the request struct, and calls xprt_transmit().
14 * - xprt_call transmits the message and installs the caller on the 14 * - xprt_transmit sends the message and installs the caller on the
15 * socket's wait list. At the same time, it installs a timer that 15 * transport's wait list. At the same time, it installs a timer that
16 * is run after the packet's timeout has expired. 16 * is run after the packet's timeout has expired.
17 * - When a packet arrives, the data_ready handler walks the list of 17 * - When a packet arrives, the data_ready handler walks the list of
18 * pending requests for that socket. If a matching XID is found, the 18 * pending requests for that transport. If a matching XID is found, the
19 * caller is woken up, and the timer removed. 19 * caller is woken up, and the timer removed.
20 * - When no reply arrives within the timeout interval, the timer is 20 * - When no reply arrives within the timeout interval, the timer is
21 * fired by the kernel and runs xprt_timer(). It either adjusts the 21 * fired by the kernel and runs xprt_timer(). It either adjusts the
@@ -33,36 +33,17 @@
33 * 33 *
34 * Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de> 34 * Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de>
35 * 35 *
36 * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com> 36 * Transport switch API copyright (C) 2005, Chuck Lever <cel@netapp.com>
37 * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com>
38 * TCP NFS related read + write fixes
39 * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
40 *
41 * Rewrite of larges part of the code in order to stabilize TCP stuff.
42 * Fix behaviour when socket buffer is full.
43 * (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
44 */ 37 */
45 38
39#include <linux/module.h>
40
46#include <linux/types.h> 41#include <linux/types.h>
47#include <linux/slab.h> 42#include <linux/interrupt.h>
48#include <linux/capability.h>
49#include <linux/sched.h>
50#include <linux/errno.h>
51#include <linux/socket.h>
52#include <linux/in.h>
53#include <linux/net.h>
54#include <linux/mm.h>
55#include <linux/udp.h>
56#include <linux/tcp.h>
57#include <linux/sunrpc/clnt.h>
58#include <linux/file.h>
59#include <linux/workqueue.h> 43#include <linux/workqueue.h>
60#include <linux/random.h> 44#include <linux/random.h>
61 45
62#include <net/sock.h> 46#include <linux/sunrpc/clnt.h>
63#include <net/checksum.h>
64#include <net/udp.h>
65#include <net/tcp.h>
66 47
67/* 48/*
68 * Local variables 49 * Local variables
@@ -73,81 +54,90 @@
73# define RPCDBG_FACILITY RPCDBG_XPRT 54# define RPCDBG_FACILITY RPCDBG_XPRT
74#endif 55#endif
75 56
76#define XPRT_MAX_BACKOFF (8)
77#define XPRT_IDLE_TIMEOUT (5*60*HZ)
78#define XPRT_MAX_RESVPORT (800)
79
80/* 57/*
81 * Local functions 58 * Local functions
82 */ 59 */
83static void xprt_request_init(struct rpc_task *, struct rpc_xprt *); 60static void xprt_request_init(struct rpc_task *, struct rpc_xprt *);
84static inline void do_xprt_reserve(struct rpc_task *); 61static inline void do_xprt_reserve(struct rpc_task *);
85static void xprt_disconnect(struct rpc_xprt *);
86static void xprt_connect_status(struct rpc_task *task); 62static void xprt_connect_status(struct rpc_task *task);
87static struct rpc_xprt * xprt_setup(int proto, struct sockaddr_in *ap,
88 struct rpc_timeout *to);
89static struct socket *xprt_create_socket(struct rpc_xprt *, int, int);
90static void xprt_bind_socket(struct rpc_xprt *, struct socket *);
91static int __xprt_get_cong(struct rpc_xprt *, struct rpc_task *); 63static int __xprt_get_cong(struct rpc_xprt *, struct rpc_task *);
92 64
93static int xprt_clear_backlog(struct rpc_xprt *xprt);
94
95#ifdef RPC_DEBUG_DATA
96/* 65/*
97 * Print the buffer contents (first 128 bytes only--just enough for 66 * The transport code maintains an estimate on the maximum number of out-
98 * diropres return). 67 * standing RPC requests, using a smoothed version of the congestion
68 * avoidance implemented in 44BSD. This is basically the Van Jacobson
69 * congestion algorithm: If a retransmit occurs, the congestion window is
70 * halved; otherwise, it is incremented by 1/cwnd when
71 *
72 * - a reply is received and
73 * - a full number of requests are outstanding and
74 * - the congestion window hasn't been updated recently.
99 */ 75 */
100static void 76#define RPC_CWNDSHIFT (8U)
101xprt_pktdump(char *msg, u32 *packet, unsigned int count) 77#define RPC_CWNDSCALE (1U << RPC_CWNDSHIFT)
102{ 78#define RPC_INITCWND RPC_CWNDSCALE
103 u8 *buf = (u8 *) packet; 79#define RPC_MAXCWND(xprt) ((xprt)->max_reqs << RPC_CWNDSHIFT)
104 int j;
105
106 dprintk("RPC: %s\n", msg);
107 for (j = 0; j < count && j < 128; j += 4) {
108 if (!(j & 31)) {
109 if (j)
110 dprintk("\n");
111 dprintk("0x%04x ", j);
112 }
113 dprintk("%02x%02x%02x%02x ",
114 buf[j], buf[j+1], buf[j+2], buf[j+3]);
115 }
116 dprintk("\n");
117}
118#else
119static inline void
120xprt_pktdump(char *msg, u32 *packet, unsigned int count)
121{
122 /* NOP */
123}
124#endif
125 80
126/* 81#define RPCXPRT_CONGESTED(xprt) ((xprt)->cong >= (xprt)->cwnd)
127 * Look up RPC transport given an INET socket 82
83/**
84 * xprt_reserve_xprt - serialize write access to transports
85 * @task: task that is requesting access to the transport
86 *
87 * This prevents mixing the payload of separate requests, and prevents
88 * transport connects from colliding with writes. No congestion control
89 * is provided.
128 */ 90 */
129static inline struct rpc_xprt * 91int xprt_reserve_xprt(struct rpc_task *task)
130xprt_from_sock(struct sock *sk)
131{ 92{
132 return (struct rpc_xprt *) sk->sk_user_data; 93 struct rpc_xprt *xprt = task->tk_xprt;
94 struct rpc_rqst *req = task->tk_rqstp;
95
96 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
97 if (task == xprt->snd_task)
98 return 1;
99 if (task == NULL)
100 return 0;
101 goto out_sleep;
102 }
103 xprt->snd_task = task;
104 if (req) {
105 req->rq_bytes_sent = 0;
106 req->rq_ntrans++;
107 }
108 return 1;
109
110out_sleep:
111 dprintk("RPC: %4d failed to lock transport %p\n",
112 task->tk_pid, xprt);
113 task->tk_timeout = 0;
114 task->tk_status = -EAGAIN;
115 if (req && req->rq_ntrans)
116 rpc_sleep_on(&xprt->resend, task, NULL, NULL);
117 else
118 rpc_sleep_on(&xprt->sending, task, NULL, NULL);
119 return 0;
133} 120}
134 121
135/* 122/*
136 * Serialize write access to sockets, in order to prevent different 123 * xprt_reserve_xprt_cong - serialize write access to transports
137 * requests from interfering with each other. 124 * @task: task that is requesting access to the transport
138 * Also prevents TCP socket connects from colliding with writes. 125 *
126 * Same as xprt_reserve_xprt, but Van Jacobson congestion control is
127 * integrated into the decision of whether a request is allowed to be
128 * woken up and given access to the transport.
139 */ 129 */
140static int 130int xprt_reserve_xprt_cong(struct rpc_task *task)
141__xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
142{ 131{
132 struct rpc_xprt *xprt = task->tk_xprt;
143 struct rpc_rqst *req = task->tk_rqstp; 133 struct rpc_rqst *req = task->tk_rqstp;
144 134
145 if (test_and_set_bit(XPRT_LOCKED, &xprt->sockstate)) { 135 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
146 if (task == xprt->snd_task) 136 if (task == xprt->snd_task)
147 return 1; 137 return 1;
148 goto out_sleep; 138 goto out_sleep;
149 } 139 }
150 if (xprt->nocong || __xprt_get_cong(xprt, task)) { 140 if (__xprt_get_cong(xprt, task)) {
151 xprt->snd_task = task; 141 xprt->snd_task = task;
152 if (req) { 142 if (req) {
153 req->rq_bytes_sent = 0; 143 req->rq_bytes_sent = 0;
@@ -156,10 +146,10 @@ __xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
156 return 1; 146 return 1;
157 } 147 }
158 smp_mb__before_clear_bit(); 148 smp_mb__before_clear_bit();
159 clear_bit(XPRT_LOCKED, &xprt->sockstate); 149 clear_bit(XPRT_LOCKED, &xprt->state);
160 smp_mb__after_clear_bit(); 150 smp_mb__after_clear_bit();
161out_sleep: 151out_sleep:
162 dprintk("RPC: %4d failed to lock socket %p\n", task->tk_pid, xprt); 152 dprintk("RPC: %4d failed to lock transport %p\n", task->tk_pid, xprt);
163 task->tk_timeout = 0; 153 task->tk_timeout = 0;
164 task->tk_status = -EAGAIN; 154 task->tk_status = -EAGAIN;
165 if (req && req->rq_ntrans) 155 if (req && req->rq_ntrans)
@@ -169,26 +159,52 @@ out_sleep:
169 return 0; 159 return 0;
170} 160}
171 161
172static inline int 162static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
173xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
174{ 163{
175 int retval; 164 int retval;
176 165
177 spin_lock_bh(&xprt->sock_lock); 166 spin_lock_bh(&xprt->transport_lock);
178 retval = __xprt_lock_write(xprt, task); 167 retval = xprt->ops->reserve_xprt(task);
179 spin_unlock_bh(&xprt->sock_lock); 168 spin_unlock_bh(&xprt->transport_lock);
180 return retval; 169 return retval;
181} 170}
182 171
172static void __xprt_lock_write_next(struct rpc_xprt *xprt)
173{
174 struct rpc_task *task;
175 struct rpc_rqst *req;
183 176
184static void 177 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
185__xprt_lock_write_next(struct rpc_xprt *xprt) 178 return;
179
180 task = rpc_wake_up_next(&xprt->resend);
181 if (!task) {
182 task = rpc_wake_up_next(&xprt->sending);
183 if (!task)
184 goto out_unlock;
185 }
186
187 req = task->tk_rqstp;
188 xprt->snd_task = task;
189 if (req) {
190 req->rq_bytes_sent = 0;
191 req->rq_ntrans++;
192 }
193 return;
194
195out_unlock:
196 smp_mb__before_clear_bit();
197 clear_bit(XPRT_LOCKED, &xprt->state);
198 smp_mb__after_clear_bit();
199}
200
201static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
186{ 202{
187 struct rpc_task *task; 203 struct rpc_task *task;
188 204
189 if (test_and_set_bit(XPRT_LOCKED, &xprt->sockstate)) 205 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
190 return; 206 return;
191 if (!xprt->nocong && RPCXPRT_CONGESTED(xprt)) 207 if (RPCXPRT_CONGESTED(xprt))
192 goto out_unlock; 208 goto out_unlock;
193 task = rpc_wake_up_next(&xprt->resend); 209 task = rpc_wake_up_next(&xprt->resend);
194 if (!task) { 210 if (!task) {
@@ -196,7 +212,7 @@ __xprt_lock_write_next(struct rpc_xprt *xprt)
196 if (!task) 212 if (!task)
197 goto out_unlock; 213 goto out_unlock;
198 } 214 }
199 if (xprt->nocong || __xprt_get_cong(xprt, task)) { 215 if (__xprt_get_cong(xprt, task)) {
200 struct rpc_rqst *req = task->tk_rqstp; 216 struct rpc_rqst *req = task->tk_rqstp;
201 xprt->snd_task = task; 217 xprt->snd_task = task;
202 if (req) { 218 if (req) {
@@ -207,87 +223,52 @@ __xprt_lock_write_next(struct rpc_xprt *xprt)
207 } 223 }
208out_unlock: 224out_unlock:
209 smp_mb__before_clear_bit(); 225 smp_mb__before_clear_bit();
210 clear_bit(XPRT_LOCKED, &xprt->sockstate); 226 clear_bit(XPRT_LOCKED, &xprt->state);
211 smp_mb__after_clear_bit(); 227 smp_mb__after_clear_bit();
212} 228}
213 229
214/* 230/**
215 * Releases the socket for use by other requests. 231 * xprt_release_xprt - allow other requests to use a transport
232 * @xprt: transport with other tasks potentially waiting
233 * @task: task that is releasing access to the transport
234 *
235 * Note that "task" can be NULL. No congestion control is provided.
216 */ 236 */
217static void 237void xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
218__xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task)
219{ 238{
220 if (xprt->snd_task == task) { 239 if (xprt->snd_task == task) {
221 xprt->snd_task = NULL; 240 xprt->snd_task = NULL;
222 smp_mb__before_clear_bit(); 241 smp_mb__before_clear_bit();
223 clear_bit(XPRT_LOCKED, &xprt->sockstate); 242 clear_bit(XPRT_LOCKED, &xprt->state);
224 smp_mb__after_clear_bit(); 243 smp_mb__after_clear_bit();
225 __xprt_lock_write_next(xprt); 244 __xprt_lock_write_next(xprt);
226 } 245 }
227} 246}
228 247
229static inline void 248/**
230xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task) 249 * xprt_release_xprt_cong - allow other requests to use a transport
231{ 250 * @xprt: transport with other tasks potentially waiting
232 spin_lock_bh(&xprt->sock_lock); 251 * @task: task that is releasing access to the transport
233 __xprt_release_write(xprt, task); 252 *
234 spin_unlock_bh(&xprt->sock_lock); 253 * Note that "task" can be NULL. Another task is awoken to use the
235} 254 * transport if the transport's congestion window allows it.
236
237/*
238 * Write data to socket.
239 */ 255 */
240static inline int 256void xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
241xprt_sendmsg(struct rpc_xprt *xprt, struct rpc_rqst *req)
242{ 257{
243 struct socket *sock = xprt->sock; 258 if (xprt->snd_task == task) {
244 struct xdr_buf *xdr = &req->rq_snd_buf; 259 xprt->snd_task = NULL;
245 struct sockaddr *addr = NULL; 260 smp_mb__before_clear_bit();
246 int addrlen = 0; 261 clear_bit(XPRT_LOCKED, &xprt->state);
247 unsigned int skip; 262 smp_mb__after_clear_bit();
248 int result; 263 __xprt_lock_write_next_cong(xprt);
249
250 if (!sock)
251 return -ENOTCONN;
252
253 xprt_pktdump("packet data:",
254 req->rq_svec->iov_base,
255 req->rq_svec->iov_len);
256
257 /* For UDP, we need to provide an address */
258 if (!xprt->stream) {
259 addr = (struct sockaddr *) &xprt->addr;
260 addrlen = sizeof(xprt->addr);
261 } 264 }
262 /* Dont repeat bytes */ 265}
263 skip = req->rq_bytes_sent;
264
265 clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
266 result = xdr_sendpages(sock, addr, addrlen, xdr, skip, MSG_DONTWAIT);
267
268 dprintk("RPC: xprt_sendmsg(%d) = %d\n", xdr->len - skip, result);
269
270 if (result >= 0)
271 return result;
272 266
273 switch (result) { 267static inline void xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task)
274 case -ECONNREFUSED: 268{
275 /* When the server has died, an ICMP port unreachable message 269 spin_lock_bh(&xprt->transport_lock);
276 * prompts ECONNREFUSED. 270 xprt->ops->release_xprt(xprt, task);
277 */ 271 spin_unlock_bh(&xprt->transport_lock);
278 case -EAGAIN:
279 break;
280 case -ECONNRESET:
281 case -ENOTCONN:
282 case -EPIPE:
283 /* connection broken */
284 if (xprt->stream)
285 result = -ENOTCONN;
286 break;
287 default:
288 printk(KERN_NOTICE "RPC: sendmsg returned error %d\n", -result);
289 }
290 return result;
291} 272}
292 273
293/* 274/*
@@ -321,26 +302,40 @@ __xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req)
321 return; 302 return;
322 req->rq_cong = 0; 303 req->rq_cong = 0;
323 xprt->cong -= RPC_CWNDSCALE; 304 xprt->cong -= RPC_CWNDSCALE;
324 __xprt_lock_write_next(xprt); 305 __xprt_lock_write_next_cong(xprt);
325} 306}
326 307
327/* 308/**
328 * Adjust RPC congestion window 309 * xprt_release_rqst_cong - housekeeping when request is complete
310 * @task: RPC request that recently completed
311 *
312 * Useful for transports that require congestion control.
313 */
314void xprt_release_rqst_cong(struct rpc_task *task)
315{
316 __xprt_put_cong(task->tk_xprt, task->tk_rqstp);
317}
318
319/**
320 * xprt_adjust_cwnd - adjust transport congestion window
321 * @task: recently completed RPC request used to adjust window
322 * @result: result code of completed RPC request
323 *
329 * We use a time-smoothed congestion estimator to avoid heavy oscillation. 324 * We use a time-smoothed congestion estimator to avoid heavy oscillation.
330 */ 325 */
331static void 326void xprt_adjust_cwnd(struct rpc_task *task, int result)
332xprt_adjust_cwnd(struct rpc_xprt *xprt, int result)
333{ 327{
334 unsigned long cwnd; 328 struct rpc_rqst *req = task->tk_rqstp;
329 struct rpc_xprt *xprt = task->tk_xprt;
330 unsigned long cwnd = xprt->cwnd;
335 331
336 cwnd = xprt->cwnd;
337 if (result >= 0 && cwnd <= xprt->cong) { 332 if (result >= 0 && cwnd <= xprt->cong) {
338 /* The (cwnd >> 1) term makes sure 333 /* The (cwnd >> 1) term makes sure
339 * the result gets rounded properly. */ 334 * the result gets rounded properly. */
340 cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd; 335 cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd;
341 if (cwnd > RPC_MAXCWND(xprt)) 336 if (cwnd > RPC_MAXCWND(xprt))
342 cwnd = RPC_MAXCWND(xprt); 337 cwnd = RPC_MAXCWND(xprt);
343 __xprt_lock_write_next(xprt); 338 __xprt_lock_write_next_cong(xprt);
344 } else if (result == -ETIMEDOUT) { 339 } else if (result == -ETIMEDOUT) {
345 cwnd >>= 1; 340 cwnd >>= 1;
346 if (cwnd < RPC_CWNDSCALE) 341 if (cwnd < RPC_CWNDSCALE)
@@ -349,11 +344,89 @@ xprt_adjust_cwnd(struct rpc_xprt *xprt, int result)
349 dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n", 344 dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n",
350 xprt->cong, xprt->cwnd, cwnd); 345 xprt->cong, xprt->cwnd, cwnd);
351 xprt->cwnd = cwnd; 346 xprt->cwnd = cwnd;
347 __xprt_put_cong(xprt, req);
348}
349
350/**
351 * xprt_wake_pending_tasks - wake all tasks on a transport's pending queue
352 * @xprt: transport with waiting tasks
353 * @status: result code to plant in each task before waking it
354 *
355 */
356void xprt_wake_pending_tasks(struct rpc_xprt *xprt, int status)
357{
358 if (status < 0)
359 rpc_wake_up_status(&xprt->pending, status);
360 else
361 rpc_wake_up(&xprt->pending);
362}
363
364/**
365 * xprt_wait_for_buffer_space - wait for transport output buffer to clear
366 * @task: task to be put to sleep
367 *
368 */
369void xprt_wait_for_buffer_space(struct rpc_task *task)
370{
371 struct rpc_rqst *req = task->tk_rqstp;
372 struct rpc_xprt *xprt = req->rq_xprt;
373
374 task->tk_timeout = req->rq_timeout;
375 rpc_sleep_on(&xprt->pending, task, NULL, NULL);
376}
377
378/**
379 * xprt_write_space - wake the task waiting for transport output buffer space
380 * @xprt: transport with waiting tasks
381 *
382 * Can be called in a soft IRQ context, so xprt_write_space never sleeps.
383 */
384void xprt_write_space(struct rpc_xprt *xprt)
385{
386 if (unlikely(xprt->shutdown))
387 return;
388
389 spin_lock_bh(&xprt->transport_lock);
390 if (xprt->snd_task) {
391 dprintk("RPC: write space: waking waiting task on xprt %p\n",
392 xprt);
393 rpc_wake_up_task(xprt->snd_task);
394 }
395 spin_unlock_bh(&xprt->transport_lock);
396}
397
398/**
399 * xprt_set_retrans_timeout_def - set a request's retransmit timeout
400 * @task: task whose timeout is to be set
401 *
402 * Set a request's retransmit timeout based on the transport's
403 * default timeout parameters. Used by transports that don't adjust
404 * the retransmit timeout based on round-trip time estimation.
405 */
406void xprt_set_retrans_timeout_def(struct rpc_task *task)
407{
408 task->tk_timeout = task->tk_rqstp->rq_timeout;
352} 409}
353 410
354/* 411/*
355 * Reset the major timeout value 412 * xprt_set_retrans_timeout_rtt - set a request's retransmit timeout
413 * @task: task whose timeout is to be set
414 *
415 * Set a request's retransmit timeout using the RTT estimator.
356 */ 416 */
417void xprt_set_retrans_timeout_rtt(struct rpc_task *task)
418{
419 int timer = task->tk_msg.rpc_proc->p_timer;
420 struct rpc_rtt *rtt = task->tk_client->cl_rtt;
421 struct rpc_rqst *req = task->tk_rqstp;
422 unsigned long max_timeout = req->rq_xprt->timeout.to_maxval;
423
424 task->tk_timeout = rpc_calc_rto(rtt, timer);
425 task->tk_timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries;
426 if (task->tk_timeout > max_timeout || task->tk_timeout == 0)
427 task->tk_timeout = max_timeout;
428}
429
357static void xprt_reset_majortimeo(struct rpc_rqst *req) 430static void xprt_reset_majortimeo(struct rpc_rqst *req)
358{ 431{
359 struct rpc_timeout *to = &req->rq_xprt->timeout; 432 struct rpc_timeout *to = &req->rq_xprt->timeout;
@@ -368,8 +441,10 @@ static void xprt_reset_majortimeo(struct rpc_rqst *req)
368 req->rq_majortimeo += jiffies; 441 req->rq_majortimeo += jiffies;
369} 442}
370 443
371/* 444/**
372 * Adjust timeout values etc for next retransmit 445 * xprt_adjust_timeout - adjust timeout values for next retransmit
446 * @req: RPC request containing parameters to use for the adjustment
447 *
373 */ 448 */
374int xprt_adjust_timeout(struct rpc_rqst *req) 449int xprt_adjust_timeout(struct rpc_rqst *req)
375{ 450{
@@ -391,9 +466,9 @@ int xprt_adjust_timeout(struct rpc_rqst *req)
391 req->rq_retries = 0; 466 req->rq_retries = 0;
392 xprt_reset_majortimeo(req); 467 xprt_reset_majortimeo(req);
393 /* Reset the RTT counters == "slow start" */ 468 /* Reset the RTT counters == "slow start" */
394 spin_lock_bh(&xprt->sock_lock); 469 spin_lock_bh(&xprt->transport_lock);
395 rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval); 470 rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval);
396 spin_unlock_bh(&xprt->sock_lock); 471 spin_unlock_bh(&xprt->transport_lock);
397 pprintk("RPC: %lu timeout\n", jiffies); 472 pprintk("RPC: %lu timeout\n", jiffies);
398 status = -ETIMEDOUT; 473 status = -ETIMEDOUT;
399 } 474 }
@@ -405,133 +480,52 @@ int xprt_adjust_timeout(struct rpc_rqst *req)
405 return status; 480 return status;
406} 481}
407 482
408/* 483static void xprt_autoclose(void *args)
409 * Close down a transport socket
410 */
411static void
412xprt_close(struct rpc_xprt *xprt)
413{
414 struct socket *sock = xprt->sock;
415 struct sock *sk = xprt->inet;
416
417 if (!sk)
418 return;
419
420 write_lock_bh(&sk->sk_callback_lock);
421 xprt->inet = NULL;
422 xprt->sock = NULL;
423
424 sk->sk_user_data = NULL;
425 sk->sk_data_ready = xprt->old_data_ready;
426 sk->sk_state_change = xprt->old_state_change;
427 sk->sk_write_space = xprt->old_write_space;
428 write_unlock_bh(&sk->sk_callback_lock);
429
430 sk->sk_no_check = 0;
431
432 sock_release(sock);
433}
434
435static void
436xprt_socket_autoclose(void *args)
437{ 484{
438 struct rpc_xprt *xprt = (struct rpc_xprt *)args; 485 struct rpc_xprt *xprt = (struct rpc_xprt *)args;
439 486
440 xprt_disconnect(xprt); 487 xprt_disconnect(xprt);
441 xprt_close(xprt); 488 xprt->ops->close(xprt);
442 xprt_release_write(xprt, NULL); 489 xprt_release_write(xprt, NULL);
443} 490}
444 491
445/* 492/**
446 * Mark a transport as disconnected 493 * xprt_disconnect - mark a transport as disconnected
494 * @xprt: transport to flag for disconnect
495 *
447 */ 496 */
448static void 497void xprt_disconnect(struct rpc_xprt *xprt)
449xprt_disconnect(struct rpc_xprt *xprt)
450{ 498{
451 dprintk("RPC: disconnected transport %p\n", xprt); 499 dprintk("RPC: disconnected transport %p\n", xprt);
452 spin_lock_bh(&xprt->sock_lock); 500 spin_lock_bh(&xprt->transport_lock);
453 xprt_clear_connected(xprt); 501 xprt_clear_connected(xprt);
454 rpc_wake_up_status(&xprt->pending, -ENOTCONN); 502 xprt_wake_pending_tasks(xprt, -ENOTCONN);
455 spin_unlock_bh(&xprt->sock_lock); 503 spin_unlock_bh(&xprt->transport_lock);
456} 504}
457 505
458/*
459 * Used to allow disconnection when we've been idle
460 */
461static void 506static void
462xprt_init_autodisconnect(unsigned long data) 507xprt_init_autodisconnect(unsigned long data)
463{ 508{
464 struct rpc_xprt *xprt = (struct rpc_xprt *)data; 509 struct rpc_xprt *xprt = (struct rpc_xprt *)data;
465 510
466 spin_lock(&xprt->sock_lock); 511 spin_lock(&xprt->transport_lock);
467 if (!list_empty(&xprt->recv) || xprt->shutdown) 512 if (!list_empty(&xprt->recv) || xprt->shutdown)
468 goto out_abort; 513 goto out_abort;
469 if (test_and_set_bit(XPRT_LOCKED, &xprt->sockstate)) 514 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
470 goto out_abort; 515 goto out_abort;
471 spin_unlock(&xprt->sock_lock); 516 spin_unlock(&xprt->transport_lock);
472 /* Let keventd close the socket */ 517 if (xprt_connecting(xprt))
473 if (test_bit(XPRT_CONNECTING, &xprt->sockstate) != 0)
474 xprt_release_write(xprt, NULL); 518 xprt_release_write(xprt, NULL);
475 else 519 else
476 schedule_work(&xprt->task_cleanup); 520 schedule_work(&xprt->task_cleanup);
477 return; 521 return;
478out_abort: 522out_abort:
479 spin_unlock(&xprt->sock_lock); 523 spin_unlock(&xprt->transport_lock);
480} 524}
481 525
482static void xprt_socket_connect(void *args) 526/**
483{ 527 * xprt_connect - schedule a transport connect operation
484 struct rpc_xprt *xprt = (struct rpc_xprt *)args; 528 * @task: RPC task that is requesting the connect
485 struct socket *sock = xprt->sock;
486 int status = -EIO;
487
488 if (xprt->shutdown || xprt->addr.sin_port == 0)
489 goto out;
490
491 /*
492 * Start by resetting any existing state
493 */
494 xprt_close(xprt);
495 sock = xprt_create_socket(xprt, xprt->prot, xprt->resvport);
496 if (sock == NULL) {
497 /* couldn't create socket or bind to reserved port;
498 * this is likely a permanent error, so cause an abort */
499 goto out;
500 }
501 xprt_bind_socket(xprt, sock);
502 xprt_sock_setbufsize(xprt);
503
504 status = 0;
505 if (!xprt->stream)
506 goto out;
507
508 /*
509 * Tell the socket layer to start connecting...
510 */
511 status = sock->ops->connect(sock, (struct sockaddr *) &xprt->addr,
512 sizeof(xprt->addr), O_NONBLOCK);
513 dprintk("RPC: %p connect status %d connected %d sock state %d\n",
514 xprt, -status, xprt_connected(xprt), sock->sk->sk_state);
515 if (status < 0) {
516 switch (status) {
517 case -EINPROGRESS:
518 case -EALREADY:
519 goto out_clear;
520 }
521 }
522out:
523 if (status < 0)
524 rpc_wake_up_status(&xprt->pending, status);
525 else
526 rpc_wake_up(&xprt->pending);
527out_clear:
528 smp_mb__before_clear_bit();
529 clear_bit(XPRT_CONNECTING, &xprt->sockstate);
530 smp_mb__after_clear_bit();
531}
532
533/*
534 * Attempt to connect a TCP socket.
535 * 529 *
536 */ 530 */
537void xprt_connect(struct rpc_task *task) 531void xprt_connect(struct rpc_task *task)
@@ -552,37 +546,19 @@ void xprt_connect(struct rpc_task *task)
552 if (!xprt_lock_write(xprt, task)) 546 if (!xprt_lock_write(xprt, task))
553 return; 547 return;
554 if (xprt_connected(xprt)) 548 if (xprt_connected(xprt))
555 goto out_write; 549 xprt_release_write(xprt, task);
550 else {
551 if (task->tk_rqstp)
552 task->tk_rqstp->rq_bytes_sent = 0;
556 553
557 if (task->tk_rqstp) 554 task->tk_timeout = xprt->connect_timeout;
558 task->tk_rqstp->rq_bytes_sent = 0; 555 rpc_sleep_on(&xprt->pending, task, xprt_connect_status, NULL);
559 556 xprt->ops->connect(task);
560 task->tk_timeout = RPC_CONNECT_TIMEOUT;
561 rpc_sleep_on(&xprt->pending, task, xprt_connect_status, NULL);
562 if (!test_and_set_bit(XPRT_CONNECTING, &xprt->sockstate)) {
563 /* Note: if we are here due to a dropped connection
564 * we delay reconnecting by RPC_REESTABLISH_TIMEOUT/HZ
565 * seconds
566 */
567 if (xprt->sock != NULL)
568 schedule_delayed_work(&xprt->sock_connect,
569 RPC_REESTABLISH_TIMEOUT);
570 else {
571 schedule_work(&xprt->sock_connect);
572 if (!RPC_IS_ASYNC(task))
573 flush_scheduled_work();
574 }
575 } 557 }
576 return; 558 return;
577 out_write:
578 xprt_release_write(xprt, task);
579} 559}
580 560
581/* 561static void xprt_connect_status(struct rpc_task *task)
582 * We arrive here when awoken from waiting on connection establishment.
583 */
584static void
585xprt_connect_status(struct rpc_task *task)
586{ 562{
587 struct rpc_xprt *xprt = task->tk_xprt; 563 struct rpc_xprt *xprt = task->tk_xprt;
588 564
@@ -592,31 +568,42 @@ xprt_connect_status(struct rpc_task *task)
592 return; 568 return;
593 } 569 }
594 570
595 /* if soft mounted, just cause this RPC to fail */
596 if (RPC_IS_SOFT(task))
597 task->tk_status = -EIO;
598
599 switch (task->tk_status) { 571 switch (task->tk_status) {
600 case -ECONNREFUSED: 572 case -ECONNREFUSED:
601 case -ECONNRESET: 573 case -ECONNRESET:
574 dprintk("RPC: %4d xprt_connect_status: server %s refused connection\n",
575 task->tk_pid, task->tk_client->cl_server);
576 break;
602 case -ENOTCONN: 577 case -ENOTCONN:
603 return; 578 dprintk("RPC: %4d xprt_connect_status: connection broken\n",
579 task->tk_pid);
580 break;
604 case -ETIMEDOUT: 581 case -ETIMEDOUT:
605 dprintk("RPC: %4d xprt_connect_status: timed out\n", 582 dprintk("RPC: %4d xprt_connect_status: connect attempt timed out\n",
606 task->tk_pid); 583 task->tk_pid);
607 break; 584 break;
608 default: 585 default:
609 printk(KERN_ERR "RPC: error %d connecting to server %s\n", 586 dprintk("RPC: %4d xprt_connect_status: error %d connecting to server %s\n",
610 -task->tk_status, task->tk_client->cl_server); 587 task->tk_pid, -task->tk_status, task->tk_client->cl_server);
588 xprt_release_write(xprt, task);
589 task->tk_status = -EIO;
590 return;
591 }
592
593 /* if soft mounted, just cause this RPC to fail */
594 if (RPC_IS_SOFT(task)) {
595 xprt_release_write(xprt, task);
596 task->tk_status = -EIO;
611 } 597 }
612 xprt_release_write(xprt, task);
613} 598}
614 599
615/* 600/**
616 * Look up the RPC request corresponding to a reply, and then lock it. 601 * xprt_lookup_rqst - find an RPC request corresponding to an XID
602 * @xprt: transport on which the original request was transmitted
603 * @xid: RPC XID of incoming reply
604 *
617 */ 605 */
618static inline struct rpc_rqst * 606struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, u32 xid)
619xprt_lookup_rqst(struct rpc_xprt *xprt, u32 xid)
620{ 607{
621 struct list_head *pos; 608 struct list_head *pos;
622 struct rpc_rqst *req = NULL; 609 struct rpc_rqst *req = NULL;
@@ -631,556 +618,68 @@ xprt_lookup_rqst(struct rpc_xprt *xprt, u32 xid)
631 return req; 618 return req;
632} 619}
633 620
634/* 621/**
635 * Complete reply received. 622 * xprt_update_rtt - update an RPC client's RTT state after receiving a reply
636 * The TCP code relies on us to remove the request from xprt->pending. 623 * @task: RPC request that recently completed
637 */ 624 *
638static void
639xprt_complete_rqst(struct rpc_xprt *xprt, struct rpc_rqst *req, int copied)
640{
641 struct rpc_task *task = req->rq_task;
642 struct rpc_clnt *clnt = task->tk_client;
643
644 /* Adjust congestion window */
645 if (!xprt->nocong) {
646 unsigned timer = task->tk_msg.rpc_proc->p_timer;
647 xprt_adjust_cwnd(xprt, copied);
648 __xprt_put_cong(xprt, req);
649 if (timer) {
650 if (req->rq_ntrans == 1)
651 rpc_update_rtt(clnt->cl_rtt, timer,
652 (long)jiffies - req->rq_xtime);
653 rpc_set_timeo(clnt->cl_rtt, timer, req->rq_ntrans - 1);
654 }
655 }
656
657#ifdef RPC_PROFILE
658 /* Profile only reads for now */
659 if (copied > 1024) {
660 static unsigned long nextstat;
661 static unsigned long pkt_rtt, pkt_len, pkt_cnt;
662
663 pkt_cnt++;
664 pkt_len += req->rq_slen + copied;
665 pkt_rtt += jiffies - req->rq_xtime;
666 if (time_before(nextstat, jiffies)) {
667 printk("RPC: %lu %ld cwnd\n", jiffies, xprt->cwnd);
668 printk("RPC: %ld %ld %ld %ld stat\n",
669 jiffies, pkt_cnt, pkt_len, pkt_rtt);
670 pkt_rtt = pkt_len = pkt_cnt = 0;
671 nextstat = jiffies + 5 * HZ;
672 }
673 }
674#endif
675
676 dprintk("RPC: %4d has input (%d bytes)\n", task->tk_pid, copied);
677 list_del_init(&req->rq_list);
678 req->rq_received = req->rq_private_buf.len = copied;
679
680 /* ... and wake up the process. */
681 rpc_wake_up_task(task);
682 return;
683}
684
685static size_t
686skb_read_bits(skb_reader_t *desc, void *to, size_t len)
687{
688 if (len > desc->count)
689 len = desc->count;
690 if (skb_copy_bits(desc->skb, desc->offset, to, len))
691 return 0;
692 desc->count -= len;
693 desc->offset += len;
694 return len;
695}
696
697static size_t
698skb_read_and_csum_bits(skb_reader_t *desc, void *to, size_t len)
699{
700 unsigned int csum2, pos;
701
702 if (len > desc->count)
703 len = desc->count;
704 pos = desc->offset;
705 csum2 = skb_copy_and_csum_bits(desc->skb, pos, to, len, 0);
706 desc->csum = csum_block_add(desc->csum, csum2, pos);
707 desc->count -= len;
708 desc->offset += len;
709 return len;
710}
711
712/*
713 * We have set things up such that we perform the checksum of the UDP
714 * packet in parallel with the copies into the RPC client iovec. -DaveM
715 */
716int
717csum_partial_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb)
718{
719 skb_reader_t desc;
720
721 desc.skb = skb;
722 desc.offset = sizeof(struct udphdr);
723 desc.count = skb->len - desc.offset;
724
725 if (skb->ip_summed == CHECKSUM_UNNECESSARY)
726 goto no_checksum;
727
728 desc.csum = csum_partial(skb->data, desc.offset, skb->csum);
729 if (xdr_partial_copy_from_skb(xdr, 0, &desc, skb_read_and_csum_bits) < 0)
730 return -1;
731 if (desc.offset != skb->len) {
732 unsigned int csum2;
733 csum2 = skb_checksum(skb, desc.offset, skb->len - desc.offset, 0);
734 desc.csum = csum_block_add(desc.csum, csum2, desc.offset);
735 }
736 if (desc.count)
737 return -1;
738 if ((unsigned short)csum_fold(desc.csum))
739 return -1;
740 return 0;
741no_checksum:
742 if (xdr_partial_copy_from_skb(xdr, 0, &desc, skb_read_bits) < 0)
743 return -1;
744 if (desc.count)
745 return -1;
746 return 0;
747}
748
749/*
750 * Input handler for RPC replies. Called from a bottom half and hence
751 * atomic.
752 */
753static void
754udp_data_ready(struct sock *sk, int len)
755{
756 struct rpc_task *task;
757 struct rpc_xprt *xprt;
758 struct rpc_rqst *rovr;
759 struct sk_buff *skb;
760 int err, repsize, copied;
761 u32 _xid, *xp;
762
763 read_lock(&sk->sk_callback_lock);
764 dprintk("RPC: udp_data_ready...\n");
765 if (!(xprt = xprt_from_sock(sk))) {
766 printk("RPC: udp_data_ready request not found!\n");
767 goto out;
768 }
769
770 dprintk("RPC: udp_data_ready client %p\n", xprt);
771
772 if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
773 goto out;
774
775 if (xprt->shutdown)
776 goto dropit;
777
778 repsize = skb->len - sizeof(struct udphdr);
779 if (repsize < 4) {
780 printk("RPC: impossible RPC reply size %d!\n", repsize);
781 goto dropit;
782 }
783
784 /* Copy the XID from the skb... */
785 xp = skb_header_pointer(skb, sizeof(struct udphdr),
786 sizeof(_xid), &_xid);
787 if (xp == NULL)
788 goto dropit;
789
790 /* Look up and lock the request corresponding to the given XID */
791 spin_lock(&xprt->sock_lock);
792 rovr = xprt_lookup_rqst(xprt, *xp);
793 if (!rovr)
794 goto out_unlock;
795 task = rovr->rq_task;
796
797 dprintk("RPC: %4d received reply\n", task->tk_pid);
798
799 if ((copied = rovr->rq_private_buf.buflen) > repsize)
800 copied = repsize;
801
802 /* Suck it into the iovec, verify checksum if not done by hw. */
803 if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb))
804 goto out_unlock;
805
806 /* Something worked... */
807 dst_confirm(skb->dst);
808
809 xprt_complete_rqst(xprt, rovr, copied);
810
811 out_unlock:
812 spin_unlock(&xprt->sock_lock);
813 dropit:
814 skb_free_datagram(sk, skb);
815 out:
816 read_unlock(&sk->sk_callback_lock);
817}
818
819/*
820 * Copy from an skb into memory and shrink the skb.
821 */
822static inline size_t
823tcp_copy_data(skb_reader_t *desc, void *p, size_t len)
824{
825 if (len > desc->count)
826 len = desc->count;
827 if (skb_copy_bits(desc->skb, desc->offset, p, len)) {
828 dprintk("RPC: failed to copy %zu bytes from skb. %zu bytes remain\n",
829 len, desc->count);
830 return 0;
831 }
832 desc->offset += len;
833 desc->count -= len;
834 dprintk("RPC: copied %zu bytes from skb. %zu bytes remain\n",
835 len, desc->count);
836 return len;
837}
838
839/*
840 * TCP read fragment marker
841 */
842static inline void
843tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc)
844{
845 size_t len, used;
846 char *p;
847
848 p = ((char *) &xprt->tcp_recm) + xprt->tcp_offset;
849 len = sizeof(xprt->tcp_recm) - xprt->tcp_offset;
850 used = tcp_copy_data(desc, p, len);
851 xprt->tcp_offset += used;
852 if (used != len)
853 return;
854 xprt->tcp_reclen = ntohl(xprt->tcp_recm);
855 if (xprt->tcp_reclen & 0x80000000)
856 xprt->tcp_flags |= XPRT_LAST_FRAG;
857 else
858 xprt->tcp_flags &= ~XPRT_LAST_FRAG;
859 xprt->tcp_reclen &= 0x7fffffff;
860 xprt->tcp_flags &= ~XPRT_COPY_RECM;
861 xprt->tcp_offset = 0;
862 /* Sanity check of the record length */
863 if (xprt->tcp_reclen < 4) {
864 printk(KERN_ERR "RPC: Invalid TCP record fragment length\n");
865 xprt_disconnect(xprt);
866 }
867 dprintk("RPC: reading TCP record fragment of length %d\n",
868 xprt->tcp_reclen);
869}
870
871static void
872tcp_check_recm(struct rpc_xprt *xprt)
873{
874 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u, tcp_flags = %lx\n",
875 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen, xprt->tcp_flags);
876 if (xprt->tcp_offset == xprt->tcp_reclen) {
877 xprt->tcp_flags |= XPRT_COPY_RECM;
878 xprt->tcp_offset = 0;
879 if (xprt->tcp_flags & XPRT_LAST_FRAG) {
880 xprt->tcp_flags &= ~XPRT_COPY_DATA;
881 xprt->tcp_flags |= XPRT_COPY_XID;
882 xprt->tcp_copied = 0;
883 }
884 }
885}
886
887/*
888 * TCP read xid
889 */
890static inline void
891tcp_read_xid(struct rpc_xprt *xprt, skb_reader_t *desc)
892{
893 size_t len, used;
894 char *p;
895
896 len = sizeof(xprt->tcp_xid) - xprt->tcp_offset;
897 dprintk("RPC: reading XID (%Zu bytes)\n", len);
898 p = ((char *) &xprt->tcp_xid) + xprt->tcp_offset;
899 used = tcp_copy_data(desc, p, len);
900 xprt->tcp_offset += used;
901 if (used != len)
902 return;
903 xprt->tcp_flags &= ~XPRT_COPY_XID;
904 xprt->tcp_flags |= XPRT_COPY_DATA;
905 xprt->tcp_copied = 4;
906 dprintk("RPC: reading reply for XID %08x\n",
907 ntohl(xprt->tcp_xid));
908 tcp_check_recm(xprt);
909}
910
911/*
912 * TCP read and complete request
913 */
914static inline void
915tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc)
916{
917 struct rpc_rqst *req;
918 struct xdr_buf *rcvbuf;
919 size_t len;
920 ssize_t r;
921
922 /* Find and lock the request corresponding to this xid */
923 spin_lock(&xprt->sock_lock);
924 req = xprt_lookup_rqst(xprt, xprt->tcp_xid);
925 if (!req) {
926 xprt->tcp_flags &= ~XPRT_COPY_DATA;
927 dprintk("RPC: XID %08x request not found!\n",
928 ntohl(xprt->tcp_xid));
929 spin_unlock(&xprt->sock_lock);
930 return;
931 }
932
933 rcvbuf = &req->rq_private_buf;
934 len = desc->count;
935 if (len > xprt->tcp_reclen - xprt->tcp_offset) {
936 skb_reader_t my_desc;
937
938 len = xprt->tcp_reclen - xprt->tcp_offset;
939 memcpy(&my_desc, desc, sizeof(my_desc));
940 my_desc.count = len;
941 r = xdr_partial_copy_from_skb(rcvbuf, xprt->tcp_copied,
942 &my_desc, tcp_copy_data);
943 desc->count -= r;
944 desc->offset += r;
945 } else
946 r = xdr_partial_copy_from_skb(rcvbuf, xprt->tcp_copied,
947 desc, tcp_copy_data);
948
949 if (r > 0) {
950 xprt->tcp_copied += r;
951 xprt->tcp_offset += r;
952 }
953 if (r != len) {
954 /* Error when copying to the receive buffer,
955 * usually because we weren't able to allocate
956 * additional buffer pages. All we can do now
957 * is turn off XPRT_COPY_DATA, so the request
958 * will not receive any additional updates,
959 * and time out.
960 * Any remaining data from this record will
961 * be discarded.
962 */
963 xprt->tcp_flags &= ~XPRT_COPY_DATA;
964 dprintk("RPC: XID %08x truncated request\n",
965 ntohl(xprt->tcp_xid));
966 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
967 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen);
968 goto out;
969 }
970
971 dprintk("RPC: XID %08x read %Zd bytes\n",
972 ntohl(xprt->tcp_xid), r);
973 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
974 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen);
975
976 if (xprt->tcp_copied == req->rq_private_buf.buflen)
977 xprt->tcp_flags &= ~XPRT_COPY_DATA;
978 else if (xprt->tcp_offset == xprt->tcp_reclen) {
979 if (xprt->tcp_flags & XPRT_LAST_FRAG)
980 xprt->tcp_flags &= ~XPRT_COPY_DATA;
981 }
982
983out:
984 if (!(xprt->tcp_flags & XPRT_COPY_DATA)) {
985 dprintk("RPC: %4d received reply complete\n",
986 req->rq_task->tk_pid);
987 xprt_complete_rqst(xprt, req, xprt->tcp_copied);
988 }
989 spin_unlock(&xprt->sock_lock);
990 tcp_check_recm(xprt);
991}
992
993/*
994 * TCP discard extra bytes from a short read
995 */
996static inline void
997tcp_read_discard(struct rpc_xprt *xprt, skb_reader_t *desc)
998{
999 size_t len;
1000
1001 len = xprt->tcp_reclen - xprt->tcp_offset;
1002 if (len > desc->count)
1003 len = desc->count;
1004 desc->count -= len;
1005 desc->offset += len;
1006 xprt->tcp_offset += len;
1007 dprintk("RPC: discarded %Zu bytes\n", len);
1008 tcp_check_recm(xprt);
1009}
1010
1011/*
1012 * TCP record receive routine
1013 * We first have to grab the record marker, then the XID, then the data.
1014 */ 625 */
1015static int 626void xprt_update_rtt(struct rpc_task *task)
1016tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb,
1017 unsigned int offset, size_t len)
1018{
1019 struct rpc_xprt *xprt = rd_desc->arg.data;
1020 skb_reader_t desc = {
1021 .skb = skb,
1022 .offset = offset,
1023 .count = len,
1024 .csum = 0
1025 };
1026
1027 dprintk("RPC: tcp_data_recv\n");
1028 do {
1029 /* Read in a new fragment marker if necessary */
1030 /* Can we ever really expect to get completely empty fragments? */
1031 if (xprt->tcp_flags & XPRT_COPY_RECM) {
1032 tcp_read_fraghdr(xprt, &desc);
1033 continue;
1034 }
1035 /* Read in the xid if necessary */
1036 if (xprt->tcp_flags & XPRT_COPY_XID) {
1037 tcp_read_xid(xprt, &desc);
1038 continue;
1039 }
1040 /* Read in the request data */
1041 if (xprt->tcp_flags & XPRT_COPY_DATA) {
1042 tcp_read_request(xprt, &desc);
1043 continue;
1044 }
1045 /* Skip over any trailing bytes on short reads */
1046 tcp_read_discard(xprt, &desc);
1047 } while (desc.count);
1048 dprintk("RPC: tcp_data_recv done\n");
1049 return len - desc.count;
1050}
1051
1052static void tcp_data_ready(struct sock *sk, int bytes)
1053{ 627{
1054 struct rpc_xprt *xprt; 628 struct rpc_rqst *req = task->tk_rqstp;
1055 read_descriptor_t rd_desc; 629 struct rpc_rtt *rtt = task->tk_client->cl_rtt;
1056 630 unsigned timer = task->tk_msg.rpc_proc->p_timer;
1057 read_lock(&sk->sk_callback_lock);
1058 dprintk("RPC: tcp_data_ready...\n");
1059 if (!(xprt = xprt_from_sock(sk))) {
1060 printk("RPC: tcp_data_ready socket info not found!\n");
1061 goto out;
1062 }
1063 if (xprt->shutdown)
1064 goto out;
1065
1066 /* We use rd_desc to pass struct xprt to tcp_data_recv */
1067 rd_desc.arg.data = xprt;
1068 rd_desc.count = 65536;
1069 tcp_read_sock(sk, &rd_desc, tcp_data_recv);
1070out:
1071 read_unlock(&sk->sk_callback_lock);
1072}
1073
1074static void
1075tcp_state_change(struct sock *sk)
1076{
1077 struct rpc_xprt *xprt;
1078 631
1079 read_lock(&sk->sk_callback_lock); 632 if (timer) {
1080 if (!(xprt = xprt_from_sock(sk))) 633 if (req->rq_ntrans == 1)
1081 goto out; 634 rpc_update_rtt(rtt, timer,
1082 dprintk("RPC: tcp_state_change client %p...\n", xprt); 635 (long)jiffies - req->rq_xtime);
1083 dprintk("RPC: state %x conn %d dead %d zapped %d\n", 636 rpc_set_timeo(rtt, timer, req->rq_ntrans - 1);
1084 sk->sk_state, xprt_connected(xprt),
1085 sock_flag(sk, SOCK_DEAD),
1086 sock_flag(sk, SOCK_ZAPPED));
1087
1088 switch (sk->sk_state) {
1089 case TCP_ESTABLISHED:
1090 spin_lock_bh(&xprt->sock_lock);
1091 if (!xprt_test_and_set_connected(xprt)) {
1092 /* Reset TCP record info */
1093 xprt->tcp_offset = 0;
1094 xprt->tcp_reclen = 0;
1095 xprt->tcp_copied = 0;
1096 xprt->tcp_flags = XPRT_COPY_RECM | XPRT_COPY_XID;
1097 rpc_wake_up(&xprt->pending);
1098 }
1099 spin_unlock_bh(&xprt->sock_lock);
1100 break;
1101 case TCP_SYN_SENT:
1102 case TCP_SYN_RECV:
1103 break;
1104 default:
1105 xprt_disconnect(xprt);
1106 break;
1107 } 637 }
1108 out:
1109 read_unlock(&sk->sk_callback_lock);
1110} 638}
1111 639
1112/* 640/**
1113 * Called when more output buffer space is available for this socket. 641 * xprt_complete_rqst - called when reply processing is complete
1114 * We try not to wake our writers until they can make "significant" 642 * @task: RPC request that recently completed
1115 * progress, otherwise we'll waste resources thrashing sock_sendmsg 643 * @copied: actual number of bytes received from the transport
1116 * with a bunch of small requests. 644 *
645 * Caller holds transport lock.
1117 */ 646 */
1118static void 647void xprt_complete_rqst(struct rpc_task *task, int copied)
1119xprt_write_space(struct sock *sk)
1120{ 648{
1121 struct rpc_xprt *xprt; 649 struct rpc_rqst *req = task->tk_rqstp;
1122 struct socket *sock;
1123
1124 read_lock(&sk->sk_callback_lock);
1125 if (!(xprt = xprt_from_sock(sk)) || !(sock = sk->sk_socket))
1126 goto out;
1127 if (xprt->shutdown)
1128 goto out;
1129
1130 /* Wait until we have enough socket memory */
1131 if (xprt->stream) {
1132 /* from net/core/stream.c:sk_stream_write_space */
1133 if (sk_stream_wspace(sk) < sk_stream_min_wspace(sk))
1134 goto out;
1135 } else {
1136 /* from net/core/sock.c:sock_def_write_space */
1137 if (!sock_writeable(sk))
1138 goto out;
1139 }
1140 650
1141 if (!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)) 651 dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
1142 goto out; 652 task->tk_pid, ntohl(req->rq_xid), copied);
1143 653
1144 spin_lock_bh(&xprt->sock_lock); 654 list_del_init(&req->rq_list);
1145 if (xprt->snd_task) 655 req->rq_received = req->rq_private_buf.len = copied;
1146 rpc_wake_up_task(xprt->snd_task); 656 rpc_wake_up_task(task);
1147 spin_unlock_bh(&xprt->sock_lock);
1148out:
1149 read_unlock(&sk->sk_callback_lock);
1150} 657}
1151 658
1152/* 659static void xprt_timer(struct rpc_task *task)
1153 * RPC receive timeout handler.
1154 */
1155static void
1156xprt_timer(struct rpc_task *task)
1157{ 660{
1158 struct rpc_rqst *req = task->tk_rqstp; 661 struct rpc_rqst *req = task->tk_rqstp;
1159 struct rpc_xprt *xprt = req->rq_xprt; 662 struct rpc_xprt *xprt = req->rq_xprt;
1160 663
1161 spin_lock(&xprt->sock_lock); 664 dprintk("RPC: %4d xprt_timer\n", task->tk_pid);
1162 if (req->rq_received)
1163 goto out;
1164
1165 xprt_adjust_cwnd(req->rq_xprt, -ETIMEDOUT);
1166 __xprt_put_cong(xprt, req);
1167 665
1168 dprintk("RPC: %4d xprt_timer (%s request)\n", 666 spin_lock(&xprt->transport_lock);
1169 task->tk_pid, req ? "pending" : "backlogged"); 667 if (!req->rq_received) {
1170 668 if (xprt->ops->timer)
1171 task->tk_status = -ETIMEDOUT; 669 xprt->ops->timer(task);
1172out: 670 task->tk_status = -ETIMEDOUT;
671 }
1173 task->tk_timeout = 0; 672 task->tk_timeout = 0;
1174 rpc_wake_up_task(task); 673 rpc_wake_up_task(task);
1175 spin_unlock(&xprt->sock_lock); 674 spin_unlock(&xprt->transport_lock);
1176} 675}
1177 676
1178/* 677/**
1179 * Place the actual RPC call. 678 * xprt_prepare_transmit - reserve the transport before sending a request
1180 * We have to copy the iovec because sendmsg fiddles with its contents. 679 * @task: RPC task about to send a request
680 *
1181 */ 681 */
1182int 682int xprt_prepare_transmit(struct rpc_task *task)
1183xprt_prepare_transmit(struct rpc_task *task)
1184{ 683{
1185 struct rpc_rqst *req = task->tk_rqstp; 684 struct rpc_rqst *req = task->tk_rqstp;
1186 struct rpc_xprt *xprt = req->rq_xprt; 685 struct rpc_xprt *xprt = req->rq_xprt;
@@ -1191,12 +690,12 @@ xprt_prepare_transmit(struct rpc_task *task)
1191 if (xprt->shutdown) 690 if (xprt->shutdown)
1192 return -EIO; 691 return -EIO;
1193 692
1194 spin_lock_bh(&xprt->sock_lock); 693 spin_lock_bh(&xprt->transport_lock);
1195 if (req->rq_received && !req->rq_bytes_sent) { 694 if (req->rq_received && !req->rq_bytes_sent) {
1196 err = req->rq_received; 695 err = req->rq_received;
1197 goto out_unlock; 696 goto out_unlock;
1198 } 697 }
1199 if (!__xprt_lock_write(xprt, task)) { 698 if (!xprt->ops->reserve_xprt(task)) {
1200 err = -EAGAIN; 699 err = -EAGAIN;
1201 goto out_unlock; 700 goto out_unlock;
1202 } 701 }
@@ -1206,39 +705,34 @@ xprt_prepare_transmit(struct rpc_task *task)
1206 goto out_unlock; 705 goto out_unlock;
1207 } 706 }
1208out_unlock: 707out_unlock:
1209 spin_unlock_bh(&xprt->sock_lock); 708 spin_unlock_bh(&xprt->transport_lock);
1210 return err; 709 return err;
1211} 710}
1212 711
1213void 712/**
1214xprt_transmit(struct rpc_task *task) 713 * xprt_transmit - send an RPC request on a transport
714 * @task: controlling RPC task
715 *
716 * We have to copy the iovec because sendmsg fiddles with its contents.
717 */
718void xprt_transmit(struct rpc_task *task)
1215{ 719{
1216 struct rpc_clnt *clnt = task->tk_client;
1217 struct rpc_rqst *req = task->tk_rqstp; 720 struct rpc_rqst *req = task->tk_rqstp;
1218 struct rpc_xprt *xprt = req->rq_xprt; 721 struct rpc_xprt *xprt = req->rq_xprt;
1219 int status, retry = 0; 722 int status;
1220
1221 723
1222 dprintk("RPC: %4d xprt_transmit(%u)\n", task->tk_pid, req->rq_slen); 724 dprintk("RPC: %4d xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
1223 725
1224 /* set up everything as needed. */
1225 /* Write the record marker */
1226 if (xprt->stream) {
1227 u32 *marker = req->rq_svec[0].iov_base;
1228
1229 *marker = htonl(0x80000000|(req->rq_slen-sizeof(*marker)));
1230 }
1231
1232 smp_rmb(); 726 smp_rmb();
1233 if (!req->rq_received) { 727 if (!req->rq_received) {
1234 if (list_empty(&req->rq_list)) { 728 if (list_empty(&req->rq_list)) {
1235 spin_lock_bh(&xprt->sock_lock); 729 spin_lock_bh(&xprt->transport_lock);
1236 /* Update the softirq receive buffer */ 730 /* Update the softirq receive buffer */
1237 memcpy(&req->rq_private_buf, &req->rq_rcv_buf, 731 memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
1238 sizeof(req->rq_private_buf)); 732 sizeof(req->rq_private_buf));
1239 /* Add request to the receive list */ 733 /* Add request to the receive list */
1240 list_add_tail(&req->rq_list, &xprt->recv); 734 list_add_tail(&req->rq_list, &xprt->recv);
1241 spin_unlock_bh(&xprt->sock_lock); 735 spin_unlock_bh(&xprt->transport_lock);
1242 xprt_reset_majortimeo(req); 736 xprt_reset_majortimeo(req);
1243 /* Turn off autodisconnect */ 737 /* Turn off autodisconnect */
1244 del_singleshot_timer_sync(&xprt->timer); 738 del_singleshot_timer_sync(&xprt->timer);
@@ -1246,40 +740,19 @@ xprt_transmit(struct rpc_task *task)
1246 } else if (!req->rq_bytes_sent) 740 } else if (!req->rq_bytes_sent)
1247 return; 741 return;
1248 742
1249 /* Continue transmitting the packet/record. We must be careful 743 status = xprt->ops->send_request(task);
1250 * to cope with writespace callbacks arriving _after_ we have 744 if (status == 0) {
1251 * called xprt_sendmsg(). 745 dprintk("RPC: %4d xmit complete\n", task->tk_pid);
1252 */ 746 spin_lock_bh(&xprt->transport_lock);
1253 while (1) { 747 xprt->ops->set_retrans_timeout(task);
1254 req->rq_xtime = jiffies; 748 /* Don't race with disconnect */
1255 status = xprt_sendmsg(xprt, req); 749 if (!xprt_connected(xprt))
1256 750 task->tk_status = -ENOTCONN;
1257 if (status < 0) 751 else if (!req->rq_received)
1258 break; 752 rpc_sleep_on(&xprt->pending, task, NULL, xprt_timer);
1259 753 xprt->ops->release_xprt(xprt, task);
1260 if (xprt->stream) { 754 spin_unlock_bh(&xprt->transport_lock);
1261 req->rq_bytes_sent += status; 755 return;
1262
1263 /* If we've sent the entire packet, immediately
1264 * reset the count of bytes sent. */
1265 if (req->rq_bytes_sent >= req->rq_slen) {
1266 req->rq_bytes_sent = 0;
1267 goto out_receive;
1268 }
1269 } else {
1270 if (status >= req->rq_slen)
1271 goto out_receive;
1272 status = -EAGAIN;
1273 break;
1274 }
1275
1276 dprintk("RPC: %4d xmit incomplete (%d left of %d)\n",
1277 task->tk_pid, req->rq_slen - req->rq_bytes_sent,
1278 req->rq_slen);
1279
1280 status = -EAGAIN;
1281 if (retry++ > 50)
1282 break;
1283 } 756 }
1284 757
1285 /* Note: at this point, task->tk_sleeping has not yet been set, 758 /* Note: at this point, task->tk_sleeping has not yet been set,
@@ -1289,60 +762,19 @@ xprt_transmit(struct rpc_task *task)
1289 task->tk_status = status; 762 task->tk_status = status;
1290 763
1291 switch (status) { 764 switch (status) {
1292 case -EAGAIN:
1293 if (test_bit(SOCK_ASYNC_NOSPACE, &xprt->sock->flags)) {
1294 /* Protect against races with xprt_write_space */
1295 spin_lock_bh(&xprt->sock_lock);
1296 /* Don't race with disconnect */
1297 if (!xprt_connected(xprt))
1298 task->tk_status = -ENOTCONN;
1299 else if (test_bit(SOCK_NOSPACE, &xprt->sock->flags)) {
1300 task->tk_timeout = req->rq_timeout;
1301 rpc_sleep_on(&xprt->pending, task, NULL, NULL);
1302 }
1303 spin_unlock_bh(&xprt->sock_lock);
1304 return;
1305 }
1306 /* Keep holding the socket if it is blocked */
1307 rpc_delay(task, HZ>>4);
1308 return;
1309 case -ECONNREFUSED: 765 case -ECONNREFUSED:
1310 task->tk_timeout = RPC_REESTABLISH_TIMEOUT;
1311 rpc_sleep_on(&xprt->sending, task, NULL, NULL); 766 rpc_sleep_on(&xprt->sending, task, NULL, NULL);
767 case -EAGAIN:
1312 case -ENOTCONN: 768 case -ENOTCONN:
1313 return; 769 return;
1314 default: 770 default:
1315 if (xprt->stream) 771 break;
1316 xprt_disconnect(xprt);
1317 } 772 }
1318 xprt_release_write(xprt, task); 773 xprt_release_write(xprt, task);
1319 return; 774 return;
1320 out_receive:
1321 dprintk("RPC: %4d xmit complete\n", task->tk_pid);
1322 /* Set the task's receive timeout value */
1323 spin_lock_bh(&xprt->sock_lock);
1324 if (!xprt->nocong) {
1325 int timer = task->tk_msg.rpc_proc->p_timer;
1326 task->tk_timeout = rpc_calc_rto(clnt->cl_rtt, timer);
1327 task->tk_timeout <<= rpc_ntimeo(clnt->cl_rtt, timer) + req->rq_retries;
1328 if (task->tk_timeout > xprt->timeout.to_maxval || task->tk_timeout == 0)
1329 task->tk_timeout = xprt->timeout.to_maxval;
1330 } else
1331 task->tk_timeout = req->rq_timeout;
1332 /* Don't race with disconnect */
1333 if (!xprt_connected(xprt))
1334 task->tk_status = -ENOTCONN;
1335 else if (!req->rq_received)
1336 rpc_sleep_on(&xprt->pending, task, NULL, xprt_timer);
1337 __xprt_release_write(xprt, task);
1338 spin_unlock_bh(&xprt->sock_lock);
1339} 775}
1340 776
1341/* 777static inline void do_xprt_reserve(struct rpc_task *task)
1342 * Reserve an RPC call slot.
1343 */
1344static inline void
1345do_xprt_reserve(struct rpc_task *task)
1346{ 778{
1347 struct rpc_xprt *xprt = task->tk_xprt; 779 struct rpc_xprt *xprt = task->tk_xprt;
1348 780
@@ -1362,22 +794,25 @@ do_xprt_reserve(struct rpc_task *task)
1362 rpc_sleep_on(&xprt->backlog, task, NULL, NULL); 794 rpc_sleep_on(&xprt->backlog, task, NULL, NULL);
1363} 795}
1364 796
1365void 797/**
1366xprt_reserve(struct rpc_task *task) 798 * xprt_reserve - allocate an RPC request slot
799 * @task: RPC task requesting a slot allocation
800 *
801 * If no more slots are available, place the task on the transport's
802 * backlog queue.
803 */
804void xprt_reserve(struct rpc_task *task)
1367{ 805{
1368 struct rpc_xprt *xprt = task->tk_xprt; 806 struct rpc_xprt *xprt = task->tk_xprt;
1369 807
1370 task->tk_status = -EIO; 808 task->tk_status = -EIO;
1371 if (!xprt->shutdown) { 809 if (!xprt->shutdown) {
1372 spin_lock(&xprt->xprt_lock); 810 spin_lock(&xprt->reserve_lock);
1373 do_xprt_reserve(task); 811 do_xprt_reserve(task);
1374 spin_unlock(&xprt->xprt_lock); 812 spin_unlock(&xprt->reserve_lock);
1375 } 813 }
1376} 814}
1377 815
1378/*
1379 * Allocate a 'unique' XID
1380 */
1381static inline u32 xprt_alloc_xid(struct rpc_xprt *xprt) 816static inline u32 xprt_alloc_xid(struct rpc_xprt *xprt)
1382{ 817{
1383 return xprt->xid++; 818 return xprt->xid++;
@@ -1388,11 +823,7 @@ static inline void xprt_init_xid(struct rpc_xprt *xprt)
1388 get_random_bytes(&xprt->xid, sizeof(xprt->xid)); 823 get_random_bytes(&xprt->xid, sizeof(xprt->xid));
1389} 824}
1390 825
1391/* 826static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
1392 * Initialize RPC request
1393 */
1394static void
1395xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
1396{ 827{
1397 struct rpc_rqst *req = task->tk_rqstp; 828 struct rpc_rqst *req = task->tk_rqstp;
1398 829
@@ -1404,124 +835,97 @@ xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
1404 req, ntohl(req->rq_xid)); 835 req, ntohl(req->rq_xid));
1405} 836}
1406 837
1407/* 838/**
1408 * Release an RPC call slot 839 * xprt_release - release an RPC request slot
840 * @task: task which is finished with the slot
841 *
1409 */ 842 */
1410void 843void xprt_release(struct rpc_task *task)
1411xprt_release(struct rpc_task *task)
1412{ 844{
1413 struct rpc_xprt *xprt = task->tk_xprt; 845 struct rpc_xprt *xprt = task->tk_xprt;
1414 struct rpc_rqst *req; 846 struct rpc_rqst *req;
1415 847
1416 if (!(req = task->tk_rqstp)) 848 if (!(req = task->tk_rqstp))
1417 return; 849 return;
1418 spin_lock_bh(&xprt->sock_lock); 850 spin_lock_bh(&xprt->transport_lock);
1419 __xprt_release_write(xprt, task); 851 xprt->ops->release_xprt(xprt, task);
1420 __xprt_put_cong(xprt, req); 852 if (xprt->ops->release_request)
853 xprt->ops->release_request(task);
1421 if (!list_empty(&req->rq_list)) 854 if (!list_empty(&req->rq_list))
1422 list_del(&req->rq_list); 855 list_del(&req->rq_list);
1423 xprt->last_used = jiffies; 856 xprt->last_used = jiffies;
1424 if (list_empty(&xprt->recv) && !xprt->shutdown) 857 if (list_empty(&xprt->recv) && !xprt->shutdown)
1425 mod_timer(&xprt->timer, xprt->last_used + XPRT_IDLE_TIMEOUT); 858 mod_timer(&xprt->timer,
1426 spin_unlock_bh(&xprt->sock_lock); 859 xprt->last_used + xprt->idle_timeout);
860 spin_unlock_bh(&xprt->transport_lock);
1427 task->tk_rqstp = NULL; 861 task->tk_rqstp = NULL;
1428 memset(req, 0, sizeof(*req)); /* mark unused */ 862 memset(req, 0, sizeof(*req)); /* mark unused */
1429 863
1430 dprintk("RPC: %4d release request %p\n", task->tk_pid, req); 864 dprintk("RPC: %4d release request %p\n", task->tk_pid, req);
1431 865
1432 spin_lock(&xprt->xprt_lock); 866 spin_lock(&xprt->reserve_lock);
1433 list_add(&req->rq_list, &xprt->free); 867 list_add(&req->rq_list, &xprt->free);
1434 xprt_clear_backlog(xprt); 868 rpc_wake_up_next(&xprt->backlog);
1435 spin_unlock(&xprt->xprt_lock); 869 spin_unlock(&xprt->reserve_lock);
1436}
1437
1438/*
1439 * Set default timeout parameters
1440 */
1441static void
1442xprt_default_timeout(struct rpc_timeout *to, int proto)
1443{
1444 if (proto == IPPROTO_UDP)
1445 xprt_set_timeout(to, 5, 5 * HZ);
1446 else
1447 xprt_set_timeout(to, 5, 60 * HZ);
1448} 870}
1449 871
1450/* 872/**
1451 * Set constant timeout 873 * xprt_set_timeout - set constant RPC timeout
874 * @to: RPC timeout parameters to set up
875 * @retr: number of retries
876 * @incr: amount of increase after each retry
877 *
1452 */ 878 */
1453void 879void xprt_set_timeout(struct rpc_timeout *to, unsigned int retr, unsigned long incr)
1454xprt_set_timeout(struct rpc_timeout *to, unsigned int retr, unsigned long incr)
1455{ 880{
1456 to->to_initval = 881 to->to_initval =
1457 to->to_increment = incr; 882 to->to_increment = incr;
1458 to->to_maxval = incr * retr; 883 to->to_maxval = to->to_initval + (incr * retr);
1459 to->to_retries = retr; 884 to->to_retries = retr;
1460 to->to_exponential = 0; 885 to->to_exponential = 0;
1461} 886}
1462 887
1463unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE; 888static struct rpc_xprt *xprt_setup(int proto, struct sockaddr_in *ap, struct rpc_timeout *to)
1464unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;
1465
1466/*
1467 * Initialize an RPC client
1468 */
1469static struct rpc_xprt *
1470xprt_setup(int proto, struct sockaddr_in *ap, struct rpc_timeout *to)
1471{ 889{
890 int result;
1472 struct rpc_xprt *xprt; 891 struct rpc_xprt *xprt;
1473 unsigned int entries;
1474 size_t slot_table_size;
1475 struct rpc_rqst *req; 892 struct rpc_rqst *req;
1476 893
1477 dprintk("RPC: setting up %s transport...\n",
1478 proto == IPPROTO_UDP? "UDP" : "TCP");
1479
1480 entries = (proto == IPPROTO_TCP)?
1481 xprt_tcp_slot_table_entries : xprt_udp_slot_table_entries;
1482
1483 if ((xprt = kmalloc(sizeof(struct rpc_xprt), GFP_KERNEL)) == NULL) 894 if ((xprt = kmalloc(sizeof(struct rpc_xprt), GFP_KERNEL)) == NULL)
1484 return ERR_PTR(-ENOMEM); 895 return ERR_PTR(-ENOMEM);
1485 memset(xprt, 0, sizeof(*xprt)); /* Nnnngh! */ 896 memset(xprt, 0, sizeof(*xprt)); /* Nnnngh! */
1486 xprt->max_reqs = entries;
1487 slot_table_size = entries * sizeof(xprt->slot[0]);
1488 xprt->slot = kmalloc(slot_table_size, GFP_KERNEL);
1489 if (xprt->slot == NULL) {
1490 kfree(xprt);
1491 return ERR_PTR(-ENOMEM);
1492 }
1493 memset(xprt->slot, 0, slot_table_size);
1494 897
1495 xprt->addr = *ap; 898 xprt->addr = *ap;
1496 xprt->prot = proto; 899
1497 xprt->stream = (proto == IPPROTO_TCP)? 1 : 0; 900 switch (proto) {
1498 if (xprt->stream) { 901 case IPPROTO_UDP:
1499 xprt->cwnd = RPC_MAXCWND(xprt); 902 result = xs_setup_udp(xprt, to);
1500 xprt->nocong = 1; 903 break;
1501 xprt->max_payload = (1U << 31) - 1; 904 case IPPROTO_TCP:
1502 } else { 905 result = xs_setup_tcp(xprt, to);
1503 xprt->cwnd = RPC_INITCWND; 906 break;
1504 xprt->max_payload = (1U << 16) - (MAX_HEADER << 3); 907 default:
908 printk(KERN_ERR "RPC: unrecognized transport protocol: %d\n",
909 proto);
910 result = -EIO;
911 break;
912 }
913 if (result) {
914 kfree(xprt);
915 return ERR_PTR(result);
1505 } 916 }
1506 spin_lock_init(&xprt->sock_lock); 917
1507 spin_lock_init(&xprt->xprt_lock); 918 spin_lock_init(&xprt->transport_lock);
1508 init_waitqueue_head(&xprt->cong_wait); 919 spin_lock_init(&xprt->reserve_lock);
1509 920
1510 INIT_LIST_HEAD(&xprt->free); 921 INIT_LIST_HEAD(&xprt->free);
1511 INIT_LIST_HEAD(&xprt->recv); 922 INIT_LIST_HEAD(&xprt->recv);
1512 INIT_WORK(&xprt->sock_connect, xprt_socket_connect, xprt); 923 INIT_WORK(&xprt->task_cleanup, xprt_autoclose, xprt);
1513 INIT_WORK(&xprt->task_cleanup, xprt_socket_autoclose, xprt);
1514 init_timer(&xprt->timer); 924 init_timer(&xprt->timer);
1515 xprt->timer.function = xprt_init_autodisconnect; 925 xprt->timer.function = xprt_init_autodisconnect;
1516 xprt->timer.data = (unsigned long) xprt; 926 xprt->timer.data = (unsigned long) xprt;
1517 xprt->last_used = jiffies; 927 xprt->last_used = jiffies;
1518 xprt->port = XPRT_MAX_RESVPORT; 928 xprt->cwnd = RPC_INITCWND;
1519
1520 /* Set timeout parameters */
1521 if (to) {
1522 xprt->timeout = *to;
1523 } else
1524 xprt_default_timeout(&xprt->timeout, xprt->prot);
1525 929
1526 rpc_init_wait_queue(&xprt->pending, "xprt_pending"); 930 rpc_init_wait_queue(&xprt->pending, "xprt_pending");
1527 rpc_init_wait_queue(&xprt->sending, "xprt_sending"); 931 rpc_init_wait_queue(&xprt->sending, "xprt_sending");
@@ -1529,139 +933,25 @@ xprt_setup(int proto, struct sockaddr_in *ap, struct rpc_timeout *to)
1529 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog"); 933 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog");
1530 934
1531 /* initialize free list */ 935 /* initialize free list */
1532 for (req = &xprt->slot[entries-1]; req >= &xprt->slot[0]; req--) 936 for (req = &xprt->slot[xprt->max_reqs-1]; req >= &xprt->slot[0]; req--)
1533 list_add(&req->rq_list, &xprt->free); 937 list_add(&req->rq_list, &xprt->free);
1534 938
1535 xprt_init_xid(xprt); 939 xprt_init_xid(xprt);
1536 940
1537 /* Check whether we want to use a reserved port */
1538 xprt->resvport = capable(CAP_NET_BIND_SERVICE) ? 1 : 0;
1539
1540 dprintk("RPC: created transport %p with %u slots\n", xprt, 941 dprintk("RPC: created transport %p with %u slots\n", xprt,
1541 xprt->max_reqs); 942 xprt->max_reqs);
1542 943
1543 return xprt; 944 return xprt;
1544} 945}
1545 946
1546/* 947/**
1547 * Bind to a reserved port 948 * xprt_create_proto - create an RPC client transport
1548 */ 949 * @proto: requested transport protocol
1549static inline int xprt_bindresvport(struct rpc_xprt *xprt, struct socket *sock) 950 * @sap: remote peer's address
1550{ 951 * @to: timeout parameters for new transport
1551 struct sockaddr_in myaddr = { 952 *
1552 .sin_family = AF_INET,
1553 };
1554 int err, port;
1555
1556 /* Were we already bound to a given port? Try to reuse it */
1557 port = xprt->port;
1558 do {
1559 myaddr.sin_port = htons(port);
1560 err = sock->ops->bind(sock, (struct sockaddr *) &myaddr,
1561 sizeof(myaddr));
1562 if (err == 0) {
1563 xprt->port = port;
1564 return 0;
1565 }
1566 if (--port == 0)
1567 port = XPRT_MAX_RESVPORT;
1568 } while (err == -EADDRINUSE && port != xprt->port);
1569
1570 printk("RPC: Can't bind to reserved port (%d).\n", -err);
1571 return err;
1572}
1573
1574static void
1575xprt_bind_socket(struct rpc_xprt *xprt, struct socket *sock)
1576{
1577 struct sock *sk = sock->sk;
1578
1579 if (xprt->inet)
1580 return;
1581
1582 write_lock_bh(&sk->sk_callback_lock);
1583 sk->sk_user_data = xprt;
1584 xprt->old_data_ready = sk->sk_data_ready;
1585 xprt->old_state_change = sk->sk_state_change;
1586 xprt->old_write_space = sk->sk_write_space;
1587 if (xprt->prot == IPPROTO_UDP) {
1588 sk->sk_data_ready = udp_data_ready;
1589 sk->sk_no_check = UDP_CSUM_NORCV;
1590 xprt_set_connected(xprt);
1591 } else {
1592 tcp_sk(sk)->nonagle = 1; /* disable Nagle's algorithm */
1593 sk->sk_data_ready = tcp_data_ready;
1594 sk->sk_state_change = tcp_state_change;
1595 xprt_clear_connected(xprt);
1596 }
1597 sk->sk_write_space = xprt_write_space;
1598
1599 /* Reset to new socket */
1600 xprt->sock = sock;
1601 xprt->inet = sk;
1602 write_unlock_bh(&sk->sk_callback_lock);
1603
1604 return;
1605}
1606
1607/*
1608 * Set socket buffer length
1609 */
1610void
1611xprt_sock_setbufsize(struct rpc_xprt *xprt)
1612{
1613 struct sock *sk = xprt->inet;
1614
1615 if (xprt->stream)
1616 return;
1617 if (xprt->rcvsize) {
1618 sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1619 sk->sk_rcvbuf = xprt->rcvsize * xprt->max_reqs * 2;
1620 }
1621 if (xprt->sndsize) {
1622 sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1623 sk->sk_sndbuf = xprt->sndsize * xprt->max_reqs * 2;
1624 sk->sk_write_space(sk);
1625 }
1626}
1627
1628/*
1629 * Datastream sockets are created here, but xprt_connect will create
1630 * and connect stream sockets.
1631 */
1632static struct socket * xprt_create_socket(struct rpc_xprt *xprt, int proto, int resvport)
1633{
1634 struct socket *sock;
1635 int type, err;
1636
1637 dprintk("RPC: xprt_create_socket(%s %d)\n",
1638 (proto == IPPROTO_UDP)? "udp" : "tcp", proto);
1639
1640 type = (proto == IPPROTO_UDP)? SOCK_DGRAM : SOCK_STREAM;
1641
1642 if ((err = sock_create_kern(PF_INET, type, proto, &sock)) < 0) {
1643 printk("RPC: can't create socket (%d).\n", -err);
1644 return NULL;
1645 }
1646
1647 /* If the caller has the capability, bind to a reserved port */
1648 if (resvport && xprt_bindresvport(xprt, sock) < 0) {
1649 printk("RPC: can't bind to reserved port.\n");
1650 goto failed;
1651 }
1652
1653 return sock;
1654
1655failed:
1656 sock_release(sock);
1657 return NULL;
1658}
1659
1660/*
1661 * Create an RPC client transport given the protocol and peer address.
1662 */ 953 */
1663struct rpc_xprt * 954struct rpc_xprt *xprt_create_proto(int proto, struct sockaddr_in *sap, struct rpc_timeout *to)
1664xprt_create_proto(int proto, struct sockaddr_in *sap, struct rpc_timeout *to)
1665{ 955{
1666 struct rpc_xprt *xprt; 956 struct rpc_xprt *xprt;
1667 957
@@ -1673,46 +963,26 @@ xprt_create_proto(int proto, struct sockaddr_in *sap, struct rpc_timeout *to)
1673 return xprt; 963 return xprt;
1674} 964}
1675 965
1676/* 966static void xprt_shutdown(struct rpc_xprt *xprt)
1677 * Prepare for transport shutdown.
1678 */
1679static void
1680xprt_shutdown(struct rpc_xprt *xprt)
1681{ 967{
1682 xprt->shutdown = 1; 968 xprt->shutdown = 1;
1683 rpc_wake_up(&xprt->sending); 969 rpc_wake_up(&xprt->sending);
1684 rpc_wake_up(&xprt->resend); 970 rpc_wake_up(&xprt->resend);
1685 rpc_wake_up(&xprt->pending); 971 xprt_wake_pending_tasks(xprt, -EIO);
1686 rpc_wake_up(&xprt->backlog); 972 rpc_wake_up(&xprt->backlog);
1687 wake_up(&xprt->cong_wait);
1688 del_timer_sync(&xprt->timer); 973 del_timer_sync(&xprt->timer);
1689
1690 /* synchronously wait for connect worker to finish */
1691 cancel_delayed_work(&xprt->sock_connect);
1692 flush_scheduled_work();
1693}
1694
1695/*
1696 * Clear the xprt backlog queue
1697 */
1698static int
1699xprt_clear_backlog(struct rpc_xprt *xprt) {
1700 rpc_wake_up_next(&xprt->backlog);
1701 wake_up(&xprt->cong_wait);
1702 return 1;
1703} 974}
1704 975
1705/* 976/**
1706 * Destroy an RPC transport, killing off all requests. 977 * xprt_destroy - destroy an RPC transport, killing off all requests.
978 * @xprt: transport to destroy
979 *
1707 */ 980 */
1708int 981int xprt_destroy(struct rpc_xprt *xprt)
1709xprt_destroy(struct rpc_xprt *xprt)
1710{ 982{
1711 dprintk("RPC: destroying transport %p\n", xprt); 983 dprintk("RPC: destroying transport %p\n", xprt);
1712 xprt_shutdown(xprt); 984 xprt_shutdown(xprt);
1713 xprt_disconnect(xprt); 985 xprt->ops->destroy(xprt);
1714 xprt_close(xprt);
1715 kfree(xprt->slot);
1716 kfree(xprt); 986 kfree(xprt);
1717 987
1718 return 0; 988 return 0;
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
new file mode 100644
index 000000000000..2e1529217e65
--- /dev/null
+++ b/net/sunrpc/xprtsock.c
@@ -0,0 +1,1252 @@
1/*
2 * linux/net/sunrpc/xprtsock.c
3 *
4 * Client-side transport implementation for sockets.
5 *
6 * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com>
7 * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com>
8 * TCP NFS related read + write fixes
9 * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
10 *
11 * Rewrite of larges part of the code in order to stabilize TCP stuff.
12 * Fix behaviour when socket buffer is full.
13 * (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
14 *
15 * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com>
16 */
17
18#include <linux/types.h>
19#include <linux/slab.h>
20#include <linux/capability.h>
21#include <linux/sched.h>
22#include <linux/pagemap.h>
23#include <linux/errno.h>
24#include <linux/socket.h>
25#include <linux/in.h>
26#include <linux/net.h>
27#include <linux/mm.h>
28#include <linux/udp.h>
29#include <linux/tcp.h>
30#include <linux/sunrpc/clnt.h>
31#include <linux/file.h>
32
33#include <net/sock.h>
34#include <net/checksum.h>
35#include <net/udp.h>
36#include <net/tcp.h>
37
38/*
39 * How many times to try sending a request on a socket before waiting
40 * for the socket buffer to clear.
41 */
42#define XS_SENDMSG_RETRY (10U)
43
44/*
45 * Time out for an RPC UDP socket connect. UDP socket connects are
46 * synchronous, but we set a timeout anyway in case of resource
47 * exhaustion on the local host.
48 */
49#define XS_UDP_CONN_TO (5U * HZ)
50
51/*
52 * Wait duration for an RPC TCP connection to be established. Solaris
53 * NFS over TCP uses 60 seconds, for example, which is in line with how
54 * long a server takes to reboot.
55 */
56#define XS_TCP_CONN_TO (60U * HZ)
57
58/*
59 * Wait duration for a reply from the RPC portmapper.
60 */
61#define XS_BIND_TO (60U * HZ)
62
63/*
64 * Delay if a UDP socket connect error occurs. This is most likely some
65 * kind of resource problem on the local host.
66 */
67#define XS_UDP_REEST_TO (2U * HZ)
68
69/*
70 * The reestablish timeout allows clients to delay for a bit before attempting
71 * to reconnect to a server that just dropped our connection.
72 *
73 * We implement an exponential backoff when trying to reestablish a TCP
74 * transport connection with the server. Some servers like to drop a TCP
75 * connection when they are overworked, so we start with a short timeout and
76 * increase over time if the server is down or not responding.
77 */
78#define XS_TCP_INIT_REEST_TO (3U * HZ)
79#define XS_TCP_MAX_REEST_TO (5U * 60 * HZ)
80
81/*
82 * TCP idle timeout; client drops the transport socket if it is idle
83 * for this long. Note that we also timeout UDP sockets to prevent
84 * holding port numbers when there is no RPC traffic.
85 */
86#define XS_IDLE_DISC_TO (5U * 60 * HZ)
87
88#ifdef RPC_DEBUG
89# undef RPC_DEBUG_DATA
90# define RPCDBG_FACILITY RPCDBG_TRANS
91#endif
92
93#ifdef RPC_DEBUG_DATA
94static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
95{
96 u8 *buf = (u8 *) packet;
97 int j;
98
99 dprintk("RPC: %s\n", msg);
100 for (j = 0; j < count && j < 128; j += 4) {
101 if (!(j & 31)) {
102 if (j)
103 dprintk("\n");
104 dprintk("0x%04x ", j);
105 }
106 dprintk("%02x%02x%02x%02x ",
107 buf[j], buf[j+1], buf[j+2], buf[j+3]);
108 }
109 dprintk("\n");
110}
111#else
112static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
113{
114 /* NOP */
115}
116#endif
117
118#define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL)
119
120static inline int xs_send_head(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base, unsigned int len)
121{
122 struct kvec iov = {
123 .iov_base = xdr->head[0].iov_base + base,
124 .iov_len = len - base,
125 };
126 struct msghdr msg = {
127 .msg_name = addr,
128 .msg_namelen = addrlen,
129 .msg_flags = XS_SENDMSG_FLAGS,
130 };
131
132 if (xdr->len > len)
133 msg.msg_flags |= MSG_MORE;
134
135 if (likely(iov.iov_len))
136 return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
137 return kernel_sendmsg(sock, &msg, NULL, 0, 0);
138}
139
140static int xs_send_tail(struct socket *sock, struct xdr_buf *xdr, unsigned int base, unsigned int len)
141{
142 struct kvec iov = {
143 .iov_base = xdr->tail[0].iov_base + base,
144 .iov_len = len - base,
145 };
146 struct msghdr msg = {
147 .msg_flags = XS_SENDMSG_FLAGS,
148 };
149
150 return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
151}
152
153/**
154 * xs_sendpages - write pages directly to a socket
155 * @sock: socket to send on
156 * @addr: UDP only -- address of destination
157 * @addrlen: UDP only -- length of destination address
158 * @xdr: buffer containing this request
159 * @base: starting position in the buffer
160 *
161 */
162static inline int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
163{
164 struct page **ppage = xdr->pages;
165 unsigned int len, pglen = xdr->page_len;
166 int err, ret = 0;
167 ssize_t (*sendpage)(struct socket *, struct page *, int, size_t, int);
168
169 if (unlikely(!sock))
170 return -ENOTCONN;
171
172 clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
173
174 len = xdr->head[0].iov_len;
175 if (base < len || (addr != NULL && base == 0)) {
176 err = xs_send_head(sock, addr, addrlen, xdr, base, len);
177 if (ret == 0)
178 ret = err;
179 else if (err > 0)
180 ret += err;
181 if (err != (len - base))
182 goto out;
183 base = 0;
184 } else
185 base -= len;
186
187 if (unlikely(pglen == 0))
188 goto copy_tail;
189 if (unlikely(base >= pglen)) {
190 base -= pglen;
191 goto copy_tail;
192 }
193 if (base || xdr->page_base) {
194 pglen -= base;
195 base += xdr->page_base;
196 ppage += base >> PAGE_CACHE_SHIFT;
197 base &= ~PAGE_CACHE_MASK;
198 }
199
200 sendpage = sock->ops->sendpage ? : sock_no_sendpage;
201 do {
202 int flags = XS_SENDMSG_FLAGS;
203
204 len = PAGE_CACHE_SIZE;
205 if (base)
206 len -= base;
207 if (pglen < len)
208 len = pglen;
209
210 if (pglen != len || xdr->tail[0].iov_len != 0)
211 flags |= MSG_MORE;
212
213 /* Hmm... We might be dealing with highmem pages */
214 if (PageHighMem(*ppage))
215 sendpage = sock_no_sendpage;
216 err = sendpage(sock, *ppage, base, len, flags);
217 if (ret == 0)
218 ret = err;
219 else if (err > 0)
220 ret += err;
221 if (err != len)
222 goto out;
223 base = 0;
224 ppage++;
225 } while ((pglen -= len) != 0);
226copy_tail:
227 len = xdr->tail[0].iov_len;
228 if (base < len) {
229 err = xs_send_tail(sock, xdr, base, len);
230 if (ret == 0)
231 ret = err;
232 else if (err > 0)
233 ret += err;
234 }
235out:
236 return ret;
237}
238
239/**
240 * xs_nospace - place task on wait queue if transmit was incomplete
241 * @task: task to put to sleep
242 *
243 */
244static void xs_nospace(struct rpc_task *task)
245{
246 struct rpc_rqst *req = task->tk_rqstp;
247 struct rpc_xprt *xprt = req->rq_xprt;
248
249 dprintk("RPC: %4d xmit incomplete (%u left of %u)\n",
250 task->tk_pid, req->rq_slen - req->rq_bytes_sent,
251 req->rq_slen);
252
253 if (test_bit(SOCK_ASYNC_NOSPACE, &xprt->sock->flags)) {
254 /* Protect against races with write_space */
255 spin_lock_bh(&xprt->transport_lock);
256
257 /* Don't race with disconnect */
258 if (!xprt_connected(xprt))
259 task->tk_status = -ENOTCONN;
260 else if (test_bit(SOCK_NOSPACE, &xprt->sock->flags))
261 xprt_wait_for_buffer_space(task);
262
263 spin_unlock_bh(&xprt->transport_lock);
264 } else
265 /* Keep holding the socket if it is blocked */
266 rpc_delay(task, HZ>>4);
267}
268
269/**
270 * xs_udp_send_request - write an RPC request to a UDP socket
271 * @task: address of RPC task that manages the state of an RPC request
272 *
273 * Return values:
274 * 0: The request has been sent
275 * EAGAIN: The socket was blocked, please call again later to
276 * complete the request
277 * ENOTCONN: Caller needs to invoke connect logic then call again
278 * other: Some other error occured, the request was not sent
279 */
280static int xs_udp_send_request(struct rpc_task *task)
281{
282 struct rpc_rqst *req = task->tk_rqstp;
283 struct rpc_xprt *xprt = req->rq_xprt;
284 struct xdr_buf *xdr = &req->rq_snd_buf;
285 int status;
286
287 xs_pktdump("packet data:",
288 req->rq_svec->iov_base,
289 req->rq_svec->iov_len);
290
291 req->rq_xtime = jiffies;
292 status = xs_sendpages(xprt->sock, (struct sockaddr *) &xprt->addr,
293 sizeof(xprt->addr), xdr, req->rq_bytes_sent);
294
295 dprintk("RPC: xs_udp_send_request(%u) = %d\n",
296 xdr->len - req->rq_bytes_sent, status);
297
298 if (likely(status >= (int) req->rq_slen))
299 return 0;
300
301 /* Still some bytes left; set up for a retry later. */
302 if (status > 0)
303 status = -EAGAIN;
304
305 switch (status) {
306 case -ENETUNREACH:
307 case -EPIPE:
308 case -ECONNREFUSED:
309 /* When the server has died, an ICMP port unreachable message
310 * prompts ECONNREFUSED. */
311 break;
312 case -EAGAIN:
313 xs_nospace(task);
314 break;
315 default:
316 dprintk("RPC: sendmsg returned unrecognized error %d\n",
317 -status);
318 break;
319 }
320
321 return status;
322}
323
324static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf)
325{
326 u32 reclen = buf->len - sizeof(rpc_fraghdr);
327 rpc_fraghdr *base = buf->head[0].iov_base;
328 *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen);
329}
330
331/**
332 * xs_tcp_send_request - write an RPC request to a TCP socket
333 * @task: address of RPC task that manages the state of an RPC request
334 *
335 * Return values:
336 * 0: The request has been sent
337 * EAGAIN: The socket was blocked, please call again later to
338 * complete the request
339 * ENOTCONN: Caller needs to invoke connect logic then call again
340 * other: Some other error occured, the request was not sent
341 *
342 * XXX: In the case of soft timeouts, should we eventually give up
343 * if sendmsg is not able to make progress?
344 */
345static int xs_tcp_send_request(struct rpc_task *task)
346{
347 struct rpc_rqst *req = task->tk_rqstp;
348 struct rpc_xprt *xprt = req->rq_xprt;
349 struct xdr_buf *xdr = &req->rq_snd_buf;
350 int status, retry = 0;
351
352 xs_encode_tcp_record_marker(&req->rq_snd_buf);
353
354 xs_pktdump("packet data:",
355 req->rq_svec->iov_base,
356 req->rq_svec->iov_len);
357
358 /* Continue transmitting the packet/record. We must be careful
359 * to cope with writespace callbacks arriving _after_ we have
360 * called sendmsg(). */
361 while (1) {
362 req->rq_xtime = jiffies;
363 status = xs_sendpages(xprt->sock, NULL, 0, xdr,
364 req->rq_bytes_sent);
365
366 dprintk("RPC: xs_tcp_send_request(%u) = %d\n",
367 xdr->len - req->rq_bytes_sent, status);
368
369 if (unlikely(status < 0))
370 break;
371
372 /* If we've sent the entire packet, immediately
373 * reset the count of bytes sent. */
374 req->rq_bytes_sent += status;
375 if (likely(req->rq_bytes_sent >= req->rq_slen)) {
376 req->rq_bytes_sent = 0;
377 return 0;
378 }
379
380 status = -EAGAIN;
381 if (retry++ > XS_SENDMSG_RETRY)
382 break;
383 }
384
385 switch (status) {
386 case -EAGAIN:
387 xs_nospace(task);
388 break;
389 case -ECONNREFUSED:
390 case -ECONNRESET:
391 case -ENOTCONN:
392 case -EPIPE:
393 status = -ENOTCONN;
394 break;
395 default:
396 dprintk("RPC: sendmsg returned unrecognized error %d\n",
397 -status);
398 xprt_disconnect(xprt);
399 break;
400 }
401
402 return status;
403}
404
405/**
406 * xs_close - close a socket
407 * @xprt: transport
408 *
409 * This is used when all requests are complete; ie, no DRC state remains
410 * on the server we want to save.
411 */
412static void xs_close(struct rpc_xprt *xprt)
413{
414 struct socket *sock = xprt->sock;
415 struct sock *sk = xprt->inet;
416
417 if (!sk)
418 return;
419
420 dprintk("RPC: xs_close xprt %p\n", xprt);
421
422 write_lock_bh(&sk->sk_callback_lock);
423 xprt->inet = NULL;
424 xprt->sock = NULL;
425
426 sk->sk_user_data = NULL;
427 sk->sk_data_ready = xprt->old_data_ready;
428 sk->sk_state_change = xprt->old_state_change;
429 sk->sk_write_space = xprt->old_write_space;
430 write_unlock_bh(&sk->sk_callback_lock);
431
432 sk->sk_no_check = 0;
433
434 sock_release(sock);
435}
436
437/**
438 * xs_destroy - prepare to shutdown a transport
439 * @xprt: doomed transport
440 *
441 */
442static void xs_destroy(struct rpc_xprt *xprt)
443{
444 dprintk("RPC: xs_destroy xprt %p\n", xprt);
445
446 cancel_delayed_work(&xprt->connect_worker);
447 flush_scheduled_work();
448
449 xprt_disconnect(xprt);
450 xs_close(xprt);
451 kfree(xprt->slot);
452}
453
454static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
455{
456 return (struct rpc_xprt *) sk->sk_user_data;
457}
458
459/**
460 * xs_udp_data_ready - "data ready" callback for UDP sockets
461 * @sk: socket with data to read
462 * @len: how much data to read
463 *
464 */
465static void xs_udp_data_ready(struct sock *sk, int len)
466{
467 struct rpc_task *task;
468 struct rpc_xprt *xprt;
469 struct rpc_rqst *rovr;
470 struct sk_buff *skb;
471 int err, repsize, copied;
472 u32 _xid, *xp;
473
474 read_lock(&sk->sk_callback_lock);
475 dprintk("RPC: xs_udp_data_ready...\n");
476 if (!(xprt = xprt_from_sock(sk)))
477 goto out;
478
479 if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
480 goto out;
481
482 if (xprt->shutdown)
483 goto dropit;
484
485 repsize = skb->len - sizeof(struct udphdr);
486 if (repsize < 4) {
487 dprintk("RPC: impossible RPC reply size %d!\n", repsize);
488 goto dropit;
489 }
490
491 /* Copy the XID from the skb... */
492 xp = skb_header_pointer(skb, sizeof(struct udphdr),
493 sizeof(_xid), &_xid);
494 if (xp == NULL)
495 goto dropit;
496
497 /* Look up and lock the request corresponding to the given XID */
498 spin_lock(&xprt->transport_lock);
499 rovr = xprt_lookup_rqst(xprt, *xp);
500 if (!rovr)
501 goto out_unlock;
502 task = rovr->rq_task;
503
504 if ((copied = rovr->rq_private_buf.buflen) > repsize)
505 copied = repsize;
506
507 /* Suck it into the iovec, verify checksum if not done by hw. */
508 if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb))
509 goto out_unlock;
510
511 /* Something worked... */
512 dst_confirm(skb->dst);
513
514 xprt_adjust_cwnd(task, copied);
515 xprt_update_rtt(task);
516 xprt_complete_rqst(task, copied);
517
518 out_unlock:
519 spin_unlock(&xprt->transport_lock);
520 dropit:
521 skb_free_datagram(sk, skb);
522 out:
523 read_unlock(&sk->sk_callback_lock);
524}
525
526static inline size_t xs_tcp_copy_data(skb_reader_t *desc, void *p, size_t len)
527{
528 if (len > desc->count)
529 len = desc->count;
530 if (skb_copy_bits(desc->skb, desc->offset, p, len)) {
531 dprintk("RPC: failed to copy %zu bytes from skb. %zu bytes remain\n",
532 len, desc->count);
533 return 0;
534 }
535 desc->offset += len;
536 desc->count -= len;
537 dprintk("RPC: copied %zu bytes from skb. %zu bytes remain\n",
538 len, desc->count);
539 return len;
540}
541
542static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc)
543{
544 size_t len, used;
545 char *p;
546
547 p = ((char *) &xprt->tcp_recm) + xprt->tcp_offset;
548 len = sizeof(xprt->tcp_recm) - xprt->tcp_offset;
549 used = xs_tcp_copy_data(desc, p, len);
550 xprt->tcp_offset += used;
551 if (used != len)
552 return;
553
554 xprt->tcp_reclen = ntohl(xprt->tcp_recm);
555 if (xprt->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
556 xprt->tcp_flags |= XPRT_LAST_FRAG;
557 else
558 xprt->tcp_flags &= ~XPRT_LAST_FRAG;
559 xprt->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
560
561 xprt->tcp_flags &= ~XPRT_COPY_RECM;
562 xprt->tcp_offset = 0;
563
564 /* Sanity check of the record length */
565 if (unlikely(xprt->tcp_reclen < 4)) {
566 dprintk("RPC: invalid TCP record fragment length\n");
567 xprt_disconnect(xprt);
568 return;
569 }
570 dprintk("RPC: reading TCP record fragment of length %d\n",
571 xprt->tcp_reclen);
572}
573
574static void xs_tcp_check_recm(struct rpc_xprt *xprt)
575{
576 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u, tcp_flags = %lx\n",
577 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen, xprt->tcp_flags);
578 if (xprt->tcp_offset == xprt->tcp_reclen) {
579 xprt->tcp_flags |= XPRT_COPY_RECM;
580 xprt->tcp_offset = 0;
581 if (xprt->tcp_flags & XPRT_LAST_FRAG) {
582 xprt->tcp_flags &= ~XPRT_COPY_DATA;
583 xprt->tcp_flags |= XPRT_COPY_XID;
584 xprt->tcp_copied = 0;
585 }
586 }
587}
588
589static inline void xs_tcp_read_xid(struct rpc_xprt *xprt, skb_reader_t *desc)
590{
591 size_t len, used;
592 char *p;
593
594 len = sizeof(xprt->tcp_xid) - xprt->tcp_offset;
595 dprintk("RPC: reading XID (%Zu bytes)\n", len);
596 p = ((char *) &xprt->tcp_xid) + xprt->tcp_offset;
597 used = xs_tcp_copy_data(desc, p, len);
598 xprt->tcp_offset += used;
599 if (used != len)
600 return;
601 xprt->tcp_flags &= ~XPRT_COPY_XID;
602 xprt->tcp_flags |= XPRT_COPY_DATA;
603 xprt->tcp_copied = 4;
604 dprintk("RPC: reading reply for XID %08x\n",
605 ntohl(xprt->tcp_xid));
606 xs_tcp_check_recm(xprt);
607}
608
609static inline void xs_tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc)
610{
611 struct rpc_rqst *req;
612 struct xdr_buf *rcvbuf;
613 size_t len;
614 ssize_t r;
615
616 /* Find and lock the request corresponding to this xid */
617 spin_lock(&xprt->transport_lock);
618 req = xprt_lookup_rqst(xprt, xprt->tcp_xid);
619 if (!req) {
620 xprt->tcp_flags &= ~XPRT_COPY_DATA;
621 dprintk("RPC: XID %08x request not found!\n",
622 ntohl(xprt->tcp_xid));
623 spin_unlock(&xprt->transport_lock);
624 return;
625 }
626
627 rcvbuf = &req->rq_private_buf;
628 len = desc->count;
629 if (len > xprt->tcp_reclen - xprt->tcp_offset) {
630 skb_reader_t my_desc;
631
632 len = xprt->tcp_reclen - xprt->tcp_offset;
633 memcpy(&my_desc, desc, sizeof(my_desc));
634 my_desc.count = len;
635 r = xdr_partial_copy_from_skb(rcvbuf, xprt->tcp_copied,
636 &my_desc, xs_tcp_copy_data);
637 desc->count -= r;
638 desc->offset += r;
639 } else
640 r = xdr_partial_copy_from_skb(rcvbuf, xprt->tcp_copied,
641 desc, xs_tcp_copy_data);
642
643 if (r > 0) {
644 xprt->tcp_copied += r;
645 xprt->tcp_offset += r;
646 }
647 if (r != len) {
648 /* Error when copying to the receive buffer,
649 * usually because we weren't able to allocate
650 * additional buffer pages. All we can do now
651 * is turn off XPRT_COPY_DATA, so the request
652 * will not receive any additional updates,
653 * and time out.
654 * Any remaining data from this record will
655 * be discarded.
656 */
657 xprt->tcp_flags &= ~XPRT_COPY_DATA;
658 dprintk("RPC: XID %08x truncated request\n",
659 ntohl(xprt->tcp_xid));
660 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
661 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen);
662 goto out;
663 }
664
665 dprintk("RPC: XID %08x read %Zd bytes\n",
666 ntohl(xprt->tcp_xid), r);
667 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
668 xprt, xprt->tcp_copied, xprt->tcp_offset, xprt->tcp_reclen);
669
670 if (xprt->tcp_copied == req->rq_private_buf.buflen)
671 xprt->tcp_flags &= ~XPRT_COPY_DATA;
672 else if (xprt->tcp_offset == xprt->tcp_reclen) {
673 if (xprt->tcp_flags & XPRT_LAST_FRAG)
674 xprt->tcp_flags &= ~XPRT_COPY_DATA;
675 }
676
677out:
678 if (!(xprt->tcp_flags & XPRT_COPY_DATA))
679 xprt_complete_rqst(req->rq_task, xprt->tcp_copied);
680 spin_unlock(&xprt->transport_lock);
681 xs_tcp_check_recm(xprt);
682}
683
684static inline void xs_tcp_read_discard(struct rpc_xprt *xprt, skb_reader_t *desc)
685{
686 size_t len;
687
688 len = xprt->tcp_reclen - xprt->tcp_offset;
689 if (len > desc->count)
690 len = desc->count;
691 desc->count -= len;
692 desc->offset += len;
693 xprt->tcp_offset += len;
694 dprintk("RPC: discarded %Zu bytes\n", len);
695 xs_tcp_check_recm(xprt);
696}
697
698static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
699{
700 struct rpc_xprt *xprt = rd_desc->arg.data;
701 skb_reader_t desc = {
702 .skb = skb,
703 .offset = offset,
704 .count = len,
705 .csum = 0
706 };
707
708 dprintk("RPC: xs_tcp_data_recv started\n");
709 do {
710 /* Read in a new fragment marker if necessary */
711 /* Can we ever really expect to get completely empty fragments? */
712 if (xprt->tcp_flags & XPRT_COPY_RECM) {
713 xs_tcp_read_fraghdr(xprt, &desc);
714 continue;
715 }
716 /* Read in the xid if necessary */
717 if (xprt->tcp_flags & XPRT_COPY_XID) {
718 xs_tcp_read_xid(xprt, &desc);
719 continue;
720 }
721 /* Read in the request data */
722 if (xprt->tcp_flags & XPRT_COPY_DATA) {
723 xs_tcp_read_request(xprt, &desc);
724 continue;
725 }
726 /* Skip over any trailing bytes on short reads */
727 xs_tcp_read_discard(xprt, &desc);
728 } while (desc.count);
729 dprintk("RPC: xs_tcp_data_recv done\n");
730 return len - desc.count;
731}
732
733/**
734 * xs_tcp_data_ready - "data ready" callback for TCP sockets
735 * @sk: socket with data to read
736 * @bytes: how much data to read
737 *
738 */
739static void xs_tcp_data_ready(struct sock *sk, int bytes)
740{
741 struct rpc_xprt *xprt;
742 read_descriptor_t rd_desc;
743
744 read_lock(&sk->sk_callback_lock);
745 dprintk("RPC: xs_tcp_data_ready...\n");
746 if (!(xprt = xprt_from_sock(sk)))
747 goto out;
748 if (xprt->shutdown)
749 goto out;
750
751 /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
752 rd_desc.arg.data = xprt;
753 rd_desc.count = 65536;
754 tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
755out:
756 read_unlock(&sk->sk_callback_lock);
757}
758
759/**
760 * xs_tcp_state_change - callback to handle TCP socket state changes
761 * @sk: socket whose state has changed
762 *
763 */
764static void xs_tcp_state_change(struct sock *sk)
765{
766 struct rpc_xprt *xprt;
767
768 read_lock(&sk->sk_callback_lock);
769 if (!(xprt = xprt_from_sock(sk)))
770 goto out;
771 dprintk("RPC: xs_tcp_state_change client %p...\n", xprt);
772 dprintk("RPC: state %x conn %d dead %d zapped %d\n",
773 sk->sk_state, xprt_connected(xprt),
774 sock_flag(sk, SOCK_DEAD),
775 sock_flag(sk, SOCK_ZAPPED));
776
777 switch (sk->sk_state) {
778 case TCP_ESTABLISHED:
779 spin_lock_bh(&xprt->transport_lock);
780 if (!xprt_test_and_set_connected(xprt)) {
781 /* Reset TCP record info */
782 xprt->tcp_offset = 0;
783 xprt->tcp_reclen = 0;
784 xprt->tcp_copied = 0;
785 xprt->tcp_flags = XPRT_COPY_RECM | XPRT_COPY_XID;
786 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
787 xprt_wake_pending_tasks(xprt, 0);
788 }
789 spin_unlock_bh(&xprt->transport_lock);
790 break;
791 case TCP_SYN_SENT:
792 case TCP_SYN_RECV:
793 break;
794 default:
795 xprt_disconnect(xprt);
796 break;
797 }
798 out:
799 read_unlock(&sk->sk_callback_lock);
800}
801
802/**
803 * xs_udp_write_space - callback invoked when socket buffer space
804 * becomes available
805 * @sk: socket whose state has changed
806 *
807 * Called when more output buffer space is available for this socket.
808 * We try not to wake our writers until they can make "significant"
809 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
810 * with a bunch of small requests.
811 */
812static void xs_udp_write_space(struct sock *sk)
813{
814 read_lock(&sk->sk_callback_lock);
815
816 /* from net/core/sock.c:sock_def_write_space */
817 if (sock_writeable(sk)) {
818 struct socket *sock;
819 struct rpc_xprt *xprt;
820
821 if (unlikely(!(sock = sk->sk_socket)))
822 goto out;
823 if (unlikely(!(xprt = xprt_from_sock(sk))))
824 goto out;
825 if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
826 goto out;
827
828 xprt_write_space(xprt);
829 }
830
831 out:
832 read_unlock(&sk->sk_callback_lock);
833}
834
835/**
836 * xs_tcp_write_space - callback invoked when socket buffer space
837 * becomes available
838 * @sk: socket whose state has changed
839 *
840 * Called when more output buffer space is available for this socket.
841 * We try not to wake our writers until they can make "significant"
842 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
843 * with a bunch of small requests.
844 */
845static void xs_tcp_write_space(struct sock *sk)
846{
847 read_lock(&sk->sk_callback_lock);
848
849 /* from net/core/stream.c:sk_stream_write_space */
850 if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
851 struct socket *sock;
852 struct rpc_xprt *xprt;
853
854 if (unlikely(!(sock = sk->sk_socket)))
855 goto out;
856 if (unlikely(!(xprt = xprt_from_sock(sk))))
857 goto out;
858 if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
859 goto out;
860
861 xprt_write_space(xprt);
862 }
863
864 out:
865 read_unlock(&sk->sk_callback_lock);
866}
867
868static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
869{
870 struct sock *sk = xprt->inet;
871
872 if (xprt->rcvsize) {
873 sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
874 sk->sk_rcvbuf = xprt->rcvsize * xprt->max_reqs * 2;
875 }
876 if (xprt->sndsize) {
877 sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
878 sk->sk_sndbuf = xprt->sndsize * xprt->max_reqs * 2;
879 sk->sk_write_space(sk);
880 }
881}
882
883/**
884 * xs_udp_set_buffer_size - set send and receive limits
885 * @xprt: generic transport
886 * @sndsize: requested size of send buffer, in bytes
887 * @rcvsize: requested size of receive buffer, in bytes
888 *
889 * Set socket send and receive buffer size limits.
890 */
891static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
892{
893 xprt->sndsize = 0;
894 if (sndsize)
895 xprt->sndsize = sndsize + 1024;
896 xprt->rcvsize = 0;
897 if (rcvsize)
898 xprt->rcvsize = rcvsize + 1024;
899
900 xs_udp_do_set_buffer_size(xprt);
901}
902
903/**
904 * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
905 * @task: task that timed out
906 *
907 * Adjust the congestion window after a retransmit timeout has occurred.
908 */
909static void xs_udp_timer(struct rpc_task *task)
910{
911 xprt_adjust_cwnd(task, -ETIMEDOUT);
912}
913
914static int xs_bindresvport(struct rpc_xprt *xprt, struct socket *sock)
915{
916 struct sockaddr_in myaddr = {
917 .sin_family = AF_INET,
918 };
919 int err;
920 unsigned short port = xprt->port;
921
922 do {
923 myaddr.sin_port = htons(port);
924 err = sock->ops->bind(sock, (struct sockaddr *) &myaddr,
925 sizeof(myaddr));
926 if (err == 0) {
927 xprt->port = port;
928 dprintk("RPC: xs_bindresvport bound to port %u\n",
929 port);
930 return 0;
931 }
932 if (port <= xprt_min_resvport)
933 port = xprt_max_resvport;
934 else
935 port--;
936 } while (err == -EADDRINUSE && port != xprt->port);
937
938 dprintk("RPC: can't bind to reserved port (%d).\n", -err);
939 return err;
940}
941
942/**
943 * xs_udp_connect_worker - set up a UDP socket
944 * @args: RPC transport to connect
945 *
946 * Invoked by a work queue tasklet.
947 */
948static void xs_udp_connect_worker(void *args)
949{
950 struct rpc_xprt *xprt = (struct rpc_xprt *) args;
951 struct socket *sock = xprt->sock;
952 int err, status = -EIO;
953
954 if (xprt->shutdown || xprt->addr.sin_port == 0)
955 goto out;
956
957 dprintk("RPC: xs_udp_connect_worker for xprt %p\n", xprt);
958
959 /* Start by resetting any existing state */
960 xs_close(xprt);
961
962 if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
963 dprintk("RPC: can't create UDP transport socket (%d).\n", -err);
964 goto out;
965 }
966
967 if (xprt->resvport && xs_bindresvport(xprt, sock) < 0) {
968 sock_release(sock);
969 goto out;
970 }
971
972 if (!xprt->inet) {
973 struct sock *sk = sock->sk;
974
975 write_lock_bh(&sk->sk_callback_lock);
976
977 sk->sk_user_data = xprt;
978 xprt->old_data_ready = sk->sk_data_ready;
979 xprt->old_state_change = sk->sk_state_change;
980 xprt->old_write_space = sk->sk_write_space;
981 sk->sk_data_ready = xs_udp_data_ready;
982 sk->sk_write_space = xs_udp_write_space;
983 sk->sk_no_check = UDP_CSUM_NORCV;
984
985 xprt_set_connected(xprt);
986
987 /* Reset to new socket */
988 xprt->sock = sock;
989 xprt->inet = sk;
990
991 write_unlock_bh(&sk->sk_callback_lock);
992 }
993 xs_udp_do_set_buffer_size(xprt);
994 status = 0;
995out:
996 xprt_wake_pending_tasks(xprt, status);
997 xprt_clear_connecting(xprt);
998}
999
1000/*
1001 * We need to preserve the port number so the reply cache on the server can
1002 * find our cached RPC replies when we get around to reconnecting.
1003 */
1004static void xs_tcp_reuse_connection(struct rpc_xprt *xprt)
1005{
1006 int result;
1007 struct socket *sock = xprt->sock;
1008 struct sockaddr any;
1009
1010 dprintk("RPC: disconnecting xprt %p to reuse port\n", xprt);
1011
1012 /*
1013 * Disconnect the transport socket by doing a connect operation
1014 * with AF_UNSPEC. This should return immediately...
1015 */
1016 memset(&any, 0, sizeof(any));
1017 any.sa_family = AF_UNSPEC;
1018 result = sock->ops->connect(sock, &any, sizeof(any), 0);
1019 if (result)
1020 dprintk("RPC: AF_UNSPEC connect return code %d\n",
1021 result);
1022}
1023
1024/**
1025 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint
1026 * @args: RPC transport to connect
1027 *
1028 * Invoked by a work queue tasklet.
1029 */
1030static void xs_tcp_connect_worker(void *args)
1031{
1032 struct rpc_xprt *xprt = (struct rpc_xprt *)args;
1033 struct socket *sock = xprt->sock;
1034 int err, status = -EIO;
1035
1036 if (xprt->shutdown || xprt->addr.sin_port == 0)
1037 goto out;
1038
1039 dprintk("RPC: xs_tcp_connect_worker for xprt %p\n", xprt);
1040
1041 if (!xprt->sock) {
1042 /* start from scratch */
1043 if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
1044 dprintk("RPC: can't create TCP transport socket (%d).\n", -err);
1045 goto out;
1046 }
1047
1048 if (xprt->resvport && xs_bindresvport(xprt, sock) < 0) {
1049 sock_release(sock);
1050 goto out;
1051 }
1052 } else
1053 /* "close" the socket, preserving the local port */
1054 xs_tcp_reuse_connection(xprt);
1055
1056 if (!xprt->inet) {
1057 struct sock *sk = sock->sk;
1058
1059 write_lock_bh(&sk->sk_callback_lock);
1060
1061 sk->sk_user_data = xprt;
1062 xprt->old_data_ready = sk->sk_data_ready;
1063 xprt->old_state_change = sk->sk_state_change;
1064 xprt->old_write_space = sk->sk_write_space;
1065 sk->sk_data_ready = xs_tcp_data_ready;
1066 sk->sk_state_change = xs_tcp_state_change;
1067 sk->sk_write_space = xs_tcp_write_space;
1068
1069 /* socket options */
1070 sk->sk_userlocks |= SOCK_BINDPORT_LOCK;
1071 sock_reset_flag(sk, SOCK_LINGER);
1072 tcp_sk(sk)->linger2 = 0;
1073 tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
1074
1075 xprt_clear_connected(xprt);
1076
1077 /* Reset to new socket */
1078 xprt->sock = sock;
1079 xprt->inet = sk;
1080
1081 write_unlock_bh(&sk->sk_callback_lock);
1082 }
1083
1084 /* Tell the socket layer to start connecting... */
1085 status = sock->ops->connect(sock, (struct sockaddr *) &xprt->addr,
1086 sizeof(xprt->addr), O_NONBLOCK);
1087 dprintk("RPC: %p connect status %d connected %d sock state %d\n",
1088 xprt, -status, xprt_connected(xprt), sock->sk->sk_state);
1089 if (status < 0) {
1090 switch (status) {
1091 case -EINPROGRESS:
1092 case -EALREADY:
1093 goto out_clear;
1094 case -ECONNREFUSED:
1095 case -ECONNRESET:
1096 /* retry with existing socket, after a delay */
1097 break;
1098 default:
1099 /* get rid of existing socket, and retry */
1100 xs_close(xprt);
1101 break;
1102 }
1103 }
1104out:
1105 xprt_wake_pending_tasks(xprt, status);
1106out_clear:
1107 xprt_clear_connecting(xprt);
1108}
1109
1110/**
1111 * xs_connect - connect a socket to a remote endpoint
1112 * @task: address of RPC task that manages state of connect request
1113 *
1114 * TCP: If the remote end dropped the connection, delay reconnecting.
1115 *
1116 * UDP socket connects are synchronous, but we use a work queue anyway
1117 * to guarantee that even unprivileged user processes can set up a
1118 * socket on a privileged port.
1119 *
1120 * If a UDP socket connect fails, the delay behavior here prevents
1121 * retry floods (hard mounts).
1122 */
1123static void xs_connect(struct rpc_task *task)
1124{
1125 struct rpc_xprt *xprt = task->tk_xprt;
1126
1127 if (xprt_test_and_set_connecting(xprt))
1128 return;
1129
1130 if (xprt->sock != NULL) {
1131 dprintk("RPC: xs_connect delayed xprt %p for %lu seconds\n",
1132 xprt, xprt->reestablish_timeout / HZ);
1133 schedule_delayed_work(&xprt->connect_worker,
1134 xprt->reestablish_timeout);
1135 xprt->reestablish_timeout <<= 1;
1136 if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
1137 xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
1138 } else {
1139 dprintk("RPC: xs_connect scheduled xprt %p\n", xprt);
1140 schedule_work(&xprt->connect_worker);
1141
1142 /* flush_scheduled_work can sleep... */
1143 if (!RPC_IS_ASYNC(task))
1144 flush_scheduled_work();
1145 }
1146}
1147
1148static struct rpc_xprt_ops xs_udp_ops = {
1149 .set_buffer_size = xs_udp_set_buffer_size,
1150 .reserve_xprt = xprt_reserve_xprt_cong,
1151 .release_xprt = xprt_release_xprt_cong,
1152 .connect = xs_connect,
1153 .send_request = xs_udp_send_request,
1154 .set_retrans_timeout = xprt_set_retrans_timeout_rtt,
1155 .timer = xs_udp_timer,
1156 .release_request = xprt_release_rqst_cong,
1157 .close = xs_close,
1158 .destroy = xs_destroy,
1159};
1160
1161static struct rpc_xprt_ops xs_tcp_ops = {
1162 .reserve_xprt = xprt_reserve_xprt,
1163 .release_xprt = xprt_release_xprt,
1164 .connect = xs_connect,
1165 .send_request = xs_tcp_send_request,
1166 .set_retrans_timeout = xprt_set_retrans_timeout_def,
1167 .close = xs_close,
1168 .destroy = xs_destroy,
1169};
1170
1171/**
1172 * xs_setup_udp - Set up transport to use a UDP socket
1173 * @xprt: transport to set up
1174 * @to: timeout parameters
1175 *
1176 */
1177int xs_setup_udp(struct rpc_xprt *xprt, struct rpc_timeout *to)
1178{
1179 size_t slot_table_size;
1180
1181 dprintk("RPC: setting up udp-ipv4 transport...\n");
1182
1183 xprt->max_reqs = xprt_udp_slot_table_entries;
1184 slot_table_size = xprt->max_reqs * sizeof(xprt->slot[0]);
1185 xprt->slot = kmalloc(slot_table_size, GFP_KERNEL);
1186 if (xprt->slot == NULL)
1187 return -ENOMEM;
1188 memset(xprt->slot, 0, slot_table_size);
1189
1190 xprt->prot = IPPROTO_UDP;
1191 xprt->port = xprt_max_resvport;
1192 xprt->tsh_size = 0;
1193 xprt->resvport = capable(CAP_NET_BIND_SERVICE) ? 1 : 0;
1194 /* XXX: header size can vary due to auth type, IPv6, etc. */
1195 xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
1196
1197 INIT_WORK(&xprt->connect_worker, xs_udp_connect_worker, xprt);
1198 xprt->bind_timeout = XS_BIND_TO;
1199 xprt->connect_timeout = XS_UDP_CONN_TO;
1200 xprt->reestablish_timeout = XS_UDP_REEST_TO;
1201 xprt->idle_timeout = XS_IDLE_DISC_TO;
1202
1203 xprt->ops = &xs_udp_ops;
1204
1205 if (to)
1206 xprt->timeout = *to;
1207 else
1208 xprt_set_timeout(&xprt->timeout, 5, 5 * HZ);
1209
1210 return 0;
1211}
1212
1213/**
1214 * xs_setup_tcp - Set up transport to use a TCP socket
1215 * @xprt: transport to set up
1216 * @to: timeout parameters
1217 *
1218 */
1219int xs_setup_tcp(struct rpc_xprt *xprt, struct rpc_timeout *to)
1220{
1221 size_t slot_table_size;
1222
1223 dprintk("RPC: setting up tcp-ipv4 transport...\n");
1224
1225 xprt->max_reqs = xprt_tcp_slot_table_entries;
1226 slot_table_size = xprt->max_reqs * sizeof(xprt->slot[0]);
1227 xprt->slot = kmalloc(slot_table_size, GFP_KERNEL);
1228 if (xprt->slot == NULL)
1229 return -ENOMEM;
1230 memset(xprt->slot, 0, slot_table_size);
1231
1232 xprt->prot = IPPROTO_TCP;
1233 xprt->port = xprt_max_resvport;
1234 xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
1235 xprt->resvport = capable(CAP_NET_BIND_SERVICE) ? 1 : 0;
1236 xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
1237
1238 INIT_WORK(&xprt->connect_worker, xs_tcp_connect_worker, xprt);
1239 xprt->bind_timeout = XS_BIND_TO;
1240 xprt->connect_timeout = XS_TCP_CONN_TO;
1241 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
1242 xprt->idle_timeout = XS_IDLE_DISC_TO;
1243
1244 xprt->ops = &xs_tcp_ops;
1245
1246 if (to)
1247 xprt->timeout = *to;
1248 else
1249 xprt_set_timeout(&xprt->timeout, 2, 60 * HZ);
1250
1251 return 0;
1252}
generated by cgit v1.2.2 (git 2.25.0) at 2025-02-02 16:01:28 -0500