xprtsock.c 45.4 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13
/*
 * linux/net/sunrpc/xprtsock.c
 *
 * Client-side transport implementation for sockets.
 *
 * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com>
 * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com>
 * TCP NFS related read + write fixes
 *  (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
 *
 * Rewrite of larges part of the code in order to stabilize TCP stuff.
 * Fix behaviour when socket buffer is full.
 *  (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
14 15
 *
 * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com>
16 17 18 19 20 21 22 23 24 25 26 27 28 29
 */

#include <linux/types.h>
#include <linux/slab.h>
#include <linux/capability.h>
#include <linux/pagemap.h>
#include <linux/errno.h>
#include <linux/socket.h>
#include <linux/in.h>
#include <linux/net.h>
#include <linux/mm.h>
#include <linux/udp.h>
#include <linux/tcp.h>
#include <linux/sunrpc/clnt.h>
30
#include <linux/sunrpc/sched.h>
31 32 33 34 35 36 37
#include <linux/file.h>

#include <net/sock.h>
#include <net/checksum.h>
#include <net/udp.h>
#include <net/tcp.h>

38 39 40 41 42 43 44 45 46
/*
 * xprtsock tunables
 */
unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;

unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;

47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132
/*
 * We can register our own files under /proc/sys/sunrpc by
 * calling register_sysctl_table() again.  The files in that
 * directory become the union of all files registered there.
 *
 * We simply need to make sure that we don't collide with
 * someone else's file names!
 */

#ifdef RPC_DEBUG

static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;

static struct ctl_table_header *sunrpc_table_header;

/*
 * FIXME: changing the UDP slot table size should also resize the UDP
 *        socket buffers for existing UDP transports
 */
static ctl_table xs_tunables_table[] = {
	{
		.ctl_name	= CTL_SLOTTABLE_UDP,
		.procname	= "udp_slot_table_entries",
		.data		= &xprt_udp_slot_table_entries,
		.maxlen		= sizeof(unsigned int),
		.mode		= 0644,
		.proc_handler	= &proc_dointvec_minmax,
		.strategy	= &sysctl_intvec,
		.extra1		= &min_slot_table_size,
		.extra2		= &max_slot_table_size
	},
	{
		.ctl_name	= CTL_SLOTTABLE_TCP,
		.procname	= "tcp_slot_table_entries",
		.data		= &xprt_tcp_slot_table_entries,
		.maxlen		= sizeof(unsigned int),
		.mode		= 0644,
		.proc_handler	= &proc_dointvec_minmax,
		.strategy	= &sysctl_intvec,
		.extra1		= &min_slot_table_size,
		.extra2		= &max_slot_table_size
	},
	{
		.ctl_name	= CTL_MIN_RESVPORT,
		.procname	= "min_resvport",
		.data		= &xprt_min_resvport,
		.maxlen		= sizeof(unsigned int),
		.mode		= 0644,
		.proc_handler	= &proc_dointvec_minmax,
		.strategy	= &sysctl_intvec,
		.extra1		= &xprt_min_resvport_limit,
		.extra2		= &xprt_max_resvport_limit
	},
	{
		.ctl_name	= CTL_MAX_RESVPORT,
		.procname	= "max_resvport",
		.data		= &xprt_max_resvport,
		.maxlen		= sizeof(unsigned int),
		.mode		= 0644,
		.proc_handler	= &proc_dointvec_minmax,
		.strategy	= &sysctl_intvec,
		.extra1		= &xprt_min_resvport_limit,
		.extra2		= &xprt_max_resvport_limit
	},
	{
		.ctl_name = 0,
	},
};

static ctl_table sunrpc_table[] = {
	{
		.ctl_name	= CTL_SUNRPC,
		.procname	= "sunrpc",
		.mode		= 0555,
		.child		= xs_tunables_table
	},
	{
		.ctl_name = 0,
	},
};

#endif

133 134 135 136 137 138
/*
 * How many times to try sending a request on a socket before waiting
 * for the socket buffer to clear.
 */
#define XS_SENDMSG_RETRY	(10U)

139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182
/*
 * Time out for an RPC UDP socket connect.  UDP socket connects are
 * synchronous, but we set a timeout anyway in case of resource
 * exhaustion on the local host.
 */
#define XS_UDP_CONN_TO		(5U * HZ)

/*
 * Wait duration for an RPC TCP connection to be established.  Solaris
 * NFS over TCP uses 60 seconds, for example, which is in line with how
 * long a server takes to reboot.
 */
#define XS_TCP_CONN_TO		(60U * HZ)

/*
 * Wait duration for a reply from the RPC portmapper.
 */
#define XS_BIND_TO		(60U * HZ)

/*
 * Delay if a UDP socket connect error occurs.  This is most likely some
 * kind of resource problem on the local host.
 */
#define XS_UDP_REEST_TO		(2U * HZ)

/*
 * The reestablish timeout allows clients to delay for a bit before attempting
 * to reconnect to a server that just dropped our connection.
 *
 * We implement an exponential backoff when trying to reestablish a TCP
 * transport connection with the server.  Some servers like to drop a TCP
 * connection when they are overworked, so we start with a short timeout and
 * increase over time if the server is down or not responding.
 */
#define XS_TCP_INIT_REEST_TO	(3U * HZ)
#define XS_TCP_MAX_REEST_TO	(5U * 60 * HZ)

/*
 * TCP idle timeout; client drops the transport socket if it is idle
 * for this long.  Note that we also timeout UDP sockets to prevent
 * holding port numbers when there is no RPC traffic.
 */
#define XS_IDLE_DISC_TO		(5U * 60 * HZ)

183 184
#ifdef RPC_DEBUG
# undef  RPC_DEBUG_DATA
185
# define RPCDBG_FACILITY	RPCDBG_TRANS
186 187 188
#endif

#ifdef RPC_DEBUG_DATA
189
static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
190
{
191 192
	u8 *buf = (u8 *) packet;
	int j;
193

194
	dprintk("RPC:       %s\n", msg);
195 196 197 198 199 200 201 202 203 204 205 206
	for (j = 0; j < count && j < 128; j += 4) {
		if (!(j & 31)) {
			if (j)
				dprintk("\n");
			dprintk("0x%04x ", j);
		}
		dprintk("%02x%02x%02x%02x ",
			buf[j], buf[j+1], buf[j+2], buf[j+3]);
	}
	dprintk("\n");
}
#else
207
static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
208 209 210 211 212
{
	/* NOP */
}
#endif

213 214
struct sock_xprt {
	struct rpc_xprt		xprt;
215 216 217 218 219 220

	/*
	 * Network layer
	 */
	struct socket *		sock;
	struct sock *		inet;
221 222 223 224 225 226 227 228 229 230 231 232

	/*
	 * State of TCP reply receive
	 */
	__be32			tcp_fraghdr,
				tcp_xid;

	u32			tcp_offset,
				tcp_reclen;

	unsigned long		tcp_copied,
				tcp_flags;
233 234 235 236

	/*
	 * Connection of transports
	 */
237
	struct delayed_work	connect_worker;
238
	struct sockaddr_storage	addr;
239
	unsigned short		port;
240 241 242 243 244 245

	/*
	 * UDP socket buffer size parameters
	 */
	size_t			rcvsize,
				sndsize;
246 247 248 249 250 251 252

