projects / deliverable / linux.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
Merge branch 'x86-urgent-for-linus' of git://git.kernel.org/pub/scm/linux/kernel...
[deliverable/linux.git] / net / sunrpc / xprtsock.c
CommitLineData
a246b010
CL		 1/*
2 * linux/net/sunrpc/xprtsock.c
3 *
4 * Client-side transport implementation for sockets.
5 *
113aa838
AC		 6 * TCP callback races fixes (C) 1998 Red Hat
7 * TCP send fixes (C) 1998 Red Hat
a246b010
CL		 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>
55aa4f58
CL		 14 *
15 * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com>
8f9d5b1a
CL		 16 *
17 * IPv6 support contributed by Gilles Quillard, Bull Open Source, 2005.
18 * <gilles.quillard@bull.net>
a246b010
CL		 19 */
20
21#include <linux/types.h>
176e21ee 22#include <linux/string.h>
a246b010 23#include <linux/slab.h>
bc25571e 24#include <linux/module.h>
a246b010 25#include <linux/capability.h>
a246b010
CL		 26#include <linux/pagemap.h>
27#include <linux/errno.h>
28#include <linux/socket.h>
29#include <linux/in.h>
30#include <linux/net.h>
31#include <linux/mm.h>
176e21ee 32#include <linux/un.h>
a246b010
CL		 33#include <linux/udp.h>
34#include <linux/tcp.h>
35#include <linux/sunrpc/clnt.h>
5976687a 36#include <linux/sunrpc/addr.h>
02107148 37#include <linux/sunrpc/sched.h>
4cfc7e60 38#include <linux/sunrpc/svcsock.h>
49c36fcc 39#include <linux/sunrpc/xprtsock.h>
a246b010 40#include <linux/file.h>
9e00abc3 41#ifdef CONFIG_SUNRPC_BACKCHANNEL
44b98efd
RL		 42#include <linux/sunrpc/bc_xprt.h>
43#endif
a246b010
CL		 44
45#include <net/sock.h>
46#include <net/checksum.h>
47#include <net/udp.h>
48#include <net/tcp.h>
49
40b5ea0c
TM		 50#include <trace/events/sunrpc.h>
51
4cfc7e60 52#include "sunrpc.h"
176e21ee
CL		 53
54static void xs_close(struct rpc_xprt *xprt);
55
c556b754
CL		 56/*
57 * xprtsock tunables
58 */
09acfea5
TM		 59static unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
60static unsigned int xprt_tcp_slot_table_entries = RPC_MIN_SLOT_TABLE;
61static unsigned int xprt_max_tcp_slot_table_entries = RPC_MAX_SLOT_TABLE;
c556b754 62
09acfea5
TM		 63static unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
64static unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
c556b754 65
7d1e8255 66#define XS_TCP_LINGER_TO (15U * HZ)
25fe6142 67static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
7d1e8255 68
fbf76683
CL		 69/*
70 * We can register our own files under /proc/sys/sunrpc by
71 * calling register_sysctl_table() again. The files in that
72 * directory become the union of all files registered there.
73 *
74 * We simply need to make sure that we don't collide with
75 * someone else's file names!
76 */
77
78#ifdef RPC_DEBUG
79
80static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
81static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
d9ba131d 82static unsigned int max_tcp_slot_table_limit = RPC_MAX_SLOT_TABLE_LIMIT;
fbf76683
CL		 83static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
84static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
85
86static struct ctl_table_header *sunrpc_table_header;
87
88/*
89 * FIXME: changing the UDP slot table size should also resize the UDP
90 * socket buffers for existing UDP transports
91 */
fe2c6338 92static struct ctl_table xs_tunables_table[] = {
fbf76683 93 {
fbf76683
CL		 94 .procname = "udp_slot_table_entries",
95 .data = &xprt_udp_slot_table_entries,
96 .maxlen = sizeof(unsigned int),
97 .mode = 0644,
6d456111 98 .proc_handler = proc_dointvec_minmax,
fbf76683
CL		 99 .extra1 = &min_slot_table_size,
100 .extra2 = &max_slot_table_size
101 },
102 {
fbf76683
CL		 103 .procname = "tcp_slot_table_entries",
104 .data = &xprt_tcp_slot_table_entries,
105 .maxlen = sizeof(unsigned int),
106 .mode = 0644,
6d456111 107 .proc_handler = proc_dointvec_minmax,
fbf76683
CL		 108 .extra1 = &min_slot_table_size,
109 .extra2 = &max_slot_table_size
110 },
d9ba131d
TM		 111 {
112 .procname = "tcp_max_slot_table_entries",
113 .data = &xprt_max_tcp_slot_table_entries,
114 .maxlen = sizeof(unsigned int),
115 .mode = 0644,
116 .proc_handler = proc_dointvec_minmax,
117 .extra1 = &min_slot_table_size,
118 .extra2 = &max_tcp_slot_table_limit
119 },
fbf76683 120 {
fbf76683
CL		 121 .procname = "min_resvport",
122 .data = &xprt_min_resvport,
123 .maxlen = sizeof(unsigned int),
124 .mode = 0644,
6d456111 125 .proc_handler = proc_dointvec_minmax,
fbf76683
CL		 126 .extra1 = &xprt_min_resvport_limit,
127 .extra2 = &xprt_max_resvport_limit
128 },
129 {
fbf76683
CL		 130 .procname = "max_resvport",
131 .data = &xprt_max_resvport,
132 .maxlen = sizeof(unsigned int),
133 .mode = 0644,
6d456111 134 .proc_handler = proc_dointvec_minmax,
fbf76683
CL		 135 .extra1 = &xprt_min_resvport_limit,
136 .extra2 = &xprt_max_resvport_limit
137 },
25fe6142
TM		 138 {
139 .procname = "tcp_fin_timeout",
140 .data = &xs_tcp_fin_timeout,
141 .maxlen = sizeof(xs_tcp_fin_timeout),
142 .mode = 0644,
6d456111 143 .proc_handler = proc_dointvec_jiffies,
fbf76683 144 },
f8572d8f 145 { },
fbf76683
CL		 146};
147
fe2c6338 148static struct ctl_table sunrpc_table[] = {
fbf76683 149 {
fbf76683
CL		 150 .procname = "sunrpc",
151 .mode = 0555,
152 .child = xs_tunables_table
153 },
f8572d8f 154 { },
fbf76683
CL		 155};
156
157#endif
158
03bf4b70
CL		 159/*
160 * Wait duration for a reply from the RPC portmapper.
161 */
162#define XS_BIND_TO (60U * HZ)
163
164/*
165 * Delay if a UDP socket connect error occurs. This is most likely some
166 * kind of resource problem on the local host.
167 */
168#define XS_UDP_REEST_TO (2U * HZ)
169
170/*
171 * The reestablish timeout allows clients to delay for a bit before attempting
172 * to reconnect to a server that just dropped our connection.
173 *
174 * We implement an exponential backoff when trying to reestablish a TCP
175 * transport connection with the server. Some servers like to drop a TCP
176 * connection when they are overworked, so we start with a short timeout and
177 * increase over time if the server is down or not responding.
178 */
179#define XS_TCP_INIT_REEST_TO (3U * HZ)
180#define XS_TCP_MAX_REEST_TO (5U * 60 * HZ)
181
182/*
183 * TCP idle timeout; client drops the transport socket if it is idle
184 * for this long. Note that we also timeout UDP sockets to prevent
185 * holding port numbers when there is no RPC traffic.
186 */
187#define XS_IDLE_DISC_TO (5U * 60 * HZ)
188
a246b010
CL		 189#ifdef RPC_DEBUG
190# undef RPC_DEBUG_DATA
9903cd1c 191# define RPCDBG_FACILITY RPCDBG_TRANS
a246b010
CL		 192#endif
193
a246b010 194#ifdef RPC_DEBUG_DATA
9903cd1c 195static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
a246b010 196{
9903cd1c
CL		 197 u8 *buf = (u8 *) packet;
198 int j;
a246b010 199
46121cf7 200 dprintk("RPC: %s\n", msg);
a246b010
CL		 201 for (j = 0; j < count && j < 128; j += 4) {
202 if (!(j & 31)) {
203 if (j)
204 dprintk("\n");
205 dprintk("0x%04x ", j);
206 }
207 dprintk("%02x%02x%02x%02x ",
208 buf[j], buf[j+1], buf[j+2], buf[j+3]);
209 }
210 dprintk("\n");
211}
212#else
9903cd1c 213static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
a246b010
CL		 214{
215 /* NOP */
216}
217#endif
218
ffc2e518
CL		 219struct sock_xprt {
220 struct rpc_xprt xprt;
ee0ac0c2
CL		 221
222 /*
223 * Network layer
224 */
225 struct socket * sock;
226 struct sock * inet;
51971139
CL		 227
228 /*
229 * State of TCP reply receive
230 */
231 __be32 tcp_fraghdr,
b76ce561
TM		 232 tcp_xid,
233 tcp_calldir;
51971139
CL		 234
235 u32 tcp_offset,
236 tcp_reclen;
237
238 unsigned long tcp_copied,
239 tcp_flags;
c8475461
CL		 240
241 /*
242 * Connection of transports
243 */
34161db6 244 struct delayed_work connect_worker;
fbfffbd5
CL		 245 struct sockaddr_storage srcaddr;
246 unsigned short srcport;
7c6e066e
CL		 247
248 /*
249 * UDP socket buffer size parameters
250 */
251 size_t rcvsize,
252 sndsize;
314dfd79
CL		 253
254 /*
255 * Saved socket callback addresses
256 */
676d2369 257 void (*old_data_ready)(struct sock *);
314dfd79
CL		 258 void (*old_state_change)(struct sock *);
259 void (*old_write_space)(struct sock *);
2118071d 260 void (*old_error_report)(struct sock *);
ffc2e518
CL		 261};
262
e136d092
CL		 263/*
264 * TCP receive state flags
265 */
266#define TCP_RCV_LAST_FRAG (1UL << 0)
267#define TCP_RCV_COPY_FRAGHDR (1UL << 1)
268#define TCP_RCV_COPY_XID (1UL << 2)
269#define TCP_RCV_COPY_DATA (1UL << 3)
f4a2e418
RL		 270#define TCP_RCV_READ_CALLDIR (1UL << 4)
271#define TCP_RCV_COPY_CALLDIR (1UL << 5)
18dca02a
RL		 272
273/*
274 * TCP RPC flags
275 */
f4a2e418 276#define TCP_RPC_REPLY (1UL << 6)
e136d092 277
2118071d
TM		 278static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
279{
280 return (struct rpc_xprt *) sk->sk_user_data;
281}
282
95392c59
CL		 283static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
284{
285 return (struct sockaddr *) &xprt->addr;
286}
287
176e21ee
CL		 288static inline struct sockaddr_un *xs_addr_un(struct rpc_xprt *xprt)
289{
290 return (struct sockaddr_un *) &xprt->addr;
291}
292
95392c59 293static inline struct sockaddr_in *xs_addr_in(struct rpc_xprt *xprt)
edb267a6 294{
95392c59
CL		 295 return (struct sockaddr_in *) &xprt->addr;
296}
297
298static inline struct sockaddr_in6 *xs_addr_in6(struct rpc_xprt *xprt)
299{
300 return (struct sockaddr_in6 *) &xprt->addr;
301}
302
c877b849 303static void xs_format_common_peer_addresses(struct rpc_xprt *xprt)
edb267a6 304{
c877b849 305 struct sockaddr *sap = xs_addr(xprt);
9dc3b095
CL		 306 struct sockaddr_in6 *sin6;
307 struct sockaddr_in *sin;
176e21ee 308 struct sockaddr_un *sun;
c877b849 309 char buf[128];
edb267a6 310
9dc3b095 311 switch (sap->sa_family) {
176e21ee
CL		 312 case AF_LOCAL:
313 sun = xs_addr_un(xprt);
314 strlcpy(buf, sun->sun_path, sizeof(buf));
315 xprt->address_strings[RPC_DISPLAY_ADDR] =
316 kstrdup(buf, GFP_KERNEL);
317 break;
9dc3b095 318 case AF_INET:
176e21ee
CL		 319 (void)rpc_ntop(sap, buf, sizeof(buf));
320 xprt->address_strings[RPC_DISPLAY_ADDR] =
321 kstrdup(buf, GFP_KERNEL);
9dc3b095 322 sin = xs_addr_in(xprt);
fc0b5791 323 snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
9dc3b095
CL		 324 break;
325 case AF_INET6:
176e21ee
CL		 326 (void)rpc_ntop(sap, buf, sizeof(buf));
327 xprt->address_strings[RPC_DISPLAY_ADDR] =
328 kstrdup(buf, GFP_KERNEL);
9dc3b095 329 sin6 = xs_addr_in6(xprt);
fc0b5791 330 snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
9dc3b095
CL		 331 break;
332 default:
333 BUG();
756805e7 334 }
176e21ee 335
9dc3b095 336 xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
edb267a6
CL		 337}
338
9dc3b095 339static void xs_format_common_peer_ports(struct rpc_xprt *xprt)
4b6473fb 340{
9dc3b095
CL		 341 struct sockaddr *sap = xs_addr(xprt);
342 char buf[128];
4b6473fb 343
81160e66 344 snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
c877b849 345 xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
4b6473fb 346
81160e66 347 snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
c877b849
CL		 348 xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
349}
4b6473fb 350
9dc3b095
CL		 351static void xs_format_peer_addresses(struct rpc_xprt *xprt,
352 const char *protocol,
353 const char *netid)
c877b849 354{
b454ae90 355 xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
b454ae90 356 xprt->address_strings[RPC_DISPLAY_NETID] = netid;
c877b849 357 xs_format_common_peer_addresses(xprt);
9dc3b095 358 xs_format_common_peer_ports(xprt);
edb267a6 359}
4b6473fb 360
9dc3b095 361static void xs_update_peer_port(struct rpc_xprt *xprt)
4b6473fb 362{
9dc3b095
CL		 363 kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]);
364 kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
4417c8c4 365
9dc3b095 366 xs_format_common_peer_ports(xprt);
edb267a6
CL		 367}
368
369static void xs_free_peer_addresses(struct rpc_xprt *xprt)
370{
33e01dc7
CL		 371 unsigned int i;
372
373 for (i = 0; i < RPC_DISPLAY_MAX; i++)
374 switch (i) {
375 case RPC_DISPLAY_PROTO:
376 case RPC_DISPLAY_NETID:
377 continue;
378 default:
379 kfree(xprt->address_strings[i]);
380 }
edb267a6
CL		 381}
382
b4b5cc85
CL		 383#define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL)
384
24c5684b 385static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
b4b5cc85 386{
b4b5cc85
CL		 387 struct msghdr msg = {
388 .msg_name = addr,
389 .msg_namelen = addrlen,
24c5684b
TM		 390 .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
391 };
392 struct kvec iov = {
393 .iov_base = vec->iov_base + base,
394 .iov_len = vec->iov_len - base,
b4b5cc85
CL		 395 };
396
24c5684b 397 if (iov.iov_len != 0)
b4b5cc85
CL		 398 return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
399 return kernel_sendmsg(sock, &msg, NULL, 0, 0);
400}
401
a6b31d18 402static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more, bool zerocopy)
b4b5cc85 403{
a6b31d18
TM		 404 ssize_t (*do_sendpage)(struct socket *sock, struct page *page,
405 int offset, size_t size, int flags);
24c5684b
TM		 406 struct page **ppage;
407 unsigned int remainder;
408 int err, sent = 0;
409
410 remainder = xdr->page_len - base;
411 base += xdr->page_base;
412 ppage = xdr->pages + (base >> PAGE_SHIFT);
413 base &= ~PAGE_MASK;
a6b31d18
TM		 414 do_sendpage = sock->ops->sendpage;
415 if (!zerocopy)
416 do_sendpage = sock_no_sendpage;
24c5684b
TM		 417 for(;;) {
418 unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
419 int flags = XS_SENDMSG_FLAGS;
b4b5cc85 420
24c5684b
TM		 421 remainder -= len;
422 if (remainder != 0 || more)
423 flags |= MSG_MORE;
a6b31d18 424 err = do_sendpage(sock, *ppage, base, len, flags);
24c5684b
TM		 425 if (remainder == 0 || err != len)
426 break;
427 sent += err;
428 ppage++;
429 base = 0;
430 }
431 if (sent == 0)
432 return err;
433 if (err > 0)
434 sent += err;
435 return sent;
b4b5cc85
CL		 436}
437
9903cd1c
CL		 438/**
439 * xs_sendpages - write pages directly to a socket
440 * @sock: socket to send on
441 * @addr: UDP only -- address of destination
442 * @addrlen: UDP only -- length of destination address
443 * @xdr: buffer containing this request
444 * @base: starting position in the buffer
a6b31d18 445 * @zerocopy: true if it is safe to use sendpage()
9903cd1c 446 *
a246b010 447 */
a6b31d18 448static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base, bool zerocopy)
a246b010 449{
24c5684b
TM		 450 unsigned int remainder = xdr->len - base;
451 int err, sent = 0;
a246b010 452
262965f5 453 if (unlikely(!sock))
fba91afb 454 return -ENOTSOCK;
262965f5
CL		 455
456 clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
24c5684b
TM		 457 if (base != 0) {
458 addr = NULL;
459 addrlen = 0;
460 }
262965f5 461
24c5684b
TM		 462 if (base < xdr->head[0].iov_len || addr != NULL) {
463 unsigned int len = xdr->head[0].iov_len - base;
464 remainder -= len;
465 err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
466 if (remainder == 0 || err != len)
a246b010 467 goto out;
24c5684b 468 sent += err;
a246b010
CL		 469 base = 0;
470 } else
24c5684b 471 base -= xdr->head[0].iov_len;
a246b010 472
24c5684b
TM		 473 if (base < xdr->page_len) {
474 unsigned int len = xdr->page_len - base;
475 remainder -= len;
a6b31d18 476 err = xs_send_pagedata(sock, xdr, base, remainder != 0, zerocopy);
24c5684b 477 if (remainder == 0 || err != len)
a246b010 478 goto out;
24c5684b 479 sent += err;
a246b010 480 base = 0;
24c5684b
TM		 481 } else
482 base -= xdr->page_len;
483
484 if (base >= xdr->tail[0].iov_len)
485 return sent;
486 err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
a246b010 487out:
24c5684b
TM		 488 if (sent == 0)
489 return err;
490 if (err > 0)
491 sent += err;
492 return sent;
a246b010
CL		 493}
494
b6ddf64f
TM		 495static void xs_nospace_callback(struct rpc_task *task)
496{
497 struct sock_xprt *transport = container_of(task->tk_rqstp->rq_xprt, struct sock_xprt, xprt);
498
499 transport->inet->sk_write_pending--;
500 clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
501}
502
9903cd1c 503/**
262965f5
CL		 504 * xs_nospace - place task on wait queue if transmit was incomplete
505 * @task: task to put to sleep
9903cd1c 506 *
a246b010 507 */
5e3771ce 508static int xs_nospace(struct rpc_task *task)
a246b010 509{
262965f5
CL		 510 struct rpc_rqst *req = task->tk_rqstp;
511 struct rpc_xprt *xprt = req->rq_xprt;
ee0ac0c2 512 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
06ea0bfe 513 struct sock *sk = transport->inet;
24ca9a84 514 int ret = -EAGAIN;
a246b010 515
46121cf7 516 dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
262965f5
CL		 517 task->tk_pid, req->rq_slen - req->rq_bytes_sent,
518 req->rq_slen);
519
b6ddf64f
TM		 520 /* Protect against races with write_space */
521 spin_lock_bh(&xprt->transport_lock);
522
523 /* Don't race with disconnect */
524 if (xprt_connected(xprt)) {
525 if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
526 /*
527 * Notify TCP that we're limited by the application
528 * window size
529 */
530 set_bit(SOCK_NOSPACE, &transport->sock->flags);
06ea0bfe 531 sk->sk_write_pending++;
b6ddf64f
TM		 532 /* ...and wait for more buffer space */
533 xprt_wait_for_buffer_space(task, xs_nospace_callback);
534 }
535 } else {
536 clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
5e3771ce 537 ret = -ENOTCONN;
b6ddf64f 538 }
262965f5 539
b6ddf64f 540 spin_unlock_bh(&xprt->transport_lock);
06ea0bfe
TM		 541
542 /* Race breaker in case memory is freed before above code is called */
543 sk->sk_write_space(sk);
5e3771ce 544 return ret;
262965f5
CL		 545}
546
61677eee
CL		 547/*
548 * Construct a stream transport record marker in @buf.
549 */
550static inline void xs_encode_stream_record_marker(struct xdr_buf *buf)
551{
552 u32 reclen = buf->len - sizeof(rpc_fraghdr);
553 rpc_fraghdr *base = buf->head[0].iov_base;
554 *base = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT | reclen);
555}
556
176e21ee
CL		 557/**
558 * xs_local_send_request - write an RPC request to an AF_LOCAL socket
559 * @task: RPC task that manages the state of an RPC request
560 *
561 * Return values:
562 * 0: The request has been sent
563 * EAGAIN: The socket was blocked, please call again later to
564 * complete the request
565 * ENOTCONN: Caller needs to invoke connect logic then call again
566 * other: Some other error occured, the request was not sent
567 */
568static int xs_local_send_request(struct rpc_task *task)
569{
570 struct rpc_rqst *req = task->tk_rqstp;
571 struct rpc_xprt *xprt = req->rq_xprt;
572 struct sock_xprt *transport =
573 container_of(xprt, struct sock_xprt, xprt);
574 struct xdr_buf *xdr = &req->rq_snd_buf;
575 int status;
576
577 xs_encode_stream_record_marker(&req->rq_snd_buf);
578
579 xs_pktdump("packet data:",
580 req->rq_svec->iov_base, req->rq_svec->iov_len);
581
582 status = xs_sendpages(transport->sock, NULL, 0,
a6b31d18 583 xdr, req->rq_bytes_sent, true);
176e21ee
CL		 584 dprintk("RPC: %s(%u) = %d\n",
585 __func__, xdr->len - req->rq_bytes_sent, status);
586 if (likely(status >= 0)) {
587 req->rq_bytes_sent += status;
588 req->rq_xmit_bytes_sent += status;
589 if (likely(req->rq_bytes_sent >= req->rq_slen)) {
590 req->rq_bytes_sent = 0;
591 return 0;
592 }
593 status = -EAGAIN;
594 }
595
596 switch (status) {
597 case -EAGAIN:
598 status = xs_nospace(task);
599 break;
600 default:
601 dprintk("RPC: sendmsg returned unrecognized error %d\n",
602 -status);
603 case -EPIPE:
604 xs_close(xprt);
605 status = -ENOTCONN;
606 }
607
608 return status;
609}
610
262965f5
CL		 611/**
612 * xs_udp_send_request - write an RPC request to a UDP socket
613 * @task: address of RPC task that manages the state of an RPC request
614 *
615 * Return values:
616 * 0: The request has been sent
617 * EAGAIN: The socket was blocked, please call again later to
618 * complete the request
619 * ENOTCONN: Caller needs to invoke connect logic then call again
25985edc 620 * other: Some other error occurred, the request was not sent
262965f5
CL		 621 */
622static int xs_udp_send_request(struct rpc_task *task)
623{
624 struct rpc_rqst *req = task->tk_rqstp;
625 struct rpc_xprt *xprt = req->rq_xprt;
ee0ac0c2 626 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
262965f5
CL		 627 struct xdr_buf *xdr = &req->rq_snd_buf;
628 int status;
a246b010 629
9903cd1c 630 xs_pktdump("packet data:",
a246b010
CL		 631 req->rq_svec->iov_base,
632 req->rq_svec->iov_len);
633
01d37c42
TM		 634 if (!xprt_bound(xprt))
635 return -ENOTCONN;
ee0ac0c2 636 status = xs_sendpages(transport->sock,
95392c59 637 xs_addr(xprt),
ee0ac0c2 638 xprt->addrlen, xdr,
a6b31d18 639 req->rq_bytes_sent, true);
a246b010 640
46121cf7 641 dprintk("RPC: xs_udp_send_request(%u) = %d\n",
262965f5 642 xdr->len - req->rq_bytes_sent, status);
a246b010 643
2199700f 644 if (status >= 0) {
d60dbb20 645 req->rq_xmit_bytes_sent += status;
2199700f
TM		 646 if (status >= req->rq_slen)
647 return 0;
648 /* Still some bytes left; set up for a retry later. */
262965f5 649 status = -EAGAIN;
2199700f 650 }
a246b010 651
262965f5 652 switch (status) {
fba91afb
TM		 653 case -ENOTSOCK:
654 status = -ENOTCONN;
655 /* Should we call xs_close() here? */
656 break;
b6ddf64f 657 case -EAGAIN:
5e3771ce 658 status = xs_nospace(task);
b6ddf64f 659 break;
c8485e4d
TM		 660 default:
661 dprintk("RPC: sendmsg returned unrecognized error %d\n",
662 -status);
262965f5
CL		 663 case -ENETUNREACH:
664 case -EPIPE:
a246b010
CL		 665 case -ECONNREFUSED:
666 /* When the server has died, an ICMP port unreachable message
9903cd1c 667 * prompts ECONNREFUSED. */
b6ddf64f 668 clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
a246b010 669 }
5fe46e9d 670
262965f5 671 return status;
a246b010
CL		 672}
673
e06799f9
TM		 674/**
675 * xs_tcp_shutdown - gracefully shut down a TCP socket
676 * @xprt: transport
677 *
678 * Initiates a graceful shutdown of the TCP socket by calling the
679 * equivalent of shutdown(SHUT_WR);
680 */
681static void xs_tcp_shutdown(struct rpc_xprt *xprt)
682{
683 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
684 struct socket *sock = transport->sock;
685
40b5ea0c 686 if (sock != NULL) {
e06799f9 687 kernel_sock_shutdown(sock, SHUT_WR);
40b5ea0c
TM		 688 trace_rpc_socket_shutdown(xprt, sock);
689 }
e06799f9
TM		 690}
691
9903cd1c 692/**
262965f5 693 * xs_tcp_send_request - write an RPC request to a TCP socket
9903cd1c
CL		 694 * @task: address of RPC task that manages the state of an RPC request
695 *
696 * Return values:
262965f5
CL		 697 * 0: The request has been sent
698 * EAGAIN: The socket was blocked, please call again later to
699 * complete the request
700 * ENOTCONN: Caller needs to invoke connect logic then call again
25985edc 701 * other: Some other error occurred, the request was not sent
9903cd1c
CL		 702 *
703 * XXX: In the case of soft timeouts, should we eventually give up
262965f5 704 * if sendmsg is not able to make progress?
9903cd1c 705 */
262965f5 706static int xs_tcp_send_request(struct rpc_task *task)
a246b010
CL		 707{
708 struct rpc_rqst *req = task->tk_rqstp;
709 struct rpc_xprt *xprt = req->rq_xprt;
ee0ac0c2 710 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
262965f5 711 struct xdr_buf *xdr = &req->rq_snd_buf;
a6b31d18 712 bool zerocopy = true;
b595bb15 713 int status;
a246b010 714
61677eee 715 xs_encode_stream_record_marker(&req->rq_snd_buf);
a246b010 716
262965f5
CL		 717 xs_pktdump("packet data:",
718 req->rq_svec->iov_base,
719 req->rq_svec->iov_len);
a6b31d18
TM		 720 /* Don't use zero copy if this is a resend. If the RPC call
721 * completes while the socket holds a reference to the pages,
722 * then we may end up resending corrupted data.
723 */
724 if (task->tk_flags & RPC_TASK_SENT)
725 zerocopy = false;
a246b010
CL		 726
727 /* Continue transmitting the packet/record. We must be careful
728 * to cope with writespace callbacks arriving _after_ we have
262965f5 729 * called sendmsg(). */
a246b010 730 while (1) {
ee0ac0c2 731 status = xs_sendpages(transport->sock,
a6b31d18
TM		 732 NULL, 0, xdr, req->rq_bytes_sent,
733 zerocopy);
a246b010 734
46121cf7 735 dprintk("RPC: xs_tcp_send_request(%u) = %d\n",
262965f5 736 xdr->len - req->rq_bytes_sent, status);
a246b010 737
262965f5 738 if (unlikely(status < 0))
a246b010 739 break;
a246b010 740
262965f5
CL		 741 /* If we've sent the entire packet, immediately
742 * reset the count of bytes sent. */
743 req->rq_bytes_sent += status;
d60dbb20 744 req->rq_xmit_bytes_sent += status;
262965f5
CL		 745 if (likely(req->rq_bytes_sent >= req->rq_slen)) {
746 req->rq_bytes_sent = 0;
747 return 0;
748 }
a246b010 749
06b4b681
TM		 750 if (status != 0)
751 continue;
a246b010 752 status = -EAGAIN;
06b4b681 753 break;
a246b010
CL		 754 }
755
262965f5 756 switch (status) {
fba91afb
TM		 757 case -ENOTSOCK:
758 status = -ENOTCONN;
759 /* Should we call xs_close() here? */
760 break;
262965f5 761 case -EAGAIN:
5e3771ce 762 status = xs_nospace(task);
262965f5 763 break;
c8485e4d
TM		 764 default:
765 dprintk("RPC: sendmsg returned unrecognized error %d\n",
766 -status);
262965f5 767 case -ECONNRESET:
2a9e1cfa
TM		 768 xs_tcp_shutdown(xprt);
769 case -ECONNREFUSED:
262965f5 770 case -ENOTCONN:
b9d2bb2e 771 case -EPIPE:
b6ddf64f 772 clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
a246b010 773 }
5fe46e9d 774
a246b010
CL		 775 return status;
776}
777
e0ab53de
TM		 778/**
779 * xs_tcp_release_xprt - clean up after a tcp transmission
780 * @xprt: transport
781 * @task: rpc task
782 *
783 * This cleans up if an error causes us to abort the transmission of a request.
784 * In this case, the socket may need to be reset in order to avoid confusing
785 * the server.
786 */
787static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
788{
789 struct rpc_rqst *req;
790
791 if (task != xprt->snd_task)
792 return;
793 if (task == NULL)
794 goto out_release;
795 req = task->tk_rqstp;
43cedbf0
TM		 796 if (req == NULL)
797 goto out_release;
e0ab53de
TM		 798 if (req->rq_bytes_sent == 0)
799 goto out_release;
800 if (req->rq_bytes_sent == req->rq_snd_buf.len)
801 goto out_release;
a4f0835c 802 set_bit(XPRT_CLOSE_WAIT, &xprt->state);
e0ab53de
TM		 803out_release:
804 xprt_release_xprt(xprt, task);
805}
806
2a9e1cfa
TM		 807static void xs_save_old_callbacks(struct sock_xprt *transport, struct sock *sk)
808{
809 transport->old_data_ready = sk->sk_data_ready;
810 transport->old_state_change = sk->sk_state_change;
811 transport->old_write_space = sk->sk_write_space;
2118071d 812 transport->old_error_report = sk->sk_error_report;
2a9e1cfa
TM		 813}
814
815static void xs_restore_old_callbacks(struct sock_xprt *transport, struct sock *sk)
816{
817 sk->sk_data_ready = transport->old_data_ready;
818 sk->sk_state_change = transport->old_state_change;
819 sk->sk_write_space = transport->old_write_space;
2118071d
TM		 820 sk->sk_error_report = transport->old_error_report;
821}
822
823/**
824 * xs_error_report - callback to handle TCP socket state errors
825 * @sk: socket
826 *
827 * Note: we don't call sock_error() since there may be a rpc_task
828 * using the socket, and so we don't want to clear sk->sk_err.
829 */
830static void xs_error_report(struct sock *sk)
831{
832 struct rpc_xprt *xprt;
833 int err;
834
835 read_lock_bh(&sk->sk_callback_lock);
836 if (!(xprt = xprt_from_sock(sk)))
837 goto out;
838
839 err = -sk->sk_err;
840 if (err == 0)
841 goto out;
842 dprintk("RPC: xs_error_report client %p, error=%d...\n",
843 xprt, -err);
e8353c76 844 trace_rpc_socket_error(xprt, sk->sk_socket, err);
2118071d
TM		 845 xprt_wake_pending_tasks(xprt, err);
846 out:
847 read_unlock_bh(&sk->sk_callback_lock);
2a9e1cfa
TM		 848}
849
fe315e76 850static void xs_reset_transport(struct sock_xprt *transport)
a246b010 851{
ee0ac0c2
CL		 852 struct socket *sock = transport->sock;
853 struct sock *sk = transport->inet;
a246b010 854
fe315e76
CL		 855 if (sk == NULL)
856 return;
9903cd1c 857
246408dc
TM		 858 transport->srcport = 0;
859
a246b010 860 write_lock_bh(&sk->sk_callback_lock);
ee0ac0c2
CL		 861 transport->inet = NULL;
862 transport->sock = NULL;
a246b010 863
9903cd1c 864 sk->sk_user_data = NULL;
2a9e1cfa
TM		 865
866 xs_restore_old_callbacks(transport, sk);
a246b010
CL		 867 write_unlock_bh(&sk->sk_callback_lock);
868
40b5ea0c 869 trace_rpc_socket_close(&transport->xprt, sock);
a246b010 870 sock_release(sock);
fe315e76
CL		 871}
872
873/**
874 * xs_close - close a socket
875 * @xprt: transport
876 *
877 * This is used when all requests are complete; ie, no DRC state remains
878 * on the server we want to save.
f75e6745
TM		 879 *
880 * The caller _must_ be holding XPRT_LOCKED in order to avoid issues with
881 * xs_reset_transport() zeroing the socket from underneath a writer.
fe315e76
CL		 882 */
883static void xs_close(struct rpc_xprt *xprt)
884{
885 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
886
887 dprintk("RPC: xs_close xprt %p\n", xprt);
888
93dc41bd
N		 889 cancel_delayed_work_sync(&transport->connect_worker);
890
fe315e76 891 xs_reset_transport(transport);
61d0a8e6 892 xprt->reestablish_timeout = 0;
fe315e76 893
4e857c58 894 smp_mb__before_atomic();
7d1e8255 895 clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
632e3bdc 896 clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
3b948ae5 897 clear_bit(XPRT_CLOSING, &xprt->state);
4e857c58 898 smp_mb__after_atomic();
62da3b24 899 xprt_disconnect_done(xprt);
a246b010
CL		 900}
901
f75e6745
TM		 902static void xs_tcp_close(struct rpc_xprt *xprt)
903{
904 if (test_and_clear_bit(XPRT_CONNECTION_CLOSE, &xprt->state))
905 xs_close(xprt);
906 else
907 xs_tcp_shutdown(xprt);
908}
909
315f3812
KM		 910static void xs_xprt_free(struct rpc_xprt *xprt)
911{
912 xs_free_peer_addresses(xprt);
913 xprt_free(xprt);
914}
915
9903cd1c
CL		 916/**
917 * xs_destroy - prepare to shutdown a transport
918 * @xprt: doomed transport
919 *
920 */
921static void xs_destroy(struct rpc_xprt *xprt)
a246b010 922{
46121cf7 923 dprintk("RPC: xs_destroy xprt %p\n", xprt);
9903cd1c 924
a1311d87 925 xs_close(xprt);
315f3812 926 xs_xprt_free(xprt);
a1311d87 927 module_put(THIS_MODULE);
a246b010
CL		 928}
929
176e21ee
CL		 930static int xs_local_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb)
931{
932 struct xdr_skb_reader desc = {
933 .skb = skb,
934 .offset = sizeof(rpc_fraghdr),
935 .count = skb->len - sizeof(rpc_fraghdr),
936 };
937
938 if (xdr_partial_copy_from_skb(xdr, 0, &desc, xdr_skb_read_bits) < 0)
939 return -1;
940 if (desc.count)
941 return -1;
942 return 0;
943}
944
945/**
946 * xs_local_data_ready - "data ready" callback for AF_LOCAL sockets
947 * @sk: socket with data to read
948 * @len: how much data to read
949 *
950 * Currently this assumes we can read the whole reply in a single gulp.
951 */
676d2369 952static void xs_local_data_ready(struct sock *sk)
176e21ee
CL		 953{
954 struct rpc_task *task;
955 struct rpc_xprt *xprt;
956 struct rpc_rqst *rovr;
957 struct sk_buff *skb;
958 int err, repsize, copied;
959 u32 _xid;
960 __be32 *xp;
961
962 read_lock_bh(&sk->sk_callback_lock);
963 dprintk("RPC: %s...\n", __func__);
964 xprt = xprt_from_sock(sk);
965 if (xprt == NULL)
966 goto out;
967
968 skb = skb_recv_datagram(sk, 0, 1, &err);
969 if (skb == NULL)
970 goto out;
971
176e21ee
CL		 972 repsize = skb->len - sizeof(rpc_fraghdr);
973 if (repsize < 4) {
974 dprintk("RPC: impossible RPC reply size %d\n", repsize);
975 goto dropit;
976 }
977
978 /* Copy the XID from the skb... */
979 xp = skb_header_pointer(skb, sizeof(rpc_fraghdr), sizeof(_xid), &_xid);
980 if (xp == NULL)
981 goto dropit;
982
983 /* Look up and lock the request corresponding to the given XID */
984 spin_lock(&xprt->transport_lock);
985 rovr = xprt_lookup_rqst(xprt, *xp);
986 if (!rovr)
987 goto out_unlock;
988 task = rovr->rq_task;
989
990 copied = rovr->rq_private_buf.buflen;
991 if (copied > repsize)
992 copied = repsize;
993
994 if (xs_local_copy_to_xdr(&rovr->rq_private_buf, skb)) {
995 dprintk("RPC: sk_buff copy failed\n");
996 goto out_unlock;
997 }
998
999 xprt_complete_rqst(task, copied);
1000
1001 out_unlock:
1002 spin_unlock(&xprt->transport_lock);
1003 dropit:
1004 skb_free_datagram(sk, skb);
1005 out:
1006 read_unlock_bh(&sk->sk_callback_lock);
1007}
1008
9903cd1c
CL		 1009/**
1010 * xs_udp_data_ready - "data ready" callback for UDP sockets
1011 * @sk: socket with data to read
1012 * @len: how much data to read
1013 *
a246b010 1014 */
676d2369 1015static void xs_udp_data_ready(struct sock *sk)
a246b010 1016{
9903cd1c
CL		 1017 struct rpc_task *task;
1018 struct rpc_xprt *xprt;
a246b010 1019 struct rpc_rqst *rovr;
9903cd1c 1020 struct sk_buff *skb;
a246b010 1021 int err, repsize, copied;
d8ed029d
AD		 1022 u32 _xid;
1023 __be32 *xp;
a246b010 1024
f064af1e 1025 read_lock_bh(&sk->sk_callback_lock);
46121cf7 1026 dprintk("RPC: xs_udp_data_ready...\n");
9903cd1c 1027 if (!(xprt = xprt_from_sock(sk)))
a246b010 1028 goto out;
a246b010
CL		 1029
1030 if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
1031 goto out;
1032
a246b010
CL		 1033 repsize = skb->len - sizeof(struct udphdr);
1034 if (repsize < 4) {
46121cf7 1035 dprintk("RPC: impossible RPC reply size %d!\n", repsize);
a246b010
CL		 1036 goto dropit;
1037 }
1038
1039 /* Copy the XID from the skb... */
1040 xp = skb_header_pointer(skb, sizeof(struct udphdr),
1041 sizeof(_xid), &_xid);
1042 if (xp == NULL)
1043 goto dropit;
1044
1045 /* Look up and lock the request corresponding to the given XID */
4a0f8c04 1046 spin_lock(&xprt->transport_lock);
a246b010
CL		 1047 rovr = xprt_lookup_rqst(xprt, *xp);
1048 if (!rovr)
1049 goto out_unlock;
1050 task = rovr->rq_task;
1051
a246b010
CL		 1052 if ((copied = rovr->rq_private_buf.buflen) > repsize)
1053 copied = repsize;
1054
1055 /* Suck it into the iovec, verify checksum if not done by hw. */
1781f7f5
HX		 1056 if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
1057 UDPX_INC_STATS_BH(sk, UDP_MIB_INERRORS);
a246b010 1058 goto out_unlock;
1781f7f5
HX		 1059 }
1060
1061 UDPX_INC_STATS_BH(sk, UDP_MIB_INDATAGRAMS);
a246b010 1062
6a24dfb6 1063 xprt_adjust_cwnd(xprt, task, copied);
1570c1e4 1064 xprt_complete_rqst(task, copied);
a246b010
CL		 1065
1066 out_unlock:
4a0f8c04 1067 spin_unlock(&xprt->transport_lock);
a246b010
CL		 1068 dropit:
1069 skb_free_datagram(sk, skb);
1070 out:
f064af1e 1071 read_unlock_bh(&sk->sk_callback_lock);
a246b010
CL		 1072}
1073
a519fc7a
TM		 1074/*
1075 * Helper function to force a TCP close if the server is sending
1076 * junk and/or it has put us in CLOSE_WAIT
1077 */
1078static void xs_tcp_force_close(struct rpc_xprt *xprt)
1079{
1080 set_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
1081 xprt_force_disconnect(xprt);
1082}
1083
dd456471 1084static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
a246b010 1085{
51971139 1086 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
a246b010
CL		 1087 size_t len, used;
1088 char *p;
1089
51971139
CL		 1090 p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
1091 len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
9d292316 1092 used = xdr_skb_read_bits(desc, p, len);
51971139 1093 transport->tcp_offset += used;
a246b010
CL		 1094 if (used != len)
1095 return;
808012fb 1096
51971139
CL		 1097 transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
1098 if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
e136d092 1099 transport->tcp_flags |= TCP_RCV_LAST_FRAG;
a246b010 1100 else
e136d092 1101 transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
51971139 1102 transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
808012fb 1103
e136d092 1104 transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
51971139 1105 transport->tcp_offset = 0;
808012fb 1106
a246b010 1107 /* Sanity check of the record length */
18dca02a 1108 if (unlikely(transport->tcp_reclen < 8)) {
46121cf7 1109 dprintk("RPC: invalid TCP record fragment length\n");
a519fc7a 1110 xs_tcp_force_close(xprt);
9903cd1c 1111 return;
a246b010 1112 }
46121cf7 1113 dprintk("RPC: reading TCP record fragment of length %d\n",
51971139 1114 transport->tcp_reclen);
a246b010
CL		 1115}
1116
51971139 1117static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
a246b010 1118{
51971139 1119 if (transport->tcp_offset == transport->tcp_reclen) {
e136d092 1120 transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
51971139 1121 transport->tcp_offset = 0;
e136d092
CL		 1122 if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
1123 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1124 transport->tcp_flags |= TCP_RCV_COPY_XID;
51971139 1125 transport->tcp_copied = 0;
a246b010
CL		 1126 }
1127 }
1128}
1129
dd456471 1130static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
a246b010
CL		 1131{
1132 size_t len, used;
1133 char *p;
1134
51971139 1135 len = sizeof(transport->tcp_xid) - transport->tcp_offset;
46121cf7 1136 dprintk("RPC: reading XID (%Zu bytes)\n", len);
51971139 1137 p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
9d292316 1138 used = xdr_skb_read_bits(desc, p, len);
51971139 1139 transport->tcp_offset += used;
a246b010
CL		 1140 if (used != len)
1141 return;
e136d092 1142 transport->tcp_flags &= ~TCP_RCV_COPY_XID;
f4a2e418 1143 transport->tcp_flags |= TCP_RCV_READ_CALLDIR;
51971139 1144 transport->tcp_copied = 4;
18dca02a
RL		 1145 dprintk("RPC: reading %s XID %08x\n",
1146 (transport->tcp_flags & TCP_RPC_REPLY) ? "reply for"
1147 : "request with",
51971139
CL		 1148 ntohl(transport->tcp_xid));
1149 xs_tcp_check_fraghdr(transport);
a246b010
CL		 1150}
1151
18dca02a
RL		 1152static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
1153 struct xdr_skb_reader *desc)
a246b010 1154{
18dca02a
RL		 1155 size_t len, used;
1156 u32 offset;
b76ce561 1157 char *p;
18dca02a
RL		 1158
1159 /*
1160 * We want transport->tcp_offset to be 8 at the end of this routine
1161 * (4 bytes for the xid and 4 bytes for the call/reply flag).
1162 * When this function is called for the first time,
1163 * transport->tcp_offset is 4 (after having already read the xid).
1164 */
1165 offset = transport->tcp_offset - sizeof(transport->tcp_xid);
b76ce561 1166 len = sizeof(transport->tcp_calldir) - offset;
18dca02a 1167 dprintk("RPC: reading CALL/REPLY flag (%Zu bytes)\n", len);
b76ce561
TM		 1168 p = ((char *) &transport->tcp_calldir) + offset;
1169 used = xdr_skb_read_bits(desc, p, len);
18dca02a
RL		 1170 transport->tcp_offset += used;
1171 if (used != len)
1172 return;
f4a2e418 1173 transport->tcp_flags &= ~TCP_RCV_READ_CALLDIR;
f4a2e418
RL		 1174 /*
1175 * We don't yet have the XDR buffer, so we will write the calldir
1176 * out after we get the buffer from the 'struct rpc_rqst'
1177 */
b76ce561
TM		 1178 switch (ntohl(transport->tcp_calldir)) {
1179 case RPC_REPLY:
1180 transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
1181 transport->tcp_flags |= TCP_RCV_COPY_DATA;
18dca02a 1182 transport->tcp_flags |= TCP_RPC_REPLY;
b76ce561
TM		 1183 break;
1184 case RPC_CALL:
1185 transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
1186 transport->tcp_flags |= TCP_RCV_COPY_DATA;
18dca02a 1187 transport->tcp_flags &= ~TCP_RPC_REPLY;
b76ce561
TM		 1188 break;
1189 default:
1190 dprintk("RPC: invalid request message type\n");
a519fc7a 1191 xs_tcp_force_close(&transport->xprt);
b76ce561 1192 }
18dca02a
RL		 1193 xs_tcp_check_fraghdr(transport);
1194}
1195
44b98efd
RL		 1196static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
1197 struct xdr_skb_reader *desc,
1198 struct rpc_rqst *req)
a246b010 1199{
44b98efd
RL		 1200 struct sock_xprt *transport =
1201 container_of(xprt, struct sock_xprt, xprt);
a246b010
CL		 1202 struct xdr_buf *rcvbuf;
1203 size_t len;
1204 ssize_t r;
1205
a246b010 1206 rcvbuf = &req->rq_private_buf;
f4a2e418
RL		 1207
1208 if (transport->tcp_flags & TCP_RCV_COPY_CALLDIR) {
1209 /*
1210 * Save the RPC direction in the XDR buffer
1211 */
f4a2e418 1212 memcpy(rcvbuf->head[0].iov_base + transport->tcp_copied,
b76ce561
TM		 1213 &transport->tcp_calldir,
1214 sizeof(transport->tcp_calldir));
1215 transport->tcp_copied += sizeof(transport->tcp_calldir);
f4a2e418 1216 transport->tcp_flags &= ~TCP_RCV_COPY_CALLDIR;
a246b010
CL		 1217 }
1218
a246b010 1219 len = desc->count;
51971139 1220 if (len > transport->tcp_reclen - transport->tcp_offset) {
dd456471 1221 struct xdr_skb_reader my_desc;
a246b010 1222
51971139 1223 len = transport->tcp_reclen - transport->tcp_offset;
a246b010
CL		 1224 memcpy(&my_desc, desc, sizeof(my_desc));
1225 my_desc.count = len;
51971139 1226 r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
9d292316 1227 &my_desc, xdr_skb_read_bits);
a246b010
CL		 1228 desc->count -= r;
1229 desc->offset += r;
1230 } else
51971139 1231 r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
9d292316 1232 desc, xdr_skb_read_bits);
a246b010
CL		 1233
1234 if (r > 0) {
51971139
CL		 1235 transport->tcp_copied += r;
1236 transport->tcp_offset += r;
a246b010
CL		 1237 }
1238 if (r != len) {
1239 /* Error when copying to the receive buffer,
1240 * usually because we weren't able to allocate
1241 * additional buffer pages. All we can do now
e136d092 1242 * is turn off TCP_RCV_COPY_DATA, so the request
a246b010
CL		 1243 * will not receive any additional updates,
1244 * and time out.
1245 * Any remaining data from this record will
1246 * be discarded.
1247 */
e136d092 1248 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
46121cf7 1249 dprintk("RPC: XID %08x truncated request\n",
51971139 1250 ntohl(transport->tcp_xid));
46121cf7
CL		 1251 dprintk("RPC: xprt = %p, tcp_copied = %lu, "
1252 "tcp_offset = %u, tcp_reclen = %u\n",
1253 xprt, transport->tcp_copied,
1254 transport->tcp_offset, transport->tcp_reclen);
44b98efd 1255 return;
a246b010
CL		 1256 }
1257
46121cf7 1258 dprintk("RPC: XID %08x read %Zd bytes\n",
51971139 1259 ntohl(transport->tcp_xid), r);
46121cf7
CL		 1260 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
1261 "tcp_reclen = %u\n", xprt, transport->tcp_copied,
1262 transport->tcp_offset, transport->tcp_reclen);
51971139
CL		 1263
1264 if (transport->tcp_copied == req->rq_private_buf.buflen)
e136d092 1265 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
51971139 1266 else if (transport->tcp_offset == transport->tcp_reclen) {
e136d092
CL		 1267 if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
1268 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
a246b010 1269 }
44b98efd
RL		 1270}
1271
1272/*
1273 * Finds the request corresponding to the RPC xid and invokes the common
1274 * tcp read code to read the data.
1275 */
1276static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
1277 struct xdr_skb_reader *desc)
1278{
1279 struct sock_xprt *transport =
1280 container_of(xprt, struct sock_xprt, xprt);
1281 struct rpc_rqst *req;
1282
1283 dprintk("RPC: read reply XID %08x\n", ntohl(transport->tcp_xid));
1284
1285 /* Find and lock the request corresponding to this xid */
1286 spin_lock(&xprt->transport_lock);
1287 req = xprt_lookup_rqst(xprt, transport->tcp_xid);
1288 if (!req) {
1289 dprintk("RPC: XID %08x request not found!\n",
1290 ntohl(transport->tcp_xid));
1291 spin_unlock(&xprt->transport_lock);
1292 return -1;
1293 }
1294
1295 xs_tcp_read_common(xprt, desc, req);
1296
e136d092 1297 if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
51971139 1298 xprt_complete_rqst(req->rq_task, transport->tcp_copied);
44b98efd 1299
4a0f8c04 1300 spin_unlock(&xprt->transport_lock);
44b98efd
RL		 1301 return 0;
1302}
1303
9e00abc3 1304#if defined(CONFIG_SUNRPC_BACKCHANNEL)
44b98efd
RL		 1305/*
1306 * Obtains an rpc_rqst previously allocated and invokes the common
1307 * tcp read code to read the data. The result is placed in the callback
1308 * queue.
1309 * If we're unable to obtain the rpc_rqst we schedule the closing of the
1310 * connection and return -1.
1311 */
2ea24497 1312static int xs_tcp_read_callback(struct rpc_xprt *xprt,
44b98efd
RL		 1313 struct xdr_skb_reader *desc)
1314{
1315 struct sock_xprt *transport =
1316 container_of(xprt, struct sock_xprt, xprt);
1317 struct rpc_rqst *req;
1318
2ea24497
TM		 1319 /* Look up and lock the request corresponding to the given XID */
1320 spin_lock(&xprt->transport_lock);
1321 req = xprt_lookup_bc_request(xprt, transport->tcp_xid);
44b98efd 1322 if (req == NULL) {
2ea24497 1323 spin_unlock(&xprt->transport_lock);
44b98efd
RL		 1324 printk(KERN_WARNING "Callback slot table overflowed\n");
1325 xprt_force_disconnect(xprt);
1326 return -1;
1327 }
1328
44b98efd
RL		 1329 dprintk("RPC: read callback XID %08x\n", ntohl(req->rq_xid));
1330 xs_tcp_read_common(xprt, desc, req);
1331
2ea24497
TM		 1332 if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
1333 xprt_complete_bc_request(req, transport->tcp_copied);
1334 spin_unlock(&xprt->transport_lock);
44b98efd
RL		 1335
1336 return 0;
1337}
1338
1339static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1340 struct xdr_skb_reader *desc)
1341{
1342 struct sock_xprt *transport =
1343 container_of(xprt, struct sock_xprt, xprt);
1344
1345 return (transport->tcp_flags & TCP_RPC_REPLY) ?
1346 xs_tcp_read_reply(xprt, desc) :
1347 xs_tcp_read_callback(xprt, desc);
1348}
1349#else
1350static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1351 struct xdr_skb_reader *desc)
1352{
1353 return xs_tcp_read_reply(xprt, desc);
1354}
9e00abc3 1355#endif /* CONFIG_SUNRPC_BACKCHANNEL */
44b98efd
RL		 1356
1357/*
1358 * Read data off the transport. This can be either an RPC_CALL or an
1359 * RPC_REPLY. Relay the processing to helper functions.
1360 */
1361static void xs_tcp_read_data(struct rpc_xprt *xprt,
1362 struct xdr_skb_reader *desc)
1363{
1364 struct sock_xprt *transport =
1365 container_of(xprt, struct sock_xprt, xprt);
1366
1367 if (_xs_tcp_read_data(xprt, desc) == 0)
1368 xs_tcp_check_fraghdr(transport);
1369 else {
1370 /*
1371 * The transport_lock protects the request handling.
1372 * There's no need to hold it to update the tcp_flags.
1373 */
1374 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1375 }
a246b010
CL		 1376}
1377
dd456471 1378static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
a246b010
CL		 1379{
1380 size_t len;
1381
51971139 1382 len = transport->tcp_reclen - transport->tcp_offset;
a246b010
CL		 1383 if (len > desc->count)
1384 len = desc->count;
1385 desc->count -= len;
1386 desc->offset += len;
51971139 1387 transport->tcp_offset += len;
46121cf7 1388 dprintk("RPC: discarded %Zu bytes\n", len);
51971139 1389 xs_tcp_check_fraghdr(transport);
a246b010
CL		 1390}
1391
9903cd1c 1392static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
a246b010
CL		 1393{
1394 struct rpc_xprt *xprt = rd_desc->arg.data;
51971139 1395 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
dd456471 1396 struct xdr_skb_reader desc = {
a246b010
CL		 1397 .skb = skb,
1398 .offset = offset,
1399 .count = len,
9903cd1c 1400 };
a246b010 1401
46121cf7 1402 dprintk("RPC: xs_tcp_data_recv started\n");
a246b010
CL		 1403 do {
1404 /* Read in a new fragment marker if necessary */
1405 /* Can we ever really expect to get completely empty fragments? */
e136d092 1406 if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
9903cd1c 1407 xs_tcp_read_fraghdr(xprt, &desc);
a246b010
CL		 1408 continue;
1409 }
1410 /* Read in the xid if necessary */
e136d092 1411 if (transport->tcp_flags & TCP_RCV_COPY_XID) {
51971139 1412 xs_tcp_read_xid(transport, &desc);
a246b010
CL		 1413 continue;
1414 }
18dca02a 1415 /* Read in the call/reply flag */
f4a2e418 1416 if (transport->tcp_flags & TCP_RCV_READ_CALLDIR) {
18dca02a
RL		 1417 xs_tcp_read_calldir(transport, &desc);
1418 continue;
1419 }
a246b010 1420 /* Read in the request data */
e136d092 1421 if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
44b98efd 1422 xs_tcp_read_data(xprt, &desc);
a246b010
CL		 1423 continue;
1424 }
1425 /* Skip over any trailing bytes on short reads */
51971139 1426 xs_tcp_read_discard(transport, &desc);
a246b010 1427 } while (desc.count);
46121cf7 1428 dprintk("RPC: xs_tcp_data_recv done\n");
a246b010
CL		 1429 return len - desc.count;
1430}
1431
9903cd1c
CL		 1432/**
1433 * xs_tcp_data_ready - "data ready" callback for TCP sockets
1434 * @sk: socket with data to read
1435 * @bytes: how much data to read
1436 *
1437 */
676d2369 1438static void xs_tcp_data_ready(struct sock *sk)
a246b010
CL		 1439{
1440 struct rpc_xprt *xprt;
1441 read_descriptor_t rd_desc;
ff2d7db8 1442 int read;
a246b010 1443
46121cf7
CL		 1444 dprintk("RPC: xs_tcp_data_ready...\n");
1445
f064af1e 1446 read_lock_bh(&sk->sk_callback_lock);
9903cd1c 1447 if (!(xprt = xprt_from_sock(sk)))
a246b010 1448 goto out;
61d0a8e6
NB		 1449 /* Any data means we had a useful conversation, so
1450 * the we don't need to delay the next reconnect
1451 */
1452 if (xprt->reestablish_timeout)
1453 xprt->reestablish_timeout = 0;
1454
9903cd1c 1455 /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
a246b010 1456 rd_desc.arg.data = xprt;
ff2d7db8
TM		 1457 do {
1458 rd_desc.count = 65536;
1459 read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
1460 } while (read > 0);
a246b010 1461out:
f064af1e 1462 read_unlock_bh(&sk->sk_callback_lock);
a246b010
CL		 1463}
1464
7d1e8255
TM		 1465/*
1466 * Do the equivalent of linger/linger2 handling for dealing with
1467 * broken servers that don't close the socket in a timely
1468 * fashion
1469 */
1470static void xs_tcp_schedule_linger_timeout(struct rpc_xprt *xprt,
1471 unsigned long timeout)
1472{
1473 struct sock_xprt *transport;
1474
1475 if (xprt_test_and_set_connecting(xprt))
1476 return;
1477 set_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1478 transport = container_of(xprt, struct sock_xprt, xprt);
1479 queue_delayed_work(rpciod_workqueue, &transport->connect_worker,
1480 timeout);
1481}
1482
1483static void xs_tcp_cancel_linger_timeout(struct rpc_xprt *xprt)
1484{
1485 struct sock_xprt *transport;
1486
1487 transport = container_of(xprt, struct sock_xprt, xprt);
1488
1489 if (!test_bit(XPRT_CONNECTION_ABORT, &xprt->state) ||
1490 !cancel_delayed_work(&transport->connect_worker))
1491 return;
1492 clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1493 xprt_clear_connecting(xprt);
1494}
1495
4bc1e68e 1496static void xs_sock_reset_connection_flags(struct rpc_xprt *xprt)
7d1e8255 1497{
4e857c58 1498 smp_mb__before_atomic();
a519fc7a
TM		 1499 clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1500 clear_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
7d1e8255
TM		 1501 clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1502 clear_bit(XPRT_CLOSING, &xprt->state);
4e857c58 1503 smp_mb__after_atomic();
4bc1e68e
TM		 1504}
1505
1506static void xs_sock_mark_closed(struct rpc_xprt *xprt)
1507{
1508 xs_sock_reset_connection_flags(xprt);
7d1e8255
TM		 1509 /* Mark transport as closed and wake up all pending tasks */
1510 xprt_disconnect_done(xprt);
1511}
1512
9903cd1c
CL		 1513/**
1514 * xs_tcp_state_change - callback to handle TCP socket state changes
1515 * @sk: socket whose state has changed
1516 *
1517 */
1518static void xs_tcp_state_change(struct sock *sk)
a246b010 1519{
9903cd1c 1520 struct rpc_xprt *xprt;
a246b010 1521
f064af1e 1522 read_lock_bh(&sk->sk_callback_lock);
a246b010
CL		 1523 if (!(xprt = xprt_from_sock(sk)))
1524 goto out;
46121cf7 1525 dprintk("RPC: xs_tcp_state_change client %p...\n", xprt);
669502ff 1526 dprintk("RPC: state %x conn %d dead %d zapped %d sk_shutdown %d\n",
46121cf7
CL		 1527 sk->sk_state, xprt_connected(xprt),
1528 sock_flag(sk, SOCK_DEAD),
669502ff
AC		 1529 sock_flag(sk, SOCK_ZAPPED),
1530 sk->sk_shutdown);
a246b010 1531
40b5ea0c 1532 trace_rpc_socket_state_change(xprt, sk->sk_socket);
a246b010
CL		 1533 switch (sk->sk_state) {
1534 case TCP_ESTABLISHED:
f064af1e 1535 spin_lock(&xprt->transport_lock);
a246b010 1536 if (!xprt_test_and_set_connected(xprt)) {
51971139
CL		 1537 struct sock_xprt *transport = container_of(xprt,
1538 struct sock_xprt, xprt);
1539
a246b010 1540 /* Reset TCP record info */
51971139
CL		 1541 transport->tcp_offset = 0;
1542 transport->tcp_reclen = 0;
1543 transport->tcp_copied = 0;
e136d092
CL		 1544 transport->tcp_flags =
1545 TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
8b71798c 1546 xprt->connect_cookie++;
51971139 1547
2a491991 1548 xprt_wake_pending_tasks(xprt, -EAGAIN);
a246b010 1549 }
f064af1e 1550 spin_unlock(&xprt->transport_lock);
a246b010 1551 break;
3b948ae5
TM		 1552 case TCP_FIN_WAIT1:
1553 /* The client initiated a shutdown of the socket */
7c1d71cf 1554 xprt->connect_cookie++;
663b8858 1555 xprt->reestablish_timeout = 0;
3b948ae5 1556 set_bit(XPRT_CLOSING, &xprt->state);
4e857c58 1557 smp_mb__before_atomic();
3b948ae5 1558 clear_bit(XPRT_CONNECTED, &xprt->state);
ef803670 1559 clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
4e857c58 1560 smp_mb__after_atomic();
25fe6142 1561 xs_tcp_schedule_linger_timeout(xprt, xs_tcp_fin_timeout);
a246b010 1562 break;
632e3bdc 1563 case TCP_CLOSE_WAIT:
3b948ae5 1564 /* The server initiated a shutdown of the socket */
7c1d71cf 1565 xprt->connect_cookie++;
d0bea455 1566 clear_bit(XPRT_CONNECTED, &xprt->state);
a519fc7a 1567 xs_tcp_force_close(xprt);
663b8858
TM		 1568 case TCP_CLOSING:
1569 /*
1570 * If the server closed down the connection, make sure that
1571 * we back off before reconnecting
1572 */
1573 if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
1574 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
3b948ae5
TM		 1575 break;
1576 case TCP_LAST_ACK:
670f9457 1577 set_bit(XPRT_CLOSING, &xprt->state);
25fe6142 1578 xs_tcp_schedule_linger_timeout(xprt, xs_tcp_fin_timeout);
4e857c58 1579 smp_mb__before_atomic();
3b948ae5 1580 clear_bit(XPRT_CONNECTED, &xprt->state);
4e857c58 1581 smp_mb__after_atomic();
3b948ae5
TM		 1582 break;
1583 case TCP_CLOSE:
7d1e8255
TM		 1584 xs_tcp_cancel_linger_timeout(xprt);
1585 xs_sock_mark_closed(xprt);
a246b010
CL		 1586 }
1587 out:
f064af1e 1588 read_unlock_bh(&sk->sk_callback_lock);
a246b010
CL		 1589}
1590
1f0fa154
IJ		 1591static void xs_write_space(struct sock *sk)
1592{
1593 struct socket *sock;
1594 struct rpc_xprt *xprt;
1595
1596 if (unlikely(!(sock = sk->sk_socket)))
1597 return;
1598 clear_bit(SOCK_NOSPACE, &sock->flags);
1599
1600 if (unlikely(!(xprt = xprt_from_sock(sk))))
1601 return;
1602 if (test_and_clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags) == 0)
1603 return;
1604
1605 xprt_write_space(xprt);
1606}
1607
9903cd1c 1608/**
c7b2cae8
CL		 1609 * xs_udp_write_space - callback invoked when socket buffer space
1610 * becomes available
9903cd1c
CL		 1611 * @sk: socket whose state has changed
1612 *
a246b010
CL		 1613 * Called when more output buffer space is available for this socket.
1614 * We try not to wake our writers until they can make "significant"
c7b2cae8 1615 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
a246b010
CL		 1616 * with a bunch of small requests.
1617 */
c7b2cae8 1618static void xs_udp_write_space(struct sock *sk)
a246b010 1619{
f064af1e 1620 read_lock_bh(&sk->sk_callback_lock);
a246b010 1621
c7b2cae8 1622 /* from net/core/sock.c:sock_def_write_space */
1f0fa154
IJ		 1623 if (sock_writeable(sk))
1624 xs_write_space(sk);
a246b010 1625
f064af1e 1626 read_unlock_bh(&sk->sk_callback_lock);
c7b2cae8 1627}
a246b010 1628
c7b2cae8
CL		 1629/**
1630 * xs_tcp_write_space - callback invoked when socket buffer space
1631 * becomes available
1632 * @sk: socket whose state has changed
1633 *
1634 * Called when more output buffer space is available for this socket.
1635 * We try not to wake our writers until they can make "significant"
1636 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1637 * with a bunch of small requests.
1638 */
1639static void xs_tcp_write_space(struct sock *sk)
1640{
f064af1e 1641 read_lock_bh(&sk->sk_callback_lock);
c7b2cae8
CL		 1642
1643 /* from net/core/stream.c:sk_stream_write_space */
64dc6130 1644 if (sk_stream_is_writeable(sk))
1f0fa154 1645 xs_write_space(sk);
b6ddf64f 1646
f064af1e 1647 read_unlock_bh(&sk->sk_callback_lock);
a246b010
CL		 1648}
1649
470056c2 1650static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
a246b010 1651{
ee0ac0c2
CL		 1652 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1653 struct sock *sk = transport->inet;
a246b010 1654
7c6e066e 1655 if (transport->rcvsize) {
a246b010 1656 sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
7c6e066e 1657 sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
a246b010 1658 }
7c6e066e 1659 if (transport->sndsize) {
a246b010 1660 sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
7c6e066e 1661 sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
a246b010
CL		 1662 sk->sk_write_space(sk);
1663 }
1664}
1665
43118c29 1666/**
470056c2 1667 * xs_udp_set_buffer_size - set send and receive limits
43118c29 1668 * @xprt: generic transport
470056c2
CL		 1669 * @sndsize: requested size of send buffer, in bytes
1670 * @rcvsize: requested size of receive buffer, in bytes
43118c29 1671 *
470056c2 1672 * Set socket send and receive buffer size limits.
43118c29 1673 */
470056c2 1674static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
43118c29 1675{
7c6e066e
CL		 1676 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1677
1678 transport->sndsize = 0;
470056c2 1679 if (sndsize)
7c6e066e
CL		 1680 transport->sndsize = sndsize + 1024;
1681 transport->rcvsize = 0;
470056c2 1682 if (rcvsize)
7c6e066e 1683 transport->rcvsize = rcvsize + 1024;
470056c2
CL		 1684
1685 xs_udp_do_set_buffer_size(xprt);
43118c29
CL		 1686}
1687
46c0ee8b
CL		 1688/**
1689 * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
1690 * @task: task that timed out
1691 *
1692 * Adjust the congestion window after a retransmit timeout has occurred.
1693 */
6a24dfb6 1694static void xs_udp_timer(struct rpc_xprt *xprt, struct rpc_task *task)
46c0ee8b 1695{
6a24dfb6 1696 xprt_adjust_cwnd(xprt, task, -ETIMEDOUT);
46c0ee8b
CL		 1697}
1698
b85d8806
CL		 1699static unsigned short xs_get_random_port(void)
1700{
1701 unsigned short range = xprt_max_resvport - xprt_min_resvport;
63862b5b 1702 unsigned short rand = (unsigned short) prandom_u32() % range;
b85d8806
CL		 1703 return rand + xprt_min_resvport;
1704}
1705
92200412
CL		 1706/**
1707 * xs_set_port - reset the port number in the remote endpoint address
1708 * @xprt: generic transport
1709 * @port: new port number
1710 *
1711 */
1712static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
1713{
46121cf7 1714 dprintk("RPC: setting port for xprt %p to %u\n", xprt, port);
c4efcb1d 1715
9dc3b095
CL		 1716 rpc_set_port(xs_addr(xprt), port);
1717 xs_update_peer_port(xprt);
92200412
CL		 1718}
1719
5d4ec932 1720static unsigned short xs_get_srcport(struct sock_xprt *transport)
67a391d7 1721{
fbfffbd5 1722 unsigned short port = transport->srcport;
67a391d7
TM		 1723
1724 if (port == 0 && transport->xprt.resvport)
1725 port = xs_get_random_port();
1726 return port;
1727}
1728
baaf4e48 1729static unsigned short xs_next_srcport(struct sock_xprt *transport, unsigned short port)
67a391d7 1730{
fbfffbd5
CL		 1731 if (transport->srcport != 0)
1732 transport->srcport = 0;
67a391d7
TM		 1733 if (!transport->xprt.resvport)
1734 return 0;
1735 if (port <= xprt_min_resvport || port > xprt_max_resvport)
1736 return xprt_max_resvport;
1737 return --port;
1738}
beb59b68 1739static int xs_bind(struct sock_xprt *transport, struct socket *sock)
a246b010 1740{
beb59b68 1741 struct sockaddr_storage myaddr;
67a391d7 1742 int err, nloop = 0;
5d4ec932 1743 unsigned short port = xs_get_srcport(transport);
67a391d7 1744 unsigned short last;
a246b010 1745
beb59b68 1746 memcpy(&myaddr, &transport->srcaddr, transport->xprt.addrlen);
a246b010 1747 do {
beb59b68
PE		 1748 rpc_set_port((struct sockaddr *)&myaddr, port);
1749 err = kernel_bind(sock, (struct sockaddr *)&myaddr,
1750 transport->xprt.addrlen);
67a391d7 1751 if (port == 0)
d3bc9a1d 1752 break;
a246b010 1753 if (err == 0) {
fbfffbd5 1754 transport->srcport = port;
d3bc9a1d 1755 break;
a246b010 1756 }
67a391d7 1757 last = port;
baaf4e48 1758 port = xs_next_srcport(transport, port);
67a391d7
TM		 1759 if (port > last)
1760 nloop++;
1761 } while (err == -EADDRINUSE && nloop != 2);
90058d37 1762
4232e863 1763 if (myaddr.ss_family == AF_INET)
beb59b68
PE		 1764 dprintk("RPC: %s %pI4:%u: %s (%d)\n", __func__,
1765 &((struct sockaddr_in *)&myaddr)->sin_addr,
1766 port, err ? "failed" : "ok", err);
1767 else
1768 dprintk("RPC: %s %pI6:%u: %s (%d)\n", __func__,
1769 &((struct sockaddr_in6 *)&myaddr)->sin6_addr,
1770 port, err ? "failed" : "ok", err);
a246b010
CL		 1771 return err;
1772}
1773
176e21ee
CL		 1774/*
1775 * We don't support autobind on AF_LOCAL sockets
1776 */
1777static void xs_local_rpcbind(struct rpc_task *task)
1778{
3dc0da27
TM		 1779 rcu_read_lock();
1780 xprt_set_bound(rcu_dereference(task->tk_client->cl_xprt));
1781 rcu_read_unlock();
176e21ee
CL		 1782}
1783
1784static void xs_local_set_port(struct rpc_xprt *xprt, unsigned short port)
1785{
1786}
beb59b68 1787
ed07536e
PZ		 1788#ifdef CONFIG_DEBUG_LOCK_ALLOC
1789static struct lock_class_key xs_key[2];
1790static struct lock_class_key xs_slock_key[2];
1791
176e21ee
CL		 1792static inline void xs_reclassify_socketu(struct socket *sock)
1793{
1794 struct sock *sk = sock->sk;
1795
176e21ee
CL		 1796 sock_lock_init_class_and_name(sk, "slock-AF_LOCAL-RPC",
1797 &xs_slock_key[1], "sk_lock-AF_LOCAL-RPC", &xs_key[1]);
1798}
1799
8945ee5e 1800static inline void xs_reclassify_socket4(struct socket *sock)
ed07536e
PZ		 1801{
1802 struct sock *sk = sock->sk;
8945ee5e 1803
8945ee5e
CL		 1804 sock_lock_init_class_and_name(sk, "slock-AF_INET-RPC",
1805 &xs_slock_key[0], "sk_lock-AF_INET-RPC", &xs_key[0]);
1806}
ed07536e 1807
8945ee5e
CL		 1808static inline void xs_reclassify_socket6(struct socket *sock)
1809{
1810 struct sock *sk = sock->sk;
ed07536e 1811
8945ee5e
CL		 1812 sock_lock_init_class_and_name(sk, "slock-AF_INET6-RPC",
1813 &xs_slock_key[1], "sk_lock-AF_INET6-RPC", &xs_key[1]);
ed07536e 1814}
6bc9638a
PE		 1815
1816static inline void xs_reclassify_socket(int family, struct socket *sock)
1817{
1b7a1819
WAA		 1818 WARN_ON_ONCE(sock_owned_by_user(sock->sk));
1819 if (sock_owned_by_user(sock->sk))
1820 return;
1821
4232e863 1822 switch (family) {
176e21ee
CL		 1823 case AF_LOCAL:
1824 xs_reclassify_socketu(sock);
1825 break;
4232e863 1826 case AF_INET:
6bc9638a 1827 xs_reclassify_socket4(sock);
4232e863
CL		 1828 break;
1829 case AF_INET6:
6bc9638a 1830 xs_reclassify_socket6(sock);
4232e863
CL		 1831 break;
1832 }
6bc9638a 1833}
ed07536e 1834#else
176e21ee
CL		 1835static inline void xs_reclassify_socketu(struct socket *sock)
1836{
1837}
1838
8945ee5e
CL		 1839static inline void xs_reclassify_socket4(struct socket *sock)
1840{
1841}
1842
1843static inline void xs_reclassify_socket6(struct socket *sock)
ed07536e
PZ		 1844{
1845}
6bc9638a
PE		 1846
1847static inline void xs_reclassify_socket(int family, struct socket *sock)
1848{
1849}
ed07536e
PZ		 1850#endif
1851
93dc41bd
N		 1852static void xs_dummy_setup_socket(struct work_struct *work)
1853{
1854}
1855
6bc9638a
PE		 1856static struct socket *xs_create_sock(struct rpc_xprt *xprt,
1857 struct sock_xprt *transport, int family, int type, int protocol)
22f79326
PE		 1858{
1859 struct socket *sock;
1860 int err;
1861
6bc9638a 1862 err = __sock_create(xprt->xprt_net, family, type, protocol, &sock, 1);
22f79326
PE		 1863 if (err < 0) {
1864 dprintk("RPC: can't create %d transport socket (%d).\n",
1865 protocol, -err);
1866 goto out;
1867 }
6bc9638a 1868 xs_reclassify_socket(family, sock);
22f79326 1869
4cea288a
BH		 1870 err = xs_bind(transport, sock);
1871 if (err) {
22f79326
PE		 1872 sock_release(sock);
1873 goto out;
1874 }
1875
1876 return sock;
1877out:
1878 return ERR_PTR(err);
1879}
1880
176e21ee
CL		 1881static int xs_local_finish_connecting(struct rpc_xprt *xprt,
1882 struct socket *sock)
1883{
1884 struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
1885 xprt);
1886
1887 if (!transport->inet) {
1888 struct sock *sk = sock->sk;
1889
1890 write_lock_bh(&sk->sk_callback_lock);
1891
1892 xs_save_old_callbacks(transport, sk);
1893
1894 sk->sk_user_data = xprt;
1895 sk->sk_data_ready = xs_local_data_ready;
1896 sk->sk_write_space = xs_udp_write_space;
2118071d 1897 sk->sk_error_report = xs_error_report;
176e21ee
CL		 1898 sk->sk_allocation = GFP_ATOMIC;
1899
1900 xprt_clear_connected(xprt);
1901
1902 /* Reset to new socket */
1903 transport->sock = sock;
1904 transport->inet = sk;
1905
1906 write_unlock_bh(&sk->sk_callback_lock);
1907 }
1908
1909 /* Tell the socket layer to start connecting... */
1910 xprt->stat.connect_count++;
1911 xprt->stat.connect_start = jiffies;
1912 return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, 0);
1913}
1914
1915/**
1916 * xs_local_setup_socket - create AF_LOCAL socket, connect to a local endpoint
1917 * @xprt: RPC transport to connect
1918 * @transport: socket transport to connect
1919 * @create_sock: function to create a socket of the correct type
176e21ee 1920 */
dc107402 1921static int xs_local_setup_socket(struct sock_xprt *transport)
176e21ee 1922{
176e21ee
CL		 1923 struct rpc_xprt *xprt = &transport->xprt;
1924 struct socket *sock;
1925 int status = -EIO;
1926
5cf02d09
JL		 1927 current->flags |= PF_FSTRANS;
1928
176e21ee
CL		 1929 clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1930 status = __sock_create(xprt->xprt_net, AF_LOCAL,
1931 SOCK_STREAM, 0, &sock, 1);
1932 if (status < 0) {
1933 dprintk("RPC: can't create AF_LOCAL "
1934 "transport socket (%d).\n", -status);
1935 goto out;
1936 }
1937 xs_reclassify_socketu(sock);
1938
1939 dprintk("RPC: worker connecting xprt %p via AF_LOCAL to %s\n",
1940 xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1941
1942 status = xs_local_finish_connecting(xprt, sock);
40b5ea0c 1943 trace_rpc_socket_connect(xprt, sock, status);
176e21ee
CL		 1944 switch (status) {
1945 case 0:
1946 dprintk("RPC: xprt %p connected to %s\n",
1947 xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1948 xprt_set_connected(xprt);
1949 break;
1950 case -ENOENT:
1951 dprintk("RPC: xprt %p: socket %s does not exist\n",
1952 xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1953 break;
4a20a988
TM		 1954 case -ECONNREFUSED:
1955 dprintk("RPC: xprt %p: connection refused for %s\n",
1956 xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1957 break;
176e21ee
CL		 1958 default:
1959 printk(KERN_ERR "%s: unhandled error (%d) connecting to %s\n",
1960 __func__, -status,
1961 xprt->address_strings[RPC_DISPLAY_ADDR]);
1962 }
1963
1964out:
1965 xprt_clear_connecting(xprt);
1966 xprt_wake_pending_tasks(xprt, status);
5cf02d09 1967 current->flags &= ~PF_FSTRANS;
dc107402
BF		 1968 return status;
1969}
1970
b6669737 1971static void xs_local_connect(struct rpc_xprt *xprt, struct rpc_task *task)
dc107402 1972{
dc107402
BF		 1973 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1974 int ret;
1975
1976 if (RPC_IS_ASYNC(task)) {
1977 /*
1978 * We want the AF_LOCAL connect to be resolved in the
1979 * filesystem namespace of the process making the rpc
1980 * call. Thus we connect synchronously.
1981 *
1982 * If we want to support asynchronous AF_LOCAL calls,
1983 * we'll need to figure out how to pass a namespace to
1984 * connect.
1985 */
1986 rpc_exit(task, -ENOTCONN);
1987 return;
1988 }
1989 ret = xs_local_setup_socket(transport);
1990 if (ret && !RPC_IS_SOFTCONN(task))
1991 msleep_interruptible(15000);
176e21ee
CL		 1992}
1993
a564b8f0
MG		 1994#ifdef CONFIG_SUNRPC_SWAP
1995static void xs_set_memalloc(struct rpc_xprt *xprt)
1996{
1997 struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
1998 xprt);
1999
2000 if (xprt->swapper)
2001 sk_set_memalloc(transport->inet);
2002}
2003
2004/**
2005 * xs_swapper - Tag this transport as being used for swap.
2006 * @xprt: transport to tag
2007 * @enable: enable/disable
2008 *
2009 */
2010int xs_swapper(struct rpc_xprt *xprt, int enable)
2011{
2012 struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
2013 xprt);
2014 int err = 0;
2015
2016 if (enable) {
2017 xprt->swapper++;
2018 xs_set_memalloc(xprt);
2019 } else if (xprt->swapper) {
2020 xprt->swapper--;
2021 sk_clear_memalloc(transport->inet);
2022 }
2023
2024 return err;
2025}
2026EXPORT_SYMBOL_GPL(xs_swapper);
2027#else
2028static void xs_set_memalloc(struct rpc_xprt *xprt)
2029{
2030}
2031#endif
2032
16be2d20
CL		 2033static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
2034{
2035 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2036
2037 if (!transport->inet) {
2038 struct sock *sk = sock->sk;
2039
2040 write_lock_bh(&sk->sk_callback_lock);
2041
2a9e1cfa
TM		 2042 xs_save_old_callbacks(transport, sk);
2043
16be2d20 2044 sk->sk_user_data = xprt;
16be2d20
CL		 2045 sk->sk_data_ready = xs_udp_data_ready;
2046 sk->sk_write_space = xs_udp_write_space;
16be2d20
CL		 2047 sk->sk_allocation = GFP_ATOMIC;
2048
2049 xprt_set_connected(xprt);
2050
2051 /* Reset to new socket */
2052 transport->sock = sock;
2053 transport->inet = sk;
2054
a564b8f0
MG		 2055 xs_set_memalloc(xprt);
2056
16be2d20
CL		 2057 write_unlock_bh(&sk->sk_callback_lock);
2058 }
2059 xs_udp_do_set_buffer_size(xprt);
2060}
2061
8c14ff2a 2062static void xs_udp_setup_socket(struct work_struct *work)
a246b010 2063{
34161db6
TM		 2064 struct sock_xprt *transport =
2065 container_of(work, struct sock_xprt, connect_worker.work);
c8475461 2066 struct rpc_xprt *xprt = &transport->xprt;
ee0ac0c2 2067 struct socket *sock = transport->sock;
b65c0310 2068 int status = -EIO;
9903cd1c 2069
5cf02d09
JL		 2070 current->flags |= PF_FSTRANS;
2071
b0d93ad5 2072 /* Start by resetting any existing state */
fe315e76 2073 xs_reset_transport(transport);
8c14ff2a
PE		 2074 sock = xs_create_sock(xprt, transport,
2075 xs_addr(xprt)->sa_family, SOCK_DGRAM, IPPROTO_UDP);
b65c0310 2076 if (IS_ERR(sock))
b0d93ad5 2077 goto out;
68e220bd 2078
c740eff8
CL		 2079 dprintk("RPC: worker connecting xprt %p via %s to "
2080 "%s (port %s)\n", xprt,
2081 xprt->address_strings[RPC_DISPLAY_PROTO],
2082 xprt->address_strings[RPC_DISPLAY_ADDR],
2083 xprt->address_strings[RPC_DISPLAY_PORT]);
68e220bd
CL		 2084
2085 xs_udp_finish_connecting(xprt, sock);
40b5ea0c 2086 trace_rpc_socket_connect(xprt, sock, 0);
b0d93ad5
CL		 2087 status = 0;
2088out:
b0d93ad5 2089 xprt_clear_connecting(xprt);
7d1e8255 2090 xprt_wake_pending_tasks(xprt, status);
5cf02d09 2091 current->flags &= ~PF_FSTRANS;
a246b010
CL		 2092}
2093
3167e12c
CL		 2094/*
2095 * We need to preserve the port number so the reply cache on the server can
2096 * find our cached RPC replies when we get around to reconnecting.
2097 */
58dddac9 2098static void xs_abort_connection(struct sock_xprt *transport)
3167e12c
CL		 2099{
2100 int result;
3167e12c
CL		 2101 struct sockaddr any;
2102
58dddac9 2103 dprintk("RPC: disconnecting xprt %p to reuse port\n", transport);
3167e12c
CL		 2104
2105 /*
2106 * Disconnect the transport socket by doing a connect operation
2107 * with AF_UNSPEC. This should return immediately...
2108 */
2109 memset(&any, 0, sizeof(any));
2110 any.sa_family = AF_UNSPEC;
ee0ac0c2 2111 result = kernel_connect(transport->sock, &any, sizeof(any), 0);
40b5ea0c
TM		 2112 trace_rpc_socket_reset_connection(&transport->xprt,
2113 transport->sock, result);
7d1e8255 2114 if (!result)
4bc1e68e
TM		 2115 xs_sock_reset_connection_flags(&transport->xprt);
2116 dprintk("RPC: AF_UNSPEC connect return code %d\n", result);
3167e12c
CL		 2117}
2118
58dddac9 2119static void xs_tcp_reuse_connection(struct sock_xprt *transport)
40d2549d
TM		 2120{
2121 unsigned int state = transport->inet->sk_state;
2122
669502ff
AC		 2123 if (state == TCP_CLOSE && transport->sock->state == SS_UNCONNECTED) {
2124 /* we don't need to abort the connection if the socket
2125 * hasn't undergone a shutdown
2126 */
2127 if (transport->inet->sk_shutdown == 0)
2128 return;
2129 dprintk("RPC: %s: TCP_CLOSEd and sk_shutdown set to %d\n",
2130 __func__, transport->inet->sk_shutdown);
2131 }
2132 if ((1 << state) & (TCPF_ESTABLISHED|TCPF_SYN_SENT)) {
2133 /* we don't need to abort the connection if the socket
2134 * hasn't undergone a shutdown
2135 */
2136 if (transport->inet->sk_shutdown == 0)
2137 return;
2138 dprintk("RPC: %s: ESTABLISHED/SYN_SENT "
2139 "sk_shutdown set to %d\n",
2140 __func__, transport->inet->sk_shutdown);
2141 }
58dddac9 2142 xs_abort_connection(transport);
40d2549d
TM		 2143}
2144
16be2d20 2145static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
a246b010 2146{
16be2d20 2147 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
fe19a96b 2148 int ret = -ENOTCONN;
edb267a6 2149
ee0ac0c2 2150 if (!transport->inet) {
b0d93ad5 2151 struct sock *sk = sock->sk;
7f260e85
TM		 2152 unsigned int keepidle = xprt->timeout->to_initval / HZ;
2153 unsigned int keepcnt = xprt->timeout->to_retries + 1;
2154 unsigned int opt_on = 1;
2155
2156 /* TCP Keepalive options */
2157 kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
2158 (char *)&opt_on, sizeof(opt_on));
2159 kernel_setsockopt(sock, SOL_TCP, TCP_KEEPIDLE,
2160 (char *)&keepidle, sizeof(keepidle));
2161 kernel_setsockopt(sock, SOL_TCP, TCP_KEEPINTVL,
2162 (char *)&keepidle, sizeof(keepidle));
2163 kernel_setsockopt(sock, SOL_TCP, TCP_KEEPCNT,
2164 (char *)&keepcnt, sizeof(keepcnt));
b0d93ad5
CL		 2165
2166 write_lock_bh(&sk->sk_callback_lock);
2167
2a9e1cfa
TM		 2168 xs_save_old_callbacks(transport, sk);
2169
b0d93ad5 2170 sk->sk_user_data = xprt;
b0d93ad5
CL		 2171 sk->sk_data_ready = xs_tcp_data_ready;
2172 sk->sk_state_change = xs_tcp_state_change;
2173 sk->sk_write_space = xs_tcp_write_space;
2118071d 2174 sk->sk_error_report = xs_error_report;
b079fa7b 2175 sk->sk_allocation = GFP_ATOMIC;
3167e12c
CL		 2176
2177 /* socket options */
2178 sk->sk_userlocks |= SOCK_BINDPORT_LOCK;
2179 sock_reset_flag(sk, SOCK_LINGER);
2180 tcp_sk(sk)->linger2 = 0;
2181 tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
b0d93ad5
CL		 2182
2183 xprt_clear_connected(xprt);
2184
2185 /* Reset to new socket */
ee0ac0c2
CL		 2186 transport->sock = sock;
2187 transport->inet = sk;
b0d93ad5
CL		 2188
2189 write_unlock_bh(&sk->sk_callback_lock);
2190 }
2191
01d37c42 2192 if (!xprt_bound(xprt))
fe19a96b 2193 goto out;
01d37c42 2194
a564b8f0
MG		 2195 xs_set_memalloc(xprt);
2196
b0d93ad5 2197 /* Tell the socket layer to start connecting... */
262ca07d
CL		 2198 xprt->stat.connect_count++;
2199 xprt->stat.connect_start = jiffies;
fe19a96b
TM		 2200 ret = kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
2201 switch (ret) {
2202 case 0:
2203 case -EINPROGRESS:
2204 /* SYN_SENT! */
fe19a96b
TM		 2205 if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2206 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2207 }
2208out:
2209 return ret;
16be2d20
CL		 2210}
2211
9903cd1c 2212/**
b61d59ff
TM		 2213 * xs_tcp_setup_socket - create a TCP socket and connect to a remote endpoint
2214 * @xprt: RPC transport to connect
2215 * @transport: socket transport to connect
2216 * @create_sock: function to create a socket of the correct type
9903cd1c
CL		 2217 *
2218 * Invoked by a work queue tasklet.
a246b010 2219 */
cdd518d5 2220static void xs_tcp_setup_socket(struct work_struct *work)
a246b010 2221{
cdd518d5
PE		 2222 struct sock_xprt *transport =
2223 container_of(work, struct sock_xprt, connect_worker.work);
ee0ac0c2 2224 struct socket *sock = transport->sock;
a9f5f0f7 2225 struct rpc_xprt *xprt = &transport->xprt;
b61d59ff 2226 int status = -EIO;
a246b010 2227
5cf02d09
JL		 2228 current->flags |= PF_FSTRANS;
2229
ee0ac0c2 2230 if (!sock) {
7d1e8255 2231 clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
cdd518d5
PE		 2232 sock = xs_create_sock(xprt, transport,
2233 xs_addr(xprt)->sa_family, SOCK_STREAM, IPPROTO_TCP);
b61d59ff
TM		 2234 if (IS_ERR(sock)) {
2235 status = PTR_ERR(sock);
3167e12c
CL		 2236 goto out;
2237 }
7d1e8255
TM		 2238 } else {
2239 int abort_and_exit;
a246b010 2240
7d1e8255
TM		 2241 abort_and_exit = test_and_clear_bit(XPRT_CONNECTION_ABORT,
2242 &xprt->state);
3167e12c 2243 /* "close" the socket, preserving the local port */
58dddac9 2244 xs_tcp_reuse_connection(transport);
a246b010 2245
7d1e8255
TM		 2246 if (abort_and_exit)
2247 goto out_eagain;
2248 }
a246b010 2249
c740eff8
CL		 2250 dprintk("RPC: worker connecting xprt %p via %s to "
2251 "%s (port %s)\n", xprt,
2252 xprt->address_strings[RPC_DISPLAY_PROTO],
2253 xprt->address_strings[RPC_DISPLAY_ADDR],
2254 xprt->address_strings[RPC_DISPLAY_PORT]);
edb267a6 2255
16be2d20 2256 status = xs_tcp_finish_connecting(xprt, sock);
40b5ea0c 2257 trace_rpc_socket_connect(xprt, sock, status);
46121cf7
CL		 2258 dprintk("RPC: %p connect status %d connected %d sock state %d\n",
2259 xprt, -status, xprt_connected(xprt),
2260 sock->sk->sk_state);
2a491991 2261 switch (status) {
f75e6745
TM		 2262 default:
2263 printk("%s: connect returned unhandled error %d\n",
2264 __func__, status);
2265 case -EADDRNOTAVAIL:
2266 /* We're probably in TIME_WAIT. Get rid of existing socket,
2267 * and retry
2268 */
a519fc7a 2269 xs_tcp_force_close(xprt);
88b5ed73 2270 break;
2a491991
TM		 2271 case 0:
2272 case -EINPROGRESS:
2273 case -EALREADY:
7d1e8255 2274 xprt_clear_connecting(xprt);
5cf02d09 2275 current->flags &= ~PF_FSTRANS;
7d1e8255 2276 return;
9fcfe0c8
TM		 2277 case -EINVAL:
2278 /* Happens, for instance, if the user specified a link
2279 * local IPv6 address without a scope-id.
2280 */
3ed5e2a2
TM		 2281 case -ECONNREFUSED:
2282 case -ECONNRESET:
2283 case -ENETUNREACH:
2284 /* retry with existing socket, after a delay */
9fcfe0c8 2285 goto out;
a246b010 2286 }
7d1e8255 2287out_eagain:
2a491991 2288 status = -EAGAIN;
a246b010 2289out:
68e220bd 2290 xprt_clear_connecting(xprt);
7d1e8255 2291 xprt_wake_pending_tasks(xprt, status);
5cf02d09 2292 current->flags &= ~PF_FSTRANS;
68e220bd 2293}
b0d93ad5 2294
9903cd1c
CL		 2295/**
2296 * xs_connect - connect a socket to a remote endpoint
1b092092 2297 * @xprt: pointer to transport structure
9903cd1c
CL		 2298 * @task: address of RPC task that manages state of connect request
2299 *
2300 * TCP: If the remote end dropped the connection, delay reconnecting.
03bf4b70
CL		 2301 *
2302 * UDP socket connects are synchronous, but we use a work queue anyway
2303 * to guarantee that even unprivileged user processes can set up a
2304 * socket on a privileged port.
2305 *
2306 * If a UDP socket connect fails, the delay behavior here prevents
2307 * retry floods (hard mounts).
9903cd1c 2308 */
1b092092 2309static void xs_connect(struct rpc_xprt *xprt, struct rpc_task *task)
a246b010 2310{
ee0ac0c2 2311 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
a246b010 2312
09a21c41 2313 if (transport->sock != NULL && !RPC_IS_SOFTCONN(task)) {
46121cf7
CL		 2314 dprintk("RPC: xs_connect delayed xprt %p for %lu "
2315 "seconds\n",
03bf4b70 2316 xprt, xprt->reestablish_timeout / HZ);
c1384c9c
TM		 2317 queue_delayed_work(rpciod_workqueue,
2318 &transport->connect_worker,
2319 xprt->reestablish_timeout);
03bf4b70 2320 xprt->reestablish_timeout <<= 1;
61d0a8e6
NB		 2321 if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2322 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
03bf4b70
CL		 2323 if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
2324 xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
b0d93ad5 2325 } else {
46121cf7 2326 dprintk("RPC: xs_connect scheduled xprt %p\n", xprt);
c1384c9c
TM		 2327 queue_delayed_work(rpciod_workqueue,
2328 &transport->connect_worker, 0);
a246b010
CL		 2329 }
2330}
2331
176e21ee
CL		 2332/**
2333 * xs_local_print_stats - display AF_LOCAL socket-specifc stats
2334 * @xprt: rpc_xprt struct containing statistics
2335 * @seq: output file
2336 *
2337 */
2338static void xs_local_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2339{
2340 long idle_time = 0;
2341
2342 if (xprt_connected(xprt))
2343 idle_time = (long)(jiffies - xprt->last_used) / HZ;
2344
2345 seq_printf(seq, "\txprt:\tlocal %lu %lu %lu %ld %lu %lu %lu "
15a45206 2346 "%llu %llu %lu %llu %llu\n",
176e21ee
CL		 2347 xprt->stat.bind_count,
2348 xprt->stat.connect_count,
2349 xprt->stat.connect_time,
2350 idle_time,
2351 xprt->stat.sends,
2352 xprt->stat.recvs,
2353 xprt->stat.bad_xids,
2354 xprt->stat.req_u,
15a45206
AA		 2355 xprt->stat.bklog_u,
2356 xprt->stat.max_slots,
2357 xprt->stat.sending_u,
2358 xprt->stat.pending_u);
176e21ee
CL		 2359}
2360
262ca07d
CL		 2361/**
2362 * xs_udp_print_stats - display UDP socket-specifc stats
2363 * @xprt: rpc_xprt struct containing statistics
2364 * @seq: output file
2365 *
2366 */
2367static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2368{
c8475461
CL		 2369 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2370
15a45206
AA		 2371 seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %llu %llu "
2372 "%lu %llu %llu\n",
fbfffbd5 2373 transport->srcport,
262ca07d
CL		 2374 xprt->stat.bind_count,
2375 xprt->stat.sends,
2376 xprt->stat.recvs,
2377 xprt->stat.bad_xids,
2378 xprt->stat.req_u,
15a45206
AA		 2379 xprt->stat.bklog_u,
2380 xprt->stat.max_slots,
2381 xprt->stat.sending_u,
2382 xprt->stat.pending_u);
262ca07d
CL		 2383}
2384
2385/**
2386 * xs_tcp_print_stats - display TCP socket-specifc stats
2387 * @xprt: rpc_xprt struct containing statistics
2388 * @seq: output file
2389 *
2390 */
2391static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2392{
c8475461 2393 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
262ca07d
CL		 2394 long idle_time = 0;
2395
2396 if (xprt_connected(xprt))
2397 idle_time = (long)(jiffies - xprt->last_used) / HZ;
2398
15a45206
AA		 2399 seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu "
2400 "%llu %llu %lu %llu %llu\n",
fbfffbd5 2401 transport->srcport,
262ca07d
CL		 2402 xprt->stat.bind_count,
2403 xprt->stat.connect_count,
2404 xprt->stat.connect_time,
2405 idle_time,
2406 xprt->stat.sends,
2407 xprt->stat.recvs,
2408 xprt->stat.bad_xids,
2409 xprt->stat.req_u,
15a45206
AA		 2410 xprt->stat.bklog_u,
2411 xprt->stat.max_slots,
2412 xprt->stat.sending_u,
2413 xprt->stat.pending_u);
262ca07d
CL		 2414}
2415
4cfc7e60
RI		 2416/*
2417 * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
2418 * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
2419 * to use the server side send routines.
2420 */
5a51f13a 2421static void *bc_malloc(struct rpc_task *task, size_t size)
4cfc7e60
RI		 2422{
2423 struct page *page;
2424 struct rpc_buffer *buf;
2425
b8a13d03
WAA		 2426 WARN_ON_ONCE(size > PAGE_SIZE - sizeof(struct rpc_buffer));
2427 if (size > PAGE_SIZE - sizeof(struct rpc_buffer))
2428 return NULL;
4cfc7e60 2429
b8a13d03 2430 page = alloc_page(GFP_KERNEL);
4cfc7e60
RI		 2431 if (!page)
2432 return NULL;
2433
2434 buf = page_address(page);
2435 buf->len = PAGE_SIZE;
2436
2437 return buf->data;
2438}
2439
2440/*
2441 * Free the space allocated in the bc_alloc routine
2442 */
5a51f13a 2443static void bc_free(void *buffer)
4cfc7e60
RI		 2444{
2445 struct rpc_buffer *buf;
2446
2447 if (!buffer)
2448 return;
2449
2450 buf = container_of(buffer, struct rpc_buffer, data);
2451 free_page((unsigned long)buf);
2452}
2453
2454/*
2455 * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
2456 * held. Borrows heavily from svc_tcp_sendto and xs_tcp_send_request.
2457 */
2458static int bc_sendto(struct rpc_rqst *req)
2459{
2460 int len;
2461 struct xdr_buf *xbufp = &req->rq_snd_buf;
2462 struct rpc_xprt *xprt = req->rq_xprt;
2463 struct sock_xprt *transport =
2464 container_of(xprt, struct sock_xprt, xprt);
2465 struct socket *sock = transport->sock;
2466 unsigned long headoff;
2467 unsigned long tailoff;
2468
61677eee 2469 xs_encode_stream_record_marker(xbufp);
4cfc7e60
RI		 2470
2471 tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
2472 headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
2473 len = svc_send_common(sock, xbufp,
2474 virt_to_page(xbufp->head[0].iov_base), headoff,
2475 xbufp->tail[0].iov_base, tailoff);
2476
2477 if (len != xbufp->len) {
2478 printk(KERN_NOTICE "Error sending entire callback!\n");
2479 len = -EAGAIN;
2480 }
2481
2482 return len;
2483}
2484
2485/*
2486 * The send routine. Borrows from svc_send
2487 */
2488static int bc_send_request(struct rpc_task *task)
2489{
2490 struct rpc_rqst *req = task->tk_rqstp;
2491 struct svc_xprt *xprt;
4cfc7e60
RI		 2492 u32 len;
2493
2494 dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
2495 /*
2496 * Get the server socket associated with this callback xprt
2497 */
2498 xprt = req->rq_xprt->bc_xprt;
4cfc7e60
RI		 2499
2500 /*
2501 * Grab the mutex to serialize data as the connection is shared
2502 * with the fore channel
2503 */
2504 if (!mutex_trylock(&xprt->xpt_mutex)) {
2505 rpc_sleep_on(&xprt->xpt_bc_pending, task, NULL);
2506 if (!mutex_trylock(&xprt->xpt_mutex))
2507 return -EAGAIN;
2508 rpc_wake_up_queued_task(&xprt->xpt_bc_pending, task);
2509 }
2510 if (test_bit(XPT_DEAD, &xprt->xpt_flags))
2511 len = -ENOTCONN;
2512 else
2513 len = bc_sendto(req);
2514 mutex_unlock(&xprt->xpt_mutex);
2515
2516 if (len > 0)
2517 len = 0;
2518
2519 return len;
2520}
2521
2522/*
2523 * The close routine. Since this is client initiated, we do nothing
2524 */
2525
2526static void bc_close(struct rpc_xprt *xprt)
2527{
4cfc7e60
RI		 2528}
2529
2530/*
2531 * The xprt destroy routine. Again, because this connection is client
2532 * initiated, we do nothing
2533 */
2534
2535static void bc_destroy(struct rpc_xprt *xprt)
2536{
47f72efa
KM		 2537 dprintk("RPC: bc_destroy xprt %p\n", xprt);
2538
2539 xs_xprt_free(xprt);
2540 module_put(THIS_MODULE);
4cfc7e60
RI		 2541}
2542
176e21ee
CL		 2543static struct rpc_xprt_ops xs_local_ops = {
2544 .reserve_xprt = xprt_reserve_xprt,
2545 .release_xprt = xs_tcp_release_xprt,
f39c1bfb 2546 .alloc_slot = xprt_alloc_slot,
176e21ee
CL		 2547 .rpcbind = xs_local_rpcbind,
2548 .set_port = xs_local_set_port,
dc107402 2549 .connect = xs_local_connect,
176e21ee
CL		 2550 .buf_alloc = rpc_malloc,
2551 .buf_free = rpc_free,
2552 .send_request = xs_local_send_request,
2553 .set_retrans_timeout = xprt_set_retrans_timeout_def,
2554 .close = xs_close,
a1311d87 2555 .destroy = xs_destroy,
176e21ee
CL		 2556 .print_stats = xs_local_print_stats,
2557};
2558
262965f5 2559static struct rpc_xprt_ops xs_udp_ops = {
43118c29 2560 .set_buffer_size = xs_udp_set_buffer_size,
12a80469 2561 .reserve_xprt = xprt_reserve_xprt_cong,
49e9a890 2562 .release_xprt = xprt_release_xprt_cong,
f39c1bfb 2563 .alloc_slot = xprt_alloc_slot,
45160d62 2564 .rpcbind = rpcb_getport_async,
92200412 2565 .set_port = xs_set_port,
262965f5 2566 .connect = xs_connect,
02107148
CL		 2567 .buf_alloc = rpc_malloc,
2568 .buf_free = rpc_free,
262965f5 2569 .send_request = xs_udp_send_request,
fe3aca29 2570 .set_retrans_timeout = xprt_set_retrans_timeout_rtt,
46c0ee8b 2571 .timer = xs_udp_timer,
a58dd398 2572 .release_request = xprt_release_rqst_cong,
262965f5
CL		 2573 .close = xs_close,
2574 .destroy = xs_destroy,
262ca07d 2575 .print_stats = xs_udp_print_stats,
262965f5
CL		 2576};
2577
2578static struct rpc_xprt_ops xs_tcp_ops = {
12a80469 2579 .reserve_xprt = xprt_reserve_xprt,
e0ab53de 2580 .release_xprt = xs_tcp_release_xprt,
f39c1bfb 2581 .alloc_slot = xprt_lock_and_alloc_slot,
45160d62 2582 .rpcbind = rpcb_getport_async,
92200412 2583 .set_port = xs_set_port,
0b9e7943 2584 .connect = xs_connect,
02107148
CL		 2585 .buf_alloc = rpc_malloc,
2586 .buf_free = rpc_free,
262965f5 2587 .send_request = xs_tcp_send_request,
fe3aca29 2588 .set_retrans_timeout = xprt_set_retrans_timeout_def,
f75e6745 2589 .close = xs_tcp_close,
9903cd1c 2590 .destroy = xs_destroy,
262ca07d 2591 .print_stats = xs_tcp_print_stats,
a246b010
CL		 2592};
2593
4cfc7e60
RI		 2594/*
2595 * The rpc_xprt_ops for the server backchannel
2596 */
2597
2598static struct rpc_xprt_ops bc_tcp_ops = {
2599 .reserve_xprt = xprt_reserve_xprt,
2600 .release_xprt = xprt_release_xprt,
84e28a30 2601 .alloc_slot = xprt_alloc_slot,
4cfc7e60
RI		 2602 .buf_alloc = bc_malloc,
2603 .buf_free = bc_free,
2604 .send_request = bc_send_request,
2605 .set_retrans_timeout = xprt_set_retrans_timeout_def,
2606 .close = bc_close,
2607 .destroy = bc_destroy,
2608 .print_stats = xs_tcp_print_stats,
2609};
2610
92476850
CL		 2611static int xs_init_anyaddr(const int family, struct sockaddr *sap)
2612{
2613 static const struct sockaddr_in sin = {
2614 .sin_family = AF_INET,
2615 .sin_addr.s_addr = htonl(INADDR_ANY),
2616 };
2617 static const struct sockaddr_in6 sin6 = {
2618 .sin6_family = AF_INET6,
2619 .sin6_addr = IN6ADDR_ANY_INIT,
2620 };
2621
2622 switch (family) {
176e21ee
CL		 2623 case AF_LOCAL:
2624 break;
92476850
CL		 2625 case AF_INET:
2626 memcpy(sap, &sin, sizeof(sin));
2627 break;
2628 case AF_INET6:
2629 memcpy(sap, &sin6, sizeof(sin6));
2630 break;
2631 default:
2632 dprintk("RPC: %s: Bad address family\n", __func__);
2633 return -EAFNOSUPPORT;
2634 }
2635 return 0;
2636}
2637
3c341b0b 2638static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
d9ba131d
TM		 2639 unsigned int slot_table_size,
2640 unsigned int max_slot_table_size)
c8541ecd
CL		 2641{
2642 struct rpc_xprt *xprt;
ffc2e518 2643 struct sock_xprt *new;
c8541ecd 2644
96802a09 2645 if (args->addrlen > sizeof(xprt->addr)) {
46121cf7 2646 dprintk("RPC: xs_setup_xprt: address too large\n");
c8541ecd
CL		 2647 return ERR_PTR(-EBADF);
2648 }
2649
d9ba131d
TM		 2650 xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
2651 max_slot_table_size);
bd1722d4 2652 if (xprt == NULL) {
46121cf7
CL		 2653 dprintk("RPC: xs_setup_xprt: couldn't allocate "
2654 "rpc_xprt\n");
c8541ecd
CL		 2655 return ERR_PTR(-ENOMEM);
2656 }
c8541ecd 2657
bd1722d4 2658 new = container_of(xprt, struct sock_xprt, xprt);
96802a09
FM		 2659 memcpy(&xprt->addr, args->dstaddr, args->addrlen);
2660 xprt->addrlen = args->addrlen;
d3bc9a1d 2661 if (args->srcaddr)
fbfffbd5 2662 memcpy(&new->srcaddr, args->srcaddr, args->addrlen);
92476850
CL		 2663 else {
2664 int err;
2665 err = xs_init_anyaddr(args->dstaddr->sa_family,
2666 (struct sockaddr *)&new->srcaddr);
2aa13531
SK		 2667 if (err != 0) {
2668 xprt_free(xprt);
92476850 2669 return ERR_PTR(err);
2aa13531 2670 }
92476850 2671 }
c8541ecd
CL		 2672
2673 return xprt;
2674}
2675
176e21ee
CL		 2676static const struct rpc_timeout xs_local_default_timeout = {
2677 .to_initval = 10 * HZ,
2678 .to_maxval = 10 * HZ,
2679 .to_retries = 2,
2680};
2681
2682/**
2683 * xs_setup_local - Set up transport to use an AF_LOCAL socket
2684 * @args: rpc transport creation arguments
2685 *
2686 * AF_LOCAL is a "tpi_cots_ord" transport, just like TCP
2687 */
2688static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
2689{
2690 struct sockaddr_un *sun = (struct sockaddr_un *)args->dstaddr;
2691 struct sock_xprt *transport;
2692 struct rpc_xprt *xprt;
2693 struct rpc_xprt *ret;
2694
d9ba131d
TM		 2695 xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
2696 xprt_max_tcp_slot_table_entries);
176e21ee
CL		 2697 if (IS_ERR(xprt))
2698 return xprt;
2699 transport = container_of(xprt, struct sock_xprt, xprt);
2700
2701 xprt->prot = 0;
2702 xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2703 xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2704
2705 xprt->bind_timeout = XS_BIND_TO;
2706 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2707 xprt->idle_timeout = XS_IDLE_DISC_TO;
2708
2709 xprt->ops = &xs_local_ops;
2710 xprt->timeout = &xs_local_default_timeout;
2711
93dc41bd
N		 2712 INIT_DELAYED_WORK(&transport->connect_worker,
2713 xs_dummy_setup_socket);
2714
176e21ee
CL		 2715 switch (sun->sun_family) {
2716 case AF_LOCAL:
2717 if (sun->sun_path[0] != '/') {
2718 dprintk("RPC: bad AF_LOCAL address: %s\n",
2719 sun->sun_path);
2720 ret = ERR_PTR(-EINVAL);
2721 goto out_err;
2722 }
2723 xprt_set_bound(xprt);
176e21ee 2724 xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
7073ea87
BF		 2725 ret = ERR_PTR(xs_local_setup_socket(transport));
2726 if (ret)
2727 goto out_err;
176e21ee
CL		 2728 break;
2729 default:
2730 ret = ERR_PTR(-EAFNOSUPPORT);
2731 goto out_err;
2732 }
2733
2734 dprintk("RPC: set up xprt to %s via AF_LOCAL\n",
2735 xprt->address_strings[RPC_DISPLAY_ADDR]);
2736
2737 if (try_module_get(THIS_MODULE))
2738 return xprt;
2739 ret = ERR_PTR(-EINVAL);
2740out_err:
315f3812 2741 xs_xprt_free(xprt);
176e21ee
CL		 2742 return ret;
2743}
2744
2881ae74
TM		 2745static const struct rpc_timeout xs_udp_default_timeout = {
2746 .to_initval = 5 * HZ,
2747 .to_maxval = 30 * HZ,
2748 .to_increment = 5 * HZ,
2749 .to_retries = 5,
2750};
2751
9903cd1c
CL		 2752/**
2753 * xs_setup_udp - Set up transport to use a UDP socket
96802a09 2754 * @args: rpc transport creation arguments
9903cd1c
CL		 2755 *
2756 */
483066d6 2757static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
a246b010 2758{
8f9d5b1a 2759 struct sockaddr *addr = args->dstaddr;
c8541ecd 2760 struct rpc_xprt *xprt;
c8475461 2761 struct sock_xprt *transport;
0a68b0be 2762 struct rpc_xprt *ret;
a246b010 2763
d9ba131d
TM		 2764 xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
2765 xprt_udp_slot_table_entries);
c8541ecd
CL		 2766 if (IS_ERR(xprt))
2767 return xprt;
c8475461 2768 transport = container_of(xprt, struct sock_xprt, xprt);
a246b010 2769
ec739ef0 2770 xprt->prot = IPPROTO_UDP;
808012fb 2771 xprt->tsh_size = 0;
a246b010
CL		 2772 /* XXX: header size can vary due to auth type, IPv6, etc. */
2773 xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
2774
03bf4b70 2775 xprt->bind_timeout = XS_BIND_TO;
03bf4b70
CL		 2776 xprt->reestablish_timeout = XS_UDP_REEST_TO;
2777 xprt->idle_timeout = XS_IDLE_DISC_TO;
a246b010 2778
262965f5 2779 xprt->ops = &xs_udp_ops;
a246b010 2780
ba7392bb 2781 xprt->timeout = &xs_udp_default_timeout;
a246b010 2782
8f9d5b1a
CL		 2783 switch (addr->sa_family) {
2784 case AF_INET:
2785 if (((struct sockaddr_in *)addr)->sin_port != htons(0))
2786 xprt_set_bound(xprt);
2787
2788 INIT_DELAYED_WORK(&transport->connect_worker,
8c14ff2a 2789 xs_udp_setup_socket);
9dc3b095 2790 xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
8f9d5b1a
CL		 2791 break;
2792 case AF_INET6:
2793 if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
2794 xprt_set_bound(xprt);
2795
2796 INIT_DELAYED_WORK(&transport->connect_worker,
8c14ff2a 2797 xs_udp_setup_socket);
9dc3b095 2798 xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
8f9d5b1a
CL		 2799 break;
2800 default:
0a68b0be
BF		 2801 ret = ERR_PTR(-EAFNOSUPPORT);
2802 goto out_err;
8f9d5b1a
CL		 2803 }
2804
c740eff8
CL		 2805 if (xprt_bound(xprt))
2806 dprintk("RPC: set up xprt to %s (port %s) via %s\n",
2807 xprt->address_strings[RPC_DISPLAY_ADDR],
2808 xprt->address_strings[RPC_DISPLAY_PORT],
2809 xprt->address_strings[RPC_DISPLAY_PROTO]);
2810 else
2811 dprintk("RPC: set up xprt to %s (autobind) via %s\n",
2812 xprt->address_strings[RPC_DISPLAY_ADDR],
2813 xprt->address_strings[RPC_DISPLAY_PROTO]);
edb267a6 2814
bc25571e
TT		 2815 if (try_module_get(THIS_MODULE))
2816 return xprt;
0a68b0be
BF		 2817 ret = ERR_PTR(-EINVAL);
2818out_err:
315f3812 2819 xs_xprt_free(xprt);
0a68b0be 2820 return ret;
a246b010
CL		 2821}
2822
2881ae74
TM		 2823static const struct rpc_timeout xs_tcp_default_timeout = {
2824 .to_initval = 60 * HZ,
2825 .to_maxval = 60 * HZ,
2826 .to_retries = 2,
2827};
2828
9903cd1c
CL		 2829/**
2830 * xs_setup_tcp - Set up transport to use a TCP socket
96802a09 2831 * @args: rpc transport creation arguments
9903cd1c
CL		 2832 *
2833 */
483066d6 2834static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
a246b010 2835{
8f9d5b1a 2836 struct sockaddr *addr = args->dstaddr;
c8541ecd 2837 struct rpc_xprt *xprt;
c8475461 2838 struct sock_xprt *transport;
0a68b0be 2839 struct rpc_xprt *ret;
b7993ceb
TM		 2840 unsigned int max_slot_table_size = xprt_max_tcp_slot_table_entries;
2841
2842 if (args->flags & XPRT_CREATE_INFINITE_SLOTS)
2843 max_slot_table_size = RPC_MAX_SLOT_TABLE_LIMIT;
a246b010 2844
d9ba131d 2845 xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
b7993ceb 2846 max_slot_table_size);
c8541ecd
CL		 2847 if (IS_ERR(xprt))
2848 return xprt;
c8475461 2849 transport = container_of(xprt, struct sock_xprt, xprt);
a246b010 2850
ec739ef0 2851 xprt->prot = IPPROTO_TCP;
808012fb 2852 xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
808012fb 2853 xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
a246b010 2854
03bf4b70 2855 xprt->bind_timeout = XS_BIND_TO;
03bf4b70
CL		 2856 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2857 xprt->idle_timeout = XS_IDLE_DISC_TO;
a246b010 2858
262965f5 2859 xprt->ops = &xs_tcp_ops;
ba7392bb 2860 xprt->timeout = &xs_tcp_default_timeout;
a246b010 2861
8f9d5b1a
CL		 2862 switch (addr->sa_family) {
2863 case AF_INET:
2864 if (((struct sockaddr_in *)addr)->sin_port != htons(0))
2865 xprt_set_bound(xprt);
2866
9dc3b095 2867 INIT_DELAYED_WORK(&transport->connect_worker,
cdd518d5 2868 xs_tcp_setup_socket);
9dc3b095 2869 xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
8f9d5b1a
CL		 2870 break;
2871 case AF_INET6:
2872 if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
2873 xprt_set_bound(xprt);
2874
9dc3b095 2875 INIT_DELAYED_WORK(&transport->connect_worker,
cdd518d5 2876 xs_tcp_setup_socket);
9dc3b095 2877 xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
8f9d5b1a
CL		 2878 break;
2879 default:
0a68b0be
BF		 2880 ret = ERR_PTR(-EAFNOSUPPORT);
2881 goto out_err;
8f9d5b1a
CL		 2882 }
2883
c740eff8
CL		 2884 if (xprt_bound(xprt))
2885 dprintk("RPC: set up xprt to %s (port %s) via %s\n",
2886 xprt->address_strings[RPC_DISPLAY_ADDR],
2887 xprt->address_strings[RPC_DISPLAY_PORT],
2888 xprt->address_strings[RPC_DISPLAY_PROTO]);
2889 else
2890 dprintk("RPC: set up xprt to %s (autobind) via %s\n",
2891 xprt->address_strings[RPC_DISPLAY_ADDR],
2892 xprt->address_strings[RPC_DISPLAY_PROTO]);
2893
bc25571e
TT		 2894 if (try_module_get(THIS_MODULE))
2895 return xprt;
0a68b0be
BF		 2896 ret = ERR_PTR(-EINVAL);
2897out_err:
315f3812 2898 xs_xprt_free(xprt);
0a68b0be 2899 return ret;
a246b010 2900}
282b32e1 2901
f300baba
AB		 2902/**
2903 * xs_setup_bc_tcp - Set up transport to use a TCP backchannel socket
2904 * @args: rpc transport creation arguments
2905 *
2906 */
2907static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
2908{
2909 struct sockaddr *addr = args->dstaddr;
2910 struct rpc_xprt *xprt;
2911 struct sock_xprt *transport;
2912 struct svc_sock *bc_sock;
0a68b0be 2913 struct rpc_xprt *ret;
f300baba 2914
d9ba131d
TM		 2915 xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
2916 xprt_tcp_slot_table_entries);
f300baba
AB		 2917 if (IS_ERR(xprt))
2918 return xprt;
2919 transport = container_of(xprt, struct sock_xprt, xprt);
2920
2921 xprt->prot = IPPROTO_TCP;
2922 xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2923 xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2924 xprt->timeout = &xs_tcp_default_timeout;
2925
2926 /* backchannel */
2927 xprt_set_bound(xprt);
2928 xprt->bind_timeout = 0;
f300baba
AB		 2929 xprt->reestablish_timeout = 0;
2930 xprt->idle_timeout = 0;
2931
f300baba
AB		 2932 xprt->ops = &bc_tcp_ops;
2933
2934 switch (addr->sa_family) {
2935 case AF_INET:
2936 xs_format_peer_addresses(xprt, "tcp",
2937 RPCBIND_NETID_TCP);
2938 break;
2939 case AF_INET6:
2940 xs_format_peer_addresses(xprt, "tcp",
2941 RPCBIND_NETID_TCP6);
2942 break;
2943 default:
0a68b0be
BF		 2944 ret = ERR_PTR(-EAFNOSUPPORT);
2945 goto out_err;
f300baba
AB		 2946 }
2947
50fa0d40
PE		 2948 dprintk("RPC: set up xprt to %s (port %s) via %s\n",
2949 xprt->address_strings[RPC_DISPLAY_ADDR],
2950 xprt->address_strings[RPC_DISPLAY_PORT],
2951 xprt->address_strings[RPC_DISPLAY_PROTO]);
f300baba 2952
99de8ea9
BF		 2953 /*
2954 * Once we've associated a backchannel xprt with a connection,
28303ca3
WM		 2955 * we want to keep it around as long as the connection lasts,
2956 * in case we need to start using it for a backchannel again;
2957 * this reference won't be dropped until bc_xprt is destroyed.
99de8ea9
BF		 2958 */
2959 xprt_get(xprt);
2960 args->bc_xprt->xpt_bc_xprt = xprt;
2961 xprt->bc_xprt = args->bc_xprt;
2962 bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
2963 transport->sock = bc_sock->sk_sock;
2964 transport->inet = bc_sock->sk_sk;
2965
f300baba
AB		 2966 /*
2967 * Since we don't want connections for the backchannel, we set
2968 * the xprt status to connected
2969 */
2970 xprt_set_connected(xprt);
2971
f300baba
AB		 2972 if (try_module_get(THIS_MODULE))
2973 return xprt;
642aab58
KM		 2974
2975 args->bc_xprt->xpt_bc_xprt = NULL;
99de8ea9 2976 xprt_put(xprt);
0a68b0be
BF		 2977 ret = ERR_PTR(-EINVAL);
2978out_err:
315f3812 2979 xs_xprt_free(xprt);
0a68b0be 2980 return ret;
f300baba
AB		 2981}
2982
176e21ee
CL		 2983static struct xprt_class xs_local_transport = {
2984 .list = LIST_HEAD_INIT(xs_local_transport.list),
2985 .name = "named UNIX socket",
2986 .owner = THIS_MODULE,
2987 .ident = XPRT_TRANSPORT_LOCAL,
2988 .setup = xs_setup_local,
2989};
2990
bc25571e
TT		 2991static struct xprt_class xs_udp_transport = {
2992 .list = LIST_HEAD_INIT(xs_udp_transport.list),
2993 .name = "udp",
2994 .owner = THIS_MODULE,
f300baba 2995 .ident = XPRT_TRANSPORT_UDP,
bc25571e
TT		 2996 .setup = xs_setup_udp,
2997};
2998
2999static struct xprt_class xs_tcp_transport = {
3000 .list = LIST_HEAD_INIT(xs_tcp_transport.list),
3001 .name = "tcp",
3002 .owner = THIS_MODULE,
f300baba 3003 .ident = XPRT_TRANSPORT_TCP,
bc25571e
TT		 3004 .setup = xs_setup_tcp,
3005};
3006
f300baba
AB		 3007static struct xprt_class xs_bc_tcp_transport = {
3008 .list = LIST_HEAD_INIT(xs_bc_tcp_transport.list),
3009 .name = "tcp NFSv4.1 backchannel",
3010 .owner = THIS_MODULE,
3011 .ident = XPRT_TRANSPORT_BC_TCP,
3012 .setup = xs_setup_bc_tcp,
3013};
3014
282b32e1 3015/**
bc25571e 3016 * init_socket_xprt - set up xprtsock's sysctls, register with RPC client
282b32e1
CL		 3017 *
3018 */
3019int init_socket_xprt(void)
3020{
fbf76683 3021#ifdef RPC_DEBUG
2b1bec5f 3022 if (!sunrpc_table_header)
0b4d4147 3023 sunrpc_table_header = register_sysctl_table(sunrpc_table);
fbf76683
CL		 3024#endif
3025
176e21ee 3026 xprt_register_transport(&xs_local_transport);
bc25571e
TT		 3027 xprt_register_transport(&xs_udp_transport);
3028 xprt_register_transport(&xs_tcp_transport);
f300baba 3029 xprt_register_transport(&xs_bc_tcp_transport);
bc25571e 3030
282b32e1
CL		 3031 return 0;
3032}
3033
3034/**
bc25571e 3035 * cleanup_socket_xprt - remove xprtsock's sysctls, unregister
282b32e1
CL		 3036 *
3037 */
3038void cleanup_socket_xprt(void)
3039{
fbf76683
CL		 3040#ifdef RPC_DEBUG
3041 if (sunrpc_table_header) {
3042 unregister_sysctl_table(sunrpc_table_header);
3043 sunrpc_table_header = NULL;
3044 }
3045#endif
bc25571e 3046
176e21ee 3047 xprt_unregister_transport(&xs_local_transport);
bc25571e
TT		 3048 xprt_unregister_transport(&xs_udp_transport);
3049 xprt_unregister_transport(&xs_tcp_transport);
f300baba 3050 xprt_unregister_transport(&xs_bc_tcp_transport);
282b32e1 3051}
cbf11071 3052
9bbb9e5a
RR		 3053static int param_set_uint_minmax(const char *val,
3054 const struct kernel_param *kp,
cbf11071
TM		 3055 unsigned int min, unsigned int max)
3056{
3057 unsigned long num;
3058 int ret;
3059
3060 if (!val)
3061 return -EINVAL;
3062 ret = strict_strtoul(val, 0, &num);
3063 if (ret == -EINVAL || num < min || num > max)
3064 return -EINVAL;
3065 *((unsigned int *)kp->arg) = num;
3066 return 0;
3067}
3068
9bbb9e5a 3069static int param_set_portnr(const char *val, const struct kernel_param *kp)
cbf11071
TM		 3070{
3071 return param_set_uint_minmax(val, kp,
3072 RPC_MIN_RESVPORT,
3073 RPC_MAX_RESVPORT);
3074}
3075
9bbb9e5a
RR		 3076static struct kernel_param_ops param_ops_portnr = {
3077 .set = param_set_portnr,
3078 .get = param_get_uint,
3079};
3080
cbf11071
TM		 3081#define param_check_portnr(name, p) \
3082 __param_check(name, p, unsigned int);
3083
3084module_param_named(min_resvport, xprt_min_resvport, portnr, 0644);
3085module_param_named(max_resvport, xprt_max_resvport, portnr, 0644);
3086
9bbb9e5a
RR		 3087static int param_set_slot_table_size(const char *val,
3088 const struct kernel_param *kp)
cbf11071
TM		 3089{
3090 return param_set_uint_minmax(val, kp,
3091 RPC_MIN_SLOT_TABLE,
3092 RPC_MAX_SLOT_TABLE);
3093}
3094
9bbb9e5a
RR		 3095static struct kernel_param_ops param_ops_slot_table_size = {
3096 .set = param_set_slot_table_size,
3097 .get = param_get_uint,
3098};
3099
cbf11071
TM		 3100#define param_check_slot_table_size(name, p) \
3101 __param_check(name, p, unsigned int);
3102
d9ba131d
TM		 3103static int param_set_max_slot_table_size(const char *val,
3104 const struct kernel_param *kp)
3105{
3106 return param_set_uint_minmax(val, kp,
3107 RPC_MIN_SLOT_TABLE,
3108 RPC_MAX_SLOT_TABLE_LIMIT);
3109}
3110
3111static struct kernel_param_ops param_ops_max_slot_table_size = {
3112 .set = param_set_max_slot_table_size,
3113 .get = param_get_uint,
3114};
3115
3116#define param_check_max_slot_table_size(name, p) \
3117 __param_check(name, p, unsigned int);
3118
cbf11071
TM		 3119module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
3120 slot_table_size, 0644);
d9ba131d
TM		 3121module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
3122 max_slot_table_size, 0644);
cbf11071
TM		 3123module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
3124 slot_table_size, 0644);
3125
Linux kernel - Deliverables
This page took 1.439308 seconds and 5 git commands to generate.