svcsock.c

   1// SPDX-License-Identifier: GPL-2.0-only
   2/*
   3 * linux/net/sunrpc/svcsock.c
   4 *
   5 * These are the RPC server socket internals.
   6 *
   7 * The server scheduling algorithm does not always distribute the load
   8 * evenly when servicing a single client. May need to modify the
   9 * svc_xprt_enqueue procedure...
  10 *
  11 * TCP support is largely untested and may be a little slow. The problem
  12 * is that we currently do two separate recvfrom's, one for the 4-byte
  13 * record length, and the second for the actual record. This could possibly
  14 * be improved by always reading a minimum size of around 100 bytes and
  15 * tucking any superfluous bytes away in a temporary store. Still, that
  16 * leaves write requests out in the rain. An alternative may be to peek at
  17 * the first skb in the queue, and if it matches the next TCP sequence
  18 * number, to extract the record marker. Yuck.
  19 *
  20 * Copyright (C) 1995, 1996 Olaf Kirch <okir@monad.swb.de>
  21 */
  22
  23#include <linux/kernel.h>
  24#include <linux/sched.h>
  25#include <linux/module.h>
  26#include <linux/errno.h>
  27#include <linux/fcntl.h>
  28#include <linux/net.h>
  29#include <linux/in.h>
  30#include <linux/inet.h>
  31#include <linux/udp.h>
  32#include <linux/tcp.h>
  33#include <linux/unistd.h>
  34#include <linux/slab.h>
  35#include <linux/netdevice.h>
  36#include <linux/skbuff.h>
  37#include <linux/file.h>
  38#include <linux/freezer.h>
  39#include <linux/bvec.h>
  40
  41#include <net/sock.h>
  42#include <net/checksum.h>
  43#include <net/ip.h>
  44#include <net/ipv6.h>
  45#include <net/udp.h>
  46#include <net/tcp.h>
  47#include <net/tcp_states.h>
  48#include <net/tls_prot.h>
  49#include <net/handshake.h>
  50#include <linux/uaccess.h>
  51#include <linux/highmem.h>
  52#include <asm/ioctls.h>
  53#include <linux/key.h>
  54
  55#include <linux/sunrpc/types.h>
  56#include <linux/sunrpc/clnt.h>
  57#include <linux/sunrpc/xdr.h>
  58#include <linux/sunrpc/msg_prot.h>
  59#include <linux/sunrpc/svcsock.h>
  60#include <linux/sunrpc/stats.h>
  61#include <linux/sunrpc/xprt.h>
  62
  63#include <trace/events/sock.h>
  64#include <trace/events/sunrpc.h>
  65
  66#include "socklib.h"
  67#include "sunrpc.h"
  68
  69#define RPCDBG_FACILITY	RPCDBG_SVCXPRT
  70
  71/*
  72 * For UDP:
  73 * 1 for header page
  74 * enough pages for RPCSVC_MAXPAYLOAD_UDP
  75 * 1 in case payload is not aligned
  76 * 1 for tail page
  77 */
  78enum {
  79	SUNRPC_MAX_UDP_SENDPAGES = 1 + RPCSVC_MAXPAYLOAD_UDP / PAGE_SIZE + 1 + 1
  80};
  81
  82/* To-do: to avoid tying up an nfsd thread while waiting for a
  83 * handshake request, the request could instead be deferred.
  84 */
  85enum {
  86	SVC_HANDSHAKE_TO	= 5U * HZ
  87};
  88
  89static struct svc_sock *svc_setup_socket(struct svc_serv *, struct socket *,
  90					 int flags);
  91static int		svc_udp_recvfrom(struct svc_rqst *);
  92static int		svc_udp_sendto(struct svc_rqst *);
  93static void		svc_sock_detach(struct svc_xprt *);
  94static void		svc_tcp_sock_detach(struct svc_xprt *);
  95static void		svc_sock_free(struct svc_xprt *);
  96
  97static struct svc_xprt *svc_create_socket(struct svc_serv *, int,
  98					  struct net *, struct sockaddr *,
  99					  int, int);
 100#ifdef CONFIG_DEBUG_LOCK_ALLOC
 101static struct lock_class_key svc_key[2];
 102static struct lock_class_key svc_slock_key[2];
 103
 104static void svc_reclassify_socket(struct socket *sock)
 105{
 106	struct sock *sk = sock->sk;
 107
 108	if (WARN_ON_ONCE(!sock_allow_reclassification(sk)))
 109		return;
 110
 111	switch (sk->sk_family) {
 112	case AF_INET:
 113		sock_lock_init_class_and_name(sk, "slock-AF_INET-NFSD",
 114					      &svc_slock_key[0],
 115					      "sk_xprt.xpt_lock-AF_INET-NFSD",
 116					      &svc_key[0]);
 117		break;
 118
 119	case AF_INET6:
 120		sock_lock_init_class_and_name(sk, "slock-AF_INET6-NFSD",
 121					      &svc_slock_key[1],
 122					      "sk_xprt.xpt_lock-AF_INET6-NFSD",
 123					      &svc_key[1]);
 124		break;
 125
 126	default:
 127		BUG();
 128	}
 129}
 130#else
 131static void svc_reclassify_socket(struct socket *sock)
 132{
 133}
 134#endif
 135
 136/**
 137 * svc_tcp_release_ctxt - Release transport-related resources
 138 * @xprt: the transport which owned the context
 139 * @ctxt: the context from rqstp->rq_xprt_ctxt or dr->xprt_ctxt
 140 *
 141 */
 142static void svc_tcp_release_ctxt(struct svc_xprt *xprt, void *ctxt)
 143{
 144}
 145
 146/**
 147 * svc_udp_release_ctxt - Release transport-related resources
 148 * @xprt: the transport which owned the context
 149 * @ctxt: the context from rqstp->rq_xprt_ctxt or dr->xprt_ctxt
 150 *
 151 */
 152static void svc_udp_release_ctxt(struct svc_xprt *xprt, void *ctxt)
 153{
 154	struct sk_buff *skb = ctxt;
 155
 156	if (skb)
 157		consume_skb(skb);
 158}
 159
 160union svc_pktinfo_u {
 161	struct in_pktinfo pkti;
 162	struct in6_pktinfo pkti6;
 163};
 164#define SVC_PKTINFO_SPACE \
 165	CMSG_SPACE(sizeof(union svc_pktinfo_u))
 166
 167static void svc_set_cmsg_data(struct svc_rqst *rqstp, struct cmsghdr *cmh)
 168{
 169	struct svc_sock *svsk =
 170		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
 171	switch (svsk->sk_sk->sk_family) {
 172	case AF_INET: {
 173			struct in_pktinfo *pki = CMSG_DATA(cmh);
 174
 175			cmh->cmsg_level = SOL_IP;
 176			cmh->cmsg_type = IP_PKTINFO;
 177			pki->ipi_ifindex = 0;
 178			pki->ipi_spec_dst.s_addr =
 179				 svc_daddr_in(rqstp)->sin_addr.s_addr;
 180			cmh->cmsg_len = CMSG_LEN(sizeof(*pki));
 181		}
 182		break;
 183
 184	case AF_INET6: {
 185			struct in6_pktinfo *pki = CMSG_DATA(cmh);
 186			struct sockaddr_in6 *daddr = svc_daddr_in6(rqstp);
 187
 188			cmh->cmsg_level = SOL_IPV6;
 189			cmh->cmsg_type = IPV6_PKTINFO;
 190			pki->ipi6_ifindex = daddr->sin6_scope_id;
 191			pki->ipi6_addr = daddr->sin6_addr;
 192			cmh->cmsg_len = CMSG_LEN(sizeof(*pki));
 193		}
 194		break;
 195	}
 196}
 197
 198static int svc_sock_result_payload(struct svc_rqst *rqstp, unsigned int offset,
 199				   unsigned int length)
 200{
 201	return 0;
 202}
 203
 204/*
 205 * Report socket names for nfsdfs
 206 */
 207static int svc_one_sock_name(struct svc_sock *svsk, char *buf, int remaining)
 208{
 209	const struct sock *sk = svsk->sk_sk;
 210	const char *proto_name = sk->sk_protocol == IPPROTO_UDP ?
 211							"udp" : "tcp";
 212	int len;
 213
 214	switch (sk->sk_family) {
 215	case PF_INET:
 216		len = snprintf(buf, remaining, "ipv4 %s %pI4 %d\n",
 217				proto_name,
 218				&inet_sk(sk)->inet_rcv_saddr,
 219				inet_sk(sk)->inet_num);
 220		break;
 221#if IS_ENABLED(CONFIG_IPV6)
 222	case PF_INET6:
 223		len = snprintf(buf, remaining, "ipv6 %s %pI6 %d\n",
 224				proto_name,
 225				&sk->sk_v6_rcv_saddr,
 226				inet_sk(sk)->inet_num);
 227		break;
 228#endif
 229	default:
 230		len = snprintf(buf, remaining, "*unknown-%d*\n",
 231				sk->sk_family);
 232	}
 233
 234	if (len >= remaining) {
 235		*buf = '\0';
 236		return -ENAMETOOLONG;
 237	}
 238	return len;
 239}
 240
 241static int
 242svc_tcp_sock_process_cmsg(struct socket *sock, struct msghdr *msg,
 243			  struct cmsghdr *cmsg, int ret)
 244{
 245	u8 content_type = tls_get_record_type(sock->sk, cmsg);
 246	u8 level, description;
 247
 248	switch (content_type) {
 249	case 0:
 250		break;
 251	case TLS_RECORD_TYPE_DATA:
 252		/* TLS sets EOR at the end of each application data
 253		 * record, even though there might be more frames
 254		 * waiting to be decrypted.
 255		 */
 256		msg->msg_flags &= ~MSG_EOR;
 257		break;
 258	case TLS_RECORD_TYPE_ALERT:
 259		tls_alert_recv(sock->sk, msg, &level, &description);
 260		ret = (level == TLS_ALERT_LEVEL_FATAL) ?
 261			-ENOTCONN : -EAGAIN;
 262		break;
 263	default:
 264		/* discard this record type */
 265		ret = -EAGAIN;
 266	}
 267	return ret;
 268}
 269
 270static int
 271svc_tcp_sock_recv_cmsg(struct socket *sock, unsigned int *msg_flags)
 272{
 273	union {
 274		struct cmsghdr	cmsg;
 275		u8		buf[CMSG_SPACE(sizeof(u8))];
 276	} u;
 277	u8 alert[2];
 278	struct kvec alert_kvec = {
 279		.iov_base = alert,
 280		.iov_len = sizeof(alert),
 281	};
 282	struct msghdr msg = {
 283		.msg_flags = *msg_flags,
 284		.msg_control = &u,
 285		.msg_controllen = sizeof(u),
 286	};
 287	int ret;
 288
 289	iov_iter_kvec(&msg.msg_iter, ITER_DEST, &alert_kvec, 1,
 290		      alert_kvec.iov_len);
 291	ret = sock_recvmsg(sock, &msg, MSG_DONTWAIT);
 292	if (ret > 0 &&
 293	    tls_get_record_type(sock->sk, &u.cmsg) == TLS_RECORD_TYPE_ALERT) {
 294		iov_iter_revert(&msg.msg_iter, ret);
 295		ret = svc_tcp_sock_process_cmsg(sock, &msg, &u.cmsg, -EAGAIN);
 296	}
 297	return ret;
 298}
 299
 300static int
 301svc_tcp_sock_recvmsg(struct svc_sock *svsk, struct msghdr *msg)
 302{
 303	int ret;
 304	struct socket *sock = svsk->sk_sock;
 305
 306	ret = sock_recvmsg(sock, msg, MSG_DONTWAIT);
 307	if (msg->msg_flags & MSG_CTRUNC) {
 308		msg->msg_flags &= ~(MSG_CTRUNC | MSG_EOR);
 309		if (ret == 0 || ret == -EIO)
 310			ret = svc_tcp_sock_recv_cmsg(sock, &msg->msg_flags);
 311	}
 312	return ret;
 313}
 314
 315#if ARCH_IMPLEMENTS_FLUSH_DCACHE_PAGE
 316static void svc_flush_bvec(const struct bio_vec *bvec, size_t size, size_t seek)
 317{
 318	struct bvec_iter bi = {
 319		.bi_size	= size + seek,
 320	};
 321	struct bio_vec bv;
 322
 323	bvec_iter_advance(bvec, &bi, seek & PAGE_MASK);
 324	for_each_bvec(bv, bvec, bi, bi)
 325		flush_dcache_page(bv.bv_page);
 326}
 327#else
 328static inline void svc_flush_bvec(const struct bio_vec *bvec, size_t size,
 329				  size_t seek)
 330{
 331}
 332#endif
 333
 334/*
 335 * Read from @rqstp's transport socket. The incoming message fills whole
 336 * pages in @rqstp's rq_pages array until the last page of the message
 337 * has been received into a partial page.
 338 */
 339static ssize_t svc_tcp_read_msg(struct svc_rqst *rqstp, size_t buflen,
 340				size_t seek)
 341{
 342	struct svc_sock *svsk =
 343		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
 344	struct bio_vec *bvec = rqstp->rq_bvec;
 345	struct msghdr msg = { NULL };
 346	unsigned int i;
 347	ssize_t len;
 348	size_t t;
 349
 350	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 351
 352	for (i = 0, t = 0; t < buflen; i++, t += PAGE_SIZE)
 353		bvec_set_page(&bvec[i], rqstp->rq_pages[i], PAGE_SIZE, 0);
 354	rqstp->rq_respages = &rqstp->rq_pages[i];
 355	rqstp->rq_next_page = rqstp->rq_respages + 1;
 356
 357	iov_iter_bvec(&msg.msg_iter, ITER_DEST, bvec, i, buflen);
 358	if (seek) {
 359		iov_iter_advance(&msg.msg_iter, seek);
 360		buflen -= seek;
 361	}
 362	len = svc_tcp_sock_recvmsg(svsk, &msg);
 363	if (len > 0)
 364		svc_flush_bvec(bvec, len, seek);
 365
 366	/* If we read a full record, then assume there may be more
 367	 * data to read (stream based sockets only!)
 368	 */
 369	if (len == buflen)
 370		set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 371
 372	return len;
 373}
 374
 375/*
 376 * Set socket snd and rcv buffer lengths
 377 */
 378static void svc_sock_setbufsize(struct svc_sock *svsk, unsigned int nreqs)
 379{
 380	unsigned int max_mesg = svsk->sk_xprt.xpt_server->sv_max_mesg;
 381	struct socket *sock = svsk->sk_sock;
 382
 383	nreqs = min(nreqs, INT_MAX / 2 / max_mesg);
 384
 385	lock_sock(sock->sk);
 386	sock->sk->sk_sndbuf = nreqs * max_mesg * 2;
 387	sock->sk->sk_rcvbuf = nreqs * max_mesg * 2;
 388	sock->sk->sk_write_space(sock->sk);
 389	release_sock(sock->sk);
 390}
 391
 392static void svc_sock_secure_port(struct svc_rqst *rqstp)
 393{
 394	if (svc_port_is_privileged(svc_addr(rqstp)))
 395		set_bit(RQ_SECURE, &rqstp->rq_flags);
 396	else
 397		clear_bit(RQ_SECURE, &rqstp->rq_flags);
 398}
 399
 400/*
 401 * INET callback when data has been received on the socket.
 402 */
 403static void svc_data_ready(struct sock *sk)
 404{
 405	struct svc_sock	*svsk = (struct svc_sock *)sk->sk_user_data;
 406
 407	trace_sk_data_ready(sk);
 408
 409	if (svsk) {
 410		/* Refer to svc_setup_socket() for details. */
 411		rmb();
 412		svsk->sk_odata(sk);
 413		trace_svcsock_data_ready(&svsk->sk_xprt, 0);
 414		if (test_bit(XPT_HANDSHAKE, &svsk->sk_xprt.xpt_flags))
 415			return;
 416		if (!test_and_set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags))
 417			svc_xprt_enqueue(&svsk->sk_xprt);
 418	}
 419}
 420
 421/*
 422 * INET callback when space is newly available on the socket.
 423 */
 424static void svc_write_space(struct sock *sk)
 425{
 426	struct svc_sock	*svsk = (struct svc_sock *)(sk->sk_user_data);
 427
 428	if (svsk) {
 429		/* Refer to svc_setup_socket() for details. */
 430		rmb();
 431		trace_svcsock_write_space(&svsk->sk_xprt, 0);
 432		svsk->sk_owspace(sk);
 433		svc_xprt_enqueue(&svsk->sk_xprt);
 434	}
 435}
 436
 437static int svc_tcp_has_wspace(struct svc_xprt *xprt)
 438{
 439	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
 440
 441	if (test_bit(XPT_LISTENER, &xprt->xpt_flags))
 442		return 1;
 443	return !test_bit(SOCK_NOSPACE, &svsk->sk_sock->flags);
 444}
 445
 446static void svc_tcp_kill_temp_xprt(struct svc_xprt *xprt)
 447{
 448	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
 449
 450	sock_no_linger(svsk->sk_sock->sk);
 451}
 452
 453/**
 454 * svc_tcp_handshake_done - Handshake completion handler
 455 * @data: address of xprt to wake
 456 * @status: status of handshake
 457 * @peerid: serial number of key containing the remote peer's identity
 458 *
 459 * If a security policy is specified as an export option, we don't
 460 * have a specific export here to check. So we set a "TLS session
 461 * is present" flag on the xprt and let an upper layer enforce local
 462 * security policy.
 463 */
 464static void svc_tcp_handshake_done(void *data, int status, key_serial_t peerid)
 465{
 466	struct svc_xprt *xprt = data;
 467	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
 468
 469	if (!status) {
 470		if (peerid != TLS_NO_PEERID)
 471			set_bit(XPT_PEER_AUTH, &xprt->xpt_flags);
 472		set_bit(XPT_TLS_SESSION, &xprt->xpt_flags);
 473	}
 474	clear_bit(XPT_HANDSHAKE, &xprt->xpt_flags);
 475	complete_all(&svsk->sk_handshake_done);
 476}
 477
 478/**
 479 * svc_tcp_handshake - Perform a transport-layer security handshake
 480 * @xprt: connected transport endpoint
 481 *
 482 */
 483static void svc_tcp_handshake(struct svc_xprt *xprt)
 484{
 485	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
 486	struct sock *sk = svsk->sk_sock->sk;
 487	struct tls_handshake_args args = {
 488		.ta_sock	= svsk->sk_sock,
 489		.ta_done	= svc_tcp_handshake_done,
 490		.ta_data	= xprt,
 491	};
 492	int ret;
 493
 494	trace_svc_tls_upcall(xprt);
 495
 496	clear_bit(XPT_TLS_SESSION, &xprt->xpt_flags);
 497	init_completion(&svsk->sk_handshake_done);
 498
 499	ret = tls_server_hello_x509(&args, GFP_KERNEL);
 500	if (ret) {
 501		trace_svc_tls_not_started(xprt);
 502		goto out_failed;
 503	}
 504
 505	ret = wait_for_completion_interruptible_timeout(&svsk->sk_handshake_done,
 506							SVC_HANDSHAKE_TO);
 507	if (ret <= 0) {
 508		if (tls_handshake_cancel(sk)) {
 509			trace_svc_tls_timed_out(xprt);
 510			goto out_close;
 511		}
 512	}
 513
 514	if (!test_bit(XPT_TLS_SESSION, &xprt->xpt_flags)) {
 515		trace_svc_tls_unavailable(xprt);
 516		goto out_close;
 517	}
 518
 519	/* Mark the transport ready in case the remote sent RPC
 520	 * traffic before the kernel received the handshake
 521	 * completion downcall.
 522	 */
 523	set_bit(XPT_DATA, &xprt->xpt_flags);
 524	svc_xprt_enqueue(xprt);
 525	return;
 526
 527out_close:
 528	set_bit(XPT_CLOSE, &xprt->xpt_flags);
 529out_failed:
 530	clear_bit(XPT_HANDSHAKE, &xprt->xpt_flags);
 531	set_bit(XPT_DATA, &xprt->xpt_flags);
 532	svc_xprt_enqueue(xprt);
 533}
 534
 535/*
 536 * See net/ipv6/ip_sockglue.c : ip_cmsg_recv_pktinfo
 537 */
 538static int svc_udp_get_dest_address4(struct svc_rqst *rqstp,
 539				     struct cmsghdr *cmh)
 540{
 541	struct in_pktinfo *pki = CMSG_DATA(cmh);
 542	struct sockaddr_in *daddr = svc_daddr_in(rqstp);
 543
 544	if (cmh->cmsg_type != IP_PKTINFO)
 545		return 0;
 546
 547	daddr->sin_family = AF_INET;
 548	daddr->sin_addr.s_addr = pki->ipi_spec_dst.s_addr;
 549	return 1;
 550}
 551
 552/*
 553 * See net/ipv6/datagram.c : ip6_datagram_recv_ctl
 554 */
 555static int svc_udp_get_dest_address6(struct svc_rqst *rqstp,
 556				     struct cmsghdr *cmh)
 557{
 558	struct in6_pktinfo *pki = CMSG_DATA(cmh);
 559	struct sockaddr_in6 *daddr = svc_daddr_in6(rqstp);
 560
 561	if (cmh->cmsg_type != IPV6_PKTINFO)
 562		return 0;
 563
 564	daddr->sin6_family = AF_INET6;
 565	daddr->sin6_addr = pki->ipi6_addr;
 566	daddr->sin6_scope_id = pki->ipi6_ifindex;
 567	return 1;
 568}
 569
 570/*
 571 * Copy the UDP datagram's destination address to the rqstp structure.
 572 * The 'destination' address in this case is the address to which the
 573 * peer sent the datagram, i.e. our local address. For multihomed
 574 * hosts, this can change from msg to msg. Note that only the IP
 575 * address changes, the port number should remain the same.
 576 */
 577static int svc_udp_get_dest_address(struct svc_rqst *rqstp,
 578				    struct cmsghdr *cmh)
 579{
 580	switch (cmh->cmsg_level) {
 581	case SOL_IP:
 582		return svc_udp_get_dest_address4(rqstp, cmh);
 583	case SOL_IPV6:
 584		return svc_udp_get_dest_address6(rqstp, cmh);
 585	}
 586
 587	return 0;
 588}
 589
 590/**
 591 * svc_udp_recvfrom - Receive a datagram from a UDP socket.
 592 * @rqstp: request structure into which to receive an RPC Call
 593 *
 594 * Called in a loop when XPT_DATA has been set.
 595 *
 596 * Returns:
 597 *   On success, the number of bytes in a received RPC Call, or
 598 *   %0 if a complete RPC Call message was not ready to return
 599 */
 600static int svc_udp_recvfrom(struct svc_rqst *rqstp)
 601{
 602	struct svc_sock	*svsk =
 603		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
 604	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
 605	struct sk_buff	*skb;
 606	union {
 607		struct cmsghdr	hdr;
 608		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
 609	} buffer;
 610	struct cmsghdr *cmh = &buffer.hdr;
 611	struct msghdr msg = {
 612		.msg_name = svc_addr(rqstp),
 613		.msg_control = cmh,
 614		.msg_controllen = sizeof(buffer),
 615		.msg_flags = MSG_DONTWAIT,
 616	};
 617	size_t len;
 618	int err;
 619
 620	if (test_and_clear_bit(XPT_CHNGBUF, &svsk->sk_xprt.xpt_flags))
 621	    /* udp sockets need large rcvbuf as all pending
 622	     * requests are still in that buffer.  sndbuf must
 623	     * also be large enough that there is enough space
 624	     * for one reply per thread.  We count all threads
 625	     * rather than threads in a particular pool, which
 626	     * provides an upper bound on the number of threads
 627	     * which will access the socket.
 628	     */
 629	    svc_sock_setbufsize(svsk, serv->sv_nrthreads + 3);
 630
 631	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 632	err = kernel_recvmsg(svsk->sk_sock, &msg, NULL,
 633			     0, 0, MSG_PEEK | MSG_DONTWAIT);
 634	if (err < 0)
 635		goto out_recv_err;
 636	skb = skb_recv_udp(svsk->sk_sk, MSG_DONTWAIT, &err);
 637	if (!skb)
 638		goto out_recv_err;
 639
 640	len = svc_addr_len(svc_addr(rqstp));
 641	rqstp->rq_addrlen = len;
 642	if (skb->tstamp == 0) {
 643		skb->tstamp = ktime_get_real();
 644		/* Don't enable netstamp, sunrpc doesn't
 645		   need that much accuracy */
 646	}
 647	sock_write_timestamp(svsk->sk_sk, skb->tstamp);
 648	set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags); /* there may be more data... */
 649
 650	len = skb->len;
 651	rqstp->rq_arg.len = len;
 652	trace_svcsock_udp_recv(&svsk->sk_xprt, len);
 653
 654	rqstp->rq_prot = IPPROTO_UDP;
 655
 656	if (!svc_udp_get_dest_address(rqstp, cmh))
 657		goto out_cmsg_err;
 658	rqstp->rq_daddrlen = svc_addr_len(svc_daddr(rqstp));
 659
 660	if (skb_is_nonlinear(skb)) {
 661		/* we have to copy */
 662		local_bh_disable();
 663		if (csum_partial_copy_to_xdr(&rqstp->rq_arg, skb))
 664			goto out_bh_enable;
 665		local_bh_enable();
 666		consume_skb(skb);
 667	} else {
 668		/* we can use it in-place */
 669		rqstp->rq_arg.head[0].iov_base = skb->data;
 670		rqstp->rq_arg.head[0].iov_len = len;
 671		if (skb_checksum_complete(skb))
 672			goto out_free;
 673		rqstp->rq_xprt_ctxt = skb;
 674	}
 675
 676	rqstp->rq_arg.page_base = 0;
 677	if (len <= rqstp->rq_arg.head[0].iov_len) {
 678		rqstp->rq_arg.head[0].iov_len = len;
 679		rqstp->rq_arg.page_len = 0;
 680		rqstp->rq_respages = rqstp->rq_pages+1;
 681	} else {
 682		rqstp->rq_arg.page_len = len - rqstp->rq_arg.head[0].iov_len;
 683		rqstp->rq_respages = rqstp->rq_pages + 1 +
 684			DIV_ROUND_UP(rqstp->rq_arg.page_len, PAGE_SIZE);
 685	}
 686	rqstp->rq_next_page = rqstp->rq_respages+1;
 687
 688	if (serv->sv_stats)
 689		serv->sv_stats->netudpcnt++;
 690
 691	svc_sock_secure_port(rqstp);
 692	svc_xprt_received(rqstp->rq_xprt);
 693	return len;
 694
 695out_recv_err:
 696	if (err != -EAGAIN) {
 697		/* possibly an icmp error */
 698		set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 699	}
 700	trace_svcsock_udp_recv_err(&svsk->sk_xprt, err);
 701	goto out_clear_busy;
 702out_cmsg_err:
 703	net_warn_ratelimited("svc: received unknown control message %d/%d; dropping RPC reply datagram\n",
 704			     cmh->cmsg_level, cmh->cmsg_type);
 705	goto out_free;
 706out_bh_enable:
 707	local_bh_enable();
 708out_free:
 709	kfree_skb(skb);
 710out_clear_busy:
 711	svc_xprt_received(rqstp->rq_xprt);
 712	return 0;
 713}
 714
 715/**
 716 * svc_udp_sendto - Send out a reply on a UDP socket
 717 * @rqstp: completed svc_rqst
 718 *
 719 * xpt_mutex ensures @rqstp's whole message is written to the socket
 720 * without interruption.
 721 *
 722 * Returns the number of bytes sent, or a negative errno.
 723 */
 724static int svc_udp_sendto(struct svc_rqst *rqstp)
 725{
 726	struct svc_xprt *xprt = rqstp->rq_xprt;
 727	struct svc_sock	*svsk = container_of(xprt, struct svc_sock, sk_xprt);
 728	struct xdr_buf *xdr = &rqstp->rq_res;
 729	union {
 730		struct cmsghdr	hdr;
 731		long		all[SVC_PKTINFO_SPACE / sizeof(long)];
 732	} buffer;
 733	struct cmsghdr *cmh = &buffer.hdr;
 734	struct msghdr msg = {
 735		.msg_name	= &rqstp->rq_addr,
 736		.msg_namelen	= rqstp->rq_addrlen,
 737		.msg_control	= cmh,
 738		.msg_flags	= MSG_SPLICE_PAGES,
 739		.msg_controllen	= sizeof(buffer),
 740	};
 741	unsigned int count;
 742	int err;
 743
 744	svc_udp_release_ctxt(xprt, rqstp->rq_xprt_ctxt);
 745	rqstp->rq_xprt_ctxt = NULL;
 746
 747	svc_set_cmsg_data(rqstp, cmh);
 748
 749	mutex_lock(&xprt->xpt_mutex);
 750
 751	if (svc_xprt_is_dead(xprt))
 752		goto out_notconn;
 753
 754	count = xdr_buf_to_bvec(svsk->sk_bvec, SUNRPC_MAX_UDP_SENDPAGES, xdr);
 755
 756	iov_iter_bvec(&msg.msg_iter, ITER_SOURCE, svsk->sk_bvec,
 757		      count, rqstp->rq_res.len);
 758	err = sock_sendmsg(svsk->sk_sock, &msg);
 759	if (err == -ECONNREFUSED) {
 760		/* ICMP error on earlier request. */
 761		iov_iter_bvec(&msg.msg_iter, ITER_SOURCE, svsk->sk_bvec,
 762			      count, rqstp->rq_res.len);
 763		err = sock_sendmsg(svsk->sk_sock, &msg);
 764	}
 765
 766	trace_svcsock_udp_send(xprt, err);
 767
 768	mutex_unlock(&xprt->xpt_mutex);
 769	return err;
 770
 771out_notconn:
 772	mutex_unlock(&xprt->xpt_mutex);
 773	return -ENOTCONN;
 774}
 775
 776static int svc_udp_has_wspace(struct svc_xprt *xprt)
 777{
 778	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
 779	struct svc_serv	*serv = xprt->xpt_server;
 780	unsigned long required;
 781
 782	/*
 783	 * Set the SOCK_NOSPACE flag before checking the available
 784	 * sock space.
 785	 */
 786	set_bit(SOCK_NOSPACE, &svsk->sk_sock->flags);
 787	required = atomic_read(&svsk->sk_xprt.xpt_reserved) + serv->sv_max_mesg;
 788	if (required*2 > sock_wspace(svsk->sk_sk))
 789		return 0;
 790	clear_bit(SOCK_NOSPACE, &svsk->sk_sock->flags);
 791	return 1;
 792}
 793
 794static struct svc_xprt *svc_udp_accept(struct svc_xprt *xprt)
 795{
 796	BUG();
 797	return NULL;
 798}
 799
 800static void svc_udp_kill_temp_xprt(struct svc_xprt *xprt)
 801{
 802}
 803
 804static struct svc_xprt *svc_udp_create(struct svc_serv *serv,
 805				       struct net *net,
 806				       struct sockaddr *sa, int salen,
 807				       int flags)
 808{
 809	return svc_create_socket(serv, IPPROTO_UDP, net, sa, salen, flags);
 810}
 811
 812static const struct svc_xprt_ops svc_udp_ops = {
 813	.xpo_create = svc_udp_create,
 814	.xpo_recvfrom = svc_udp_recvfrom,
 815	.xpo_sendto = svc_udp_sendto,
 816	.xpo_result_payload = svc_sock_result_payload,
 817	.xpo_release_ctxt = svc_udp_release_ctxt,
 818	.xpo_detach = svc_sock_detach,
 819	.xpo_free = svc_sock_free,
 820	.xpo_has_wspace = svc_udp_has_wspace,
 821	.xpo_accept = svc_udp_accept,
 822	.xpo_kill_temp_xprt = svc_udp_kill_temp_xprt,
 823};
 824
 825static struct svc_xprt_class svc_udp_class = {
 826	.xcl_name = "udp",
 827	.xcl_owner = THIS_MODULE,
 828	.xcl_ops = &svc_udp_ops,
 829	.xcl_max_payload = RPCSVC_MAXPAYLOAD_UDP,
 830	.xcl_ident = XPRT_TRANSPORT_UDP,
 831};
 832
 833static void svc_udp_init(struct svc_sock *svsk, struct svc_serv *serv)
 834{
 835	svc_xprt_init(sock_net(svsk->sk_sock->sk), &svc_udp_class,
 836		      &svsk->sk_xprt, serv);
 837	clear_bit(XPT_CACHE_AUTH, &svsk->sk_xprt.xpt_flags);
 838	svsk->sk_sk->sk_data_ready = svc_data_ready;
 839	svsk->sk_sk->sk_write_space = svc_write_space;
 840
 841	/* initialise setting must have enough space to
 842	 * receive and respond to one request.
 843	 * svc_udp_recvfrom will re-adjust if necessary
 844	 */
 845	svc_sock_setbufsize(svsk, 3);
 846
 847	/* data might have come in before data_ready set up */
 848	set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
 849	set_bit(XPT_CHNGBUF, &svsk->sk_xprt.xpt_flags);
 850	set_bit(XPT_RPCB_UNREG, &svsk->sk_xprt.xpt_flags);
 851
 852	/* make sure we get destination address info */
 853	switch (svsk->sk_sk->sk_family) {
 854	case AF_INET:
 855		ip_sock_set_pktinfo(svsk->sk_sock->sk);
 856		break;
 857	case AF_INET6:
 858		ip6_sock_set_recvpktinfo(svsk->sk_sock->sk);
 859		break;
 860	default:
 861		BUG();
 862	}
 863}
 864
 865/*
 866 * A data_ready event on a listening socket means there's a connection
 867 * pending. Do not use state_change as a substitute for it.
 868 */
 869static void svc_tcp_listen_data_ready(struct sock *sk)
 870{
 871	struct svc_sock	*svsk = (struct svc_sock *)sk->sk_user_data;
 872
 873	trace_sk_data_ready(sk);
 874
 875	/*
 876	 * This callback may called twice when a new connection
 877	 * is established as a child socket inherits everything
 878	 * from a parent LISTEN socket.
 879	 * 1) data_ready method of the parent socket will be called
 880	 *    when one of child sockets become ESTABLISHED.
 881	 * 2) data_ready method of the child socket may be called
 882	 *    when it receives data before the socket is accepted.
 883	 * In case of 2, we should ignore it silently and DO NOT
 884	 * dereference svsk.
 885	 */
 886	if (sk->sk_state != TCP_LISTEN)
 887		return;
 888
 889	if (svsk) {
 890		/* Refer to svc_setup_socket() for details. */
 891		rmb();
 892		svsk->sk_odata(sk);
 893		set_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags);
 894		svc_xprt_enqueue(&svsk->sk_xprt);
 895	}
 896}
 897
 898/*
 899 * A state change on a connected socket means it's dying or dead.
 900 */
 901static void svc_tcp_state_change(struct sock *sk)
 902{
 903	struct svc_sock	*svsk = (struct svc_sock *)sk->sk_user_data;
 904
 905	if (svsk) {
 906		/* Refer to svc_setup_socket() for details. */
 907		rmb();
 908		svsk->sk_ostate(sk);
 909		trace_svcsock_tcp_state(&svsk->sk_xprt, svsk->sk_sock);
 910		if (sk->sk_state != TCP_ESTABLISHED)
 911			svc_xprt_deferred_close(&svsk->sk_xprt);
 912	}
 913}
 914
 915/*
 916 * Accept a TCP connection
 917 */
 918static struct svc_xprt *svc_tcp_accept(struct svc_xprt *xprt)
 919{
 920	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
 921	struct sockaddr_storage addr;
 922	struct sockaddr	*sin = (struct sockaddr *) &addr;
 923	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
 924	struct socket	*sock = svsk->sk_sock;
 925	struct socket	*newsock;
 926	struct svc_sock	*newsvsk;
 927	int		err, slen;
 928
 929	if (!sock)
 930		return NULL;
 931
 932	clear_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags);
 933	err = kernel_accept(sock, &newsock, O_NONBLOCK);
 934	if (err < 0) {
 935		if (err != -EAGAIN)
 936			trace_svcsock_accept_err(xprt, serv->sv_name, err);
 937		return NULL;
 938	}
 939	if (IS_ERR(sock_alloc_file(newsock, O_NONBLOCK, NULL)))
 940		return NULL;
 941
 942	set_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags);
 943
 944	err = kernel_getpeername(newsock, sin);
 945	if (err < 0) {
 946		trace_svcsock_getpeername_err(xprt, serv->sv_name, err);
 947		goto failed;		/* aborted connection or whatever */
 948	}
 949	slen = err;
 950
 951	/* Reset the inherited callbacks before calling svc_setup_socket */
 952	newsock->sk->sk_state_change = svsk->sk_ostate;
 953	newsock->sk->sk_data_ready = svsk->sk_odata;
 954	newsock->sk->sk_write_space = svsk->sk_owspace;
 955
 956	/* make sure that a write doesn't block forever when
 957	 * low on memory
 958	 */
 959	newsock->sk->sk_sndtimeo = HZ*30;
 960
 961	newsvsk = svc_setup_socket(serv, newsock,
 962				 (SVC_SOCK_ANONYMOUS | SVC_SOCK_TEMPORARY));
 963	if (IS_ERR(newsvsk))
 964		goto failed;
 965	svc_xprt_set_remote(&newsvsk->sk_xprt, sin, slen);
 966	err = kernel_getsockname(newsock, sin);
 967	slen = err;
 968	if (unlikely(err < 0))
 969		slen = offsetof(struct sockaddr, sa_data);
 970	svc_xprt_set_local(&newsvsk->sk_xprt, sin, slen);
 971
 972	if (sock_is_loopback(newsock->sk))
 973		set_bit(XPT_LOCAL, &newsvsk->sk_xprt.xpt_flags);
 974	else
 975		clear_bit(XPT_LOCAL, &newsvsk->sk_xprt.xpt_flags);
 976	if (serv->sv_stats)
 977		serv->sv_stats->nettcpconn++;
 978
 979	return &newsvsk->sk_xprt;
 980
 981failed:
 982	sockfd_put(newsock);
 983	return NULL;
 984}
 985
 986static size_t svc_tcp_restore_pages(struct svc_sock *svsk,
 987				    struct svc_rqst *rqstp)
 988{
 989	size_t len = svsk->sk_datalen;
 990	unsigned int i, npages;
 991
 992	if (!len)
 993		return 0;
 994	npages = (len + PAGE_SIZE - 1) >> PAGE_SHIFT;
 995	for (i = 0; i < npages; i++) {
 996		if (rqstp->rq_pages[i] != NULL)
 997			put_page(rqstp->rq_pages[i]);
 998		BUG_ON(svsk->sk_pages[i] == NULL);
 999		rqstp->rq_pages[i] = svsk->sk_pages[i];
1000		svsk->sk_pages[i] = NULL;
1001	}
1002	rqstp->rq_arg.head[0].iov_base = page_address(rqstp->rq_pages[0]);
1003	return len;
1004}
1005
1006static void svc_tcp_save_pages(struct svc_sock *svsk, struct svc_rqst *rqstp)
1007{
1008	unsigned int i, len, npages;
1009
1010	if (svsk->sk_datalen == 0)
1011		return;
1012	len = svsk->sk_datalen;
1013	npages = (len + PAGE_SIZE - 1) >> PAGE_SHIFT;
1014	for (i = 0; i < npages; i++) {
1015		svsk->sk_pages[i] = rqstp->rq_pages[i];
1016		rqstp->rq_pages[i] = NULL;
1017	}
1018}
1019
1020static void svc_tcp_clear_pages(struct svc_sock *svsk)
1021{
1022	unsigned int i, len, npages;
1023
1024	if (svsk->sk_datalen == 0)
1025		goto out;
1026	len = svsk->sk_datalen;
1027	npages = (len + PAGE_SIZE - 1) >> PAGE_SHIFT;
1028	for (i = 0; i < npages; i++) {
1029		if (svsk->sk_pages[i] == NULL) {
1030			WARN_ON_ONCE(1);
1031			continue;
1032		}
1033		put_page(svsk->sk_pages[i]);
1034		svsk->sk_pages[i] = NULL;
1035	}
1036out:
1037	svsk->sk_tcplen = 0;
1038	svsk->sk_datalen = 0;
1039}
1040
1041/*
1042 * Receive fragment record header into sk_marker.
1043 */
1044static ssize_t svc_tcp_read_marker(struct svc_sock *svsk,
1045				   struct svc_rqst *rqstp)
1046{
1047	ssize_t want, len;
1048
1049	/* If we haven't gotten the record length yet,
1050	 * get the next four bytes.
1051	 */
1052	if (svsk->sk_tcplen < sizeof(rpc_fraghdr)) {
1053		struct msghdr	msg = { NULL };
1054		struct kvec	iov;
1055
1056		want = sizeof(rpc_fraghdr) - svsk->sk_tcplen;
1057		iov.iov_base = ((char *)&svsk->sk_marker) + svsk->sk_tcplen;
1058		iov.iov_len  = want;
1059		iov_iter_kvec(&msg.msg_iter, ITER_DEST, &iov, 1, want);
1060		len = svc_tcp_sock_recvmsg(svsk, &msg);
1061		if (len < 0)
1062			return len;
1063		svsk->sk_tcplen += len;
1064		if (len < want) {
1065			/* call again to read the remaining bytes */
1066			goto err_short;
1067		}
1068		trace_svcsock_marker(&svsk->sk_xprt, svsk->sk_marker);
1069		if (svc_sock_reclen(svsk) + svsk->sk_datalen >
1070		    svsk->sk_xprt.xpt_server->sv_max_mesg)
1071			goto err_too_large;
1072	}
1073	return svc_sock_reclen(svsk);
1074
1075err_too_large:
1076	net_notice_ratelimited("svc: %s oversized RPC fragment (%u octets) from %pISpc\n",
1077			       svsk->sk_xprt.xpt_server->sv_name,
1078			       svc_sock_reclen(svsk),
1079			       (struct sockaddr *)&svsk->sk_xprt.xpt_remote);
1080	svc_xprt_deferred_close(&svsk->sk_xprt);
1081err_short:
1082	return -EAGAIN;
1083}
1084
1085static int receive_cb_reply(struct svc_sock *svsk, struct svc_rqst *rqstp)
1086{
1087	struct rpc_xprt *bc_xprt = svsk->sk_xprt.xpt_bc_xprt;
1088	struct rpc_rqst *req = NULL;
1089	struct kvec *src, *dst;
1090	__be32 *p = (__be32 *)rqstp->rq_arg.head[0].iov_base;
1091	__be32 xid = *p;
1092
1093	if (!bc_xprt)
1094		return -EAGAIN;
1095	spin_lock(&bc_xprt->queue_lock);
1096	req = xprt_lookup_rqst(bc_xprt, xid);
1097	if (!req)
1098		goto unlock_eagain;
1099
1100	memcpy(&req->rq_private_buf, &req->rq_rcv_buf, sizeof(struct xdr_buf));
1101	/*
1102	 * XXX!: cheating for now!  Only copying HEAD.
1103	 * But we know this is good enough for now (in fact, for any
1104	 * callback reply in the forseeable future).
1105	 */
1106	dst = &req->rq_private_buf.head[0];
1107	src = &rqstp->rq_arg.head[0];
1108	if (dst->iov_len < src->iov_len)
1109		goto unlock_eagain; /* whatever; just giving up. */
1110	memcpy(dst->iov_base, src->iov_base, src->iov_len);
1111	xprt_complete_rqst(req->rq_task, rqstp->rq_arg.len);
1112	rqstp->rq_arg.len = 0;
1113	spin_unlock(&bc_xprt->queue_lock);
1114	return 0;
1115unlock_eagain:
1116	spin_unlock(&bc_xprt->queue_lock);
1117	return -EAGAIN;
1118}
1119
1120static void svc_tcp_fragment_received(struct svc_sock *svsk)
1121{
1122	/* If we have more data, signal svc_xprt_enqueue() to try again */
1123	svsk->sk_tcplen = 0;
1124	svsk->sk_marker = xdr_zero;
1125}
1126
1127/**
1128 * svc_tcp_recvfrom - Receive data from a TCP socket
1129 * @rqstp: request structure into which to receive an RPC Call
1130 *
1131 * Called in a loop when XPT_DATA has been set.
1132 *
1133 * Read the 4-byte stream record marker, then use the record length
1134 * in that marker to set up exactly the resources needed to receive
1135 * the next RPC message into @rqstp.
1136 *
1137 * Returns:
1138 *   On success, the number of bytes in a received RPC Call, or
1139 *   %0 if a complete RPC Call message was not ready to return
1140 *
1141 * The zero return case handles partial receives and callback Replies.
1142 * The state of a partial receive is preserved in the svc_sock for
1143 * the next call to svc_tcp_recvfrom.
1144 */
1145static int svc_tcp_recvfrom(struct svc_rqst *rqstp)
1146{
1147	struct svc_sock	*svsk =
1148		container_of(rqstp->rq_xprt, struct svc_sock, sk_xprt);
1149	struct svc_serv	*serv = svsk->sk_xprt.xpt_server;
1150	size_t want, base;
1151	ssize_t len;
1152	__be32 *p;
1153	__be32 calldir;
1154
1155	clear_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
1156	len = svc_tcp_read_marker(svsk, rqstp);
1157	if (len < 0)
1158		goto error;
1159
1160	base = svc_tcp_restore_pages(svsk, rqstp);
1161	want = len - (svsk->sk_tcplen - sizeof(rpc_fraghdr));
1162	len = svc_tcp_read_msg(rqstp, base + want, base);
1163	if (len >= 0) {
1164		trace_svcsock_tcp_recv(&svsk->sk_xprt, len);
1165		svsk->sk_tcplen += len;
1166		svsk->sk_datalen += len;
1167	}
1168	if (len != want || !svc_sock_final_rec(svsk))
1169		goto err_incomplete;
1170	if (svsk->sk_datalen < 8)
1171		goto err_nuts;
1172
1173	rqstp->rq_arg.len = svsk->sk_datalen;
1174	rqstp->rq_arg.page_base = 0;
1175	if (rqstp->rq_arg.len <= rqstp->rq_arg.head[0].iov_len) {
1176		rqstp->rq_arg.head[0].iov_len = rqstp->rq_arg.len;
1177		rqstp->rq_arg.page_len = 0;
1178	} else
1179		rqstp->rq_arg.page_len = rqstp->rq_arg.len - rqstp->rq_arg.head[0].iov_len;
1180
1181	rqstp->rq_xprt_ctxt   = NULL;
1182	rqstp->rq_prot	      = IPPROTO_TCP;
1183	if (test_bit(XPT_LOCAL, &svsk->sk_xprt.xpt_flags))
1184		set_bit(RQ_LOCAL, &rqstp->rq_flags);
1185	else
1186		clear_bit(RQ_LOCAL, &rqstp->rq_flags);
1187
1188	p = (__be32 *)rqstp->rq_arg.head[0].iov_base;
1189	calldir = p[1];
1190	if (calldir)
1191		len = receive_cb_reply(svsk, rqstp);
1192
1193	/* Reset TCP read info */
1194	svsk->sk_datalen = 0;
1195	svc_tcp_fragment_received(svsk);
1196
1197	if (len < 0)
1198		goto error;
1199
1200	svc_xprt_copy_addrs(rqstp, &svsk->sk_xprt);
1201	if (serv->sv_stats)
1202		serv->sv_stats->nettcpcnt++;
1203
1204	svc_sock_secure_port(rqstp);
1205	svc_xprt_received(rqstp->rq_xprt);
1206	return rqstp->rq_arg.len;
1207
1208err_incomplete:
1209	svc_tcp_save_pages(svsk, rqstp);
1210	if (len < 0 && len != -EAGAIN)
1211		goto err_delete;
1212	if (len == want)
1213		svc_tcp_fragment_received(svsk);
1214	else
1215		trace_svcsock_tcp_recv_short(&svsk->sk_xprt,
1216				svc_sock_reclen(svsk),
1217				svsk->sk_tcplen - sizeof(rpc_fraghdr));
1218	goto err_noclose;
1219error:
1220	if (len != -EAGAIN)
1221		goto err_delete;
1222	trace_svcsock_tcp_recv_eagain(&svsk->sk_xprt, 0);
1223	goto err_noclose;
1224err_nuts:
1225	svsk->sk_datalen = 0;
1226err_delete:
1227	trace_svcsock_tcp_recv_err(&svsk->sk_xprt, len);
1228	svc_xprt_deferred_close(&svsk->sk_xprt);
1229err_noclose:
1230	svc_xprt_received(rqstp->rq_xprt);
1231	return 0;	/* record not complete */
1232}
1233
1234/*
1235 * MSG_SPLICE_PAGES is used exclusively to reduce the number of
1236 * copy operations in this path. Therefore the caller must ensure
1237 * that the pages backing @xdr are unchanging.
1238 */
1239static int svc_tcp_sendmsg(struct svc_sock *svsk, struct svc_rqst *rqstp,
1240			   rpc_fraghdr marker)
1241{
1242	struct msghdr msg = {
1243		.msg_flags	= MSG_SPLICE_PAGES,
1244	};
1245	unsigned int count;
1246	void *buf;
1247	int ret;
1248
1249	/* The stream record marker is copied into a temporary page
1250	 * fragment buffer so that it can be included in sk_bvec.
1251	 */
1252	buf = page_frag_alloc(&svsk->sk_frag_cache, sizeof(marker),
1253			      GFP_KERNEL);
1254	if (!buf)
1255		return -ENOMEM;
1256	memcpy(buf, &marker, sizeof(marker));
1257	bvec_set_virt(svsk->sk_bvec, buf, sizeof(marker));
1258
1259	count = xdr_buf_to_bvec(svsk->sk_bvec + 1, rqstp->rq_maxpages,
1260				&rqstp->rq_res);
1261
1262	iov_iter_bvec(&msg.msg_iter, ITER_SOURCE, svsk->sk_bvec,
1263		      1 + count, sizeof(marker) + rqstp->rq_res.len);
1264	ret = sock_sendmsg(svsk->sk_sock, &msg);
1265	page_frag_free(buf);
1266	return ret;
1267}
1268
1269/**
1270 * svc_tcp_sendto - Send out a reply on a TCP socket
1271 * @rqstp: completed svc_rqst
1272 *
1273 * xpt_mutex ensures @rqstp's whole message is written to the socket
1274 * without interruption.
1275 *
1276 * Returns the number of bytes sent, or a negative errno.
1277 */
1278static int svc_tcp_sendto(struct svc_rqst *rqstp)
1279{
1280	struct svc_xprt *xprt = rqstp->rq_xprt;
1281	struct svc_sock	*svsk = container_of(xprt, struct svc_sock, sk_xprt);
1282	struct xdr_buf *xdr = &rqstp->rq_res;
1283	rpc_fraghdr marker = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT |
1284					 (u32)xdr->len);
1285	int sent;
1286
1287	svc_tcp_release_ctxt(xprt, rqstp->rq_xprt_ctxt);
1288	rqstp->rq_xprt_ctxt = NULL;
1289
1290	mutex_lock(&xprt->xpt_mutex);
1291	if (svc_xprt_is_dead(xprt))
1292		goto out_notconn;
1293	sent = svc_tcp_sendmsg(svsk, rqstp, marker);
1294	trace_svcsock_tcp_send(xprt, sent);
1295	if (sent < 0 || sent != (xdr->len + sizeof(marker)))
1296		goto out_close;
1297	mutex_unlock(&xprt->xpt_mutex);
1298	return sent;
1299
1300out_notconn:
1301	mutex_unlock(&xprt->xpt_mutex);
1302	return -ENOTCONN;
1303out_close:
1304	pr_notice("rpc-srv/tcp: %s: %s %d when sending %zu bytes - shutting down socket\n",
1305		  xprt->xpt_server->sv_name,
1306		  (sent < 0) ? "got error" : "sent",
1307		  sent, xdr->len + sizeof(marker));
1308	svc_xprt_deferred_close(xprt);
1309	mutex_unlock(&xprt->xpt_mutex);
1310	return -EAGAIN;
1311}
1312
1313static struct svc_xprt *svc_tcp_create(struct svc_serv *serv,
1314				       struct net *net,
1315				       struct sockaddr *sa, int salen,
1316				       int flags)
1317{
1318	return svc_create_socket(serv, IPPROTO_TCP, net, sa, salen, flags);
1319}
1320
1321static const struct svc_xprt_ops svc_tcp_ops = {
1322	.xpo_create = svc_tcp_create,
1323	.xpo_recvfrom = svc_tcp_recvfrom,
1324	.xpo_sendto = svc_tcp_sendto,
1325	.xpo_result_payload = svc_sock_result_payload,
1326	.xpo_release_ctxt = svc_tcp_release_ctxt,
1327	.xpo_detach = svc_tcp_sock_detach,
1328	.xpo_free = svc_sock_free,
1329	.xpo_has_wspace = svc_tcp_has_wspace,
1330	.xpo_accept = svc_tcp_accept,
1331	.xpo_kill_temp_xprt = svc_tcp_kill_temp_xprt,
1332	.xpo_handshake = svc_tcp_handshake,
1333};
1334
1335static struct svc_xprt_class svc_tcp_class = {
1336	.xcl_name = "tcp",
1337	.xcl_owner = THIS_MODULE,
1338	.xcl_ops = &svc_tcp_ops,
1339	.xcl_max_payload = RPCSVC_MAXPAYLOAD_TCP,
1340	.xcl_ident = XPRT_TRANSPORT_TCP,
1341};
1342
1343void svc_init_xprt_sock(void)
1344{
1345	svc_reg_xprt_class(&svc_tcp_class);
1346	svc_reg_xprt_class(&svc_udp_class);
1347}
1348
1349void svc_cleanup_xprt_sock(void)
1350{
1351	svc_unreg_xprt_class(&svc_tcp_class);
1352	svc_unreg_xprt_class(&svc_udp_class);
1353}
1354
1355static void svc_tcp_init(struct svc_sock *svsk, struct svc_serv *serv)
1356{
1357	struct sock	*sk = svsk->sk_sk;
1358
1359	svc_xprt_init(sock_net(svsk->sk_sock->sk), &svc_tcp_class,
1360		      &svsk->sk_xprt, serv);
1361	set_bit(XPT_CACHE_AUTH, &svsk->sk_xprt.xpt_flags);
1362	set_bit(XPT_CONG_CTRL, &svsk->sk_xprt.xpt_flags);
1363	if (sk->sk_state == TCP_LISTEN) {
1364		strcpy(svsk->sk_xprt.xpt_remotebuf, "listener");
1365		set_bit(XPT_LISTENER, &svsk->sk_xprt.xpt_flags);
1366		set_bit(XPT_RPCB_UNREG, &svsk->sk_xprt.xpt_flags);
1367		sk->sk_data_ready = svc_tcp_listen_data_ready;
1368		set_bit(XPT_CONN, &svsk->sk_xprt.xpt_flags);
1369	} else {
1370		sk->sk_state_change = svc_tcp_state_change;
1371		sk->sk_data_ready = svc_data_ready;
1372		sk->sk_write_space = svc_write_space;
1373
1374		svsk->sk_marker = xdr_zero;
1375		svsk->sk_tcplen = 0;
1376		svsk->sk_datalen = 0;
1377		memset(&svsk->sk_pages[0], 0,
1378		       svsk->sk_maxpages * sizeof(struct page *));
1379
1380		tcp_sock_set_nodelay(sk);
1381
1382		set_bit(XPT_DATA, &svsk->sk_xprt.xpt_flags);
1383		switch (sk->sk_state) {
1384		case TCP_SYN_RECV:
1385		case TCP_ESTABLISHED:
1386			break;
1387		default:
1388			svc_xprt_deferred_close(&svsk->sk_xprt);
1389		}
1390	}
1391}
1392
1393void svc_sock_update_bufs(struct svc_serv *serv)
1394{
1395	/*
1396	 * The number of server threads has changed. Update
1397	 * rcvbuf and sndbuf accordingly on all sockets
1398	 */
1399	struct svc_sock *svsk;
1400
1401	spin_lock_bh(&serv->sv_lock);
1402	list_for_each_entry(svsk, &serv->sv_permsocks, sk_xprt.xpt_list)
1403		set_bit(XPT_CHNGBUF, &svsk->sk_xprt.xpt_flags);
1404	spin_unlock_bh(&serv->sv_lock);
1405}
1406
1407static int svc_sock_sendpages(struct svc_serv *serv, struct socket *sock, int flags)
1408{
1409	switch (sock->type) {
1410	case SOCK_STREAM:
1411		/* +1 for TCP record marker */
1412		if (flags & SVC_SOCK_TEMPORARY)
1413			return svc_serv_maxpages(serv) + 1;
1414		return 0;
1415	case SOCK_DGRAM:
1416		return SUNRPC_MAX_UDP_SENDPAGES;
1417	}
1418	return -EINVAL;
1419}
1420
1421/*
1422 * Initialize socket for RPC use and create svc_sock struct
1423 */
1424static struct svc_sock *svc_setup_socket(struct svc_serv *serv,
1425						struct socket *sock,
1426						int flags)
1427{
1428	struct svc_sock	*svsk;
1429	struct sock	*inet;
1430	int		pmap_register = !(flags & SVC_SOCK_ANONYMOUS);
1431	int		sendpages;
1432	unsigned long	pages;
1433
1434	sendpages = svc_sock_sendpages(serv, sock, flags);
1435	if (sendpages < 0)
1436		return ERR_PTR(sendpages);
1437
1438	pages = svc_serv_maxpages(serv);
1439	svsk = kzalloc(struct_size(svsk, sk_pages, pages), GFP_KERNEL);
1440	if (!svsk)
1441		return ERR_PTR(-ENOMEM);
1442
1443	if (sendpages) {
1444		svsk->sk_bvec = kcalloc(sendpages, sizeof(*svsk->sk_bvec), GFP_KERNEL);
1445		if (!svsk->sk_bvec) {
1446			kfree(svsk);
1447			return ERR_PTR(-ENOMEM);
1448		}
1449	}
1450
1451	svsk->sk_maxpages = pages;
1452
1453	inet = sock->sk;
1454
1455	if (pmap_register) {
1456		int err;
1457
1458		err = svc_register(serv, sock_net(sock->sk), inet->sk_family,
1459				     inet->sk_protocol,
1460				     ntohs(inet_sk(inet)->inet_sport));
1461		if (err < 0) {
1462			kfree(svsk->sk_bvec);
1463			kfree(svsk);
1464			return ERR_PTR(err);
1465		}
1466	}
1467
1468	svsk->sk_sock = sock;
1469	svsk->sk_sk = inet;
1470	svsk->sk_ostate = inet->sk_state_change;
1471	svsk->sk_odata = inet->sk_data_ready;
1472	svsk->sk_owspace = inet->sk_write_space;
1473	/*
1474	 * This barrier is necessary in order to prevent race condition
1475	 * with svc_data_ready(), svc_tcp_listen_data_ready(), and others
1476	 * when calling callbacks above.
1477	 */
1478	wmb();
1479	inet->sk_user_data = svsk;
1480
1481	/* Initialize the socket */
1482	if (sock->type == SOCK_DGRAM)
1483		svc_udp_init(svsk, serv);
1484	else
1485		svc_tcp_init(svsk, serv);
1486
1487	trace_svcsock_new(svsk, sock);
1488	return svsk;
1489}
1490
1491/**
1492 * svc_addsock - add a listener socket to an RPC service
1493 * @serv: pointer to RPC service to which to add a new listener
1494 * @net: caller's network namespace
1495 * @fd: file descriptor of the new listener
1496 * @name_return: pointer to buffer to fill in with name of listener
1497 * @len: size of the buffer
1498 * @cred: credential
1499 *
1500 * Fills in socket name and returns positive length of name if successful.
1501 * Name is terminated with '\n'.  On error, returns a negative errno
1502 * value.
1503 */
1504int svc_addsock(struct svc_serv *serv, struct net *net, const int fd,
1505		char *name_return, const size_t len, const struct cred *cred)
1506{
1507	int err = 0;
1508	struct socket *so = sockfd_lookup(fd, &err);
1509	struct svc_sock *svsk = NULL;
1510	struct sockaddr_storage addr;
1511	struct sockaddr *sin = (struct sockaddr *)&addr;
1512	int salen;
1513
1514	if (!so)
1515		return err;
1516	err = -EINVAL;
1517	if (sock_net(so->sk) != net)
1518		goto out;
1519	err = -EAFNOSUPPORT;
1520	if ((so->sk->sk_family != PF_INET) && (so->sk->sk_family != PF_INET6))
1521		goto out;
1522	err =  -EPROTONOSUPPORT;
1523	if (so->sk->sk_protocol != IPPROTO_TCP &&
1524	    so->sk->sk_protocol != IPPROTO_UDP)
1525		goto out;
1526	err = -EISCONN;
1527	if (so->state > SS_UNCONNECTED)
1528		goto out;
1529	err = -ENOENT;
1530	if (!try_module_get(THIS_MODULE))
1531		goto out;
1532	svsk = svc_setup_socket(serv, so, SVC_SOCK_DEFAULTS);
1533	if (IS_ERR(svsk)) {
1534		module_put(THIS_MODULE);
1535		err = PTR_ERR(svsk);
1536		goto out;
1537	}
1538	salen = kernel_getsockname(svsk->sk_sock, sin);
1539	if (salen >= 0)
1540		svc_xprt_set_local(&svsk->sk_xprt, sin, salen);
1541	svsk->sk_xprt.xpt_cred = get_cred(cred);
1542	svc_add_new_perm_xprt(serv, &svsk->sk_xprt);
1543	return svc_one_sock_name(svsk, name_return, len);
1544out:
1545	sockfd_put(so);
1546	return err;
1547}
1548EXPORT_SYMBOL_GPL(svc_addsock);
1549
1550/*
1551 * Create socket for RPC service.
1552 */
1553static struct svc_xprt *svc_create_socket(struct svc_serv *serv,
1554					  int protocol,
1555					  struct net *net,
1556					  struct sockaddr *sin, int len,
1557					  int flags)
1558{
1559	struct svc_sock	*svsk;
1560	struct socket	*sock;
1561	int		error;
1562	int		type;
1563	struct sockaddr_storage addr;
1564	struct sockaddr *newsin = (struct sockaddr *)&addr;
1565	int		newlen;
1566	int		family;
1567
1568	if (protocol != IPPROTO_UDP && protocol != IPPROTO_TCP) {
1569		printk(KERN_WARNING "svc: only UDP and TCP "
1570				"sockets supported\n");
1571		return ERR_PTR(-EINVAL);
1572	}
1573
1574	type = (protocol == IPPROTO_UDP)? SOCK_DGRAM : SOCK_STREAM;
1575	switch (sin->sa_family) {
1576	case AF_INET6:
1577		family = PF_INET6;
1578		break;
1579	case AF_INET:
1580		family = PF_INET;
1581		break;
1582	default:
1583		return ERR_PTR(-EINVAL);
1584	}
1585
1586	error = __sock_create(net, family, type, protocol, &sock, 1);
1587	if (error < 0)
1588		return ERR_PTR(error);
1589
1590	svc_reclassify_socket(sock);
1591
1592	/*
1593	 * If this is an PF_INET6 listener, we want to avoid
1594	 * getting requests from IPv4 remotes.  Those should
1595	 * be shunted to a PF_INET listener via rpcbind.
1596	 */
1597	if (family == PF_INET6)
1598		ip6_sock_set_v6only(sock->sk);
1599	if (type == SOCK_STREAM)
1600		sock->sk->sk_reuse = SK_CAN_REUSE; /* allow address reuse */
1601	error = kernel_bind(sock, (struct sockaddr_unsized *)sin, len);
1602	if (error < 0)
1603		goto bummer;
1604
1605	error = kernel_getsockname(sock, newsin);
1606	if (error < 0)
1607		goto bummer;
1608	newlen = error;
1609
1610	if (protocol == IPPROTO_TCP) {
1611		sk_net_refcnt_upgrade(sock->sk);
1612		if ((error = kernel_listen(sock, SOMAXCONN)) < 0)
1613			goto bummer;
1614	}
1615
1616	svsk = svc_setup_socket(serv, sock, flags);
1617	if (IS_ERR(svsk)) {
1618		error = PTR_ERR(svsk);
1619		goto bummer;
1620	}
1621	svc_xprt_set_local(&svsk->sk_xprt, newsin, newlen);
1622	return (struct svc_xprt *)svsk;
1623bummer:
1624	sock_release(sock);
1625	return ERR_PTR(error);
1626}
1627
1628/*
1629 * Detach the svc_sock from the socket so that no
1630 * more callbacks occur.
1631 */
1632static void svc_sock_detach(struct svc_xprt *xprt)
1633{
1634	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
1635	struct sock *sk = svsk->sk_sk;
1636
1637	/* put back the old socket callbacks */
1638	lock_sock(sk);
1639	sk->sk_state_change = svsk->sk_ostate;
1640	sk->sk_data_ready = svsk->sk_odata;
1641	sk->sk_write_space = svsk->sk_owspace;
1642	sk->sk_user_data = NULL;
1643	release_sock(sk);
1644}
1645
1646/*
1647 * Disconnect the socket, and reset the callbacks
1648 */
1649static void svc_tcp_sock_detach(struct svc_xprt *xprt)
1650{
1651	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
1652
1653	tls_handshake_close(svsk->sk_sock);
1654
1655	svc_sock_detach(xprt);
1656
1657	if (!test_bit(XPT_LISTENER, &xprt->xpt_flags)) {
1658		svc_tcp_clear_pages(svsk);
1659		kernel_sock_shutdown(svsk->sk_sock, SHUT_RDWR);
1660	}
1661}
1662
1663/*
1664 * Free the svc_sock's socket resources and the svc_sock itself.
1665 */
1666static void svc_sock_free(struct svc_xprt *xprt)
1667{
1668	struct svc_sock *svsk = container_of(xprt, struct svc_sock, sk_xprt);
1669	struct socket *sock = svsk->sk_sock;
1670
1671	trace_svcsock_free(svsk, sock);
1672
1673	tls_handshake_cancel(sock->sk);
1674	if (sock->file)
1675		sockfd_put(sock);
1676	else
1677		sock_release(sock);
1678
1679	page_frag_cache_drain(&svsk->sk_frag_cache);
1680	kfree(svsk->sk_bvec);
1681	kfree(svsk);
1682}