	/*
	 * Saved socket callback addresses
	 */
	void			(*old_data_ready)(struct sock *, int);
	void			(*old_state_change)(struct sock *);
	void			(*old_write_space)(struct sock *);
253 254
};

255 256 257 258 259 260 261 262
/*
 * TCP receive state flags
 */
#define TCP_RCV_LAST_FRAG	(1UL << 0)
#define TCP_RCV_COPY_FRAGHDR	(1UL << 1)
#define TCP_RCV_COPY_XID	(1UL << 2)
#define TCP_RCV_COPY_DATA	(1UL << 3)

263
static void xs_format_ipv4_peer_addresses(struct rpc_xprt *xprt)
264 265 266 267 268 269
{
	struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr;
	char *buf;

	buf = kzalloc(20, GFP_KERNEL);
	if (buf) {
270
		snprintf(buf, 20, NIPQUAD_FMT,
271 272 273 274 275 276 277 278 279 280 281
				NIPQUAD(addr->sin_addr.s_addr));
	}
	xprt->address_strings[RPC_DISPLAY_ADDR] = buf;

	buf = kzalloc(8, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 8, "%u",
				ntohs(addr->sin_port));
	}
	xprt->address_strings[RPC_DISPLAY_PORT] = buf;

282 283 284 285 286 287 288 289
	buf = kzalloc(8, GFP_KERNEL);
	if (buf) {
		if (xprt->prot == IPPROTO_UDP)
			snprintf(buf, 8, "udp");
		else
			snprintf(buf, 8, "tcp");
	}
	xprt->address_strings[RPC_DISPLAY_PROTO] = buf;
290 291 292

	buf = kzalloc(48, GFP_KERNEL);
	if (buf) {
293
		snprintf(buf, 48, "addr="NIPQUAD_FMT" port=%u proto=%s",
294 295 296 297 298
			NIPQUAD(addr->sin_addr.s_addr),
			ntohs(addr->sin_port),
			xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
	}
	xprt->address_strings[RPC_DISPLAY_ALL] = buf;
299 300 301 302 303 304 305 306 307 308 309 310 311 312

	buf = kzalloc(10, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 10, "%02x%02x%02x%02x",
				NIPQUAD(addr->sin_addr.s_addr));
	}
	xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = buf;

	buf = kzalloc(8, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 8, "%4hx",
				ntohs(addr->sin_port));
	}
	xprt->address_strings[RPC_DISPLAY_HEX_PORT] = buf;
313 314
}

315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366
static void xs_format_ipv6_peer_addresses(struct rpc_xprt *xprt)
{
	struct sockaddr_in6 *addr = (struct sockaddr_in6 *) &xprt->addr;
	char *buf;

	buf = kzalloc(40, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 40, NIP6_FMT,
				NIP6(addr->sin6_addr));
	}
	xprt->address_strings[RPC_DISPLAY_ADDR] = buf;

	buf = kzalloc(8, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 8, "%u",
				ntohs(addr->sin6_port));
	}
	xprt->address_strings[RPC_DISPLAY_PORT] = buf;

	buf = kzalloc(8, GFP_KERNEL);
	if (buf) {
		if (xprt->prot == IPPROTO_UDP)
			snprintf(buf, 8, "udp");
		else
			snprintf(buf, 8, "tcp");
	}
	xprt->address_strings[RPC_DISPLAY_PROTO] = buf;

	buf = kzalloc(64, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 64, "addr="NIP6_FMT" port=%u proto=%s",
				NIP6(addr->sin6_addr),
				ntohs(addr->sin6_port),
				xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
	}
	xprt->address_strings[RPC_DISPLAY_ALL] = buf;

	buf = kzalloc(36, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 36, NIP6_SEQFMT,
				NIP6(addr->sin6_addr));
	}
	xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = buf;

	buf = kzalloc(8, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 8, "%4hx",
				ntohs(addr->sin6_port));
	}
	xprt->address_strings[RPC_DISPLAY_HEX_PORT] = buf;
}

367 368
static void xs_free_peer_addresses(struct rpc_xprt *xprt)
{
369 370 371 372
	int i;

	for (i = 0; i < RPC_DISPLAY_MAX; i++)
		kfree(xprt->address_strings[i]);
373 374
}

375 376
#define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)

T
Trond Myklebust 已提交
377
static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
378 379 380 381
{
	struct msghdr msg = {
		.msg_name	= addr,
		.msg_namelen	= addrlen,
T
Trond Myklebust 已提交
382 383 384 385 386
		.msg_flags	= XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
	};
	struct kvec iov = {
		.iov_base	= vec->iov_base + base,
		.iov_len	= vec->iov_len - base,
387 388
	};

T
Trond Myklebust 已提交
389
	if (iov.iov_len != 0)
390 391 392 393
		return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
	return kernel_sendmsg(sock, &msg, NULL, 0, 0);
}

T
Trond Myklebust 已提交
394
static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
395
{
T
Trond Myklebust 已提交
396 397 398 399 400 401 402 403 404 405 406
	struct page **ppage;
	unsigned int remainder;
	int err, sent = 0;

	remainder = xdr->page_len - base;
	base += xdr->page_base;
	ppage = xdr->pages + (base >> PAGE_SHIFT);
	base &= ~PAGE_MASK;
	for(;;) {
		unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
		int flags = XS_SENDMSG_FLAGS;
407

T
Trond Myklebust 已提交
408 409 410 411 412 413 414 415 416 417 418 419 420 421 422
		remainder -= len;
		if (remainder != 0 || more)
			flags |= MSG_MORE;
		err = sock->ops->sendpage(sock, *ppage, base, len, flags);
		if (remainder == 0 || err != len)
			break;
		sent += err;
		ppage++;
		base = 0;
	}
	if (sent == 0)
		return err;
	if (err > 0)
		sent += err;
	return sent;
423 424
}

425 426 427 428 429 430 431 432
/**
 * xs_sendpages - write pages directly to a socket
 * @sock: socket to send on
 * @addr: UDP only -- address of destination
 * @addrlen: UDP only -- length of destination address
 * @xdr: buffer containing this request
 * @base: starting position in the buffer
 *
433
 */
T
Trond Myklebust 已提交
434
static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
435
{
T
Trond Myklebust 已提交
436 437
	unsigned int remainder = xdr->len - base;
	int err, sent = 0;
438

439 440 441 442
	if (unlikely(!sock))
		return -ENOTCONN;

	clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
T
Trond Myklebust 已提交
443 444 445 446
	if (base != 0) {
		addr = NULL;
		addrlen = 0;
	}
447

T
Trond Myklebust 已提交
448 449 450 451 452
	if (base < xdr->head[0].iov_len || addr != NULL) {
		unsigned int len = xdr->head[0].iov_len - base;
		remainder -= len;
		err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
		if (remainder == 0 || err != len)
453
			goto out;
T
Trond Myklebust 已提交
454
		sent += err;
455 456
		base = 0;
	} else
T
Trond Myklebust 已提交
457
		base -= xdr->head[0].iov_len;
458

T
Trond Myklebust 已提交
459 460 461 462 463
	if (base < xdr->page_len) {
		unsigned int len = xdr->page_len - base;
		remainder -= len;
		err = xs_send_pagedata(sock, xdr, base, remainder != 0);
		if (remainder == 0 || err != len)
464
			goto out;
T
Trond Myklebust 已提交
465
		sent += err;
466
		base = 0;
T
Trond Myklebust 已提交
467 468 469 470 471 472
	} else
		base -= xdr->page_len;

	if (base >= xdr->tail[0].iov_len)
		return sent;
	err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
473
out:
T
Trond Myklebust 已提交
474 475 476 477 478
	if (sent == 0)
		return err;
	if (err > 0)
		sent += err;
	return sent;
479 480
}

