xref: /openbmc/linux/net/sunrpc/xprtsock.c (revision d6c686c0)
1 // SPDX-License-Identifier: GPL-2.0
2 /*
3  * linux/net/sunrpc/xprtsock.c
4  *
5  * Client-side transport implementation for sockets.
6  *
7  * TCP callback races fixes (C) 1998 Red Hat
8  * TCP send fixes (C) 1998 Red Hat
9  * TCP NFS related read + write fixes
10  *  (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
11  *
12  * Rewrite of larges part of the code in order to stabilize TCP stuff.
13  * Fix behaviour when socket buffer is full.
14  *  (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
15  *
16  * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com>
17  *
18  * IPv6 support contributed by Gilles Quillard, Bull Open Source, 2005.
19  *   <gilles.quillard@bull.net>
20  */
21 
22 #include <linux/types.h>
23 #include <linux/string.h>
24 #include <linux/slab.h>
25 #include <linux/module.h>
26 #include <linux/capability.h>
27 #include <linux/pagemap.h>
28 #include <linux/errno.h>
29 #include <linux/socket.h>
30 #include <linux/in.h>
31 #include <linux/net.h>
32 #include <linux/mm.h>
33 #include <linux/un.h>
34 #include <linux/udp.h>
35 #include <linux/tcp.h>
36 #include <linux/sunrpc/clnt.h>
37 #include <linux/sunrpc/addr.h>
38 #include <linux/sunrpc/sched.h>
39 #include <linux/sunrpc/svcsock.h>
40 #include <linux/sunrpc/xprtsock.h>
41 #include <linux/file.h>
42 #ifdef CONFIG_SUNRPC_BACKCHANNEL
43 #include <linux/sunrpc/bc_xprt.h>
44 #endif
45 
46 #include <net/sock.h>
47 #include <net/checksum.h>
48 #include <net/udp.h>
49 #include <net/tcp.h>
50 #include <net/tls_prot.h>
51 #include <net/handshake.h>
52 
53 #include <linux/bvec.h>
54 #include <linux/highmem.h>
55 #include <linux/uio.h>
56 #include <linux/sched/mm.h>
57 
58 #include <trace/events/sock.h>
59 #include <trace/events/sunrpc.h>
60 
61 #include "socklib.h"
62 #include "sunrpc.h"
63 
64 static void xs_close(struct rpc_xprt *xprt);
65 static void xs_set_srcport(struct sock_xprt *transport, struct socket *sock);
66 static void xs_tcp_set_socket_timeouts(struct rpc_xprt *xprt,
67 		struct socket *sock);
68 
69 /*
70  * xprtsock tunables
71  */
72 static unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
73 static unsigned int xprt_tcp_slot_table_entries = RPC_MIN_SLOT_TABLE;
74 static unsigned int xprt_max_tcp_slot_table_entries = RPC_MAX_SLOT_TABLE;
75 
76 static unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
77 static unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
78 
79 #define XS_TCP_LINGER_TO	(15U * HZ)
80 static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
81 
82 /*
83  * We can register our own files under /proc/sys/sunrpc by
84  * calling register_sysctl() again.  The files in that
85  * directory become the union of all files registered there.
86  *
87  * We simply need to make sure that we don't collide with
88  * someone else's file names!
89  */
90 
91 static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
92 static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
93 static unsigned int max_tcp_slot_table_limit = RPC_MAX_SLOT_TABLE_LIMIT;
94 static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
95 static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
96 
97 static struct ctl_table_header *sunrpc_table_header;
98 
99 static struct xprt_class xs_local_transport;
100 static struct xprt_class xs_udp_transport;
101 static struct xprt_class xs_tcp_transport;
102 static struct xprt_class xs_tcp_tls_transport;
103 static struct xprt_class xs_bc_tcp_transport;
104 
105 /*
106  * FIXME: changing the UDP slot table size should also resize the UDP
107  *        socket buffers for existing UDP transports
108  */
109 static struct ctl_table xs_tunables_table[] = {
110 	{
111 		.procname	= "udp_slot_table_entries",
112 		.data		= &xprt_udp_slot_table_entries,
113 		.maxlen		= sizeof(unsigned int),
114 		.mode		= 0644,
115 		.proc_handler	= proc_dointvec_minmax,
116 		.extra1		= &min_slot_table_size,
117 		.extra2		= &max_slot_table_size
118 	},
119 	{
120 		.procname	= "tcp_slot_table_entries",
121 		.data		= &xprt_tcp_slot_table_entries,
122 		.maxlen		= sizeof(unsigned int),
123 		.mode		= 0644,
124 		.proc_handler	= proc_dointvec_minmax,
125 		.extra1		= &min_slot_table_size,
126 		.extra2		= &max_slot_table_size
127 	},
128 	{
129 		.procname	= "tcp_max_slot_table_entries",
130 		.data		= &xprt_max_tcp_slot_table_entries,
131 		.maxlen		= sizeof(unsigned int),
132 		.mode		= 0644,
133 		.proc_handler	= proc_dointvec_minmax,
134 		.extra1		= &min_slot_table_size,
135 		.extra2		= &max_tcp_slot_table_limit
136 	},
137 	{
138 		.procname	= "min_resvport",
139 		.data		= &xprt_min_resvport,
140 		.maxlen		= sizeof(unsigned int),
141 		.mode		= 0644,
142 		.proc_handler	= proc_dointvec_minmax,
143 		.extra1		= &xprt_min_resvport_limit,
144 		.extra2		= &xprt_max_resvport_limit
145 	},
146 	{
147 		.procname	= "max_resvport",
148 		.data		= &xprt_max_resvport,
149 		.maxlen		= sizeof(unsigned int),
150 		.mode		= 0644,
151 		.proc_handler	= proc_dointvec_minmax,
152 		.extra1		= &xprt_min_resvport_limit,
153 		.extra2		= &xprt_max_resvport_limit
154 	},
155 	{
156 		.procname	= "tcp_fin_timeout",
157 		.data		= &xs_tcp_fin_timeout,
158 		.maxlen		= sizeof(xs_tcp_fin_timeout),
159 		.mode		= 0644,
160 		.proc_handler	= proc_dointvec_jiffies,
161 	},
162 	{ },
163 };
164 
165 /*
166  * Wait duration for a reply from the RPC portmapper.
167  */
168 #define XS_BIND_TO		(60U * HZ)
169 
170 /*
171  * Delay if a UDP socket connect error occurs.  This is most likely some
172  * kind of resource problem on the local host.
173  */
174 #define XS_UDP_REEST_TO		(2U * HZ)
175 
176 /*
177  * The reestablish timeout allows clients to delay for a bit before attempting
178  * to reconnect to a server that just dropped our connection.
179  *
180  * We implement an exponential backoff when trying to reestablish a TCP
181  * transport connection with the server.  Some servers like to drop a TCP
182  * connection when they are overworked, so we start with a short timeout and
183  * increase over time if the server is down or not responding.
184  */
185 #define XS_TCP_INIT_REEST_TO	(3U * HZ)
186 
187 /*
188  * TCP idle timeout; client drops the transport socket if it is idle
189  * for this long.  Note that we also timeout UDP sockets to prevent
190  * holding port numbers when there is no RPC traffic.
191  */
192 #define XS_IDLE_DISC_TO		(5U * 60 * HZ)
193 
194 /*
195  * TLS handshake timeout.
196  */
197 #define XS_TLS_HANDSHAKE_TO	(10U * HZ)
198 
199 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
200 # undef  RPC_DEBUG_DATA
201 # define RPCDBG_FACILITY	RPCDBG_TRANS
202 #endif
203 
204 #ifdef RPC_DEBUG_DATA
xs_pktdump(char * msg,u32 * packet,unsigned int count)205 static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
206 {
207 	u8 *buf = (u8 *) packet;
208 	int j;
209 
210 	dprintk("RPC:       %s\n", msg);
211 	for (j = 0; j < count && j < 128; j += 4) {
212 		if (!(j & 31)) {
213 			if (j)
214 				dprintk("\n");
215 			dprintk("0x%04x ", j);
216 		}
217 		dprintk("%02x%02x%02x%02x ",
218 			buf[j], buf[j+1], buf[j+2], buf[j+3]);
219 	}
220 	dprintk("\n");
221 }
222 #else
xs_pktdump(char * msg,u32 * packet,unsigned int count)223 static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
224 {
225 	/* NOP */
226 }
227 #endif
228 
xprt_from_sock(struct sock * sk)229 static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
230 {
231 	return (struct rpc_xprt *) sk->sk_user_data;
232 }
233 
xs_addr(struct rpc_xprt * xprt)234 static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
235 {
236 	return (struct sockaddr *) &xprt->addr;
237 }
238 
xs_addr_un(struct rpc_xprt * xprt)239 static inline struct sockaddr_un *xs_addr_un(struct rpc_xprt *xprt)
240 {
241 	return (struct sockaddr_un *) &xprt->addr;
242 }
243 
xs_addr_in(struct rpc_xprt * xprt)244 static inline struct sockaddr_in *xs_addr_in(struct rpc_xprt *xprt)
245 {
246 	return (struct sockaddr_in *) &xprt->addr;
247 }
248 
xs_addr_in6(struct rpc_xprt * xprt)249 static inline struct sockaddr_in6 *xs_addr_in6(struct rpc_xprt *xprt)
250 {
251 	return (struct sockaddr_in6 *) &xprt->addr;
252 }
253 
xs_format_common_peer_addresses(struct rpc_xprt * xprt)254 static void xs_format_common_peer_addresses(struct rpc_xprt *xprt)
255 {
256 	struct sockaddr *sap = xs_addr(xprt);
257 	struct sockaddr_in6 *sin6;
258 	struct sockaddr_in *sin;
259 	struct sockaddr_un *sun;
260 	char buf[128];
261 
262 	switch (sap->sa_family) {
263 	case AF_LOCAL:
264 		sun = xs_addr_un(xprt);
265 		if (sun->sun_path[0]) {
266 			strscpy(buf, sun->sun_path, sizeof(buf));
267 		} else {
268 			buf[0] = '@';
269 			strscpy(buf+1, sun->sun_path+1, sizeof(buf)-1);
270 		}
271 		xprt->address_strings[RPC_DISPLAY_ADDR] =
272 						kstrdup(buf, GFP_KERNEL);
273 		break;
274 	case AF_INET:
275 		(void)rpc_ntop(sap, buf, sizeof(buf));
276 		xprt->address_strings[RPC_DISPLAY_ADDR] =
277 						kstrdup(buf, GFP_KERNEL);
278 		sin = xs_addr_in(xprt);
279 		snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
280 		break;
281 	case AF_INET6:
282 		(void)rpc_ntop(sap, buf, sizeof(buf));
283 		xprt->address_strings[RPC_DISPLAY_ADDR] =
284 						kstrdup(buf, GFP_KERNEL);
285 		sin6 = xs_addr_in6(xprt);
286 		snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
287 		break;
288 	default:
289 		BUG();
290 	}
291 
292 	xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
293 }
294 
xs_format_common_peer_ports(struct rpc_xprt * xprt)295 static void xs_format_common_peer_ports(struct rpc_xprt *xprt)
296 {
297 	struct sockaddr *sap = xs_addr(xprt);
298 	char buf[128];
299 
300 	snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
301 	xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
302 
303 	snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
304 	xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
305 }
306 
xs_format_peer_addresses(struct rpc_xprt * xprt,const char * protocol,const char * netid)307 static void xs_format_peer_addresses(struct rpc_xprt *xprt,
308 				     const char *protocol,
309 				     const char *netid)
310 {
311 	xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
312 	xprt->address_strings[RPC_DISPLAY_NETID] = netid;
313 	xs_format_common_peer_addresses(xprt);
314 	xs_format_common_peer_ports(xprt);
315 }
316 
xs_update_peer_port(struct rpc_xprt * xprt)317 static void xs_update_peer_port(struct rpc_xprt *xprt)
318 {
319 	kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]);
320 	kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
321 
322 	xs_format_common_peer_ports(xprt);
323 }
324 
xs_free_peer_addresses(struct rpc_xprt * xprt)325 static void xs_free_peer_addresses(struct rpc_xprt *xprt)
326 {
327 	unsigned int i;
328 
329 	for (i = 0; i < RPC_DISPLAY_MAX; i++)
330 		switch (i) {
331 		case RPC_DISPLAY_PROTO:
332 		case RPC_DISPLAY_NETID:
333 			continue;
334 		default:
335 			kfree(xprt->address_strings[i]);
336 		}
337 }
338 
339 static size_t
xs_alloc_sparse_pages(struct xdr_buf * buf,size_t want,gfp_t gfp)340 xs_alloc_sparse_pages(struct xdr_buf *buf, size_t want, gfp_t gfp)
341 {
342 	size_t i,n;
343 
344 	if (!want || !(buf->flags & XDRBUF_SPARSE_PAGES))
345 		return want;
346 	n = (buf->page_base + want + PAGE_SIZE - 1) >> PAGE_SHIFT;
347 	for (i = 0; i < n; i++) {
348 		if (buf->pages[i])
349 			continue;
350 		buf->bvec[i].bv_page = buf->pages[i] = alloc_page(gfp);
351 		if (!buf->pages[i]) {
352 			i *= PAGE_SIZE;
353 			return i > buf->page_base ? i - buf->page_base : 0;
354 		}
355 	}
356 	return want;
357 }
358 
359 static int
xs_sock_process_cmsg(struct socket * sock,struct msghdr * msg,struct cmsghdr * cmsg,int ret)360 xs_sock_process_cmsg(struct socket *sock, struct msghdr *msg,
361 		     struct cmsghdr *cmsg, int ret)
362 {
363 	u8 content_type = tls_get_record_type(sock->sk, cmsg);
364 	u8 level, description;
365 
366 	switch (content_type) {
367 	case 0:
368 		break;
369 	case TLS_RECORD_TYPE_DATA:
370 		/* TLS sets EOR at the end of each application data
371 		 * record, even though there might be more frames
372 		 * waiting to be decrypted.
373 		 */
374 		msg->msg_flags &= ~MSG_EOR;
375 		break;
376 	case TLS_RECORD_TYPE_ALERT:
377 		tls_alert_recv(sock->sk, msg, &level, &description);
378 		ret = (level == TLS_ALERT_LEVEL_FATAL) ?
379 			-EACCES : -EAGAIN;
380 		break;
381 	default:
382 		/* discard this record type */
383 		ret = -EAGAIN;
384 	}
385 	return ret;
386 }
387 
388 static int
xs_sock_recv_cmsg(struct socket * sock,struct msghdr * msg,int flags)389 xs_sock_recv_cmsg(struct socket *sock, struct msghdr *msg, int flags)
390 {
391 	union {
392 		struct cmsghdr	cmsg;
393 		u8		buf[CMSG_SPACE(sizeof(u8))];
394 	} u;
395 	int ret;
396 
397 	msg->msg_control = &u;
398 	msg->msg_controllen = sizeof(u);
399 	ret = sock_recvmsg(sock, msg, flags);
400 	if (msg->msg_controllen != sizeof(u))
401 		ret = xs_sock_process_cmsg(sock, msg, &u.cmsg, ret);
402 	return ret;
403 }
404 
405 static ssize_t
xs_sock_recvmsg(struct socket * sock,struct msghdr * msg,int flags,size_t seek)406 xs_sock_recvmsg(struct socket *sock, struct msghdr *msg, int flags, size_t seek)
407 {
408 	ssize_t ret;
409 	if (seek != 0)
410 		iov_iter_advance(&msg->msg_iter, seek);
411 	ret = xs_sock_recv_cmsg(sock, msg, flags);
412 	return ret > 0 ? ret + seek : ret;
413 }
414 
415 static ssize_t
xs_read_kvec(struct socket * sock,struct msghdr * msg,int flags,struct kvec * kvec,size_t count,size_t seek)416 xs_read_kvec(struct socket *sock, struct msghdr *msg, int flags,
417 		struct kvec *kvec, size_t count, size_t seek)
418 {
419 	iov_iter_kvec(&msg->msg_iter, ITER_DEST, kvec, 1, count);
420 	return xs_sock_recvmsg(sock, msg, flags, seek);
421 }
422 
423 static ssize_t
xs_read_bvec(struct socket * sock,struct msghdr * msg,int flags,struct bio_vec * bvec,unsigned long nr,size_t count,size_t seek)424 xs_read_bvec(struct socket *sock, struct msghdr *msg, int flags,
425 		struct bio_vec *bvec, unsigned long nr, size_t count,
426 		size_t seek)
427 {
428 	iov_iter_bvec(&msg->msg_iter, ITER_DEST, bvec, nr, count);
429 	return xs_sock_recvmsg(sock, msg, flags, seek);
430 }
431 
432 static ssize_t
xs_read_discard(struct socket * sock,struct msghdr * msg,int flags,size_t count)433 xs_read_discard(struct socket *sock, struct msghdr *msg, int flags,
434 		size_t count)
435 {
436 	iov_iter_discard(&msg->msg_iter, ITER_DEST, count);
437 	return xs_sock_recv_cmsg(sock, msg, flags);
438 }
439 
440 #if ARCH_IMPLEMENTS_FLUSH_DCACHE_PAGE
441 static void
xs_flush_bvec(const struct bio_vec * bvec,size_t count,size_t seek)442 xs_flush_bvec(const struct bio_vec *bvec, size_t count, size_t seek)
443 {
444 	struct bvec_iter bi = {
445 		.bi_size = count,
446 	};
447 	struct bio_vec bv;
448 
449 	bvec_iter_advance(bvec, &bi, seek & PAGE_MASK);
450 	for_each_bvec(bv, bvec, bi, bi)
451 		flush_dcache_page(bv.bv_page);
452 }
453 #else
454 static inline void
xs_flush_bvec(const struct bio_vec * bvec,size_t count,size_t seek)455 xs_flush_bvec(const struct bio_vec *bvec, size_t count, size_t seek)
456 {
457 }
458 #endif
459 
460 static ssize_t
xs_read_xdr_buf(struct socket * sock,struct msghdr * msg,int flags,struct xdr_buf * buf,size_t count,size_t seek,size_t * read)461 xs_read_xdr_buf(struct socket *sock, struct msghdr *msg, int flags,
462 		struct xdr_buf *buf, size_t count, size_t seek, size_t *read)
463 {
464 	size_t want, seek_init = seek, offset = 0;
465 	ssize_t ret;
466 
467 	want = min_t(size_t, count, buf->head[0].iov_len);
468 	if (seek < want) {
469 		ret = xs_read_kvec(sock, msg, flags, &buf->head[0], want, seek);
470 		if (ret <= 0)
471 			goto sock_err;
472 		offset += ret;
473 		if (offset == count || msg->msg_flags & (MSG_EOR|MSG_TRUNC))
474 			goto out;
475 		if (ret != want)
476 			goto out;
477 		seek = 0;
478 	} else {
479 		seek -= want;
480 		offset += want;
481 	}
482 
483 	want = xs_alloc_sparse_pages(
484 		buf, min_t(size_t, count - offset, buf->page_len),
485 		GFP_KERNEL | __GFP_NORETRY | __GFP_NOWARN);
486 	if (seek < want) {
487 		ret = xs_read_bvec(sock, msg, flags, buf->bvec,
488 				xdr_buf_pagecount(buf),
489 				want + buf->page_base,
490 				seek + buf->page_base);
491 		if (ret <= 0)
492 			goto sock_err;
493 		xs_flush_bvec(buf->bvec, ret, seek + buf->page_base);
494 		ret -= buf->page_base;
495 		offset += ret;
496 		if (offset == count || msg->msg_flags & (MSG_EOR|MSG_TRUNC))
497 			goto out;
498 		if (ret != want)
499 			goto out;
500 		seek = 0;
501 	} else {
502 		seek -= want;
503 		offset += want;
504 	}
505 
506 	want = min_t(size_t, count - offset, buf->tail[0].iov_len);
507 	if (seek < want) {
508 		ret = xs_read_kvec(sock, msg, flags, &buf->tail[0], want, seek);
509 		if (ret <= 0)
510 			goto sock_err;
511 		offset += ret;
512 		if (offset == count || msg->msg_flags & (MSG_EOR|MSG_TRUNC))
513 			goto out;
514 		if (ret != want)
515 			goto out;
516 	} else if (offset < seek_init)
517 		offset = seek_init;
518 	ret = -EMSGSIZE;
519 out:
520 	*read = offset - seek_init;
521 	return ret;
522 sock_err:
523 	offset += seek;
524 	goto out;
525 }
526 
527 static void
xs_read_header(struct sock_xprt * transport,struct xdr_buf * buf)528 xs_read_header(struct sock_xprt *transport, struct xdr_buf *buf)
529 {
530 	if (!transport->recv.copied) {
531 		if (buf->head[0].iov_len >= transport->recv.offset)
532 			memcpy(buf->head[0].iov_base,
533 					&transport->recv.xid,
534 					transport->recv.offset);
535 		transport->recv.copied = transport->recv.offset;
536 	}
537 }
538 
539 static bool
xs_read_stream_request_done(struct sock_xprt * transport)540 xs_read_stream_request_done(struct sock_xprt *transport)
541 {
542 	return transport->recv.fraghdr & cpu_to_be32(RPC_LAST_STREAM_FRAGMENT);
543 }
544 
545 static void
xs_read_stream_check_eor(struct sock_xprt * transport,struct msghdr * msg)546 xs_read_stream_check_eor(struct sock_xprt *transport,
547 		struct msghdr *msg)
548 {
549 	if (xs_read_stream_request_done(transport))
550 		msg->msg_flags |= MSG_EOR;
551 }
552 
553 static ssize_t
xs_read_stream_request(struct sock_xprt * transport,struct msghdr * msg,int flags,struct rpc_rqst * req)554 xs_read_stream_request(struct sock_xprt *transport, struct msghdr *msg,
555 		int flags, struct rpc_rqst *req)
556 {
557 	struct xdr_buf *buf = &req->rq_private_buf;
558 	size_t want, read;
559 	ssize_t ret;
560 
561 	xs_read_header(transport, buf);
562 
563 	want = transport->recv.len - transport->recv.offset;
564 	if (want != 0) {
565 		ret = xs_read_xdr_buf(transport->sock, msg, flags, buf,
566 				transport->recv.copied + want,
567 				transport->recv.copied,
568 				&read);
569 		transport->recv.offset += read;
570 		transport->recv.copied += read;
571 	}
572 
573 	if (transport->recv.offset == transport->recv.len)
574 		xs_read_stream_check_eor(transport, msg);
575 
576 	if (want == 0)
577 		return 0;
578 
579 	switch (ret) {
580 	default:
581 		break;
582 	case -EFAULT:
583 	case -EMSGSIZE:
584 		msg->msg_flags |= MSG_TRUNC;
585 		return read;
586 	case 0:
587 		return -ESHUTDOWN;
588 	}
589 	return ret < 0 ? ret : read;
590 }
591 
592 static size_t
xs_read_stream_headersize(bool isfrag)593 xs_read_stream_headersize(bool isfrag)
594 {
595 	if (isfrag)
596 		return sizeof(__be32);
597 	return 3 * sizeof(__be32);
598 }
599 
600 static ssize_t
xs_read_stream_header(struct sock_xprt * transport,struct msghdr * msg,int flags,size_t want,size_t seek)601 xs_read_stream_header(struct sock_xprt *transport, struct msghdr *msg,
602 		int flags, size_t want, size_t seek)
603 {
604 	struct kvec kvec = {
605 		.iov_base = &transport->recv.fraghdr,
606 		.iov_len = want,
607 	};
608 	return xs_read_kvec(transport->sock, msg, flags, &kvec, want, seek);
609 }
610 
611 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
612 static ssize_t
xs_read_stream_call(struct sock_xprt * transport,struct msghdr * msg,int flags)613 xs_read_stream_call(struct sock_xprt *transport, struct msghdr *msg, int flags)
614 {
615 	struct rpc_xprt *xprt = &transport->xprt;
616 	struct rpc_rqst *req;
617 	ssize_t ret;
618 
619 	/* Is this transport associated with the backchannel? */
620 	if (!xprt->bc_serv)
621 		return -ESHUTDOWN;
622 
623 	/* Look up and lock the request corresponding to the given XID */
624 	req = xprt_lookup_bc_request(xprt, transport->recv.xid);
625 	if (!req) {
626 		printk(KERN_WARNING "Callback slot table overflowed\n");
627 		return -ESHUTDOWN;
628 	}
629 	if (transport->recv.copied && !req->rq_private_buf.len)
630 		return -ESHUTDOWN;
631 
632 	ret = xs_read_stream_request(transport, msg, flags, req);
633 	if (msg->msg_flags & (MSG_EOR|MSG_TRUNC))
634 		xprt_complete_bc_request(req, transport->recv.copied);
635 	else
636 		req->rq_private_buf.len = transport->recv.copied;
637 
638 	return ret;
639 }
640 #else /* CONFIG_SUNRPC_BACKCHANNEL */
641 static ssize_t
xs_read_stream_call(struct sock_xprt * transport,struct msghdr * msg,int flags)642 xs_read_stream_call(struct sock_xprt *transport, struct msghdr *msg, int flags)
643 {
644 	return -ESHUTDOWN;
645 }
646 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
647 
648 static ssize_t
xs_read_stream_reply(struct sock_xprt * transport,struct msghdr * msg,int flags)649 xs_read_stream_reply(struct sock_xprt *transport, struct msghdr *msg, int flags)
650 {
651 	struct rpc_xprt *xprt = &transport->xprt;
652 	struct rpc_rqst *req;
653 	ssize_t ret = 0;
654 
655 	/* Look up and lock the request corresponding to the given XID */
656 	spin_lock(&xprt->queue_lock);
657 	req = xprt_lookup_rqst(xprt, transport->recv.xid);
658 	if (!req || (transport->recv.copied && !req->rq_private_buf.len)) {
659 		msg->msg_flags |= MSG_TRUNC;
660 		goto out;
661 	}
662 	xprt_pin_rqst(req);
663 	spin_unlock(&xprt->queue_lock);
664 
665 	ret = xs_read_stream_request(transport, msg, flags, req);
666 
667 	spin_lock(&xprt->queue_lock);
668 	if (msg->msg_flags & (MSG_EOR|MSG_TRUNC))
669 		xprt_complete_rqst(req->rq_task, transport->recv.copied);
670 	else
671 		req->rq_private_buf.len = transport->recv.copied;
672 	xprt_unpin_rqst(req);
673 out:
674 	spin_unlock(&xprt->queue_lock);
675 	return ret;
676 }
677 
678 static ssize_t
xs_read_stream(struct sock_xprt * transport,int flags)679 xs_read_stream(struct sock_xprt *transport, int flags)
680 {
681 	struct msghdr msg = { 0 };
682 	size_t want, read = 0;
683 	ssize_t ret = 0;
684 
685 	if (transport->recv.len == 0) {
686 		want = xs_read_stream_headersize(transport->recv.copied != 0);
687 		ret = xs_read_stream_header(transport, &msg, flags, want,
688 				transport->recv.offset);
689 		if (ret <= 0)
690 			goto out_err;
691 		transport->recv.offset = ret;
692 		if (transport->recv.offset != want)
693 			return transport->recv.offset;
694 		transport->recv.len = be32_to_cpu(transport->recv.fraghdr) &
695 			RPC_FRAGMENT_SIZE_MASK;
696 		transport->recv.offset -= sizeof(transport->recv.fraghdr);
697 		read = ret;
698 	}
699 
700 	switch (be32_to_cpu(transport->recv.calldir)) {
701 	default:
702 		msg.msg_flags |= MSG_TRUNC;
703 		break;
704 	case RPC_CALL:
705 		ret = xs_read_stream_call(transport, &msg, flags);
706 		break;
707 	case RPC_REPLY:
708 		ret = xs_read_stream_reply(transport, &msg, flags);
709 	}
710 	if (msg.msg_flags & MSG_TRUNC) {
711 		transport->recv.calldir = cpu_to_be32(-1);
712 		transport->recv.copied = -1;
713 	}
714 	if (ret < 0)
715 		goto out_err;
716 	read += ret;
717 	if (transport->recv.offset < transport->recv.len) {
718 		if (!(msg.msg_flags & MSG_TRUNC))
719 			return read;
720 		msg.msg_flags = 0;
721 		ret = xs_read_discard(transport->sock, &msg, flags,
722 				transport->recv.len - transport->recv.offset);
723 		if (ret <= 0)
724 			goto out_err;
725 		transport->recv.offset += ret;
726 		read += ret;
727 		if (transport->recv.offset != transport->recv.len)
728 			return read;
729 	}
730 	if (xs_read_stream_request_done(transport)) {
731 		trace_xs_stream_read_request(transport);
732 		transport->recv.copied = 0;
733 	}
734 	transport->recv.offset = 0;
735 	transport->recv.len = 0;
736 	return read;
737 out_err:
738 	return ret != 0 ? ret : -ESHUTDOWN;
739 }
740 
xs_poll_socket(struct sock_xprt * transport)741 static __poll_t xs_poll_socket(struct sock_xprt *transport)
742 {
743 	return transport->sock->ops->poll(transport->file, transport->sock,
744 			NULL);
745 }
746 
xs_poll_socket_readable(struct sock_xprt * transport)747 static bool xs_poll_socket_readable(struct sock_xprt *transport)
748 {
749 	__poll_t events = xs_poll_socket(transport);
750 
751 	return (events & (EPOLLIN | EPOLLRDNORM)) && !(events & EPOLLRDHUP);
752 }
753 
xs_poll_check_readable(struct sock_xprt * transport)754 static void xs_poll_check_readable(struct sock_xprt *transport)
755 {
756 
757 	clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
758 	if (test_bit(XPRT_SOCK_IGNORE_RECV, &transport->sock_state))
759 		return;
760 	if (!xs_poll_socket_readable(transport))
761 		return;
762 	if (!test_and_set_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
763 		queue_work(xprtiod_workqueue, &transport->recv_worker);
764 }
765 
xs_stream_data_receive(struct sock_xprt * transport)766 static void xs_stream_data_receive(struct sock_xprt *transport)
767 {
768 	size_t read = 0;
769 	ssize_t ret = 0;
770 
771 	mutex_lock(&transport->recv_mutex);
772 	if (transport->sock == NULL)
773 		goto out;
774 	for (;;) {
775 		ret = xs_read_stream(transport, MSG_DONTWAIT);
776 		if (ret < 0)
777 			break;
778 		read += ret;
779 		cond_resched();
780 	}
781 	if (ret == -ESHUTDOWN)
782 		kernel_sock_shutdown(transport->sock, SHUT_RDWR);
783 	else if (ret == -EACCES)
784 		xprt_wake_pending_tasks(&transport->xprt, -EACCES);
785 	else
786 		xs_poll_check_readable(transport);
787 out:
788 	mutex_unlock(&transport->recv_mutex);
789 	trace_xs_stream_read_data(&transport->xprt, ret, read);
790 }
791 
xs_stream_data_receive_workfn(struct work_struct * work)792 static void xs_stream_data_receive_workfn(struct work_struct *work)
793 {
794 	struct sock_xprt *transport =
795 		container_of(work, struct sock_xprt, recv_worker);
796 	unsigned int pflags = memalloc_nofs_save();
797 
798 	xs_stream_data_receive(transport);
799 	memalloc_nofs_restore(pflags);
800 }
801 
802 static void
xs_stream_reset_connect(struct sock_xprt * transport)803 xs_stream_reset_connect(struct sock_xprt *transport)
804 {
805 	transport->recv.offset = 0;
806 	transport->recv.len = 0;
807 	transport->recv.copied = 0;
808 	transport->xmit.offset = 0;
809 }
810 
811 static void
xs_stream_start_connect(struct sock_xprt * transport)812 xs_stream_start_connect(struct sock_xprt *transport)
813 {
814 	transport->xprt.stat.connect_count++;
815 	transport->xprt.stat.connect_start = jiffies;
816 }
817 
818 #define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)
819 
820 /**
821  * xs_nospace - handle transmit was incomplete
822  * @req: pointer to RPC request
823  * @transport: pointer to struct sock_xprt
824  *
825  */
xs_nospace(struct rpc_rqst * req,struct sock_xprt * transport)826 static int xs_nospace(struct rpc_rqst *req, struct sock_xprt *transport)
827 {
828 	struct rpc_xprt *xprt = &transport->xprt;
829 	struct sock *sk = transport->inet;
830 	int ret = -EAGAIN;
831 
832 	trace_rpc_socket_nospace(req, transport);
833 
834 	/* Protect against races with write_space */
835 	spin_lock(&xprt->transport_lock);
836 
837 	/* Don't race with disconnect */
838 	if (xprt_connected(xprt)) {
839 		/* wait for more buffer space */
840 		set_bit(XPRT_SOCK_NOSPACE, &transport->sock_state);
841 		set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
842 		sk->sk_write_pending++;
843 		xprt_wait_for_buffer_space(xprt);
844 	} else
845 		ret = -ENOTCONN;
846 
847 	spin_unlock(&xprt->transport_lock);
848 	return ret;
849 }
850 
xs_sock_nospace(struct rpc_rqst * req)851 static int xs_sock_nospace(struct rpc_rqst *req)
852 {
853 	struct sock_xprt *transport =
854 		container_of(req->rq_xprt, struct sock_xprt, xprt);
855 	struct sock *sk = transport->inet;
856 	int ret = -EAGAIN;
857 
858 	lock_sock(sk);
859 	if (!sock_writeable(sk))
860 		ret = xs_nospace(req, transport);
861 	release_sock(sk);
862 	return ret;
863 }
864 
xs_stream_nospace(struct rpc_rqst * req,bool vm_wait)865 static int xs_stream_nospace(struct rpc_rqst *req, bool vm_wait)
866 {
867 	struct sock_xprt *transport =
868 		container_of(req->rq_xprt, struct sock_xprt, xprt);
869 	struct sock *sk = transport->inet;
870 	int ret = -EAGAIN;
871 
872 	if (vm_wait)
873 		return -ENOBUFS;
874 	lock_sock(sk);
875 	if (!sk_stream_memory_free(sk))
876 		ret = xs_nospace(req, transport);
877 	release_sock(sk);
878 	return ret;
879 }
880 
xs_stream_prepare_request(struct rpc_rqst * req,struct xdr_buf * buf)881 static int xs_stream_prepare_request(struct rpc_rqst *req, struct xdr_buf *buf)
882 {
883 	return xdr_alloc_bvec(buf, rpc_task_gfp_mask());
884 }
885 
886 /*
887  * Determine if the previous message in the stream was aborted before it
888  * could complete transmission.
889  */
890 static bool
xs_send_request_was_aborted(struct sock_xprt * transport,struct rpc_rqst * req)891 xs_send_request_was_aborted(struct sock_xprt *transport, struct rpc_rqst *req)
892 {
893 	return transport->xmit.offset != 0 && req->rq_bytes_sent == 0;
894 }
895 
896 /*
897  * Return the stream record marker field for a record of length < 2^31-1
898  */
899 static rpc_fraghdr
xs_stream_record_marker(struct xdr_buf * xdr)900 xs_stream_record_marker(struct xdr_buf *xdr)
901 {
902 	if (!xdr->len)
903 		return 0;
904 	return cpu_to_be32(RPC_LAST_STREAM_FRAGMENT | (u32)xdr->len);
905 }
906 
907 /**
908  * xs_local_send_request - write an RPC request to an AF_LOCAL socket
909  * @req: pointer to RPC request
910  *
911  * Return values:
912  *        0:	The request has been sent
913  *   EAGAIN:	The socket was blocked, please call again later to
914  *		complete the request
915  * ENOTCONN:	Caller needs to invoke connect logic then call again
916  *    other:	Some other error occurred, the request was not sent
917  */
xs_local_send_request(struct rpc_rqst * req)918 static int xs_local_send_request(struct rpc_rqst *req)
919 {
920 	struct rpc_xprt *xprt = req->rq_xprt;
921 	struct sock_xprt *transport =
922 				container_of(xprt, struct sock_xprt, xprt);
923 	struct xdr_buf *xdr = &req->rq_snd_buf;
924 	rpc_fraghdr rm = xs_stream_record_marker(xdr);
925 	unsigned int msglen = rm ? req->rq_slen + sizeof(rm) : req->rq_slen;
926 	struct msghdr msg = {
927 		.msg_flags	= XS_SENDMSG_FLAGS,
928 	};
929 	bool vm_wait;
930 	unsigned int sent;
931 	int status;
932 
933 	/* Close the stream if the previous transmission was incomplete */
934 	if (xs_send_request_was_aborted(transport, req)) {
935 		xprt_force_disconnect(xprt);
936 		return -ENOTCONN;
937 	}
938 
939 	xs_pktdump("packet data:",
940 			req->rq_svec->iov_base, req->rq_svec->iov_len);
941 
942 	vm_wait = sk_stream_is_writeable(transport->inet) ? true : false;
943 
944 	req->rq_xtime = ktime_get();
945 	status = xprt_sock_sendmsg(transport->sock, &msg, xdr,
946 				   transport->xmit.offset, rm, &sent);
947 	dprintk("RPC:       %s(%u) = %d\n",
948 			__func__, xdr->len - transport->xmit.offset, status);
949 
950 	if (likely(sent > 0) || status == 0) {
951 		transport->xmit.offset += sent;
952 		req->rq_bytes_sent = transport->xmit.offset;
953 		if (likely(req->rq_bytes_sent >= msglen)) {
954 			req->rq_xmit_bytes_sent += transport->xmit.offset;
955 			transport->xmit.offset = 0;
956 			return 0;
957 		}
958 		status = -EAGAIN;
959 		vm_wait = false;
960 	}
961 
962 	switch (status) {
963 	case -EAGAIN:
964 		status = xs_stream_nospace(req, vm_wait);
965 		break;
966 	default:
967 		dprintk("RPC:       sendmsg returned unrecognized error %d\n",
968 			-status);
969 		fallthrough;
970 	case -EPIPE:
971 		xprt_force_disconnect(xprt);
972 		status = -ENOTCONN;
973 	}
974 
975 	return status;
976 }
977 
978 /**
979  * xs_udp_send_request - write an RPC request to a UDP socket
980  * @req: pointer to RPC request
981  *
982  * Return values:
983  *        0:	The request has been sent
984  *   EAGAIN:	The socket was blocked, please call again later to
985  *		complete the request
986  * ENOTCONN:	Caller needs to invoke connect logic then call again
987  *    other:	Some other error occurred, the request was not sent
988  */
xs_udp_send_request(struct rpc_rqst * req)989 static int xs_udp_send_request(struct rpc_rqst *req)
990 {
991 	struct rpc_xprt *xprt = req->rq_xprt;
992 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
993 	struct xdr_buf *xdr = &req->rq_snd_buf;
994 	struct msghdr msg = {
995 		.msg_name	= xs_addr(xprt),
996 		.msg_namelen	= xprt->addrlen,
997 		.msg_flags	= XS_SENDMSG_FLAGS,
998 	};
999 	unsigned int sent;
1000 	int status;
1001 
1002 	xs_pktdump("packet data:",
1003 				req->rq_svec->iov_base,
1004 				req->rq_svec->iov_len);
1005 
1006 	if (!xprt_bound(xprt))
1007 		return -ENOTCONN;
1008 
1009 	if (!xprt_request_get_cong(xprt, req))
1010 		return -EBADSLT;
1011 
1012 	status = xdr_alloc_bvec(xdr, rpc_task_gfp_mask());
1013 	if (status < 0)
1014 		return status;
1015 	req->rq_xtime = ktime_get();
1016 	status = xprt_sock_sendmsg(transport->sock, &msg, xdr, 0, 0, &sent);
1017 
1018 	dprintk("RPC:       xs_udp_send_request(%u) = %d\n",
1019 			xdr->len, status);
1020 
1021 	/* firewall is blocking us, don't return -EAGAIN or we end up looping */
1022 	if (status == -EPERM)
1023 		goto process_status;
1024 
1025 	if (status == -EAGAIN && sock_writeable(transport->inet))
1026 		status = -ENOBUFS;
1027 
1028 	if (sent > 0 || status == 0) {
1029 		req->rq_xmit_bytes_sent += sent;
1030 		if (sent >= req->rq_slen)
1031 			return 0;
1032 		/* Still some bytes left; set up for a retry later. */
1033 		status = -EAGAIN;
1034 	}
1035 
1036 process_status:
1037 	switch (status) {
1038 	case -ENOTSOCK:
1039 		status = -ENOTCONN;
1040 		/* Should we call xs_close() here? */
1041 		break;
1042 	case -EAGAIN:
1043 		status = xs_sock_nospace(req);
1044 		break;
1045 	case -ENETUNREACH:
1046 	case -ENOBUFS:
1047 	case -EPIPE:
1048 	case -ECONNREFUSED:
1049 	case -EPERM:
1050 		/* When the server has died, an ICMP port unreachable message
1051 		 * prompts ECONNREFUSED. */
1052 		break;
1053 	default:
1054 		dprintk("RPC:       sendmsg returned unrecognized error %d\n",
1055 			-status);
1056 	}
1057 
1058 	return status;
1059 }
1060 
1061 /**
1062  * xs_tcp_send_request - write an RPC request to a TCP socket
1063  * @req: pointer to RPC request
1064  *
1065  * Return values:
1066  *        0:	The request has been sent
1067  *   EAGAIN:	The socket was blocked, please call again later to
1068  *		complete the request
1069  * ENOTCONN:	Caller needs to invoke connect logic then call again
1070  *    other:	Some other error occurred, the request was not sent
1071  *
1072  * XXX: In the case of soft timeouts, should we eventually give up
1073  *	if sendmsg is not able to make progress?
1074  */
xs_tcp_send_request(struct rpc_rqst * req)1075 static int xs_tcp_send_request(struct rpc_rqst *req)
1076 {
1077 	struct rpc_xprt *xprt = req->rq_xprt;
1078 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1079 	struct xdr_buf *xdr = &req->rq_snd_buf;
1080 	rpc_fraghdr rm = xs_stream_record_marker(xdr);
1081 	unsigned int msglen = rm ? req->rq_slen + sizeof(rm) : req->rq_slen;
1082 	struct msghdr msg = {
1083 		.msg_flags	= XS_SENDMSG_FLAGS,
1084 	};
1085 	bool vm_wait;
1086 	unsigned int sent;
1087 	int status;
1088 
1089 	/* Close the stream if the previous transmission was incomplete */
1090 	if (xs_send_request_was_aborted(transport, req)) {
1091 		if (transport->sock != NULL)
1092 			kernel_sock_shutdown(transport->sock, SHUT_RDWR);
1093 		return -ENOTCONN;
1094 	}
1095 	if (!transport->inet)
1096 		return -ENOTCONN;
1097 
1098 	xs_pktdump("packet data:",
1099 				req->rq_svec->iov_base,
1100 				req->rq_svec->iov_len);
1101 
1102 	if (test_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state))
1103 		xs_tcp_set_socket_timeouts(xprt, transport->sock);
1104 
1105 	xs_set_srcport(transport, transport->sock);
1106 
1107 	/* Continue transmitting the packet/record. We must be careful
1108 	 * to cope with writespace callbacks arriving _after_ we have
1109 	 * called sendmsg(). */
1110 	req->rq_xtime = ktime_get();
1111 	tcp_sock_set_cork(transport->inet, true);
1112 
1113 	vm_wait = sk_stream_is_writeable(transport->inet) ? true : false;
1114 
1115 	do {
1116 		status = xprt_sock_sendmsg(transport->sock, &msg, xdr,
1117 					   transport->xmit.offset, rm, &sent);
1118 
1119 		dprintk("RPC:       xs_tcp_send_request(%u) = %d\n",
1120 				xdr->len - transport->xmit.offset, status);
1121 
1122 		/* If we've sent the entire packet, immediately
1123 		 * reset the count of bytes sent. */
1124 		transport->xmit.offset += sent;
1125 		req->rq_bytes_sent = transport->xmit.offset;
1126 		if (likely(req->rq_bytes_sent >= msglen)) {
1127 			req->rq_xmit_bytes_sent += transport->xmit.offset;
1128 			transport->xmit.offset = 0;
1129 			if (atomic_long_read(&xprt->xmit_queuelen) == 1)
1130 				tcp_sock_set_cork(transport->inet, false);
1131 			return 0;
1132 		}
1133 
1134 		WARN_ON_ONCE(sent == 0 && status == 0);
1135 
1136 		if (sent > 0)
1137 			vm_wait = false;
1138 
1139 	} while (status == 0);
1140 
1141 	switch (status) {
1142 	case -ENOTSOCK:
1143 		status = -ENOTCONN;
1144 		/* Should we call xs_close() here? */
1145 		break;
1146 	case -EAGAIN:
1147 		status = xs_stream_nospace(req, vm_wait);
1148 		break;
1149 	case -ECONNRESET:
1150 	case -ECONNREFUSED:
1151 	case -ENOTCONN:
1152 	case -EADDRINUSE:
1153 	case -ENOBUFS:
1154 	case -EPIPE:
1155 		break;
1156 	default:
1157 		dprintk("RPC:       sendmsg returned unrecognized error %d\n",
1158 			-status);
1159 	}
1160 
1161 	return status;
1162 }
1163 
xs_save_old_callbacks(struct sock_xprt * transport,struct sock * sk)1164 static void xs_save_old_callbacks(struct sock_xprt *transport, struct sock *sk)
1165 {
1166 	transport->old_data_ready = sk->sk_data_ready;
1167 	transport->old_state_change = sk->sk_state_change;
1168 	transport->old_write_space = sk->sk_write_space;
1169 	transport->old_error_report = sk->sk_error_report;
1170 }
1171 
xs_restore_old_callbacks(struct sock_xprt * transport,struct sock * sk)1172 static void xs_restore_old_callbacks(struct sock_xprt *transport, struct sock *sk)
1173 {
1174 	sk->sk_data_ready = transport->old_data_ready;
1175 	sk->sk_state_change = transport->old_state_change;
1176 	sk->sk_write_space = transport->old_write_space;
1177 	sk->sk_error_report = transport->old_error_report;
1178 }
1179 
xs_sock_reset_state_flags(struct rpc_xprt * xprt)1180 static void xs_sock_reset_state_flags(struct rpc_xprt *xprt)
1181 {
1182 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1183 
1184 	clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
1185 	clear_bit(XPRT_SOCK_WAKE_ERROR, &transport->sock_state);
1186 	clear_bit(XPRT_SOCK_WAKE_WRITE, &transport->sock_state);
1187 	clear_bit(XPRT_SOCK_WAKE_DISCONNECT, &transport->sock_state);
1188 	clear_bit(XPRT_SOCK_NOSPACE, &transport->sock_state);
1189 }
1190 
xs_run_error_worker(struct sock_xprt * transport,unsigned int nr)1191 static void xs_run_error_worker(struct sock_xprt *transport, unsigned int nr)
1192 {
1193 	set_bit(nr, &transport->sock_state);
1194 	queue_work(xprtiod_workqueue, &transport->error_worker);
1195 }
1196 
xs_sock_reset_connection_flags(struct rpc_xprt * xprt)1197 static void xs_sock_reset_connection_flags(struct rpc_xprt *xprt)
1198 {
1199 	xprt->connect_cookie++;
1200 	smp_mb__before_atomic();
1201 	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1202 	clear_bit(XPRT_CLOSING, &xprt->state);
1203 	xs_sock_reset_state_flags(xprt);
1204 	smp_mb__after_atomic();
1205 }
1206 
1207 /**
1208  * xs_error_report - callback to handle TCP socket state errors
1209  * @sk: socket
1210  *
1211  * Note: we don't call sock_error() since there may be a rpc_task
1212  * using the socket, and so we don't want to clear sk->sk_err.
1213  */
xs_error_report(struct sock * sk)1214 static void xs_error_report(struct sock *sk)
1215 {
1216 	struct sock_xprt *transport;
1217 	struct rpc_xprt *xprt;
1218 
1219 	if (!(xprt = xprt_from_sock(sk)))
1220 		return;
1221 
1222 	transport = container_of(xprt, struct sock_xprt, xprt);
1223 	transport->xprt_err = -sk->sk_err;
1224 	if (transport->xprt_err == 0)
1225 		return;
1226 	dprintk("RPC:       xs_error_report client %p, error=%d...\n",
1227 			xprt, -transport->xprt_err);
1228 	trace_rpc_socket_error(xprt, sk->sk_socket, transport->xprt_err);
1229 
1230 	/* barrier ensures xprt_err is set before XPRT_SOCK_WAKE_ERROR */
1231 	smp_mb__before_atomic();
1232 	xs_run_error_worker(transport, XPRT_SOCK_WAKE_ERROR);
1233 }
1234 
xs_reset_transport(struct sock_xprt * transport)1235 static void xs_reset_transport(struct sock_xprt *transport)
1236 {
1237 	struct socket *sock = transport->sock;
1238 	struct sock *sk = transport->inet;
1239 	struct rpc_xprt *xprt = &transport->xprt;
1240 	struct file *filp = transport->file;
1241 
1242 	if (sk == NULL)
1243 		return;
1244 	/*
1245 	 * Make sure we're calling this in a context from which it is safe
1246 	 * to call __fput_sync(). In practice that means rpciod and the
1247 	 * system workqueue.
1248 	 */
1249 	if (!(current->flags & PF_WQ_WORKER)) {
1250 		WARN_ON_ONCE(1);
1251 		set_bit(XPRT_CLOSE_WAIT, &xprt->state);
1252 		return;
1253 	}
1254 
1255 	if (atomic_read(&transport->xprt.swapper))
1256 		sk_clear_memalloc(sk);
1257 
1258 	tls_handshake_cancel(sk);
1259 
1260 	kernel_sock_shutdown(sock, SHUT_RDWR);
1261 
1262 	mutex_lock(&transport->recv_mutex);
1263 	lock_sock(sk);
1264 	transport->inet = NULL;
1265 	transport->sock = NULL;
1266 	transport->file = NULL;
1267 
1268 	sk->sk_user_data = NULL;
1269 
1270 	xs_restore_old_callbacks(transport, sk);
1271 	xprt_clear_connected(xprt);
1272 	xs_sock_reset_connection_flags(xprt);
1273 	/* Reset stream record info */
1274 	xs_stream_reset_connect(transport);
1275 	release_sock(sk);
1276 	mutex_unlock(&transport->recv_mutex);
1277 
1278 	trace_rpc_socket_close(xprt, sock);
1279 	__fput_sync(filp);
1280 
1281 	xprt_disconnect_done(xprt);
1282 }
1283 
1284 /**
1285  * xs_close - close a socket
1286  * @xprt: transport
1287  *
1288  * This is used when all requests are complete; ie, no DRC state remains
1289  * on the server we want to save.
1290  *
1291  * The caller _must_ be holding XPRT_LOCKED in order to avoid issues with
1292  * xs_reset_transport() zeroing the socket from underneath a writer.
1293  */
xs_close(struct rpc_xprt * xprt)1294 static void xs_close(struct rpc_xprt *xprt)
1295 {
1296 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1297 
1298 	dprintk("RPC:       xs_close xprt %p\n", xprt);
1299 
1300 	if (transport->sock)
1301 		tls_handshake_close(transport->sock);
1302 	xs_reset_transport(transport);
1303 	xprt->reestablish_timeout = 0;
1304 }
1305 
xs_inject_disconnect(struct rpc_xprt * xprt)1306 static void xs_inject_disconnect(struct rpc_xprt *xprt)
1307 {
1308 	dprintk("RPC:       injecting transport disconnect on xprt=%p\n",
1309 		xprt);
1310 	xprt_disconnect_done(xprt);
1311 }
1312 
xs_xprt_free(struct rpc_xprt * xprt)1313 static void xs_xprt_free(struct rpc_xprt *xprt)
1314 {
1315 	xs_free_peer_addresses(xprt);
1316 	xprt_free(xprt);
1317 }
1318 
1319 /**
1320  * xs_destroy - prepare to shutdown a transport
1321  * @xprt: doomed transport
1322  *
1323  */
xs_destroy(struct rpc_xprt * xprt)1324 static void xs_destroy(struct rpc_xprt *xprt)
1325 {
1326 	struct sock_xprt *transport = container_of(xprt,
1327 			struct sock_xprt, xprt);
1328 	dprintk("RPC:       xs_destroy xprt %p\n", xprt);
1329 
1330 	cancel_delayed_work_sync(&transport->connect_worker);
1331 	xs_close(xprt);
1332 	cancel_work_sync(&transport->recv_worker);
1333 	cancel_work_sync(&transport->error_worker);
1334 	xs_xprt_free(xprt);
1335 	module_put(THIS_MODULE);
1336 }
1337 
1338 /**
1339  * xs_udp_data_read_skb - receive callback for UDP sockets
1340  * @xprt: transport
1341  * @sk: socket
1342  * @skb: skbuff
1343  *
1344  */
xs_udp_data_read_skb(struct rpc_xprt * xprt,struct sock * sk,struct sk_buff * skb)1345 static void xs_udp_data_read_skb(struct rpc_xprt *xprt,
1346 		struct sock *sk,
1347 		struct sk_buff *skb)
1348 {
1349 	struct rpc_task *task;
1350 	struct rpc_rqst *rovr;
1351 	int repsize, copied;
1352 	u32 _xid;
1353 	__be32 *xp;
1354 
1355 	repsize = skb->len;
1356 	if (repsize < 4) {
1357 		dprintk("RPC:       impossible RPC reply size %d!\n", repsize);
1358 		return;
1359 	}
1360 
1361 	/* Copy the XID from the skb... */
1362 	xp = skb_header_pointer(skb, 0, sizeof(_xid), &_xid);
1363 	if (xp == NULL)
1364 		return;
1365 
1366 	/* Look up and lock the request corresponding to the given XID */
1367 	spin_lock(&xprt->queue_lock);
1368 	rovr = xprt_lookup_rqst(xprt, *xp);
1369 	if (!rovr)
1370 		goto out_unlock;
1371 	xprt_pin_rqst(rovr);
1372 	xprt_update_rtt(rovr->rq_task);
1373 	spin_unlock(&xprt->queue_lock);
1374 	task = rovr->rq_task;
1375 
1376 	if ((copied = rovr->rq_private_buf.buflen) > repsize)
1377 		copied = repsize;
1378 
1379 	/* Suck it into the iovec, verify checksum if not done by hw. */
1380 	if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
1381 		spin_lock(&xprt->queue_lock);
1382 		__UDPX_INC_STATS(sk, UDP_MIB_INERRORS);
1383 		goto out_unpin;
1384 	}
1385 
1386 
1387 	spin_lock(&xprt->transport_lock);
1388 	xprt_adjust_cwnd(xprt, task, copied);
1389 	spin_unlock(&xprt->transport_lock);
1390 	spin_lock(&xprt->queue_lock);
1391 	xprt_complete_rqst(task, copied);
1392 	__UDPX_INC_STATS(sk, UDP_MIB_INDATAGRAMS);
1393 out_unpin:
1394 	xprt_unpin_rqst(rovr);
1395  out_unlock:
1396 	spin_unlock(&xprt->queue_lock);
1397 }
1398 
xs_udp_data_receive(struct sock_xprt * transport)1399 static void xs_udp_data_receive(struct sock_xprt *transport)
1400 {
1401 	struct sk_buff *skb;
1402 	struct sock *sk;
1403 	int err;
1404 
1405 	mutex_lock(&transport->recv_mutex);
1406 	sk = transport->inet;
1407 	if (sk == NULL)
1408 		goto out;
1409 	for (;;) {
1410 		skb = skb_recv_udp(sk, MSG_DONTWAIT, &err);
1411 		if (skb == NULL)
1412 			break;
1413 		xs_udp_data_read_skb(&transport->xprt, sk, skb);
1414 		consume_skb(skb);
1415 		cond_resched();
1416 	}
1417 	xs_poll_check_readable(transport);
1418 out:
1419 	mutex_unlock(&transport->recv_mutex);
1420 }
1421 
xs_udp_data_receive_workfn(struct work_struct * work)1422 static void xs_udp_data_receive_workfn(struct work_struct *work)
1423 {
1424 	struct sock_xprt *transport =
1425 		container_of(work, struct sock_xprt, recv_worker);
1426 	unsigned int pflags = memalloc_nofs_save();
1427 
1428 	xs_udp_data_receive(transport);
1429 	memalloc_nofs_restore(pflags);
1430 }
1431 
1432 /**
1433  * xs_data_ready - "data ready" callback for sockets
1434  * @sk: socket with data to read
1435  *
1436  */
xs_data_ready(struct sock * sk)1437 static void xs_data_ready(struct sock *sk)
1438 {
1439 	struct rpc_xprt *xprt;
1440 
1441 	trace_sk_data_ready(sk);
1442 
1443 	xprt = xprt_from_sock(sk);
1444 	if (xprt != NULL) {
1445 		struct sock_xprt *transport = container_of(xprt,
1446 				struct sock_xprt, xprt);
1447 
1448 		trace_xs_data_ready(xprt);
1449 
1450 		transport->old_data_ready(sk);
1451 
1452 		if (test_bit(XPRT_SOCK_IGNORE_RECV, &transport->sock_state))
1453 			return;
1454 
1455 		/* Any data means we had a useful conversation, so
1456 		 * then we don't need to delay the next reconnect
1457 		 */
1458 		if (xprt->reestablish_timeout)
1459 			xprt->reestablish_timeout = 0;
1460 		if (!test_and_set_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
1461 			queue_work(xprtiod_workqueue, &transport->recv_worker);
1462 	}
1463 }
1464 
1465 /*
1466  * Helper function to force a TCP close if the server is sending
1467  * junk and/or it has put us in CLOSE_WAIT
1468  */
xs_tcp_force_close(struct rpc_xprt * xprt)1469 static void xs_tcp_force_close(struct rpc_xprt *xprt)
1470 {
1471 	xprt_force_disconnect(xprt);
1472 }
1473 
1474 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
xs_tcp_bc_maxpayload(struct rpc_xprt * xprt)1475 static size_t xs_tcp_bc_maxpayload(struct rpc_xprt *xprt)
1476 {
1477 	return PAGE_SIZE;
1478 }
1479 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
1480 
1481 /**
1482  * xs_local_state_change - callback to handle AF_LOCAL socket state changes
1483  * @sk: socket whose state has changed
1484  *
1485  */
xs_local_state_change(struct sock * sk)1486 static void xs_local_state_change(struct sock *sk)
1487 {
1488 	struct rpc_xprt *xprt;
1489 	struct sock_xprt *transport;
1490 
1491 	if (!(xprt = xprt_from_sock(sk)))
1492 		return;
1493 	transport = container_of(xprt, struct sock_xprt, xprt);
1494 	if (sk->sk_shutdown & SHUTDOWN_MASK) {
1495 		clear_bit(XPRT_CONNECTED, &xprt->state);
1496 		/* Trigger the socket release */
1497 		xs_run_error_worker(transport, XPRT_SOCK_WAKE_DISCONNECT);
1498 	}
1499 }
1500 
1501 /**
1502  * xs_tcp_state_change - callback to handle TCP socket state changes
1503  * @sk: socket whose state has changed
1504  *
1505  */
xs_tcp_state_change(struct sock * sk)1506 static void xs_tcp_state_change(struct sock *sk)
1507 {
1508 	struct rpc_xprt *xprt;
1509 	struct sock_xprt *transport;
1510 
1511 	if (!(xprt = xprt_from_sock(sk)))
1512 		return;
1513 	dprintk("RPC:       xs_tcp_state_change client %p...\n", xprt);
1514 	dprintk("RPC:       state %x conn %d dead %d zapped %d sk_shutdown %d\n",
1515 			sk->sk_state, xprt_connected(xprt),
1516 			sock_flag(sk, SOCK_DEAD),
1517 			sock_flag(sk, SOCK_ZAPPED),
1518 			sk->sk_shutdown);
1519 
1520 	transport = container_of(xprt, struct sock_xprt, xprt);
1521 	trace_rpc_socket_state_change(xprt, sk->sk_socket);
1522 	switch (sk->sk_state) {
1523 	case TCP_ESTABLISHED:
1524 		if (!xprt_test_and_set_connected(xprt)) {
1525 			xprt->connect_cookie++;
1526 			clear_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
1527 			xprt_clear_connecting(xprt);
1528 
1529 			xprt->stat.connect_count++;
1530 			xprt->stat.connect_time += (long)jiffies -
1531 						   xprt->stat.connect_start;
1532 			xs_run_error_worker(transport, XPRT_SOCK_WAKE_PENDING);
1533 		}
1534 		break;
1535 	case TCP_FIN_WAIT1:
1536 		/* The client initiated a shutdown of the socket */
1537 		xprt->connect_cookie++;
1538 		xprt->reestablish_timeout = 0;
1539 		set_bit(XPRT_CLOSING, &xprt->state);
1540 		smp_mb__before_atomic();
1541 		clear_bit(XPRT_CONNECTED, &xprt->state);
1542 		clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1543 		smp_mb__after_atomic();
1544 		break;
1545 	case TCP_CLOSE_WAIT:
1546 		/* The server initiated a shutdown of the socket */
1547 		xprt->connect_cookie++;
1548 		clear_bit(XPRT_CONNECTED, &xprt->state);
1549 		xs_run_error_worker(transport, XPRT_SOCK_WAKE_DISCONNECT);
1550 		fallthrough;
1551 	case TCP_CLOSING:
1552 		/*
1553 		 * If the server closed down the connection, make sure that
1554 		 * we back off before reconnecting
1555 		 */
1556 		if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
1557 			xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
1558 		break;
1559 	case TCP_LAST_ACK:
1560 		set_bit(XPRT_CLOSING, &xprt->state);
1561 		smp_mb__before_atomic();
1562 		clear_bit(XPRT_CONNECTED, &xprt->state);
1563 		smp_mb__after_atomic();
1564 		break;
1565 	case TCP_CLOSE:
1566 		if (test_and_clear_bit(XPRT_SOCK_CONNECTING,
1567 					&transport->sock_state))
1568 			xprt_clear_connecting(xprt);
1569 		clear_bit(XPRT_CLOSING, &xprt->state);
1570 		/* Trigger the socket release */
1571 		xs_run_error_worker(transport, XPRT_SOCK_WAKE_DISCONNECT);
1572 	}
1573 }
1574 
xs_write_space(struct sock * sk)1575 static void xs_write_space(struct sock *sk)
1576 {
1577 	struct sock_xprt *transport;
1578 	struct rpc_xprt *xprt;
1579 
1580 	if (!sk->sk_socket)
1581 		return;
1582 	clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
1583 
1584 	if (unlikely(!(xprt = xprt_from_sock(sk))))
1585 		return;
1586 	transport = container_of(xprt, struct sock_xprt, xprt);
1587 	if (!test_and_clear_bit(XPRT_SOCK_NOSPACE, &transport->sock_state))
1588 		return;
1589 	xs_run_error_worker(transport, XPRT_SOCK_WAKE_WRITE);
1590 	sk->sk_write_pending--;
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 	/* from net/core/sock.c:sock_def_write_space */
1606 	if (sock_writeable(sk))
1607 		xs_write_space(sk);
1608 }
1609 
1610 /**
1611  * xs_tcp_write_space - callback invoked when socket buffer space
1612  *                             becomes available
1613  * @sk: socket whose state has changed
1614  *
1615  * Called when more output buffer space is available for this socket.
1616  * We try not to wake our writers until they can make "significant"
1617  * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1618  * with a bunch of small requests.
1619  */
xs_tcp_write_space(struct sock * sk)1620 static void xs_tcp_write_space(struct sock *sk)
1621 {
1622 	/* from net/core/stream.c:sk_stream_write_space */
1623 	if (sk_stream_is_writeable(sk))
1624 		xs_write_space(sk);
1625 }
1626 
xs_udp_do_set_buffer_size(struct rpc_xprt * xprt)1627 static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1628 {
1629 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1630 	struct sock *sk = transport->inet;
1631 
1632 	if (transport->rcvsize) {
1633 		sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1634 		sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1635 	}
1636 	if (transport->sndsize) {
1637 		sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1638 		sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1639 		sk->sk_write_space(sk);
1640 	}
1641 }
1642 
1643 /**
1644  * xs_udp_set_buffer_size - set send and receive limits
1645  * @xprt: generic transport
1646  * @sndsize: requested size of send buffer, in bytes
1647  * @rcvsize: requested size of receive buffer, in bytes
1648  *
1649  * Set socket send and receive buffer size limits.
1650  */
xs_udp_set_buffer_size(struct rpc_xprt * xprt,size_t sndsize,size_t rcvsize)1651 static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1652 {
1653 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1654 
1655 	transport->sndsize = 0;
1656 	if (sndsize)
1657 		transport->sndsize = sndsize + 1024;
1658 	transport->rcvsize = 0;
1659 	if (rcvsize)
1660 		transport->rcvsize = rcvsize + 1024;
1661 
1662 	xs_udp_do_set_buffer_size(xprt);
1663 }
1664 
1665 /**
1666  * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
1667  * @xprt: controlling transport
1668  * @task: task that timed out
1669  *
1670  * Adjust the congestion window after a retransmit timeout has occurred.
1671  */
xs_udp_timer(struct rpc_xprt * xprt,struct rpc_task * task)1672 static void xs_udp_timer(struct rpc_xprt *xprt, struct rpc_task *task)
1673 {
1674 	spin_lock(&xprt->transport_lock);
1675 	xprt_adjust_cwnd(xprt, task, -ETIMEDOUT);
1676 	spin_unlock(&xprt->transport_lock);
1677 }
1678 
xs_get_random_port(void)1679 static int xs_get_random_port(void)
1680 {
1681 	unsigned short min = xprt_min_resvport, max = xprt_max_resvport;
1682 	unsigned short range;
1683 	unsigned short rand;
1684 
1685 	if (max < min)
1686 		return -EADDRINUSE;
1687 	range = max - min + 1;
1688 	rand = get_random_u32_below(range);
1689 	return rand + min;
1690 }
1691 
xs_sock_getport(struct socket * sock)1692 static unsigned short xs_sock_getport(struct socket *sock)
1693 {
1694 	struct sockaddr_storage buf;
1695 	unsigned short port = 0;
1696 
1697 	if (kernel_getsockname(sock, (struct sockaddr *)&buf) < 0)
1698 		goto out;
1699 	switch (buf.ss_family) {
1700 	case AF_INET6:
1701 		port = ntohs(((struct sockaddr_in6 *)&buf)->sin6_port);
1702 		break;
1703 	case AF_INET:
1704 		port = ntohs(((struct sockaddr_in *)&buf)->sin_port);
1705 	}
1706 out:
1707 	return port;
1708 }
1709 
1710 /**
1711  * xs_set_port - reset the port number in the remote endpoint address
1712  * @xprt: generic transport
1713  * @port: new port number
1714  *
1715  */
xs_set_port(struct rpc_xprt * xprt,unsigned short port)1716 static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
1717 {
1718 	dprintk("RPC:       setting port for xprt %p to %u\n", xprt, port);
1719 
1720 	rpc_set_port(xs_addr(xprt), port);
1721 	xs_update_peer_port(xprt);
1722 }
1723 
xs_set_srcport(struct sock_xprt * transport,struct socket * sock)1724 static void xs_set_srcport(struct sock_xprt *transport, struct socket *sock)
1725 {
1726 	if (transport->srcport == 0 && transport->xprt.reuseport)
1727 		transport->srcport = xs_sock_getport(sock);
1728 }
1729 
xs_get_srcport(struct sock_xprt * transport)1730 static int xs_get_srcport(struct sock_xprt *transport)
1731 {
1732 	int port = transport->srcport;
1733 
1734 	if (port == 0 && transport->xprt.resvport)
1735 		port = xs_get_random_port();
1736 	return port;
1737 }
1738 
xs_sock_srcport(struct rpc_xprt * xprt)1739 static unsigned short xs_sock_srcport(struct rpc_xprt *xprt)
1740 {
1741 	struct sock_xprt *sock = container_of(xprt, struct sock_xprt, xprt);
1742 	unsigned short ret = 0;
1743 	mutex_lock(&sock->recv_mutex);
1744 	if (sock->sock)
1745 		ret = xs_sock_getport(sock->sock);
1746 	mutex_unlock(&sock->recv_mutex);
1747 	return ret;
1748 }
1749 
xs_sock_srcaddr(struct rpc_xprt * xprt,char * buf,size_t buflen)1750 static int xs_sock_srcaddr(struct rpc_xprt *xprt, char *buf, size_t buflen)
1751 {
1752 	struct sock_xprt *sock = container_of(xprt, struct sock_xprt, xprt);
1753 	union {
1754 		struct sockaddr sa;
1755 		struct sockaddr_storage st;
1756 	} saddr;
1757 	int ret = -ENOTCONN;
1758 
1759 	mutex_lock(&sock->recv_mutex);
1760 	if (sock->sock) {
1761 		ret = kernel_getsockname(sock->sock, &saddr.sa);
1762 		if (ret >= 0)
1763 			ret = snprintf(buf, buflen, "%pISc", &saddr.sa);
1764 	}
1765 	mutex_unlock(&sock->recv_mutex);
1766 	return ret;
1767 }
1768 
xs_next_srcport(struct sock_xprt * transport,unsigned short port)1769 static unsigned short xs_next_srcport(struct sock_xprt *transport, unsigned short port)
1770 {
1771 	if (transport->srcport != 0)
1772 		transport->srcport = 0;
1773 	if (!transport->xprt.resvport)
1774 		return 0;
1775 	if (port <= xprt_min_resvport || port > xprt_max_resvport)
1776 		return xprt_max_resvport;
1777 	return --port;
1778 }
xs_bind(struct sock_xprt * transport,struct socket * sock)1779 static int xs_bind(struct sock_xprt *transport, struct socket *sock)
1780 {
1781 	struct sockaddr_storage myaddr;
1782 	int err, nloop = 0;
1783 	int port = xs_get_srcport(transport);
1784 	unsigned short last;
1785 
1786 	/*
1787 	 * If we are asking for any ephemeral port (i.e. port == 0 &&
1788 	 * transport->xprt.resvport == 0), don't bind.  Let the local
1789 	 * port selection happen implicitly when the socket is used
1790 	 * (for example at connect time).
1791 	 *
1792 	 * This ensures that we can continue to establish TCP
1793 	 * connections even when all local ephemeral ports are already
1794 	 * a part of some TCP connection.  This makes no difference
1795 	 * for UDP sockets, but also doesn't harm them.
1796 	 *
1797 	 * If we're asking for any reserved port (i.e. port == 0 &&
1798 	 * transport->xprt.resvport == 1) xs_get_srcport above will
1799 	 * ensure that port is non-zero and we will bind as needed.
1800 	 */
1801 	if (port <= 0)
1802 		return port;
1803 
1804 	memcpy(&myaddr, &transport->srcaddr, transport->xprt.addrlen);
1805 	do {
1806 		rpc_set_port((struct sockaddr *)&myaddr, port);
1807 		err = kernel_bind(sock, (struct sockaddr *)&myaddr,
1808 				transport->xprt.addrlen);
1809 		if (err == 0) {
1810 			if (transport->xprt.reuseport)
1811 				transport->srcport = port;
1812 			break;
1813 		}
1814 		last = port;
1815 		port = xs_next_srcport(transport, port);
1816 		if (port > last)
1817 			nloop++;
1818 	} while (err == -EADDRINUSE && nloop != 2);
1819 
1820 	if (myaddr.ss_family == AF_INET)
1821 		dprintk("RPC:       %s %pI4:%u: %s (%d)\n", __func__,
1822 				&((struct sockaddr_in *)&myaddr)->sin_addr,
1823 				port, err ? "failed" : "ok", err);
1824 	else
1825 		dprintk("RPC:       %s %pI6:%u: %s (%d)\n", __func__,
1826 				&((struct sockaddr_in6 *)&myaddr)->sin6_addr,
1827 				port, err ? "failed" : "ok", err);
1828 	return err;
1829 }
1830 
1831 /*
1832  * We don't support autobind on AF_LOCAL sockets
1833  */
xs_local_rpcbind(struct rpc_task * task)1834 static void xs_local_rpcbind(struct rpc_task *task)
1835 {
1836 	xprt_set_bound(task->tk_xprt);
1837 }
1838 
xs_local_set_port(struct rpc_xprt * xprt,unsigned short port)1839 static void xs_local_set_port(struct rpc_xprt *xprt, unsigned short port)
1840 {
1841 }
1842 
1843 #ifdef CONFIG_DEBUG_LOCK_ALLOC
1844 static struct lock_class_key xs_key[3];
1845 static struct lock_class_key xs_slock_key[3];
1846 
xs_reclassify_socketu(struct socket * sock)1847 static inline void xs_reclassify_socketu(struct socket *sock)
1848 {
1849 	struct sock *sk = sock->sk;
1850 
1851 	sock_lock_init_class_and_name(sk, "slock-AF_LOCAL-RPC",
1852 		&xs_slock_key[0], "sk_lock-AF_LOCAL-RPC", &xs_key[0]);
1853 }
1854 
xs_reclassify_socket4(struct socket * sock)1855 static inline void xs_reclassify_socket4(struct socket *sock)
1856 {
1857 	struct sock *sk = sock->sk;
1858 
1859 	sock_lock_init_class_and_name(sk, "slock-AF_INET-RPC",
1860 		&xs_slock_key[1], "sk_lock-AF_INET-RPC", &xs_key[1]);
1861 }
1862 
xs_reclassify_socket6(struct socket * sock)1863 static inline void xs_reclassify_socket6(struct socket *sock)
1864 {
1865 	struct sock *sk = sock->sk;
1866 
1867 	sock_lock_init_class_and_name(sk, "slock-AF_INET6-RPC",
1868 		&xs_slock_key[2], "sk_lock-AF_INET6-RPC", &xs_key[2]);
1869 }
1870 
xs_reclassify_socket(int family,struct socket * sock)1871 static inline void xs_reclassify_socket(int family, struct socket *sock)
1872 {
1873 	if (WARN_ON_ONCE(!sock_allow_reclassification(sock->sk)))
1874 		return;
1875 
1876 	switch (family) {
1877 	case AF_LOCAL:
1878 		xs_reclassify_socketu(sock);
1879 		break;
1880 	case AF_INET:
1881 		xs_reclassify_socket4(sock);
1882 		break;
1883 	case AF_INET6:
1884 		xs_reclassify_socket6(sock);
1885 		break;
1886 	}
1887 }
1888 #else
xs_reclassify_socket(int family,struct socket * sock)1889 static inline void xs_reclassify_socket(int family, struct socket *sock)
1890 {
1891 }
1892 #endif
1893 
xs_dummy_setup_socket(struct work_struct * work)1894 static void xs_dummy_setup_socket(struct work_struct *work)
1895 {
1896 }
1897 
xs_create_sock(struct rpc_xprt * xprt,struct sock_xprt * transport,int family,int type,int protocol,bool reuseport)1898 static struct socket *xs_create_sock(struct rpc_xprt *xprt,
1899 		struct sock_xprt *transport, int family, int type,
1900 		int protocol, bool reuseport)
1901 {
1902 	struct file *filp;
1903 	struct socket *sock;
1904 	int err;
1905 
1906 	err = __sock_create(xprt->xprt_net, family, type, protocol, &sock, 1);
1907 	if (err < 0) {
1908 		dprintk("RPC:       can't create %d transport socket (%d).\n",
1909 				protocol, -err);
1910 		goto out;
1911 	}
1912 	xs_reclassify_socket(family, sock);
1913 
1914 	if (reuseport)
1915 		sock_set_reuseport(sock->sk);
1916 
1917 	err = xs_bind(transport, sock);
1918 	if (err) {
1919 		sock_release(sock);
1920 		goto out;
1921 	}
1922 
1923 	filp = sock_alloc_file(sock, O_NONBLOCK, NULL);
1924 	if (IS_ERR(filp))
1925 		return ERR_CAST(filp);
1926 	transport->file = filp;
1927 
1928 	return sock;
1929 out:
1930 	return ERR_PTR(err);
1931 }
1932 
xs_local_finish_connecting(struct rpc_xprt * xprt,struct socket * sock)1933 static int xs_local_finish_connecting(struct rpc_xprt *xprt,
1934 				      struct socket *sock)
1935 {
1936 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
1937 									xprt);
1938 
1939 	if (!transport->inet) {
1940 		struct sock *sk = sock->sk;
1941 
1942 		lock_sock(sk);
1943 
1944 		xs_save_old_callbacks(transport, sk);
1945 
1946 		sk->sk_user_data = xprt;
1947 		sk->sk_data_ready = xs_data_ready;
1948 		sk->sk_write_space = xs_udp_write_space;
1949 		sk->sk_state_change = xs_local_state_change;
1950 		sk->sk_error_report = xs_error_report;
1951 		sk->sk_use_task_frag = false;
1952 
1953 		xprt_clear_connected(xprt);
1954 
1955 		/* Reset to new socket */
1956 		transport->sock = sock;
1957 		transport->inet = sk;
1958 
1959 		release_sock(sk);
1960 	}
1961 
1962 	xs_stream_start_connect(transport);
1963 
1964 	return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, 0);
1965 }
1966 
1967 /**
1968  * xs_local_setup_socket - create AF_LOCAL socket, connect to a local endpoint
1969  * @transport: socket transport to connect
1970  */
xs_local_setup_socket(struct sock_xprt * transport)1971 static int xs_local_setup_socket(struct sock_xprt *transport)
1972 {
1973 	struct rpc_xprt *xprt = &transport->xprt;
1974 	struct file *filp;
1975 	struct socket *sock;
1976 	int status;
1977 
1978 	status = __sock_create(xprt->xprt_net, AF_LOCAL,
1979 					SOCK_STREAM, 0, &sock, 1);
1980 	if (status < 0) {
1981 		dprintk("RPC:       can't create AF_LOCAL "
1982 			"transport socket (%d).\n", -status);
1983 		goto out;
1984 	}
1985 	xs_reclassify_socket(AF_LOCAL, sock);
1986 
1987 	filp = sock_alloc_file(sock, O_NONBLOCK, NULL);
1988 	if (IS_ERR(filp)) {
1989 		status = PTR_ERR(filp);
1990 		goto out;
1991 	}
1992 	transport->file = filp;
1993 
1994 	dprintk("RPC:       worker connecting xprt %p via AF_LOCAL to %s\n",
1995 			xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1996 
1997 	status = xs_local_finish_connecting(xprt, sock);
1998 	trace_rpc_socket_connect(xprt, sock, status);
1999 	switch (status) {
2000 	case 0:
2001 		dprintk("RPC:       xprt %p connected to %s\n",
2002 				xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
2003 		xprt->stat.connect_count++;
2004 		xprt->stat.connect_time += (long)jiffies -
2005 					   xprt->stat.connect_start;
2006 		xprt_set_connected(xprt);
2007 		break;
2008 	case -ENOBUFS:
2009 		break;
2010 	case -ENOENT:
2011 		dprintk("RPC:       xprt %p: socket %s does not exist\n",
2012 				xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
2013 		break;
2014 	case -ECONNREFUSED:
2015 		dprintk("RPC:       xprt %p: connection refused for %s\n",
2016 				xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
2017 		break;
2018 	default:
2019 		printk(KERN_ERR "%s: unhandled error (%d) connecting to %s\n",
2020 				__func__, -status,
2021 				xprt->address_strings[RPC_DISPLAY_ADDR]);
2022 	}
2023 
2024 out:
2025 	xprt_clear_connecting(xprt);
2026 	xprt_wake_pending_tasks(xprt, status);
2027 	return status;
2028 }
2029 
xs_local_connect(struct rpc_xprt * xprt,struct rpc_task * task)2030 static void xs_local_connect(struct rpc_xprt *xprt, struct rpc_task *task)
2031 {
2032 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2033 	int ret;
2034 
2035 	if (transport->file)
2036 		goto force_disconnect;
2037 
2038 	if (RPC_IS_ASYNC(task)) {
2039 		/*
2040 		 * We want the AF_LOCAL connect to be resolved in the
2041 		 * filesystem namespace of the process making the rpc
2042 		 * call.  Thus we connect synchronously.
2043 		 *
2044 		 * If we want to support asynchronous AF_LOCAL calls,
2045 		 * we'll need to figure out how to pass a namespace to
2046 		 * connect.
2047 		 */
2048 		rpc_task_set_rpc_status(task, -ENOTCONN);
2049 		goto out_wake;
2050 	}
2051 	ret = xs_local_setup_socket(transport);
2052 	if (ret && !RPC_IS_SOFTCONN(task))
2053 		msleep_interruptible(15000);
2054 	return;
2055 force_disconnect:
2056 	xprt_force_disconnect(xprt);
2057 out_wake:
2058 	xprt_clear_connecting(xprt);
2059 	xprt_wake_pending_tasks(xprt, -ENOTCONN);
2060 }
2061 
2062 #if IS_ENABLED(CONFIG_SUNRPC_SWAP)
2063 /*
2064  * Note that this should be called with XPRT_LOCKED held, or recv_mutex
2065  * held, or when we otherwise know that we have exclusive access to the
2066  * socket, to guard against races with xs_reset_transport.
2067  */
xs_set_memalloc(struct rpc_xprt * xprt)2068 static void xs_set_memalloc(struct rpc_xprt *xprt)
2069 {
2070 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
2071 			xprt);
2072 
2073 	/*
2074 	 * If there's no sock, then we have nothing to set. The
2075 	 * reconnecting process will get it for us.
2076 	 */
2077 	if (!transport->inet)
2078 		return;
2079 	if (atomic_read(&xprt->swapper))
2080 		sk_set_memalloc(transport->inet);
2081 }
2082 
2083 /**
2084  * xs_enable_swap - Tag this transport as being used for swap.
2085  * @xprt: transport to tag
2086  *
2087  * Take a reference to this transport on behalf of the rpc_clnt, and
2088  * optionally mark it for swapping if it wasn't already.
2089  */
2090 static int
xs_enable_swap(struct rpc_xprt * xprt)2091 xs_enable_swap(struct rpc_xprt *xprt)
2092 {
2093 	struct sock_xprt *xs = container_of(xprt, struct sock_xprt, xprt);
2094 
2095 	mutex_lock(&xs->recv_mutex);
2096 	if (atomic_inc_return(&xprt->swapper) == 1 &&
2097 	    xs->inet)
2098 		sk_set_memalloc(xs->inet);
2099 	mutex_unlock(&xs->recv_mutex);
2100 	return 0;
2101 }
2102 
2103 /**
2104  * xs_disable_swap - Untag this transport as being used for swap.
2105  * @xprt: transport to tag
2106  *
2107  * Drop a "swapper" reference to this xprt on behalf of the rpc_clnt. If the
2108  * swapper refcount goes to 0, untag the socket as a memalloc socket.
2109  */
2110 static void
xs_disable_swap(struct rpc_xprt * xprt)2111 xs_disable_swap(struct rpc_xprt *xprt)
2112 {
2113 	struct sock_xprt *xs = container_of(xprt, struct sock_xprt, xprt);
2114 
2115 	mutex_lock(&xs->recv_mutex);
2116 	if (atomic_dec_and_test(&xprt->swapper) &&
2117 	    xs->inet)
2118 		sk_clear_memalloc(xs->inet);
2119 	mutex_unlock(&xs->recv_mutex);
2120 }
2121 #else
xs_set_memalloc(struct rpc_xprt * xprt)2122 static void xs_set_memalloc(struct rpc_xprt *xprt)
2123 {
2124 }
2125 
2126 static int
xs_enable_swap(struct rpc_xprt * xprt)2127 xs_enable_swap(struct rpc_xprt *xprt)
2128 {
2129 	return -EINVAL;
2130 }
2131 
2132 static void
xs_disable_swap(struct rpc_xprt * xprt)2133 xs_disable_swap(struct rpc_xprt *xprt)
2134 {
2135 }
2136 #endif
2137 
xs_udp_finish_connecting(struct rpc_xprt * xprt,struct socket * sock)2138 static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
2139 {
2140 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2141 
2142 	if (!transport->inet) {
2143 		struct sock *sk = sock->sk;
2144 
2145 		lock_sock(sk);
2146 
2147 		xs_save_old_callbacks(transport, sk);
2148 
2149 		sk->sk_user_data = xprt;
2150 		sk->sk_data_ready = xs_data_ready;
2151 		sk->sk_write_space = xs_udp_write_space;
2152 		sk->sk_use_task_frag = false;
2153 
2154 		xprt_set_connected(xprt);
2155 
2156 		/* Reset to new socket */
2157 		transport->sock = sock;
2158 		transport->inet = sk;
2159 
2160 		xs_set_memalloc(xprt);
2161 
2162 		release_sock(sk);
2163 	}
2164 	xs_udp_do_set_buffer_size(xprt);
2165 
2166 	xprt->stat.connect_start = jiffies;
2167 }
2168 
xs_udp_setup_socket(struct work_struct * work)2169 static void xs_udp_setup_socket(struct work_struct *work)
2170 {
2171 	struct sock_xprt *transport =
2172 		container_of(work, struct sock_xprt, connect_worker.work);
2173 	struct rpc_xprt *xprt = &transport->xprt;
2174 	struct socket *sock;
2175 	int status = -EIO;
2176 	unsigned int pflags = current->flags;
2177 
2178 	if (atomic_read(&xprt->swapper))
2179 		current->flags |= PF_MEMALLOC;
2180 	sock = xs_create_sock(xprt, transport,
2181 			xs_addr(xprt)->sa_family, SOCK_DGRAM,
2182 			IPPROTO_UDP, false);
2183 	if (IS_ERR(sock))
2184 		goto out;
2185 
2186 	dprintk("RPC:       worker connecting xprt %p via %s to "
2187 				"%s (port %s)\n", xprt,
2188 			xprt->address_strings[RPC_DISPLAY_PROTO],
2189 			xprt->address_strings[RPC_DISPLAY_ADDR],
2190 			xprt->address_strings[RPC_DISPLAY_PORT]);
2191 
2192 	xs_udp_finish_connecting(xprt, sock);
2193 	trace_rpc_socket_connect(xprt, sock, 0);
2194 	status = 0;
2195 out:
2196 	xprt_clear_connecting(xprt);
2197 	xprt_unlock_connect(xprt, transport);
2198 	xprt_wake_pending_tasks(xprt, status);
2199 	current_restore_flags(pflags, PF_MEMALLOC);
2200 }
2201 
2202 /**
2203  * xs_tcp_shutdown - gracefully shut down a TCP socket
2204  * @xprt: transport
2205  *
2206  * Initiates a graceful shutdown of the TCP socket by calling the
2207  * equivalent of shutdown(SHUT_RDWR);
2208  */
xs_tcp_shutdown(struct rpc_xprt * xprt)2209 static void xs_tcp_shutdown(struct rpc_xprt *xprt)
2210 {
2211 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2212 	struct socket *sock = transport->sock;
2213 	int skst = transport->inet ? transport->inet->sk_state : TCP_CLOSE;
2214 
2215 	if (sock == NULL)
2216 		return;
2217 	if (!xprt->reuseport) {
2218 		xs_close(xprt);
2219 		return;
2220 	}
2221 	switch (skst) {
2222 	case TCP_FIN_WAIT1:
2223 	case TCP_FIN_WAIT2:
2224 	case TCP_LAST_ACK:
2225 		break;
2226 	case TCP_ESTABLISHED:
2227 	case TCP_CLOSE_WAIT:
2228 		kernel_sock_shutdown(sock, SHUT_RDWR);
2229 		trace_rpc_socket_shutdown(xprt, sock);
2230 		break;
2231 	default:
2232 		xs_reset_transport(transport);
2233 	}
2234 }
2235 
xs_tcp_set_socket_timeouts(struct rpc_xprt * xprt,struct socket * sock)2236 static void xs_tcp_set_socket_timeouts(struct rpc_xprt *xprt,
2237 		struct socket *sock)
2238 {
2239 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2240 	struct net *net = sock_net(sock->sk);
2241 	unsigned long connect_timeout;
2242 	unsigned long syn_retries;
2243 	unsigned int keepidle;
2244 	unsigned int keepcnt;
2245 	unsigned int timeo;
2246 	unsigned long t;
2247 
2248 	spin_lock(&xprt->transport_lock);
2249 	keepidle = DIV_ROUND_UP(xprt->timeout->to_initval, HZ);
2250 	keepcnt = xprt->timeout->to_retries + 1;
2251 	timeo = jiffies_to_msecs(xprt->timeout->to_initval) *
2252 		(xprt->timeout->to_retries + 1);
2253 	clear_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state);
2254 	spin_unlock(&xprt->transport_lock);
2255 
2256 	/* TCP Keepalive options */
2257 	sock_set_keepalive(sock->sk);
2258 	tcp_sock_set_keepidle(sock->sk, keepidle);
2259 	tcp_sock_set_keepintvl(sock->sk, keepidle);
2260 	tcp_sock_set_keepcnt(sock->sk, keepcnt);
2261 
2262 	/* TCP user timeout (see RFC5482) */
2263 	tcp_sock_set_user_timeout(sock->sk, timeo);
2264 
2265 	/* Connect timeout */
2266 	connect_timeout = max_t(unsigned long,
2267 				DIV_ROUND_UP(xprt->connect_timeout, HZ), 1);
2268 	syn_retries = max_t(unsigned long,
2269 			    READ_ONCE(net->ipv4.sysctl_tcp_syn_retries), 1);
2270 	for (t = 0; t <= syn_retries && (1UL << t) < connect_timeout; t++)
2271 		;
2272 	if (t <= syn_retries)
2273 		tcp_sock_set_syncnt(sock->sk, t - 1);
2274 }
2275 
xs_tcp_do_set_connect_timeout(struct rpc_xprt * xprt,unsigned long connect_timeout)2276 static void xs_tcp_do_set_connect_timeout(struct rpc_xprt *xprt,
2277 					  unsigned long connect_timeout)
2278 {
2279 	struct sock_xprt *transport =
2280 		container_of(xprt, struct sock_xprt, xprt);
2281 	struct rpc_timeout to;
2282 	unsigned long initval;
2283 
2284 	memcpy(&to, xprt->timeout, sizeof(to));
2285 	/* Arbitrary lower limit */
2286 	initval = max_t(unsigned long, connect_timeout, XS_TCP_INIT_REEST_TO);
2287 	to.to_initval = initval;
2288 	to.to_maxval = initval;
2289 	to.to_retries = 0;
2290 	memcpy(&transport->tcp_timeout, &to, sizeof(transport->tcp_timeout));
2291 	xprt->timeout = &transport->tcp_timeout;
2292 	xprt->connect_timeout = connect_timeout;
2293 }
2294 
xs_tcp_set_connect_timeout(struct rpc_xprt * xprt,unsigned long connect_timeout,unsigned long reconnect_timeout)2295 static void xs_tcp_set_connect_timeout(struct rpc_xprt *xprt,
2296 		unsigned long connect_timeout,
2297 		unsigned long reconnect_timeout)
2298 {
2299 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2300 
2301 	spin_lock(&xprt->transport_lock);
2302 	if (reconnect_timeout < xprt->max_reconnect_timeout)
2303 		xprt->max_reconnect_timeout = reconnect_timeout;
2304 	if (connect_timeout < xprt->connect_timeout)
2305 		xs_tcp_do_set_connect_timeout(xprt, connect_timeout);
2306 	set_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state);
2307 	spin_unlock(&xprt->transport_lock);
2308 }
2309 
xs_tcp_finish_connecting(struct rpc_xprt * xprt,struct socket * sock)2310 static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
2311 {
2312 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2313 
2314 	if (!transport->inet) {
2315 		struct sock *sk = sock->sk;
2316 
2317 		/* Avoid temporary address, they are bad for long-lived
2318 		 * connections such as NFS mounts.
2319 		 * RFC4941, section 3.6 suggests that:
2320 		 *    Individual applications, which have specific
2321 		 *    knowledge about the normal duration of connections,
2322 		 *    MAY override this as appropriate.
2323 		 */
2324 		if (xs_addr(xprt)->sa_family == PF_INET6) {
2325 			ip6_sock_set_addr_preferences(sk,
2326 				IPV6_PREFER_SRC_PUBLIC);
2327 		}
2328 
2329 		xs_tcp_set_socket_timeouts(xprt, sock);
2330 		tcp_sock_set_nodelay(sk);
2331 
2332 		lock_sock(sk);
2333 
2334 		xs_save_old_callbacks(transport, sk);
2335 
2336 		sk->sk_user_data = xprt;
2337 		sk->sk_data_ready = xs_data_ready;
2338 		sk->sk_state_change = xs_tcp_state_change;
2339 		sk->sk_write_space = xs_tcp_write_space;
2340 		sk->sk_error_report = xs_error_report;
2341 		sk->sk_use_task_frag = false;
2342 
2343 		/* socket options */
2344 		sock_reset_flag(sk, SOCK_LINGER);
2345 
2346 		xprt_clear_connected(xprt);
2347 
2348 		/* Reset to new socket */
2349 		transport->sock = sock;
2350 		transport->inet = sk;
2351 
2352 		release_sock(sk);
2353 	}
2354 
2355 	if (!xprt_bound(xprt))
2356 		return -ENOTCONN;
2357 
2358 	xs_set_memalloc(xprt);
2359 
2360 	xs_stream_start_connect(transport);
2361 
2362 	/* Tell the socket layer to start connecting... */
2363 	set_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
2364 	return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
2365 }
2366 
2367 /**
2368  * xs_tcp_setup_socket - create a TCP socket and connect to a remote endpoint
2369  * @work: queued work item
2370  *
2371  * Invoked by a work queue tasklet.
2372  */
xs_tcp_setup_socket(struct work_struct * work)2373 static void xs_tcp_setup_socket(struct work_struct *work)
2374 {
2375 	struct sock_xprt *transport =
2376 		container_of(work, struct sock_xprt, connect_worker.work);
2377 	struct socket *sock = transport->sock;
2378 	struct rpc_xprt *xprt = &transport->xprt;
2379 	int status;
2380 	unsigned int pflags = current->flags;
2381 
2382 	if (atomic_read(&xprt->swapper))
2383 		current->flags |= PF_MEMALLOC;
2384 
2385 	if (xprt_connected(xprt))
2386 		goto out;
2387 	if (test_and_clear_bit(XPRT_SOCK_CONNECT_SENT,
2388 			       &transport->sock_state) ||
2389 	    !sock) {
2390 		xs_reset_transport(transport);
2391 		sock = xs_create_sock(xprt, transport, xs_addr(xprt)->sa_family,
2392 				      SOCK_STREAM, IPPROTO_TCP, true);
2393 		if (IS_ERR(sock)) {
2394 			xprt_wake_pending_tasks(xprt, PTR_ERR(sock));
2395 			goto out;
2396 		}
2397 	}
2398 
2399 	dprintk("RPC:       worker connecting xprt %p via %s to "
2400 				"%s (port %s)\n", xprt,
2401 			xprt->address_strings[RPC_DISPLAY_PROTO],
2402 			xprt->address_strings[RPC_DISPLAY_ADDR],
2403 			xprt->address_strings[RPC_DISPLAY_PORT]);
2404 
2405 	status = xs_tcp_finish_connecting(xprt, sock);
2406 	trace_rpc_socket_connect(xprt, sock, status);
2407 	dprintk("RPC:       %p connect status %d connected %d sock state %d\n",
2408 			xprt, -status, xprt_connected(xprt),
2409 			sock->sk->sk_state);
2410 	switch (status) {
2411 	case 0:
2412 	case -EINPROGRESS:
2413 		/* SYN_SENT! */
2414 		set_bit(XPRT_SOCK_CONNECT_SENT, &transport->sock_state);
2415 		if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2416 			xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2417 		fallthrough;
2418 	case -EALREADY:
2419 		goto out_unlock;
2420 	case -EADDRNOTAVAIL:
2421 		/* Source port number is unavailable. Try a new one! */
2422 		transport->srcport = 0;
2423 		status = -EAGAIN;
2424 		break;
2425 	case -EPERM:
2426 		/* Happens, for instance, if a BPF program is preventing
2427 		 * the connect. Remap the error so upper layers can better
2428 		 * deal with it.
2429 		 */
2430 		status = -ECONNREFUSED;
2431 		fallthrough;
2432 	case -EINVAL:
2433 		/* Happens, for instance, if the user specified a link
2434 		 * local IPv6 address without a scope-id.
2435 		 */
2436 	case -ECONNREFUSED:
2437 	case -ECONNRESET:
2438 	case -ENETDOWN:
2439 	case -ENETUNREACH:
2440 	case -EHOSTUNREACH:
2441 	case -EADDRINUSE:
2442 	case -ENOBUFS:
2443 		break;
2444 	default:
2445 		printk("%s: connect returned unhandled error %d\n",
2446 			__func__, status);
2447 		status = -EAGAIN;
2448 	}
2449 
2450 	/* xs_tcp_force_close() wakes tasks with a fixed error code.
2451 	 * We need to wake them first to ensure the correct error code.
2452 	 */
2453 	xprt_wake_pending_tasks(xprt, status);
2454 	xs_tcp_force_close(xprt);
2455 out:
2456 	xprt_clear_connecting(xprt);
2457 out_unlock:
2458 	xprt_unlock_connect(xprt, transport);
2459 	current_restore_flags(pflags, PF_MEMALLOC);
2460 }
2461 
2462 /*
2463  * Transfer the connected socket to @upper_transport, then mark that
2464  * xprt CONNECTED.
2465  */
xs_tcp_tls_finish_connecting(struct rpc_xprt * lower_xprt,struct sock_xprt * upper_transport)2466 static int xs_tcp_tls_finish_connecting(struct rpc_xprt *lower_xprt,
2467 					struct sock_xprt *upper_transport)
2468 {
2469 	struct sock_xprt *lower_transport =
2470 			container_of(lower_xprt, struct sock_xprt, xprt);
2471 	struct rpc_xprt *upper_xprt = &upper_transport->xprt;
2472 
2473 	if (!upper_transport->inet) {
2474 		struct socket *sock = lower_transport->sock;
2475 		struct sock *sk = sock->sk;
2476 
2477 		/* Avoid temporary address, they are bad for long-lived
2478 		 * connections such as NFS mounts.
2479 		 * RFC4941, section 3.6 suggests that:
2480 		 *    Individual applications, which have specific
2481 		 *    knowledge about the normal duration of connections,
2482 		 *    MAY override this as appropriate.
2483 		 */
2484 		if (xs_addr(upper_xprt)->sa_family == PF_INET6)
2485 			ip6_sock_set_addr_preferences(sk, IPV6_PREFER_SRC_PUBLIC);
2486 
2487 		xs_tcp_set_socket_timeouts(upper_xprt, sock);
2488 		tcp_sock_set_nodelay(sk);
2489 
2490 		lock_sock(sk);
2491 
2492 		/* @sk is already connected, so it now has the RPC callbacks.
2493 		 * Reach into @lower_transport to save the original ones.
2494 		 */
2495 		upper_transport->old_data_ready = lower_transport->old_data_ready;
2496 		upper_transport->old_state_change = lower_transport->old_state_change;
2497 		upper_transport->old_write_space = lower_transport->old_write_space;
2498 		upper_transport->old_error_report = lower_transport->old_error_report;
2499 		sk->sk_user_data = upper_xprt;
2500 
2501 		/* socket options */
2502 		sock_reset_flag(sk, SOCK_LINGER);
2503 
2504 		xprt_clear_connected(upper_xprt);
2505 
2506 		upper_transport->sock = sock;
2507 		upper_transport->inet = sk;
2508 		upper_transport->file = lower_transport->file;
2509 
2510 		release_sock(sk);
2511 
2512 		/* Reset lower_transport before shutting down its clnt */
2513 		mutex_lock(&lower_transport->recv_mutex);
2514 		lower_transport->inet = NULL;
2515 		lower_transport->sock = NULL;
2516 		lower_transport->file = NULL;
2517 
2518 		xprt_clear_connected(lower_xprt);
2519 		xs_sock_reset_connection_flags(lower_xprt);
2520 		xs_stream_reset_connect(lower_transport);
2521 		mutex_unlock(&lower_transport->recv_mutex);
2522 	}
2523 
2524 	if (!xprt_bound(upper_xprt))
2525 		return -ENOTCONN;
2526 
2527 	xs_set_memalloc(upper_xprt);
2528 
2529 	if (!xprt_test_and_set_connected(upper_xprt)) {
2530 		upper_xprt->connect_cookie++;
2531 		clear_bit(XPRT_SOCK_CONNECTING, &upper_transport->sock_state);
2532 		xprt_clear_connecting(upper_xprt);
2533 
2534 		upper_xprt->stat.connect_count++;
2535 		upper_xprt->stat.connect_time += (long)jiffies -
2536 					   upper_xprt->stat.connect_start;
2537 		xs_run_error_worker(upper_transport, XPRT_SOCK_WAKE_PENDING);
2538 	}
2539 	return 0;
2540 }
2541 
2542 /**
2543  * xs_tls_handshake_done - TLS handshake completion handler
2544  * @data: address of xprt to wake
2545  * @status: status of handshake
2546  * @peerid: serial number of key containing the remote's identity
2547  *
2548  */
xs_tls_handshake_done(void * data,int status,key_serial_t peerid)2549 static void xs_tls_handshake_done(void *data, int status, key_serial_t peerid)
2550 {
2551 	struct rpc_xprt *lower_xprt = data;
2552 	struct sock_xprt *lower_transport =
2553 				container_of(lower_xprt, struct sock_xprt, xprt);
2554 
2555 	lower_transport->xprt_err = status ? -EACCES : 0;
2556 	complete(&lower_transport->handshake_done);
2557 	xprt_put(lower_xprt);
2558 }
2559 
xs_tls_handshake_sync(struct rpc_xprt * lower_xprt,struct xprtsec_parms * xprtsec)2560 static int xs_tls_handshake_sync(struct rpc_xprt *lower_xprt, struct xprtsec_parms *xprtsec)
2561 {
2562 	struct sock_xprt *lower_transport =
2563 				container_of(lower_xprt, struct sock_xprt, xprt);
2564 	struct tls_handshake_args args = {
2565 		.ta_sock	= lower_transport->sock,
2566 		.ta_done	= xs_tls_handshake_done,
2567 		.ta_data	= xprt_get(lower_xprt),
2568 		.ta_peername	= lower_xprt->servername,
2569 	};
2570 	struct sock *sk = lower_transport->inet;
2571 	int rc;
2572 
2573 	init_completion(&lower_transport->handshake_done);
2574 	set_bit(XPRT_SOCK_IGNORE_RECV, &lower_transport->sock_state);
2575 	lower_transport->xprt_err = -ETIMEDOUT;
2576 	switch (xprtsec->policy) {
2577 	case RPC_XPRTSEC_TLS_ANON:
2578 		rc = tls_client_hello_anon(&args, GFP_KERNEL);
2579 		if (rc)
2580 			goto out_put_xprt;
2581 		break;
2582 	case RPC_XPRTSEC_TLS_X509:
2583 		args.ta_my_cert = xprtsec->cert_serial;
2584 		args.ta_my_privkey = xprtsec->privkey_serial;
2585 		rc = tls_client_hello_x509(&args, GFP_KERNEL);
2586 		if (rc)
2587 			goto out_put_xprt;
2588 		break;
2589 	default:
2590 		rc = -EACCES;
2591 		goto out_put_xprt;
2592 	}
2593 
2594 	rc = wait_for_completion_interruptible_timeout(&lower_transport->handshake_done,
2595 						       XS_TLS_HANDSHAKE_TO);
2596 	if (rc <= 0) {
2597 		if (!tls_handshake_cancel(sk)) {
2598 			if (rc == 0)
2599 				rc = -ETIMEDOUT;
2600 			goto out_put_xprt;
2601 		}
2602 	}
2603 
2604 	rc = lower_transport->xprt_err;
2605 
2606 out:
2607 	xs_stream_reset_connect(lower_transport);
2608 	clear_bit(XPRT_SOCK_IGNORE_RECV, &lower_transport->sock_state);
2609 	return rc;
2610 
2611 out_put_xprt:
2612 	xprt_put(lower_xprt);
2613 	goto out;
2614 }
2615 
2616 /**
2617  * xs_tcp_tls_setup_socket - establish a TLS session on a TCP socket
2618  * @work: queued work item
2619  *
2620  * Invoked by a work queue tasklet.
2621  *
2622  * For RPC-with-TLS, there is a two-stage connection process.
2623  *
2624  * The "upper-layer xprt" is visible to the RPC consumer. Once it has
2625  * been marked connected, the consumer knows that a TCP connection and
2626  * a TLS session have been established.
2627  *
2628  * A "lower-layer xprt", created in this function, handles the mechanics
2629  * of connecting the TCP socket, performing the RPC_AUTH_TLS probe, and
2630  * then driving the TLS handshake. Once all that is complete, the upper
2631  * layer xprt is marked connected.
2632  */
xs_tcp_tls_setup_socket(struct work_struct * work)2633 static void xs_tcp_tls_setup_socket(struct work_struct *work)
2634 {
2635 	struct sock_xprt *upper_transport =
2636 		container_of(work, struct sock_xprt, connect_worker.work);
2637 	struct rpc_clnt *upper_clnt = upper_transport->clnt;
2638 	struct rpc_xprt *upper_xprt = &upper_transport->xprt;
2639 	struct rpc_create_args args = {
2640 		.net		= upper_xprt->xprt_net,
2641 		.protocol	= upper_xprt->prot,
2642 		.address	= (struct sockaddr *)&upper_xprt->addr,
2643 		.addrsize	= upper_xprt->addrlen,
2644 		.timeout	= upper_clnt->cl_timeout,
2645 		.servername	= upper_xprt->servername,
2646 		.program	= upper_clnt->cl_program,
2647 		.prognumber	= upper_clnt->cl_prog,
2648 		.version	= upper_clnt->cl_vers,
2649 		.authflavor	= RPC_AUTH_TLS,
2650 		.cred		= upper_clnt->cl_cred,
2651 		.xprtsec	= {
2652 			.policy		= RPC_XPRTSEC_NONE,
2653 		},
2654 		.stats		= upper_clnt->cl_stats,
2655 	};
2656 	unsigned int pflags = current->flags;
2657 	struct rpc_clnt *lower_clnt;
2658 	struct rpc_xprt *lower_xprt;
2659 	int status;
2660 
2661 	if (atomic_read(&upper_xprt->swapper))
2662 		current->flags |= PF_MEMALLOC;
2663 
2664 	xs_stream_start_connect(upper_transport);
2665 
2666 	/* This implicitly sends an RPC_AUTH_TLS probe */
2667 	lower_clnt = rpc_create(&args);
2668 	if (IS_ERR(lower_clnt)) {
2669 		trace_rpc_tls_unavailable(upper_clnt, upper_xprt);
2670 		clear_bit(XPRT_SOCK_CONNECTING, &upper_transport->sock_state);
2671 		xprt_clear_connecting(upper_xprt);
2672 		xprt_wake_pending_tasks(upper_xprt, PTR_ERR(lower_clnt));
2673 		xs_run_error_worker(upper_transport, XPRT_SOCK_WAKE_PENDING);
2674 		goto out_unlock;
2675 	}
2676 
2677 	/* RPC_AUTH_TLS probe was successful. Try a TLS handshake on
2678 	 * the lower xprt.
2679 	 */
2680 	rcu_read_lock();
2681 	lower_xprt = rcu_dereference(lower_clnt->cl_xprt);
2682 	rcu_read_unlock();
2683 
2684 	if (wait_on_bit_lock(&lower_xprt->state, XPRT_LOCKED, TASK_KILLABLE))
2685 		goto out_unlock;
2686 
2687 	status = xs_tls_handshake_sync(lower_xprt, &upper_xprt->xprtsec);
2688 	if (status) {
2689 		trace_rpc_tls_not_started(upper_clnt, upper_xprt);
2690 		goto out_close;
2691 	}
2692 
2693 	status = xs_tcp_tls_finish_connecting(lower_xprt, upper_transport);
2694 	if (status)
2695 		goto out_close;
2696 	xprt_release_write(lower_xprt, NULL);
2697 
2698 	trace_rpc_socket_connect(upper_xprt, upper_transport->sock, 0);
2699 	if (!xprt_test_and_set_connected(upper_xprt)) {
2700 		upper_xprt->connect_cookie++;
2701 		clear_bit(XPRT_SOCK_CONNECTING, &upper_transport->sock_state);
2702 		xprt_clear_connecting(upper_xprt);
2703 
2704 		upper_xprt->stat.connect_count++;
2705 		upper_xprt->stat.connect_time += (long)jiffies -
2706 					   upper_xprt->stat.connect_start;
2707 		xs_run_error_worker(upper_transport, XPRT_SOCK_WAKE_PENDING);
2708 	}
2709 	rpc_shutdown_client(lower_clnt);
2710 
2711 out_unlock:
2712 	current_restore_flags(pflags, PF_MEMALLOC);
2713 	upper_transport->clnt = NULL;
2714 	xprt_unlock_connect(upper_xprt, upper_transport);
2715 	return;
2716 
2717 out_close:
2718 	xprt_release_write(lower_xprt, NULL);
2719 	rpc_shutdown_client(lower_clnt);
2720 
2721 	/* xprt_force_disconnect() wakes tasks with a fixed tk_status code.
2722 	 * Wake them first here to ensure they get our tk_status code.
2723 	 */
2724 	xprt_wake_pending_tasks(upper_xprt, status);
2725 	xs_tcp_force_close(upper_xprt);
2726 	xprt_clear_connecting(upper_xprt);
2727 	goto out_unlock;
2728 }
2729 
2730 /**
2731  * xs_connect - connect a socket to a remote endpoint
2732  * @xprt: pointer to transport structure
2733  * @task: address of RPC task that manages state of connect request
2734  *
2735  * TCP: If the remote end dropped the connection, delay reconnecting.
2736  *
2737  * UDP socket connects are synchronous, but we use a work queue anyway
2738  * to guarantee that even unprivileged user processes can set up a
2739  * socket on a privileged port.
2740  *
2741  * If a UDP socket connect fails, the delay behavior here prevents
2742  * retry floods (hard mounts).
2743  */
xs_connect(struct rpc_xprt * xprt,struct rpc_task * task)2744 static void xs_connect(struct rpc_xprt *xprt, struct rpc_task *task)
2745 {
2746 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2747 	unsigned long delay = 0;
2748 
2749 	WARN_ON_ONCE(!xprt_lock_connect(xprt, task, transport));
2750 
2751 	if (transport->sock != NULL) {
2752 		dprintk("RPC:       xs_connect delayed xprt %p for %lu "
2753 			"seconds\n", xprt, xprt->reestablish_timeout / HZ);
2754 
2755 		delay = xprt_reconnect_delay(xprt);
2756 		xprt_reconnect_backoff(xprt, XS_TCP_INIT_REEST_TO);
2757 
2758 	} else
2759 		dprintk("RPC:       xs_connect scheduled xprt %p\n", xprt);
2760 
2761 	transport->clnt = task->tk_client;
2762 	queue_delayed_work(xprtiod_workqueue,
2763 			&transport->connect_worker,
2764 			delay);
2765 }
2766 
xs_wake_disconnect(struct sock_xprt * transport)2767 static void xs_wake_disconnect(struct sock_xprt *transport)
2768 {
2769 	if (test_and_clear_bit(XPRT_SOCK_WAKE_DISCONNECT, &transport->sock_state))
2770 		xs_tcp_force_close(&transport->xprt);
2771 }
2772 
xs_wake_write(struct sock_xprt * transport)2773 static void xs_wake_write(struct sock_xprt *transport)
2774 {
2775 	if (test_and_clear_bit(XPRT_SOCK_WAKE_WRITE, &transport->sock_state))
2776 		xprt_write_space(&transport->xprt);
2777 }
2778 
xs_wake_error(struct sock_xprt * transport)2779 static void xs_wake_error(struct sock_xprt *transport)
2780 {
2781 	int sockerr;
2782 
2783 	if (!test_bit(XPRT_SOCK_WAKE_ERROR, &transport->sock_state))
2784 		return;
2785 	mutex_lock(&transport->recv_mutex);
2786 	if (transport->sock == NULL)
2787 		goto out;
2788 	if (!test_and_clear_bit(XPRT_SOCK_WAKE_ERROR, &transport->sock_state))
2789 		goto out;
2790 	sockerr = xchg(&transport->xprt_err, 0);
2791 	if (sockerr < 0)
2792 		xprt_wake_pending_tasks(&transport->xprt, sockerr);
2793 out:
2794 	mutex_unlock(&transport->recv_mutex);
2795 }
2796 
xs_wake_pending(struct sock_xprt * transport)2797 static void xs_wake_pending(struct sock_xprt *transport)
2798 {
2799 	if (test_and_clear_bit(XPRT_SOCK_WAKE_PENDING, &transport->sock_state))
2800 		xprt_wake_pending_tasks(&transport->xprt, -EAGAIN);
2801 }
2802 
xs_error_handle(struct work_struct * work)2803 static void xs_error_handle(struct work_struct *work)
2804 {
2805 	struct sock_xprt *transport = container_of(work,
2806 			struct sock_xprt, error_worker);
2807 
2808 	xs_wake_disconnect(transport);
2809 	xs_wake_write(transport);
2810 	xs_wake_error(transport);
2811 	xs_wake_pending(transport);
2812 }
2813 
2814 /**
2815  * xs_local_print_stats - display AF_LOCAL socket-specific stats
2816  * @xprt: rpc_xprt struct containing statistics
2817  * @seq: output file
2818  *
2819  */
xs_local_print_stats(struct rpc_xprt * xprt,struct seq_file * seq)2820 static void xs_local_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2821 {
2822 	long idle_time = 0;
2823 
2824 	if (xprt_connected(xprt))
2825 		idle_time = (long)(jiffies - xprt->last_used) / HZ;
2826 
2827 	seq_printf(seq, "\txprt:\tlocal %lu %lu %lu %ld %lu %lu %lu "
2828 			"%llu %llu %lu %llu %llu\n",
2829 			xprt->stat.bind_count,
2830 			xprt->stat.connect_count,
2831 			xprt->stat.connect_time / HZ,
2832 			idle_time,
2833 			xprt->stat.sends,
2834 			xprt->stat.recvs,
2835 			xprt->stat.bad_xids,
2836 			xprt->stat.req_u,
2837 			xprt->stat.bklog_u,
2838 			xprt->stat.max_slots,
2839 			xprt->stat.sending_u,
2840 			xprt->stat.pending_u);
2841 }
2842 
2843 /**
2844  * xs_udp_print_stats - display UDP socket-specific stats
2845  * @xprt: rpc_xprt struct containing statistics
2846  * @seq: output file
2847  *
2848  */
xs_udp_print_stats(struct rpc_xprt * xprt,struct seq_file * seq)2849 static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2850 {
2851 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2852 
2853 	seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %llu %llu "
2854 			"%lu %llu %llu\n",
2855 			transport->srcport,
2856 			xprt->stat.bind_count,
2857 			xprt->stat.sends,
2858 			xprt->stat.recvs,
2859 			xprt->stat.bad_xids,
2860 			xprt->stat.req_u,
2861 			xprt->stat.bklog_u,
2862 			xprt->stat.max_slots,
2863 			xprt->stat.sending_u,
2864 			xprt->stat.pending_u);
2865 }
2866 
2867 /**
2868  * xs_tcp_print_stats - display TCP socket-specific stats
2869  * @xprt: rpc_xprt struct containing statistics
2870  * @seq: output file
2871  *
2872  */
xs_tcp_print_stats(struct rpc_xprt * xprt,struct seq_file * seq)2873 static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2874 {
2875 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2876 	long idle_time = 0;
2877 
2878 	if (xprt_connected(xprt))
2879 		idle_time = (long)(jiffies - xprt->last_used) / HZ;
2880 
2881 	seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu "
2882 			"%llu %llu %lu %llu %llu\n",
2883 			transport->srcport,
2884 			xprt->stat.bind_count,
2885 			xprt->stat.connect_count,
2886 			xprt->stat.connect_time / HZ,
2887 			idle_time,
2888 			xprt->stat.sends,
2889 			xprt->stat.recvs,
2890 			xprt->stat.bad_xids,
2891 			xprt->stat.req_u,
2892 			xprt->stat.bklog_u,
2893 			xprt->stat.max_slots,
2894 			xprt->stat.sending_u,
2895 			xprt->stat.pending_u);
2896 }
2897 
2898 /*
2899  * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
2900  * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
2901  * to use the server side send routines.
2902  */
bc_malloc(struct rpc_task * task)2903 static int bc_malloc(struct rpc_task *task)
2904 {
2905 	struct rpc_rqst *rqst = task->tk_rqstp;
2906 	size_t size = rqst->rq_callsize;
2907 	struct page *page;
2908 	struct rpc_buffer *buf;
2909 
2910 	if (size > PAGE_SIZE - sizeof(struct rpc_buffer)) {
2911 		WARN_ONCE(1, "xprtsock: large bc buffer request (size %zu)\n",
2912 			  size);
2913 		return -EINVAL;
2914 	}
2915 
2916 	page = alloc_page(GFP_KERNEL | __GFP_NORETRY | __GFP_NOWARN);
2917 	if (!page)
2918 		return -ENOMEM;
2919 
2920 	buf = page_address(page);
2921 	buf->len = PAGE_SIZE;
2922 
2923 	rqst->rq_buffer = buf->data;
2924 	rqst->rq_rbuffer = (char *)rqst->rq_buffer + rqst->rq_callsize;
2925 	return 0;
2926 }
2927 
2928 /*
2929  * Free the space allocated in the bc_alloc routine
2930  */
bc_free(struct rpc_task * task)2931 static void bc_free(struct rpc_task *task)
2932 {
2933 	void *buffer = task->tk_rqstp->rq_buffer;
2934 	struct rpc_buffer *buf;
2935 
2936 	buf = container_of(buffer, struct rpc_buffer, data);
2937 	free_page((unsigned long)buf);
2938 }
2939 
bc_sendto(struct rpc_rqst * req)2940 static int bc_sendto(struct rpc_rqst *req)
2941 {
2942 	struct xdr_buf *xdr = &req->rq_snd_buf;
2943 	struct sock_xprt *transport =
2944 			container_of(req->rq_xprt, struct sock_xprt, xprt);
2945 	struct msghdr msg = {
2946 		.msg_flags	= 0,
2947 	};
2948 	rpc_fraghdr marker = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT |
2949 					 (u32)xdr->len);
2950 	unsigned int sent = 0;
2951 	int err;
2952 
2953 	req->rq_xtime = ktime_get();
2954 	err = xdr_alloc_bvec(xdr, rpc_task_gfp_mask());
2955 	if (err < 0)
2956 		return err;
2957 	err = xprt_sock_sendmsg(transport->sock, &msg, xdr, 0, marker, &sent);
2958 	xdr_free_bvec(xdr);
2959 	if (err < 0 || sent != (xdr->len + sizeof(marker)))
2960 		return -EAGAIN;
2961 	return sent;
2962 }
2963 
2964 /**
2965  * bc_send_request - Send a backchannel Call on a TCP socket
2966  * @req: rpc_rqst containing Call message to be sent
2967  *
2968  * xpt_mutex ensures @rqstp's whole message is written to the socket
2969  * without interruption.
2970  *
2971  * Return values:
2972  *   %0 if the message was sent successfully
2973  *   %ENOTCONN if the message was not sent
2974  */
bc_send_request(struct rpc_rqst * req)2975 static int bc_send_request(struct rpc_rqst *req)
2976 {
2977 	struct svc_xprt	*xprt;
2978 	int len;
2979 
2980 	/*
2981 	 * Get the server socket associated with this callback xprt
2982 	 */
2983 	xprt = req->rq_xprt->bc_xprt;
2984 
2985 	/*
2986 	 * Grab the mutex to serialize data as the connection is shared
2987 	 * with the fore channel
2988 	 */
2989 	mutex_lock(&xprt->xpt_mutex);
2990 	if (test_bit(XPT_DEAD, &xprt->xpt_flags))
2991 		len = -ENOTCONN;
2992 	else
2993 		len = bc_sendto(req);
2994 	mutex_unlock(&xprt->xpt_mutex);
2995 
2996 	if (len > 0)
2997 		len = 0;
2998 
2999 	return len;
3000 }
3001 
3002 /*
3003  * The close routine. Since this is client initiated, we do nothing
3004  */
3005 
bc_close(struct rpc_xprt * xprt)3006 static void bc_close(struct rpc_xprt *xprt)
3007 {
3008 	xprt_disconnect_done(xprt);
3009 }
3010 
3011 /*
3012  * The xprt destroy routine. Again, because this connection is client
3013  * initiated, we do nothing
3014  */
3015 
bc_destroy(struct rpc_xprt * xprt)3016 static void bc_destroy(struct rpc_xprt *xprt)
3017 {
3018 	dprintk("RPC:       bc_destroy xprt %p\n", xprt);
3019 
3020 	xs_xprt_free(xprt);
3021 	module_put(THIS_MODULE);
3022 }
3023 
3024 static const struct rpc_xprt_ops xs_local_ops = {
3025 	.reserve_xprt		= xprt_reserve_xprt,
3026 	.release_xprt		= xprt_release_xprt,
3027 	.alloc_slot		= xprt_alloc_slot,
3028 	.free_slot		= xprt_free_slot,
3029 	.rpcbind		= xs_local_rpcbind,
3030 	.set_port		= xs_local_set_port,
3031 	.connect		= xs_local_connect,
3032 	.buf_alloc		= rpc_malloc,
3033 	.buf_free		= rpc_free,
3034 	.prepare_request	= xs_stream_prepare_request,
3035 	.send_request		= xs_local_send_request,
3036 	.wait_for_reply_request	= xprt_wait_for_reply_request_def,
3037 	.close			= xs_close,
3038 	.destroy		= xs_destroy,
3039 	.print_stats		= xs_local_print_stats,
3040 	.enable_swap		= xs_enable_swap,
3041 	.disable_swap		= xs_disable_swap,
3042 };
3043 
3044 static const struct rpc_xprt_ops xs_udp_ops = {
3045 	.set_buffer_size	= xs_udp_set_buffer_size,
3046 	.reserve_xprt		= xprt_reserve_xprt_cong,
3047 	.release_xprt		= xprt_release_xprt_cong,
3048 	.alloc_slot		= xprt_alloc_slot,
3049 	.free_slot		= xprt_free_slot,
3050 	.rpcbind		= rpcb_getport_async,
3051 	.set_port		= xs_set_port,
3052 	.connect		= xs_connect,
3053 	.get_srcaddr		= xs_sock_srcaddr,
3054 	.get_srcport		= xs_sock_srcport,
3055 	.buf_alloc		= rpc_malloc,
3056 	.buf_free		= rpc_free,
3057 	.send_request		= xs_udp_send_request,
3058 	.wait_for_reply_request	= xprt_wait_for_reply_request_rtt,
3059 	.timer			= xs_udp_timer,
3060 	.release_request	= xprt_release_rqst_cong,
3061 	.close			= xs_close,
3062 	.destroy		= xs_destroy,
3063 	.print_stats		= xs_udp_print_stats,
3064 	.enable_swap		= xs_enable_swap,
3065 	.disable_swap		= xs_disable_swap,
3066 	.inject_disconnect	= xs_inject_disconnect,
3067 };
3068 
3069 static const struct rpc_xprt_ops xs_tcp_ops = {
3070 	.reserve_xprt		= xprt_reserve_xprt,
3071 	.release_xprt		= xprt_release_xprt,
3072 	.alloc_slot		= xprt_alloc_slot,
3073 	.free_slot		= xprt_free_slot,
3074 	.rpcbind		= rpcb_getport_async,
3075 	.set_port		= xs_set_port,
3076 	.connect		= xs_connect,
3077 	.get_srcaddr		= xs_sock_srcaddr,
3078 	.get_srcport		= xs_sock_srcport,
3079 	.buf_alloc		= rpc_malloc,
3080 	.buf_free		= rpc_free,
3081 	.prepare_request	= xs_stream_prepare_request,
3082 	.send_request		= xs_tcp_send_request,
3083 	.wait_for_reply_request	= xprt_wait_for_reply_request_def,
3084 	.close			= xs_tcp_shutdown,
3085 	.destroy		= xs_destroy,
3086 	.set_connect_timeout	= xs_tcp_set_connect_timeout,
3087 	.print_stats		= xs_tcp_print_stats,
3088 	.enable_swap		= xs_enable_swap,
3089 	.disable_swap		= xs_disable_swap,
3090 	.inject_disconnect	= xs_inject_disconnect,
3091 #ifdef CONFIG_SUNRPC_BACKCHANNEL
3092 	.bc_setup		= xprt_setup_bc,
3093 	.bc_maxpayload		= xs_tcp_bc_maxpayload,
3094 	.bc_num_slots		= xprt_bc_max_slots,
3095 	.bc_free_rqst		= xprt_free_bc_rqst,
3096 	.bc_destroy		= xprt_destroy_bc,
3097 #endif
3098 };
3099 
3100 /*
3101  * The rpc_xprt_ops for the server backchannel
3102  */
3103 
3104 static const struct rpc_xprt_ops bc_tcp_ops = {
3105 	.reserve_xprt		= xprt_reserve_xprt,
3106 	.release_xprt		= xprt_release_xprt,
3107 	.alloc_slot		= xprt_alloc_slot,
3108 	.free_slot		= xprt_free_slot,
3109 	.buf_alloc		= bc_malloc,
3110 	.buf_free		= bc_free,
3111 	.send_request		= bc_send_request,
3112 	.wait_for_reply_request	= xprt_wait_for_reply_request_def,
3113 	.close			= bc_close,
3114 	.destroy		= bc_destroy,
3115 	.print_stats		= xs_tcp_print_stats,
3116 	.enable_swap		= xs_enable_swap,
3117 	.disable_swap		= xs_disable_swap,
3118 	.inject_disconnect	= xs_inject_disconnect,
3119 };
3120 
xs_init_anyaddr(const int family,struct sockaddr * sap)3121 static int xs_init_anyaddr(const int family, struct sockaddr *sap)
3122 {
3123 	static const struct sockaddr_in sin = {
3124 		.sin_family		= AF_INET,
3125 		.sin_addr.s_addr	= htonl(INADDR_ANY),
3126 	};
3127 	static const struct sockaddr_in6 sin6 = {
3128 		.sin6_family		= AF_INET6,
3129 		.sin6_addr		= IN6ADDR_ANY_INIT,
3130 	};
3131 
3132 	switch (family) {
3133 	case AF_LOCAL:
3134 		break;
3135 	case AF_INET:
3136 		memcpy(sap, &sin, sizeof(sin));
3137 		break;
3138 	case AF_INET6:
3139 		memcpy(sap, &sin6, sizeof(sin6));
3140 		break;
3141 	default:
3142 		dprintk("RPC:       %s: Bad address family\n", __func__);
3143 		return -EAFNOSUPPORT;
3144 	}
3145 	return 0;
3146 }
3147 
xs_setup_xprt(struct xprt_create * args,unsigned int slot_table_size,unsigned int max_slot_table_size)3148 static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
3149 				      unsigned int slot_table_size,
3150 				      unsigned int max_slot_table_size)
3151 {
3152 	struct rpc_xprt *xprt;
3153 	struct sock_xprt *new;
3154 
3155 	if (args->addrlen > sizeof(xprt->addr)) {
3156 		dprintk("RPC:       xs_setup_xprt: address too large\n");
3157 		return ERR_PTR(-EBADF);
3158 	}
3159 
3160 	xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
3161 			max_slot_table_size);
3162 	if (xprt == NULL) {
3163 		dprintk("RPC:       xs_setup_xprt: couldn't allocate "
3164 				"rpc_xprt\n");
3165 		return ERR_PTR(-ENOMEM);
3166 	}
3167 
3168 	new = container_of(xprt, struct sock_xprt, xprt);
3169 	mutex_init(&new->recv_mutex);
3170 	memcpy(&xprt->addr, args->dstaddr, args->addrlen);
3171 	xprt->addrlen = args->addrlen;
3172 	if (args->srcaddr)
3173 		memcpy(&new->srcaddr, args->srcaddr, args->addrlen);
3174 	else {
3175 		int err;
3176 		err = xs_init_anyaddr(args->dstaddr->sa_family,
3177 					(struct sockaddr *)&new->srcaddr);
3178 		if (err != 0) {
3179 			xprt_free(xprt);
3180 			return ERR_PTR(err);
3181 		}
3182 	}
3183 
3184 	return xprt;
3185 }
3186 
3187 static const struct rpc_timeout xs_local_default_timeout = {
3188 	.to_initval = 10 * HZ,
3189 	.to_maxval = 10 * HZ,
3190 	.to_retries = 2,
3191 };
3192 
3193 /**
3194  * xs_setup_local - Set up transport to use an AF_LOCAL socket
3195  * @args: rpc transport creation arguments
3196  *
3197  * AF_LOCAL is a "tpi_cots_ord" transport, just like TCP
3198  */
xs_setup_local(struct xprt_create * args)3199 static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
3200 {
3201 	struct sockaddr_un *sun = (struct sockaddr_un *)args->dstaddr;
3202 	struct sock_xprt *transport;
3203 	struct rpc_xprt *xprt;
3204 	struct rpc_xprt *ret;
3205 
3206 	xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3207 			xprt_max_tcp_slot_table_entries);
3208 	if (IS_ERR(xprt))
3209 		return xprt;
3210 	transport = container_of(xprt, struct sock_xprt, xprt);
3211 
3212 	xprt->prot = 0;
3213 	xprt->xprt_class = &xs_local_transport;
3214 	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3215 
3216 	xprt->bind_timeout = XS_BIND_TO;
3217 	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
3218 	xprt->idle_timeout = XS_IDLE_DISC_TO;
3219 
3220 	xprt->ops = &xs_local_ops;
3221 	xprt->timeout = &xs_local_default_timeout;
3222 
3223 	INIT_WORK(&transport->recv_worker, xs_stream_data_receive_workfn);
3224 	INIT_WORK(&transport->error_worker, xs_error_handle);
3225 	INIT_DELAYED_WORK(&transport->connect_worker, xs_dummy_setup_socket);
3226 
3227 	switch (sun->sun_family) {
3228 	case AF_LOCAL:
3229 		if (sun->sun_path[0] != '/' && sun->sun_path[0] != '\0') {
3230 			dprintk("RPC:       bad AF_LOCAL address: %s\n",
3231 					sun->sun_path);
3232 			ret = ERR_PTR(-EINVAL);
3233 			goto out_err;
3234 		}
3235 		xprt_set_bound(xprt);
3236 		xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
3237 		break;
3238 	default:
3239 		ret = ERR_PTR(-EAFNOSUPPORT);
3240 		goto out_err;
3241 	}
3242 
3243 	dprintk("RPC:       set up xprt to %s via AF_LOCAL\n",
3244 			xprt->address_strings[RPC_DISPLAY_ADDR]);
3245 
3246 	if (try_module_get(THIS_MODULE))
3247 		return xprt;
3248 	ret = ERR_PTR(-EINVAL);
3249 out_err:
3250 	xs_xprt_free(xprt);
3251 	return ret;
3252 }
3253 
3254 static const struct rpc_timeout xs_udp_default_timeout = {
3255 	.to_initval = 5 * HZ,
3256 	.to_maxval = 30 * HZ,
3257 	.to_increment = 5 * HZ,
3258 	.to_retries = 5,
3259 };
3260 
3261 /**
3262  * xs_setup_udp - Set up transport to use a UDP socket
3263  * @args: rpc transport creation arguments
3264  *
3265  */
xs_setup_udp(struct xprt_create * args)3266 static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
3267 {
3268 	struct sockaddr *addr = args->dstaddr;
3269 	struct rpc_xprt *xprt;
3270 	struct sock_xprt *transport;
3271 	struct rpc_xprt *ret;
3272 
3273 	xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
3274 			xprt_udp_slot_table_entries);
3275 	if (IS_ERR(xprt))
3276 		return xprt;
3277 	transport = container_of(xprt, struct sock_xprt, xprt);
3278 
3279 	xprt->prot = IPPROTO_UDP;
3280 	xprt->xprt_class = &xs_udp_transport;
3281 	/* XXX: header size can vary due to auth type, IPv6, etc. */
3282 	xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
3283 
3284 	xprt->bind_timeout = XS_BIND_TO;
3285 	xprt->reestablish_timeout = XS_UDP_REEST_TO;
3286 	xprt->idle_timeout = XS_IDLE_DISC_TO;
3287 
3288 	xprt->ops = &xs_udp_ops;
3289 
3290 	xprt->timeout = &xs_udp_default_timeout;
3291 
3292 	INIT_WORK(&transport->recv_worker, xs_udp_data_receive_workfn);
3293 	INIT_WORK(&transport->error_worker, xs_error_handle);
3294 	INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_setup_socket);
3295 
3296 	switch (addr->sa_family) {
3297 	case AF_INET:
3298 		if (((struct sockaddr_in *)addr)->sin_port != htons(0))
3299 			xprt_set_bound(xprt);
3300 
3301 		xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
3302 		break;
3303 	case AF_INET6:
3304 		if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
3305 			xprt_set_bound(xprt);
3306 
3307 		xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
3308 		break;
3309 	default:
3310 		ret = ERR_PTR(-EAFNOSUPPORT);
3311 		goto out_err;
3312 	}
3313 
3314 	if (xprt_bound(xprt))
3315 		dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3316 				xprt->address_strings[RPC_DISPLAY_ADDR],
3317 				xprt->address_strings[RPC_DISPLAY_PORT],
3318 				xprt->address_strings[RPC_DISPLAY_PROTO]);
3319 	else
3320 		dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
3321 				xprt->address_strings[RPC_DISPLAY_ADDR],
3322 				xprt->address_strings[RPC_DISPLAY_PROTO]);
3323 
3324 	if (try_module_get(THIS_MODULE))
3325 		return xprt;
3326 	ret = ERR_PTR(-EINVAL);
3327 out_err:
3328 	xs_xprt_free(xprt);
3329 	return ret;
3330 }
3331 
3332 static const struct rpc_timeout xs_tcp_default_timeout = {
3333 	.to_initval = 60 * HZ,
3334 	.to_maxval = 60 * HZ,
3335 	.to_retries = 2,
3336 };
3337 
3338 /**
3339  * xs_setup_tcp - Set up transport to use a TCP socket
3340  * @args: rpc transport creation arguments
3341  *
3342  */
xs_setup_tcp(struct xprt_create * args)3343 static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
3344 {
3345 	struct sockaddr *addr = args->dstaddr;
3346 	struct rpc_xprt *xprt;
3347 	struct sock_xprt *transport;
3348 	struct rpc_xprt *ret;
3349 	unsigned int max_slot_table_size = xprt_max_tcp_slot_table_entries;
3350 
3351 	if (args->flags & XPRT_CREATE_INFINITE_SLOTS)
3352 		max_slot_table_size = RPC_MAX_SLOT_TABLE_LIMIT;
3353 
3354 	xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3355 			max_slot_table_size);
3356 	if (IS_ERR(xprt))
3357 		return xprt;
3358 	transport = container_of(xprt, struct sock_xprt, xprt);
3359 
3360 	xprt->prot = IPPROTO_TCP;
3361 	xprt->xprt_class = &xs_tcp_transport;
3362 	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3363 
3364 	xprt->bind_timeout = XS_BIND_TO;
3365 	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
3366 	xprt->idle_timeout = XS_IDLE_DISC_TO;
3367 
3368 	xprt->ops = &xs_tcp_ops;
3369 	xprt->timeout = &xs_tcp_default_timeout;
3370 
3371 	xprt->max_reconnect_timeout = xprt->timeout->to_maxval;
3372 	if (args->reconnect_timeout)
3373 		xprt->max_reconnect_timeout = args->reconnect_timeout;
3374 
3375 	xprt->connect_timeout = xprt->timeout->to_initval *
3376 		(xprt->timeout->to_retries + 1);
3377 	if (args->connect_timeout)
3378 		xs_tcp_do_set_connect_timeout(xprt, args->connect_timeout);
3379 
3380 	INIT_WORK(&transport->recv_worker, xs_stream_data_receive_workfn);
3381 	INIT_WORK(&transport->error_worker, xs_error_handle);
3382 	INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_setup_socket);
3383 
3384 	switch (addr->sa_family) {
3385 	case AF_INET:
3386 		if (((struct sockaddr_in *)addr)->sin_port != htons(0))
3387 			xprt_set_bound(xprt);
3388 
3389 		xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
3390 		break;
3391 	case AF_INET6:
3392 		if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
3393 			xprt_set_bound(xprt);
3394 
3395 		xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
3396 		break;
3397 	default:
3398 		ret = ERR_PTR(-EAFNOSUPPORT);
3399 		goto out_err;
3400 	}
3401 
3402 	if (xprt_bound(xprt))
3403 		dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3404 				xprt->address_strings[RPC_DISPLAY_ADDR],
3405 				xprt->address_strings[RPC_DISPLAY_PORT],
3406 				xprt->address_strings[RPC_DISPLAY_PROTO]);
3407 	else
3408 		dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
3409 				xprt->address_strings[RPC_DISPLAY_ADDR],
3410 				xprt->address_strings[RPC_DISPLAY_PROTO]);
3411 
3412 	if (try_module_get(THIS_MODULE))
3413 		return xprt;
3414 	ret = ERR_PTR(-EINVAL);
3415 out_err:
3416 	xs_xprt_free(xprt);
3417 	return ret;
3418 }
3419 
3420 /**
3421  * xs_setup_tcp_tls - Set up transport to use a TCP with TLS
3422  * @args: rpc transport creation arguments
3423  *
3424  */
xs_setup_tcp_tls(struct xprt_create * args)3425 static struct rpc_xprt *xs_setup_tcp_tls(struct xprt_create *args)
3426 {
3427 	struct sockaddr *addr = args->dstaddr;
3428 	struct rpc_xprt *xprt;
3429 	struct sock_xprt *transport;
3430 	struct rpc_xprt *ret;
3431 	unsigned int max_slot_table_size = xprt_max_tcp_slot_table_entries;
3432 
3433 	if (args->flags & XPRT_CREATE_INFINITE_SLOTS)
3434 		max_slot_table_size = RPC_MAX_SLOT_TABLE_LIMIT;
3435 
3436 	xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3437 			     max_slot_table_size);
3438 	if (IS_ERR(xprt))
3439 		return xprt;
3440 	transport = container_of(xprt, struct sock_xprt, xprt);
3441 
3442 	xprt->prot = IPPROTO_TCP;
3443 	xprt->xprt_class = &xs_tcp_transport;
3444 	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3445 
3446 	xprt->bind_timeout = XS_BIND_TO;
3447 	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
3448 	xprt->idle_timeout = XS_IDLE_DISC_TO;
3449 
3450 	xprt->ops = &xs_tcp_ops;
3451 	xprt->timeout = &xs_tcp_default_timeout;
3452 
3453 	xprt->max_reconnect_timeout = xprt->timeout->to_maxval;
3454 	xprt->connect_timeout = xprt->timeout->to_initval *
3455 		(xprt->timeout->to_retries + 1);
3456 
3457 	INIT_WORK(&transport->recv_worker, xs_stream_data_receive_workfn);
3458 	INIT_WORK(&transport->error_worker, xs_error_handle);
3459 
3460 	switch (args->xprtsec.policy) {
3461 	case RPC_XPRTSEC_TLS_ANON:
3462 	case RPC_XPRTSEC_TLS_X509:
3463 		xprt->xprtsec = args->xprtsec;
3464 		INIT_DELAYED_WORK(&transport->connect_worker,
3465 				  xs_tcp_tls_setup_socket);
3466 		break;
3467 	default:
3468 		ret = ERR_PTR(-EACCES);
3469 		goto out_err;
3470 	}
3471 
3472 	switch (addr->sa_family) {
3473 	case AF_INET:
3474 		if (((struct sockaddr_in *)addr)->sin_port != htons(0))
3475 			xprt_set_bound(xprt);
3476 
3477 		xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
3478 		break;
3479 	case AF_INET6:
3480 		if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
3481 			xprt_set_bound(xprt);
3482 
3483 		xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
3484 		break;
3485 	default:
3486 		ret = ERR_PTR(-EAFNOSUPPORT);
3487 		goto out_err;
3488 	}
3489 
3490 	if (xprt_bound(xprt))
3491 		dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3492 			xprt->address_strings[RPC_DISPLAY_ADDR],
3493 			xprt->address_strings[RPC_DISPLAY_PORT],
3494 			xprt->address_strings[RPC_DISPLAY_PROTO]);
3495 	else
3496 		dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
3497 			xprt->address_strings[RPC_DISPLAY_ADDR],
3498 			xprt->address_strings[RPC_DISPLAY_PROTO]);
3499 
3500 	if (try_module_get(THIS_MODULE))
3501 		return xprt;
3502 	ret = ERR_PTR(-EINVAL);
3503 out_err:
3504 	xs_xprt_free(xprt);
3505 	return ret;
3506 }
3507 
3508 /**
3509  * xs_setup_bc_tcp - Set up transport to use a TCP backchannel socket
3510  * @args: rpc transport creation arguments
3511  *
3512  */
xs_setup_bc_tcp(struct xprt_create * args)3513 static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
3514 {
3515 	struct sockaddr *addr = args->dstaddr;
3516 	struct rpc_xprt *xprt;
3517 	struct sock_xprt *transport;
3518 	struct svc_sock *bc_sock;
3519 	struct rpc_xprt *ret;
3520 
3521 	xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3522 			xprt_tcp_slot_table_entries);
3523 	if (IS_ERR(xprt))
3524 		return xprt;
3525 	transport = container_of(xprt, struct sock_xprt, xprt);
3526 
3527 	xprt->prot = IPPROTO_TCP;
3528 	xprt->xprt_class = &xs_bc_tcp_transport;
3529 	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3530 	xprt->timeout = &xs_tcp_default_timeout;
3531 
3532 	/* backchannel */
3533 	xprt_set_bound(xprt);
3534 	xprt->bind_timeout = 0;
3535 	xprt->reestablish_timeout = 0;
3536 	xprt->idle_timeout = 0;
3537 
3538 	xprt->ops = &bc_tcp_ops;
3539 
3540 	switch (addr->sa_family) {
3541 	case AF_INET:
3542 		xs_format_peer_addresses(xprt, "tcp",
3543 					 RPCBIND_NETID_TCP);
3544 		break;
3545 	case AF_INET6:
3546 		xs_format_peer_addresses(xprt, "tcp",
3547 				   RPCBIND_NETID_TCP6);
3548 		break;
3549 	default:
3550 		ret = ERR_PTR(-EAFNOSUPPORT);
3551 		goto out_err;
3552 	}
3553 
3554 	dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3555 			xprt->address_strings[RPC_DISPLAY_ADDR],
3556 			xprt->address_strings[RPC_DISPLAY_PORT],
3557 			xprt->address_strings[RPC_DISPLAY_PROTO]);
3558 
3559 	/*
3560 	 * Once we've associated a backchannel xprt with a connection,
3561 	 * we want to keep it around as long as the connection lasts,
3562 	 * in case we need to start using it for a backchannel again;
3563 	 * this reference won't be dropped until bc_xprt is destroyed.
3564 	 */
3565 	xprt_get(xprt);
3566 	args->bc_xprt->xpt_bc_xprt = xprt;
3567 	xprt->bc_xprt = args->bc_xprt;
3568 	bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
3569 	transport->sock = bc_sock->sk_sock;
3570 	transport->inet = bc_sock->sk_sk;
3571 
3572 	/*
3573 	 * Since we don't want connections for the backchannel, we set
3574 	 * the xprt status to connected
3575 	 */
3576 	xprt_set_connected(xprt);
3577 
3578 	if (try_module_get(THIS_MODULE))
3579 		return xprt;
3580 
3581 	args->bc_xprt->xpt_bc_xprt = NULL;
3582 	args->bc_xprt->xpt_bc_xps = NULL;
3583 	xprt_put(xprt);
3584 	ret = ERR_PTR(-EINVAL);
3585 out_err:
3586 	xs_xprt_free(xprt);
3587 	return ret;
3588 }
3589 
3590 static struct xprt_class	xs_local_transport = {
3591 	.list		= LIST_HEAD_INIT(xs_local_transport.list),
3592 	.name		= "named UNIX socket",
3593 	.owner		= THIS_MODULE,
3594 	.ident		= XPRT_TRANSPORT_LOCAL,
3595 	.setup		= xs_setup_local,
3596 	.netid		= { "" },
3597 };
3598 
3599 static struct xprt_class	xs_udp_transport = {
3600 	.list		= LIST_HEAD_INIT(xs_udp_transport.list),
3601 	.name		= "udp",
3602 	.owner		= THIS_MODULE,
3603 	.ident		= XPRT_TRANSPORT_UDP,
3604 	.setup		= xs_setup_udp,
3605 	.netid		= { "udp", "udp6", "" },
3606 };
3607 
3608 static struct xprt_class	xs_tcp_transport = {
3609 	.list		= LIST_HEAD_INIT(xs_tcp_transport.list),
3610 	.name		= "tcp",
3611 	.owner		= THIS_MODULE,
3612 	.ident		= XPRT_TRANSPORT_TCP,
3613 	.setup		= xs_setup_tcp,
3614 	.netid		= { "tcp", "tcp6", "" },
3615 };
3616 
3617 static struct xprt_class	xs_tcp_tls_transport = {
3618 	.list		= LIST_HEAD_INIT(xs_tcp_tls_transport.list),
3619 	.name		= "tcp-with-tls",
3620 	.owner		= THIS_MODULE,
3621 	.ident		= XPRT_TRANSPORT_TCP_TLS,
3622 	.setup		= xs_setup_tcp_tls,
3623 	.netid		= { "tcp", "tcp6", "" },
3624 };
3625 
3626 static struct xprt_class	xs_bc_tcp_transport = {
3627 	.list		= LIST_HEAD_INIT(xs_bc_tcp_transport.list),
3628 	.name		= "tcp NFSv4.1 backchannel",
3629 	.owner		= THIS_MODULE,
3630 	.ident		= XPRT_TRANSPORT_BC_TCP,
3631 	.setup		= xs_setup_bc_tcp,
3632 	.netid		= { "" },
3633 };
3634 
3635 /**
3636  * init_socket_xprt - set up xprtsock's sysctls, register with RPC client
3637  *
3638  */
init_socket_xprt(void)3639 int init_socket_xprt(void)
3640 {
3641 	if (!sunrpc_table_header)
3642 		sunrpc_table_header = register_sysctl("sunrpc", xs_tunables_table);
3643 
3644 	xprt_register_transport(&xs_local_transport);
3645 	xprt_register_transport(&xs_udp_transport);
3646 	xprt_register_transport(&xs_tcp_transport);
3647 	xprt_register_transport(&xs_tcp_tls_transport);
3648 	xprt_register_transport(&xs_bc_tcp_transport);
3649 
3650 	return 0;
3651 }
3652 
3653 /**
3654  * cleanup_socket_xprt - remove xprtsock's sysctls, unregister
3655  *
3656  */
cleanup_socket_xprt(void)3657 void cleanup_socket_xprt(void)
3658 {
3659 	if (sunrpc_table_header) {
3660 		unregister_sysctl_table(sunrpc_table_header);
3661 		sunrpc_table_header = NULL;
3662 	}
3663 
3664 	xprt_unregister_transport(&xs_local_transport);
3665 	xprt_unregister_transport(&xs_udp_transport);
3666 	xprt_unregister_transport(&xs_tcp_transport);
3667 	xprt_unregister_transport(&xs_tcp_tls_transport);
3668 	xprt_unregister_transport(&xs_bc_tcp_transport);
3669 }
3670 
param_set_portnr(const char * val,const struct kernel_param * kp)3671 static int param_set_portnr(const char *val, const struct kernel_param *kp)
3672 {
3673 	return param_set_uint_minmax(val, kp,
3674 			RPC_MIN_RESVPORT,
3675 			RPC_MAX_RESVPORT);
3676 }
3677 
3678 static const struct kernel_param_ops param_ops_portnr = {
3679 	.set = param_set_portnr,
3680 	.get = param_get_uint,
3681 };
3682 
3683 #define param_check_portnr(name, p) \
3684 	__param_check(name, p, unsigned int);
3685 
3686 module_param_named(min_resvport, xprt_min_resvport, portnr, 0644);
3687 module_param_named(max_resvport, xprt_max_resvport, portnr, 0644);
3688 
param_set_slot_table_size(const char * val,const struct kernel_param * kp)3689 static int param_set_slot_table_size(const char *val,
3690 				     const struct kernel_param *kp)
3691 {
3692 	return param_set_uint_minmax(val, kp,
3693 			RPC_MIN_SLOT_TABLE,
3694 			RPC_MAX_SLOT_TABLE);
3695 }
3696 
3697 static const struct kernel_param_ops param_ops_slot_table_size = {
3698 	.set = param_set_slot_table_size,
3699 	.get = param_get_uint,
3700 };
3701 
3702 #define param_check_slot_table_size(name, p) \
3703 	__param_check(name, p, unsigned int);
3704 
param_set_max_slot_table_size(const char * val,const struct kernel_param * kp)3705 static int param_set_max_slot_table_size(const char *val,
3706 				     const struct kernel_param *kp)
3707 {
3708 	return param_set_uint_minmax(val, kp,
3709 			RPC_MIN_SLOT_TABLE,
3710 			RPC_MAX_SLOT_TABLE_LIMIT);
3711 }
3712 
3713 static const struct kernel_param_ops param_ops_max_slot_table_size = {
3714 	.set = param_set_max_slot_table_size,
3715 	.get = param_get_uint,
3716 };
3717 
3718 #define param_check_max_slot_table_size(name, p) \
3719 	__param_check(name, p, unsigned int);
3720 
3721 module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
3722 		   slot_table_size, 0644);
3723 module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
3724 		   max_slot_table_size, 0644);
3725 module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
3726 		   slot_table_size, 0644);
3727