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