481
/**
482 483
 * xs_nospace - place task on wait queue if transmit was incomplete
 * @task: task to put to sleep
484
 *
485
 */
486
static void xs_nospace(struct rpc_task *task)
487
{
488 489
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
490
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
491

492
	dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
493 494 495
			task->tk_pid, req->rq_slen - req->rq_bytes_sent,
			req->rq_slen);

496
	if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
497 498 499 500 501 502
		/* Protect against races with write_space */
		spin_lock_bh(&xprt->transport_lock);

		/* Don't race with disconnect */
		if (!xprt_connected(xprt))
			task->tk_status = -ENOTCONN;
503
		else if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526
			xprt_wait_for_buffer_space(task);

		spin_unlock_bh(&xprt->transport_lock);
	} else
		/* Keep holding the socket if it is blocked */
		rpc_delay(task, HZ>>4);
}

/**
 * xs_udp_send_request - write an RPC request to a UDP socket
 * @task: address of RPC task that manages the state of an RPC request
 *
 * Return values:
 *        0:	The request has been sent
 *   EAGAIN:	The socket was blocked, please call again later to
 *		complete the request
 * ENOTCONN:	Caller needs to invoke connect logic then call again
 *    other:	Some other error occured, the request was not sent
 */
static int xs_udp_send_request(struct rpc_task *task)
{
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
527
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
528 529
	struct xdr_buf *xdr = &req->rq_snd_buf;
	int status;
530

531
	xs_pktdump("packet data:",
532 533 534
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);

535
	req->rq_xtime = jiffies;
536 537 538 539
	status = xs_sendpages(transport->sock,
			      (struct sockaddr *) &xprt->addr,
			      xprt->addrlen, xdr,
			      req->rq_bytes_sent);
540

541
	dprintk("RPC:       xs_udp_send_request(%u) = %d\n",
542
			xdr->len - req->rq_bytes_sent, status);
543

544 545
	if (likely(status >= (int) req->rq_slen))
		return 0;
546

547 548 549
	/* Still some bytes left; set up for a retry later. */
	if (status > 0)
		status = -EAGAIN;
550

551 552 553
	switch (status) {
	case -ENETUNREACH:
	case -EPIPE:
554 555
	case -ECONNREFUSED:
		/* When the server has died, an ICMP port unreachable message
556
		 * prompts ECONNREFUSED. */
557
		break;
558 559
	case -EAGAIN:
		xs_nospace(task);
560 561
		break;
	default:
562
		dprintk("RPC:       sendmsg returned unrecognized error %d\n",
563
			-status);
564
		break;
565
	}
566 567

	return status;
568 569
}

570 571 572 573 574 575 576
static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf)
{
	u32 reclen = buf->len - sizeof(rpc_fraghdr);
	rpc_fraghdr *base = buf->head[0].iov_base;
	*base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen);
}

577
/**
578
 * xs_tcp_send_request - write an RPC request to a TCP socket
579 580 581
 * @task: address of RPC task that manages the state of an RPC request
 *
 * Return values:
582 583 584 585 586
 *        0:	The request has been sent
 *   EAGAIN:	The socket was blocked, please call again later to
 *		complete the request
 * ENOTCONN:	Caller needs to invoke connect logic then call again
 *    other:	Some other error occured, the request was not sent
587 588
 *
 * XXX: In the case of soft timeouts, should we eventually give up
589
 *	if sendmsg is not able to make progress?
590
 */
591
static int xs_tcp_send_request(struct rpc_task *task)
592 593 594
{
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
595
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
596
	struct xdr_buf *xdr = &req->rq_snd_buf;
597 598
	int status;
	unsigned int retry = 0;
599

600
	xs_encode_tcp_record_marker(&req->rq_snd_buf);
601

602 603 604
	xs_pktdump("packet data:",
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);
605 606 607

	/* Continue transmitting the packet/record. We must be careful
	 * to cope with writespace callbacks arriving _after_ we have
608
	 * called sendmsg(). */
609 610
	while (1) {
		req->rq_xtime = jiffies;
611 612
		status = xs_sendpages(transport->sock,
					NULL, 0, xdr, req->rq_bytes_sent);
613

614
		dprintk("RPC:       xs_tcp_send_request(%u) = %d\n",
615
				xdr->len - req->rq_bytes_sent, status);
616

617
		if (unlikely(status < 0))
618 619
			break;

620 621 622
		/* If we've sent the entire packet, immediately
		 * reset the count of bytes sent. */
		req->rq_bytes_sent += status;
623
		task->tk_bytes_sent += status;
624 625 626 627
		if (likely(req->rq_bytes_sent >= req->rq_slen)) {
			req->rq_bytes_sent = 0;
			return 0;
		}
628 629

		status = -EAGAIN;
630
		if (retry++ > XS_SENDMSG_RETRY)
631 632 633
			break;
	}

634 635 636 637 638 639 640 641 642 643 644
	switch (status) {
	case -EAGAIN:
		xs_nospace(task);
		break;
	case -ECONNREFUSED:
	case -ECONNRESET:
	case -ENOTCONN:
	case -EPIPE:
		status = -ENOTCONN;
		break;
	default:
645
		dprintk("RPC:       sendmsg returned unrecognized error %d\n",
646
			-status);
647
		xprt_disconnect(xprt);
648
		break;
649
	}
650

651 652 653
	return status;
}

654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680
/**
 * xs_tcp_release_xprt - clean up after a tcp transmission
 * @xprt: transport
 * @task: rpc task
 *
 * This cleans up if an error causes us to abort the transmission of a request.
 * In this case, the socket may need to be reset in order to avoid confusing
 * the server.
 */
static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
{
	struct rpc_rqst *req;

	if (task != xprt->snd_task)
		return;
	if (task == NULL)
		goto out_release;
	req = task->tk_rqstp;
	if (req->rq_bytes_sent == 0)
		goto out_release;
	if (req->rq_bytes_sent == req->rq_snd_buf.len)
		goto out_release;
	set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state);
out_release:
	xprt_release_xprt(xprt, task);
}

681 682 683 684
/**
 * xs_close - close a socket
 * @xprt: transport
 *
685 686
 * This is used when all requests are complete; ie, no DRC state remains
 * on the server we want to save.
687
 */
688
static void xs_close(struct rpc_xprt *xprt)
689
{
690 691 692
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct socket *sock = transport->sock;
	struct sock *sk = transport->inet;
693 694

	if (!sk)
695
		goto clear_close_wait;
696

697
	dprintk("RPC:       xs_close xprt %p\n", xprt);
698

699
	write_lock_bh(&sk->sk_callback_lock);
700 701
	transport->inet = NULL;
	transport->sock = NULL;
702

703
	sk->sk_user_data = NULL;
704 705 706
	sk->sk_data_ready = transport->old_data_ready;
	sk->sk_state_change = transport->old_state_change;
	sk->sk_write_space = transport->old_write_space;
707 708
	write_unlock_bh(&sk->sk_callback_lock);

709
	sk->sk_no_check = 0;
710 711

	sock_release(sock);
712 713 714 715
clear_close_wait:
	smp_mb__before_clear_bit();
	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
	smp_mb__after_clear_bit();
716 717
}

718 719 720 721 722 723
/**
 * xs_destroy - prepare to shutdown a transport
 * @xprt: doomed transport
 *
 */
static void xs_destroy(struct rpc_xprt *xprt)
724
{
725 726
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

727
	dprintk("RPC:       xs_destroy xprt %p\n", xprt);
728

729
	cancel_rearming_delayed_work(&transport->connect_worker);
730 731

	xprt_disconnect(xprt);
732
	xs_close(xprt);
733
	xs_free_peer_addresses(xprt);
734
	kfree(xprt->slot);
735
	kfree(xprt);
736 737
}

738 739 740 741 742 743 744 745 746 747
static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
{
	return (struct rpc_xprt *) sk->sk_user_data;
}

/**
 * xs_udp_data_ready - "data ready" callback for UDP sockets
 * @sk: socket with data to read
 * @len: how much data to read
 *
748
 */
749
static void xs_udp_data_ready(struct sock *sk, int len)
750
{
751 752
	struct rpc_task *task;
	struct rpc_xprt *xprt;
753
	struct rpc_rqst *rovr;
754
	struct sk_buff *skb;
755
	int err, repsize, copied;
756 757
	u32 _xid;
	__be32 *xp;
758 759

	read_lock(&sk->sk_callback_lock);
760
	dprintk("RPC:       xs_udp_data_ready...\n");
761
	if (!(xprt = xprt_from_sock(sk)))
762 763 764 765 766 767 768 769 770 771
		goto out;

	if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
		goto out;

	if (xprt->shutdown)
		goto dropit;

	repsize = skb->len - sizeof(struct udphdr);
	if (repsize < 4) {
772
		dprintk("RPC:       impossible RPC reply size %d!\n", repsize);
773 774 775 776 777 778 779 780 781 782
		goto dropit;
	}

	/* Copy the XID from the skb... */
	xp = skb_header_pointer(skb, sizeof(struct udphdr),
				sizeof(_xid), &_xid);
	if (xp == NULL)
		goto dropit;

	/* Look up and lock the request corresponding to the given XID */
C
Chuck Lever 已提交
783
	spin_lock(&xprt->transport_lock);
784 785 786 787 788 789 790 791 792 793 794 795 796 797 798
	rovr = xprt_lookup_rqst(xprt, *xp);
	if (!rovr)
		goto out_unlock;
	task = rovr->rq_task;

	if ((copied = rovr->rq_private_buf.buflen) > repsize)
		copied = repsize;

	/* Suck it into the iovec, verify checksum if not done by hw. */
	if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb))
		goto out_unlock;

	/* Something worked... */
	dst_confirm(skb->dst);

799 800 801
	xprt_adjust_cwnd(task, copied);
	xprt_update_rtt(task);
	xprt_complete_rqst(task, copied);
802 803

 out_unlock:
C
Chuck Lever 已提交
804
	spin_unlock(&xprt->transport_lock);
805 806 807 808 809 810
 dropit:
	skb_free_datagram(sk, skb);
 out:
	read_unlock(&sk->sk_callback_lock);
}

811
static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
812
{
813
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
814 815 816
	size_t len, used;
	char *p;

817 818
	p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
	len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
819
	used = xdr_skb_read_bits(desc, p, len);
820
	transport->tcp_offset += used;
821 822
	if (used != len)
		return;
823

824 825
	transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
	if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
826
		transport->tcp_flags |= TCP_RCV_LAST_FRAG;
827
	else
828
		transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
829
	transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
830

831
	transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
832
	transport->tcp_offset = 0;
833

834
	/* Sanity check of the record length */
835
	if (unlikely(transport->tcp_reclen < 4)) {
836
		dprintk("RPC:       invalid TCP record fragment length\n");
837
		xprt_disconnect(xprt);
838
		return;
839
	}
840
	dprintk("RPC:       reading TCP record fragment of length %d\n",
841
			transport->tcp_reclen);
842 843
}

844
static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
845
{
846
	if (transport->tcp_offset == transport->tcp_reclen) {
847
		transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
848
		transport->tcp_offset = 0;
849 850 851
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
			transport->tcp_flags |= TCP_RCV_COPY_XID;
852
			transport->tcp_copied = 0;
853 854 855 856
		}
	}
}

857
static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
858 859 860 861
{
	size_t len, used;
	char *p;

862
	len = sizeof(transport->tcp_xid) - transport->tcp_offset;
863
	dprintk("RPC:       reading XID (%Zu bytes)\n", len);
864
	p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
865
	used = xdr_skb_read_bits(desc, p, len);
866
	transport->tcp_offset += used;
867 868
	if (used != len)
		return;
869 870
	transport->tcp_flags &= ~TCP_RCV_COPY_XID;
	transport->tcp_flags |= TCP_RCV_COPY_DATA;
871
	transport->tcp_copied = 4;
872
	dprintk("RPC:       reading reply for XID %08x\n",
873 874
			ntohl(transport->tcp_xid));
	xs_tcp_check_fraghdr(transport);
875 876
}

877
static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
878
{
879
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
880 881 882 883 884 885
	struct rpc_rqst *req;
	struct xdr_buf *rcvbuf;
	size_t len;
	ssize_t r;

	/* Find and lock the request corresponding to this xid */
C
Chuck Lever 已提交
886
	spin_lock(&xprt->transport_lock);
887
	req = xprt_lookup_rqst(xprt, transport->tcp_xid);
888
	if (!req) {
889
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
890
		dprintk("RPC:       XID %08x request not found!\n",
891
				ntohl(transport->tcp_xid));
C
Chuck Lever 已提交
892
		spin_unlock(&xprt->transport_lock);
893 894 895 896 897
		return;
	}

	rcvbuf = &req->rq_private_buf;
	len = desc->count;
898
	if (len > transport->tcp_reclen - transport->tcp_offset) {
899
		struct xdr_skb_reader my_desc;
900

901
		len = transport->tcp_reclen - transport->tcp_offset;
902 903
		memcpy(&my_desc, desc, sizeof(my_desc));
		my_desc.count = len;
904
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
905
					  &my_desc, xdr_skb_read_bits);
906 907 908
		desc->count -= r;
		desc->offset += r;
	} else
909
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
910
					  desc, xdr_skb_read_bits);
911 912

	if (r > 0) {
913 914
		transport->tcp_copied += r;
		transport->tcp_offset += r;
915 916 917 918 919
	}
	if (r != len) {
		/* Error when copying to the receive buffer,
		 * usually because we weren't able to allocate
		 * additional buffer pages. All we can do now
920
		 * is turn off TCP_RCV_COPY_DATA, so the request
921 922 923 924 925
		 * will not receive any additional updates,
		 * and time out.
		 * Any remaining data from this record will
		 * be discarded.
		 */
926
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
927
		dprintk("RPC:       XID %08x truncated request\n",
928
				ntohl(transport->tcp_xid));
929 930 931 932
		dprintk("RPC:       xprt = %p, tcp_copied = %lu, "
				"tcp_offset = %u, tcp_reclen = %u\n",
				xprt, transport->tcp_copied,
				transport->tcp_offset, transport->tcp_reclen);
933 934 935
		goto out;
	}

936
	dprintk("RPC:       XID %08x read %Zd bytes\n",
937
			ntohl(transport->tcp_xid), r);
938 939 940
	dprintk("RPC:       xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
			"tcp_reclen = %u\n", xprt, transport->tcp_copied,
			transport->tcp_offset, transport->tcp_reclen);
941 942

	if (transport->tcp_copied == req->rq_private_buf.buflen)
943
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
944
	else if (transport->tcp_offset == transport->tcp_reclen) {
945 946
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
947 948 949
	}

out:
950
	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
951
		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
C
Chuck Lever 已提交
952
	spin_unlock(&xprt->transport_lock);
953
	xs_tcp_check_fraghdr(transport);
954 955
}

956
static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
957 958 959
{
	size_t len;

960
	len = transport->tcp_reclen - transport->tcp_offset;
961 962 963 964
	if (len > desc->count)
		len = desc->count;
	desc->count -= len;
	desc->offset += len;
965
	transport->tcp_offset += len;
966
	dprintk("RPC:       discarded %Zu bytes\n", len);
967
	xs_tcp_check_fraghdr(transport);
968 969
}

970
static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
971 972
{
	struct rpc_xprt *xprt = rd_desc->arg.data;
973
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
974
	struct xdr_skb_reader desc = {
975 976 977
		.skb	= skb,
		.offset	= offset,
		.count	= len,
978
	};
979

980
	dprintk("RPC:       xs_tcp_data_recv started\n");
981 982 983
	do {
		/* Read in a new fragment marker if necessary */
		/* Can we ever really expect to get completely empty fragments? */
984
		if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
985
			xs_tcp_read_fraghdr(xprt, &desc);
986 987 988
			continue;
		}
		/* Read in the xid if necessary */
989
		if (transport->tcp_flags & TCP_RCV_COPY_XID) {
990
			xs_tcp_read_xid(transport, &desc);
991 992 993
			continue;
		}
		/* Read in the request data */
994
		if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
995
			xs_tcp_read_request(xprt, &desc);
996 997 998
			continue;
		}
		/* Skip over any trailing bytes on short reads */
999
		xs_tcp_read_discard(transport, &desc);
1000
	} while (desc.count);
1001
	dprintk("RPC:       xs_tcp_data_recv done\n");
1002 1003 1004
	return len - desc.count;
}

1005 1006 1007 1008 1009 1010 1011
/**
 * xs_tcp_data_ready - "data ready" callback for TCP sockets
 * @sk: socket with data to read
 * @bytes: how much data to read
 *
 */
static void xs_tcp_data_ready(struct sock *sk, int bytes)
1012 1013 1014 1015
{
	struct rpc_xprt *xprt;
	read_descriptor_t rd_desc;

1016 1017
	dprintk("RPC:       xs_tcp_data_ready...\n");

1018
	read_lock(&sk->sk_callback_lock);
1019
	if (!(xprt = xprt_from_sock(sk)))
1020 1021 1022 1023
		goto out;
	if (xprt->shutdown)
		goto out;

1024
	/* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
1025 1026
	rd_desc.arg.data = xprt;
	rd_desc.count = 65536;
1027
	tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
1028 1029 1030 1031
out:
	read_unlock(&sk->sk_callback_lock);
}

1032 1033 1034 1035 1036 1037
/**
 * xs_tcp_state_change - callback to handle TCP socket state changes
 * @sk: socket whose state has changed
 *
 */
static void xs_tcp_state_change(struct sock *sk)
1038
{
1039
	struct rpc_xprt *xprt;
1040 1041 1042 1043

	read_lock(&sk->sk_callback_lock);
	if (!(xprt = xprt_from_sock(sk)))
		goto out;
1044 1045 1046 1047 1048
	dprintk("RPC:       xs_tcp_state_change client %p...\n", xprt);
	dprintk("RPC:       state %x conn %d dead %d zapped %d\n",
			sk->sk_state, xprt_connected(xprt),
			sock_flag(sk, SOCK_DEAD),
			sock_flag(sk, SOCK_ZAPPED));
1049 1050 1051

	switch (sk->sk_state) {
	case TCP_ESTABLISHED:
C
Chuck Lever 已提交
1052
		spin_lock_bh(&xprt->transport_lock);
1053
		if (!xprt_test_and_set_connected(xprt)) {
1054 1055 1056
			struct sock_xprt *transport = container_of(xprt,
					struct sock_xprt, xprt);

1057
			/* Reset TCP record info */
1058 1059 1060
			transport->tcp_offset = 0;
			transport->tcp_reclen = 0;
			transport->tcp_copied = 0;
1061 1062
			transport->tcp_flags =
				TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
1063

1064
			xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
1065
			xprt_wake_pending_tasks(xprt, 0);
1066
		}
C
Chuck Lever 已提交
1067
		spin_unlock_bh(&xprt->transport_lock);
1068 1069 1070 1071
		break;
	case TCP_SYN_SENT:
	case TCP_SYN_RECV:
		break;
1072 1073 1074 1075
	case TCP_CLOSE_WAIT:
		/* Try to schedule an autoclose RPC calls */
		set_bit(XPRT_CLOSE_WAIT, &xprt->state);
		if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
1076
			queue_work(rpciod_workqueue, &xprt->task_cleanup);
1077 1078 1079 1080 1081 1082 1083
	default:
		xprt_disconnect(xprt);
	}
 out:
	read_unlock(&sk->sk_callback_lock);
}

1084
/**
1085 1086
 * xs_udp_write_space - callback invoked when socket buffer space
 *                             becomes available
1087 1088
 * @sk: socket whose state has changed
 *
1089 1090
 * Called when more output buffer space is available for this socket.
 * We try not to wake our writers until they can make "significant"
1091
 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1092 1093
 * with a bunch of small requests.
 */
1094
static void xs_udp_write_space(struct sock *sk)
1095 1096 1097
{
	read_lock(&sk->sk_callback_lock);

1098 1099 1100 1101 1102 1103
	/* from net/core/sock.c:sock_def_write_space */
	if (sock_writeable(sk)) {
		struct socket *sock;
		struct rpc_xprt *xprt;

		if (unlikely(!(sock = sk->sk_socket)))
1104
			goto out;
1105 1106 1107
		if (unlikely(!(xprt = xprt_from_sock(sk))))
			goto out;
		if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
1108
			goto out;
1109 1110

		xprt_write_space(xprt);
1111 1112
	}

1113 1114 1115
 out:
	read_unlock(&sk->sk_callback_lock);
}
1116

1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146
/**
 * xs_tcp_write_space - callback invoked when socket buffer space
 *                             becomes available
 * @sk: socket whose state has changed
 *
 * Called when more output buffer space is available for this socket.
 * We try not to wake our writers until they can make "significant"
 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
 * with a bunch of small requests.
 */
static void xs_tcp_write_space(struct sock *sk)
{
	read_lock(&sk->sk_callback_lock);

	/* from net/core/stream.c:sk_stream_write_space */
	if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
		struct socket *sock;
		struct rpc_xprt *xprt;

		if (unlikely(!(sock = sk->sk_socket)))
			goto out;
		if (unlikely(!(xprt = xprt_from_sock(sk))))
			goto out;
		if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
			goto out;

		xprt_write_space(xprt);
	}

 out:
1147 1148 1149
	read_unlock(&sk->sk_callback_lock);
}

1150
static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1151
{
1152 1153
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct sock *sk = transport->inet;
1154

1155
	if (transport->rcvsize) {
1156
		sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1157
		sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1158
	}
1159
	if (transport->sndsize) {
1160
		sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1161
		sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1162 1163 1164 1165
		sk->sk_write_space(sk);
	}
}

1166
/**
1167
 * xs_udp_set_buffer_size - set send and receive limits
1168
 * @xprt: generic transport
1169 1170
 * @sndsize: requested size of send buffer, in bytes
 * @rcvsize: requested size of receive buffer, in bytes
1171
 *
1172
 * Set socket send and receive buffer size limits.
1173
 */
1174
static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1175
{
1176 1177 1178
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

	transport->sndsize = 0;
1179
	if (sndsize)
1180 1181
		transport->sndsize = sndsize + 1024;
	transport->rcvsize = 0;
1182
	if (rcvsize)
1183
		transport->rcvsize = rcvsize + 1024;
1184 1185

	xs_udp_do_set_buffer_size(xprt);
1186 1187
}

1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198
/**
 * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
 * @task: task that timed out
 *
 * Adjust the congestion window after a retransmit timeout has occurred.
 */
static void xs_udp_timer(struct rpc_task *task)
{
	xprt_adjust_cwnd(task, -ETIMEDOUT);
}

1199 1200 1201 1202 1203 1204 1205
static unsigned short xs_get_random_port(void)
{
	unsigned short range = xprt_max_resvport - xprt_min_resvport;
	unsigned short rand = (unsigned short) net_random() % range;
	return rand + xprt_min_resvport;
}

1206 1207 1208 1209 1210 1211 1212 1213
/**
 * xs_set_port - reset the port number in the remote endpoint address
 * @xprt: generic transport
 * @port: new port number
 *
 */
static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
{
1214
	struct sockaddr *addr = (struct sockaddr *) &xprt->addr;
1215

1216
	dprintk("RPC:       setting port for xprt %p to %u\n", xprt, port);
1217

1218 1219 1220 1221 1222 1223 1224 1225 1226 1227
	switch (addr->sa_family) {
	case AF_INET:
		((struct sockaddr_in *)addr)->sin_port = htons(port);
		break;
	case AF_INET6:
		((struct sockaddr_in6 *)addr)->sin6_port = htons(port);
		break;
	default:
		BUG();
	}
1228 1229
}

1230
static int xs_bind4(struct sock_xprt *transport, struct socket *sock)
1231 1232 1233 1234
{
	struct sockaddr_in myaddr = {
		.sin_family = AF_INET,
	};
1235
	struct sockaddr_in *sa;
1236
	int err;
1237
	unsigned short port = transport->port;
1238

1239 1240 1241 1242
	if (!transport->xprt.resvport)
		port = 0;
	sa = (struct sockaddr_in *)&transport->addr;
	myaddr.sin_addr = sa->sin_addr;
1243 1244
	do {
		myaddr.sin_port = htons(port);
1245
		err = kernel_bind(sock, (struct sockaddr *) &myaddr,
1246
						sizeof(myaddr));
1247 1248
		if (!transport->xprt.resvport)
			break;
1249
		if (err == 0) {
1250
			transport->port = port;
1251
			break;
1252
		}
1253 1254 1255 1256
		if (port <= xprt_min_resvport)
			port = xprt_max_resvport;
		else
			port--;
1257
	} while (err == -EADDRINUSE && port != transport->port);
1258 1259 1260
	dprintk("RPC:       %s "NIPQUAD_FMT":%u: %s (%d)\n",
			__FUNCTION__, NIPQUAD(myaddr.sin_addr),
			port, err ? "failed" : "ok", err);
1261 1262 1263
	return err;
}

1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296
static int xs_bind6(struct sock_xprt *transport, struct socket *sock)
{
	struct sockaddr_in6 myaddr = {
		.sin6_family = AF_INET6,
	};
	struct sockaddr_in6 *sa;
	int err;
	unsigned short port = transport->port;

	if (!transport->xprt.resvport)
		port = 0;
	sa = (struct sockaddr_in6 *)&transport->addr;
	myaddr.sin6_addr = sa->sin6_addr;
	do {
		myaddr.sin6_port = htons(port);
		err = kernel_bind(sock, (struct sockaddr *) &myaddr,
						sizeof(myaddr));
		if (!transport->xprt.resvport)
			break;
		if (err == 0) {
			transport->port = port;
			break;
		}
		if (port <= xprt_min_resvport)
			port = xprt_max_resvport;
		else
			port--;
	} while (err == -EADDRINUSE && port != transport->port);
	dprintk("RPC:       xs_bind6 "NIP6_FMT":%u: %s (%d)\n",
		NIP6(myaddr.sin6_addr), port, err ? "failed" : "ok", err);
	return err;
}

1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325
#ifdef CONFIG_DEBUG_LOCK_ALLOC
static struct lock_class_key xs_key[2];
static struct lock_class_key xs_slock_key[2];

static inline void xs_reclassify_socket(struct socket *sock)
{
	struct sock *sk = sock->sk;
	BUG_ON(sk->sk_lock.owner != NULL);
	switch (sk->sk_family) {
	case AF_INET:
		sock_lock_init_class_and_name(sk, "slock-AF_INET-NFS",
			&xs_slock_key[0], "sk_lock-AF_INET-NFS", &xs_key[0]);
		break;

	case AF_INET6:
		sock_lock_init_class_and_name(sk, "slock-AF_INET6-NFS",
			&xs_slock_key[1], "sk_lock-AF_INET6-NFS", &xs_key[1]);
		break;

	default:
		BUG();
	}
}
#else
static inline void xs_reclassify_socket(struct socket *sock)
{
}
#endif

1326 1327
/**
 * xs_udp_connect_worker - set up a UDP socket
1328
 * @work: RPC transport to connect
1329 1330 1331
 *
 * Invoked by a work queue tasklet.
 */
1332
static void xs_udp_connect_worker(struct work_struct *work)
1333
{
1334 1335
	struct sock_xprt *transport =
		container_of(work, struct sock_xprt, connect_worker.work);
1336
	struct rpc_xprt *xprt = &transport->xprt;
1337
	struct socket *sock = transport->sock;
1338
	int err, status = -EIO;
1339

1340
	if (xprt->shutdown || !xprt_bound(xprt))
1341
		goto out;
1342

1343 1344
	/* Start by resetting any existing state */
	xs_close(xprt);
1345

1346
	if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
1347
		dprintk("RPC:       can't create UDP transport socket (%d).\n", -err);
1348 1349
		goto out;
	}
1350
	xs_reclassify_socket(sock);
1351

1352
	if (xs_bind4(transport, sock)) {
1353 1354 1355
		sock_release(sock);
		goto out;
	}
1356

1357
	dprintk("RPC:       worker connecting xprt %p to address: %s\n",
1358
			xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
1359

1360
	if (!transport->inet) {
1361
		struct sock *sk = sock->sk;
1362

1363
		write_lock_bh(&sk->sk_callback_lock);
1364

1365
		sk->sk_user_data = xprt;
1366 1367 1368
		transport->old_data_ready = sk->sk_data_ready;
		transport->old_state_change = sk->sk_state_change;
		transport->old_write_space = sk->sk_write_space;
1369
		sk->sk_data_ready = xs_udp_data_ready;
1370
		sk->sk_write_space = xs_udp_write_space;
1371
		sk->sk_no_check = UDP_CSUM_NORCV;
1372
		sk->sk_allocation = GFP_ATOMIC;
1373

1374 1375
		xprt_set_connected(xprt);

1376
		/* Reset to new socket */
1377 1378
		transport->sock = sock;
		transport->inet = sk;
1379

1380 1381
		write_unlock_bh(&sk->sk_callback_lock);
	}
1382
	xs_udp_do_set_buffer_size(xprt);
1383 1384 1385 1386
	status = 0;
out:
	xprt_wake_pending_tasks(xprt, status);
	xprt_clear_connecting(xprt);
1387 1388
}

1389 1390 1391 1392 1393 1394 1395
/*
 * We need to preserve the port number so the reply cache on the server can
 * find our cached RPC replies when we get around to reconnecting.
 */
static void xs_tcp_reuse_connection(struct rpc_xprt *xprt)
{
	int result;
1396
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1397 1398
	struct sockaddr any;

1399
	dprintk("RPC:       disconnecting xprt %p to reuse port\n", xprt);
1400 1401 1402 1403 1404 1405 1406

	/*
	 * Disconnect the transport socket by doing a connect operation
	 * with AF_UNSPEC.  This should return immediately...
	 */
	memset(&any, 0, sizeof(any));
	any.sa_family = AF_UNSPEC;
1407
	result = kernel_connect(transport->sock, &any, sizeof(any), 0);
1408
	if (result)
1409
		dprintk("RPC:       AF_UNSPEC connect return code %d\n",
1410 1411 1412
				result);
}

1413
/**
1414
 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint
1415
 * @work: RPC transport to connect
1416 1417
 *
 * Invoked by a work queue tasklet.
1418
 */
1419
static void xs_tcp_connect_worker(struct work_struct *work)
1420
{
1421 1422
	struct sock_xprt *transport =
		container_of(work, struct sock_xprt, connect_worker.work);
1423
	struct rpc_xprt *xprt = &transport->xprt;
1424
	struct socket *sock = transport->sock;
1425
	int err, status = -EIO;
1426

1427
	if (xprt->shutdown || !xprt_bound(xprt))
1428 1429
		goto out;

1430
	if (!sock) {
1431 1432
		/* start from scratch */
		if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
1433 1434
			dprintk("RPC:       can't create TCP transport "
					"socket (%d).\n", -err);
1435 1436
			goto out;
		}
1437
		xs_reclassify_socket(sock);
1438

1439
		if (xs_bind4(transport, sock)) {
1440 1441 1442 1443 1444 1445
			sock_release(sock);
			goto out;
		}
	} else
		/* "close" the socket, preserving the local port */
		xs_tcp_reuse_connection(xprt);
1446

1447
	dprintk("RPC:       worker connecting xprt %p to address: %s\n",
1448
			xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
1449

1450
	if (!transport->inet) {
1451 1452 1453 1454 1455
		struct sock *sk = sock->sk;

		write_lock_bh(&sk->sk_callback_lock);

		sk->sk_user_data = xprt;
1456 1457 1458
		transport->old_data_ready = sk->sk_data_ready;
		transport->old_state_change = sk->sk_state_change;
		transport->old_write_space = sk->sk_write_space;
1459 1460 1461
		sk->sk_data_ready = xs_tcp_data_ready;
		sk->sk_state_change = xs_tcp_state_change;
		sk->sk_write_space = xs_tcp_write_space;
1462
		sk->sk_allocation = GFP_ATOMIC;
1463 1464 1465 1466 1467 1468

		/* socket options */
		sk->sk_userlocks |= SOCK_BINDPORT_LOCK;
		sock_reset_flag(sk, SOCK_LINGER);
		tcp_sk(sk)->linger2 = 0;
		tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
1469 1470 1471 1472

		xprt_clear_connected(xprt);

		/* Reset to new socket */
1473 1474
		transport->sock = sock;
		transport->inet = sk;
1475 1476 1477 1478 1479

		write_unlock_bh(&sk->sk_callback_lock);
	}

	/* Tell the socket layer to start connecting... */
1480 1481
	xprt->stat.connect_count++;
	xprt->stat.connect_start = jiffies;
1482
	status = kernel_connect(sock, (struct sockaddr *) &xprt->addr,
1483
			xprt->addrlen, O_NONBLOCK);
1484 1485 1486
	dprintk("RPC:       %p connect status %d connected %d sock state %d\n",
			xprt, -status, xprt_connected(xprt),
			sock->sk->sk_state);
1487 1488 1489 1490 1491
	if (status < 0) {
		switch (status) {
			case -EINPROGRESS:
			case -EALREADY:
				goto out_clear;
1492 1493 1494 1495 1496 1497 1498 1499
			case -ECONNREFUSED:
			case -ECONNRESET:
				/* retry with existing socket, after a delay */
				break;
			default:
				/* get rid of existing socket, and retry */
				xs_close(xprt);
				break;
1500 1501 1502
		}
	}
out:
1503
	xprt_wake_pending_tasks(xprt, status);
1504
out_clear:
1505
	xprt_clear_connecting(xprt);
1506 1507
}

1508 1509 1510 1511 1512
/**
 * xs_connect - connect a socket to a remote endpoint
 * @task: address of RPC task that manages state of connect request
 *
 * TCP: If the remote end dropped the connection, delay reconnecting.
1513 1514 1515 1516 1517 1518 1519
 *
 * UDP socket connects are synchronous, but we use a work queue anyway
 * to guarantee that even unprivileged user processes can set up a
 * socket on a privileged port.
 *
 * If a UDP socket connect fails, the delay behavior here prevents
 * retry floods (hard mounts).
1520 1521
 */
static void xs_connect(struct rpc_task *task)
1522 1523
{
	struct rpc_xprt *xprt = task->tk_xprt;
1524
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1525

1526 1527 1528
	if (xprt_test_and_set_connecting(xprt))
		return;

1529
	if (transport->sock != NULL) {
1530 1531
		dprintk("RPC:       xs_connect delayed xprt %p for %lu "
				"seconds\n",
1532
				xprt, xprt->reestablish_timeout / HZ);
1533 1534 1535
		queue_delayed_work(rpciod_workqueue,
				   &transport->connect_worker,
				   xprt->reestablish_timeout);
1536 1537 1538
		xprt->reestablish_timeout <<= 1;
		if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
			xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
1539
	} else {
1540
		dprintk("RPC:       xs_connect scheduled xprt %p\n", xprt);
1541 1542
		queue_delayed_work(rpciod_workqueue,
				   &transport->connect_worker, 0);
1543 1544 1545
	}
}

1546 1547 1548 1549 1550 1551 1552 1553
/**
 * xs_udp_print_stats - display UDP socket-specifc stats
 * @xprt: rpc_xprt struct containing statistics
 * @seq: output file
 *
 */
static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
{
1554 1555
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

1556
	seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
1557
			transport->port,
1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573
			xprt->stat.bind_count,
			xprt->stat.sends,
			xprt->stat.recvs,
			xprt->stat.bad_xids,
			xprt->stat.req_u,
			xprt->stat.bklog_u);
}

/**
 * xs_tcp_print_stats - display TCP socket-specifc stats
 * @xprt: rpc_xprt struct containing statistics
 * @seq: output file
 *
 */
static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
{
1574
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1575 1576 1577 1578 1579 1580
	long idle_time = 0;

	if (xprt_connected(xprt))
		idle_time = (long)(jiffies - xprt->last_used) / HZ;

	seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n",
1581
			transport->port,
1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592
			xprt->stat.bind_count,
			xprt->stat.connect_count,
			xprt->stat.connect_time,
			idle_time,
			xprt->stat.sends,
			xprt->stat.recvs,
			xprt->stat.bad_xids,
			xprt->stat.req_u,
			xprt->stat.bklog_u);
}

1593
static struct rpc_xprt_ops xs_udp_ops = {
1594
	.set_buffer_size	= xs_udp_set_buffer_size,
1595
	.reserve_xprt		= xprt_reserve_xprt_cong,
1596
	.release_xprt		= xprt_release_xprt_cong,
1597
	.rpcbind		= rpcb_getport_async,
1598
	.set_port		= xs_set_port,
1599
	.connect		= xs_connect,
1600 1601
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1602
	.send_request		= xs_udp_send_request,
1603
	.set_retrans_timeout	= xprt_set_retrans_timeout_rtt,
1604
	.timer			= xs_udp_timer,
1605
	.release_request	= xprt_release_rqst_cong,
1606 1607
	.close			= xs_close,
	.destroy		= xs_destroy,
1608
	.print_stats		= xs_udp_print_stats,
1609 1610 1611
};

static struct rpc_xprt_ops xs_tcp_ops = {
1612
	.reserve_xprt		= xprt_reserve_xprt,
1613
	.release_xprt		= xs_tcp_release_xprt,
1614
	.rpcbind		= rpcb_getport_async,
1615
	.set_port		= xs_set_port,
1616
	.connect		= xs_connect,
1617 1618
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1619
	.send_request		= xs_tcp_send_request,
1620
	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
1621 1622
	.close			= xs_close,
	.destroy		= xs_destroy,
1623
	.print_stats		= xs_tcp_print_stats,
1624 1625
};

1626
static struct rpc_xprt *xs_setup_xprt(struct rpc_xprtsock_create *args, unsigned int slot_table_size)
1627 1628
{
	struct rpc_xprt *xprt;
1629
	struct sock_xprt *new;
1630

1631
	if (args->addrlen > sizeof(xprt->addr)) {
1632
		dprintk("RPC:       xs_setup_xprt: address too large\n");
1633 1634 1635
		return ERR_PTR(-EBADF);
	}

1636 1637
	new = kzalloc(sizeof(*new), GFP_KERNEL);
	if (new == NULL) {
1638 1639
		dprintk("RPC:       xs_setup_xprt: couldn't allocate "
				"rpc_xprt\n");
1640 1641
		return ERR_PTR(-ENOMEM);
	}
1642
	xprt = &new->xprt;
1643 1644 1645 1646 1647

	xprt->max_reqs = slot_table_size;
	xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL);
	if (xprt->slot == NULL) {
		kfree(xprt);
1648 1649
		dprintk("RPC:       xs_setup_xprt: couldn't allocate slot "
				"table\n");
1650 1651 1652
		return ERR_PTR(-ENOMEM);
	}

1653 1654
	memcpy(&xprt->addr, args->dstaddr, args->addrlen);
	xprt->addrlen = args->addrlen;
1655 1656
	if (args->srcaddr)
		memcpy(&new->addr, args->srcaddr, args->addrlen);
1657
	new->port = xs_get_random_port();
1658 1659 1660 1661

	return xprt;
}

1662 1663
/**
 * xs_setup_udp - Set up transport to use a UDP socket
1664
 * @args: rpc transport creation arguments
1665 1666
 *
 */
1667
struct rpc_xprt *xs_setup_udp(struct rpc_xprtsock_create *args)
1668
{
1669
	struct rpc_xprt *xprt;
1670
	struct sock_xprt *transport;
1671

1672
	xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries);
1673 1674
	if (IS_ERR(xprt))
		return xprt;
1675
	transport = container_of(xprt, struct sock_xprt, xprt);
1676

1677
	if (ntohs(((struct sockaddr_in *)args->dstaddr)->sin_port) != 0)
1678 1679 1680
		xprt_set_bound(xprt);

	xprt->prot = IPPROTO_UDP;
1681
	xprt->tsh_size = 0;
1682 1683 1684
	/* XXX: header size can vary due to auth type, IPv6, etc. */
	xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);

1685
	INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_connect_worker);
1686 1687 1688 1689
	xprt->bind_timeout = XS_BIND_TO;
	xprt->connect_timeout = XS_UDP_CONN_TO;
	xprt->reestablish_timeout = XS_UDP_REEST_TO;
	xprt->idle_timeout = XS_IDLE_DISC_TO;
1690

1691
	xprt->ops = &xs_udp_ops;
1692

1693 1694
	if (args->timeout)
		xprt->timeout = *args->timeout;
1695
	else
1696
		xprt_set_timeout(&xprt->timeout, 5, 5 * HZ);
1697

1698
	xs_format_ipv4_peer_addresses(xprt);
1699
	dprintk("RPC:       set up transport to address %s\n",
1700
			xprt->address_strings[RPC_DISPLAY_ALL]);
1701

1702
	return xprt;
1703 1704
}

1705 1706
/**
 * xs_setup_tcp - Set up transport to use a TCP socket
1707
 * @args: rpc transport creation arguments
1708 1709
 *
 */
1710
struct rpc_xprt *xs_setup_tcp(struct rpc_xprtsock_create *args)
1711
{
1712
	struct rpc_xprt *xprt;
1713
	struct sock_xprt *transport;
1714

1715
	xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
1716 1717
	if (IS_ERR(xprt))
		return xprt;
1718
	transport = container_of(xprt, struct sock_xprt, xprt);
1719

1720
	if (ntohs(((struct sockaddr_in *)args->dstaddr)->sin_port) != 0)
1721 1722 1723
		xprt_set_bound(xprt);

	xprt->prot = IPPROTO_TCP;
1724 1725
	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
1726

1727
	INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker);
1728 1729 1730 1731
	xprt->bind_timeout = XS_BIND_TO;
	xprt->connect_timeout = XS_TCP_CONN_TO;
	xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
	xprt->idle_timeout = XS_IDLE_DISC_TO;
1732

1733
	xprt->ops = &xs_tcp_ops;
1734

1735 1736
	if (args->timeout)
		xprt->timeout = *args->timeout;
1737
	else
1738
		xprt_set_timeout(&xprt->timeout, 2, 60 * HZ);
1739

1740
	xs_format_ipv4_peer_addresses(xprt);
1741
	dprintk("RPC:       set up transport to address %s\n",
1742
			xprt->address_strings[RPC_DISPLAY_ALL]);
1743

1744
	return xprt;
1745
}
1746 1747

/**
1748
 * init_socket_xprt - set up xprtsock's sysctls
1749 1750 1751 1752
 *
 */
int init_socket_xprt(void)
{
1753
#ifdef RPC_DEBUG
1754
	if (!sunrpc_table_header)
1755
		sunrpc_table_header = register_sysctl_table(sunrpc_table);
1756 1757
#endif

1758 1759 1760 1761
	return 0;
}

/**
1762
 * cleanup_socket_xprt - remove xprtsock's sysctls
1763 1764 1765 1766
 *
 */
void cleanup_socket_xprt(void)
{
1767 1768 1769 1770 1771 1772
#ifdef RPC_DEBUG
	if (sunrpc_table_header) {
		unregister_sysctl_table(sunrpc_table_header);
		sunrpc_table_header = NULL;
	}
#endif
1773
}