xprtsock.c 42.9 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 30
 */

#include <linux/types.h>
#include <linux/slab.h>
#include <linux/capability.h>
#include <linux/sched.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>
31
#include <linux/sunrpc/sched.h>
32 33 34 35 36 37 38
#include <linux/file.h>

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

39 40 41 42 43 44 45 46 47
/*
 * 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;

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 133
/*
 * 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

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

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 183
/*
 * 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)

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

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

195
	dprintk("RPC:       %s\n", msg);
196 197 198 199 200 201 202 203 204 205 206 207
	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
208
static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
209 210 211 212 213
{
	/* NOP */
}
#endif

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

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

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

	u32			tcp_offset,
				tcp_reclen;

	unsigned long		tcp_copied,
				tcp_flags;
234 235 236 237

	/*
	 * Connection of transports
	 */
238
	struct delayed_work	connect_worker;
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 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303
static void xs_format_peer_addresses(struct rpc_xprt *xprt)
{
	struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr;
	char *buf;

	buf = kzalloc(20, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 20, "%u.%u.%u.%u",
				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;

	if (xprt->prot == IPPROTO_UDP)
		xprt->address_strings[RPC_DISPLAY_PROTO] = "udp";
	else
		xprt->address_strings[RPC_DISPLAY_PROTO] = "tcp";

	buf = kzalloc(48, GFP_KERNEL);
	if (buf) {
		snprintf(buf, 48, "addr=%u.%u.%u.%u port=%u proto=%s",
			NIPQUAD(addr->sin_addr.s_addr),
			ntohs(addr->sin_port),
			xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
	}
	xprt->address_strings[RPC_DISPLAY_ALL] = buf;
}

static void xs_free_peer_addresses(struct rpc_xprt *xprt)
{
	kfree(xprt->address_strings[RPC_DISPLAY_ADDR]);
	kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
	kfree(xprt->address_strings[RPC_DISPLAY_ALL]);
}

304 305
#define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)

T
Trond Myklebust 已提交
306
static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
307 308 309 310
{
	struct msghdr msg = {
		.msg_name	= addr,
		.msg_namelen	= addrlen,
T
Trond Myklebust 已提交
311 312 313 314 315
		.msg_flags	= XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
	};
	struct kvec iov = {
		.iov_base	= vec->iov_base + base,
		.iov_len	= vec->iov_len - base,
316 317
	};

T
Trond Myklebust 已提交
318
	if (iov.iov_len != 0)
319 320 321 322
		return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
	return kernel_sendmsg(sock, &msg, NULL, 0, 0);
}

T
Trond Myklebust 已提交
323
static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
324
{
T
Trond Myklebust 已提交
325 326 327 328 329 330 331 332 333 334 335
	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;
336

T
Trond Myklebust 已提交
337 338 339 340 341 342 343 344 345 346 347 348 349 350 351
		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;
352 353
}

354 355 356 357 358 359 360 361
/**
 * 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
 *
362
 */
T
Trond Myklebust 已提交
363
static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
364
{
T
Trond Myklebust 已提交
365 366
	unsigned int remainder = xdr->len - base;
	int err, sent = 0;
367

368 369 370 371
	if (unlikely(!sock))
		return -ENOTCONN;

	clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
T
Trond Myklebust 已提交
372 373 374 375
	if (base != 0) {
		addr = NULL;
		addrlen = 0;
	}
376

T
Trond Myklebust 已提交
377 378 379 380 381
	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)
382
			goto out;
T
Trond Myklebust 已提交
383
		sent += err;
384 385
		base = 0;
	} else
T
Trond Myklebust 已提交
386
		base -= xdr->head[0].iov_len;
387

T
Trond Myklebust 已提交
388 389 390 391 392
	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)
393
			goto out;
T
Trond Myklebust 已提交
394
		sent += err;
395
		base = 0;
T
Trond Myklebust 已提交
396 397 398 399 400 401
	} 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);
402
out:
T
Trond Myklebust 已提交
403 404 405 406 407
	if (sent == 0)
		return err;
	if (err > 0)
		sent += err;
	return sent;
408 409
}

410
/**
411 412
 * xs_nospace - place task on wait queue if transmit was incomplete
 * @task: task to put to sleep
413
 *
414
 */
415
static void xs_nospace(struct rpc_task *task)
416
{
417 418
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
419
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
420

421
	dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
422 423 424
			task->tk_pid, req->rq_slen - req->rq_bytes_sent,
			req->rq_slen);

425
	if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
426 427 428 429 430 431
		/* 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;
432
		else if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455
			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;
456
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
457 458
	struct xdr_buf *xdr = &req->rq_snd_buf;
	int status;
459

460
	xs_pktdump("packet data:",
461 462 463
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);

464
	req->rq_xtime = jiffies;
465 466 467 468
	status = xs_sendpages(transport->sock,
			      (struct sockaddr *) &xprt->addr,
			      xprt->addrlen, xdr,
			      req->rq_bytes_sent);
469

470
	dprintk("RPC:       xs_udp_send_request(%u) = %d\n",
471
			xdr->len - req->rq_bytes_sent, status);
472

473 474
	if (likely(status >= (int) req->rq_slen))
		return 0;
475

476 477 478
	/* Still some bytes left; set up for a retry later. */
	if (status > 0)
		status = -EAGAIN;
479

480 481 482
	switch (status) {
	case -ENETUNREACH:
	case -EPIPE:
483 484
	case -ECONNREFUSED:
		/* When the server has died, an ICMP port unreachable message
485
		 * prompts ECONNREFUSED. */
486
		break;
487 488
	case -EAGAIN:
		xs_nospace(task);
489 490
		break;
	default:
491
		dprintk("RPC:       sendmsg returned unrecognized error %d\n",
492
			-status);
493
		break;
494
	}
495 496

	return status;
497 498
}

499 500 501 502 503 504 505
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);
}

506
/**
507
 * xs_tcp_send_request - write an RPC request to a TCP socket
508 509 510
 * @task: address of RPC task that manages the state of an RPC request
 *
 * Return values:
511 512 513 514 515
 *        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
516 517
 *
 * XXX: In the case of soft timeouts, should we eventually give up
518
 *	if sendmsg is not able to make progress?
519
 */
520
static int xs_tcp_send_request(struct rpc_task *task)
521 522 523
{
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
524
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
525
	struct xdr_buf *xdr = &req->rq_snd_buf;
526 527
	int status, retry = 0;

528
	xs_encode_tcp_record_marker(&req->rq_snd_buf);
529

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

	/* Continue transmitting the packet/record. We must be careful
	 * to cope with writespace callbacks arriving _after_ we have
536
	 * called sendmsg(). */
537 538
	while (1) {
		req->rq_xtime = jiffies;
539 540
		status = xs_sendpages(transport->sock,
					NULL, 0, xdr, req->rq_bytes_sent);
541

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

545
		if (unlikely(status < 0))
546 547
			break;

548 549 550
		/* If we've sent the entire packet, immediately
		 * reset the count of bytes sent. */
		req->rq_bytes_sent += status;
551
		task->tk_bytes_sent += status;
552 553 554 555
		if (likely(req->rq_bytes_sent >= req->rq_slen)) {
			req->rq_bytes_sent = 0;
			return 0;
		}
556 557

		status = -EAGAIN;
558
		if (retry++ > XS_SENDMSG_RETRY)
559 560 561
			break;
	}

562 563 564 565 566 567 568 569 570 571 572
	switch (status) {
	case -EAGAIN:
		xs_nospace(task);
		break;
	case -ECONNREFUSED:
	case -ECONNRESET:
	case -ENOTCONN:
	case -EPIPE:
		status = -ENOTCONN;
		break;
	default:
573
		dprintk("RPC:       sendmsg returned unrecognized error %d\n",
574
			-status);
575
		xprt_disconnect(xprt);
576
		break;
577
	}
578

579 580 581
	return status;
}

582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608
/**
 * 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);
}

609 610 611 612
/**
 * xs_close - close a socket
 * @xprt: transport
 *
613 614
 * This is used when all requests are complete; ie, no DRC state remains
 * on the server we want to save.
615
 */
616
static void xs_close(struct rpc_xprt *xprt)
617
{
618 619 620
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct socket *sock = transport->sock;
	struct sock *sk = transport->inet;
621 622

	if (!sk)
623
		goto clear_close_wait;
624

625
	dprintk("RPC:       xs_close xprt %p\n", xprt);
626

627
	write_lock_bh(&sk->sk_callback_lock);
628 629
	transport->inet = NULL;
	transport->sock = NULL;
630

631
	sk->sk_user_data = NULL;
632 633 634
	sk->sk_data_ready = transport->old_data_ready;
	sk->sk_state_change = transport->old_state_change;
	sk->sk_write_space = transport->old_write_space;
635 636
	write_unlock_bh(&sk->sk_callback_lock);

637
	sk->sk_no_check = 0;
638 639

	sock_release(sock);
640 641 642 643
clear_close_wait:
	smp_mb__before_clear_bit();
	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
	smp_mb__after_clear_bit();
644 645
}

646 647 648 649 650 651
/**
 * xs_destroy - prepare to shutdown a transport
 * @xprt: doomed transport
 *
 */
static void xs_destroy(struct rpc_xprt *xprt)
652
{
653 654
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

655
	dprintk("RPC:       xs_destroy xprt %p\n", xprt);
656

657
	cancel_delayed_work(&transport->connect_worker);
658 659 660
	flush_scheduled_work();

	xprt_disconnect(xprt);
661
	xs_close(xprt);
662
	xs_free_peer_addresses(xprt);
663
	kfree(xprt->slot);
664
	kfree(xprt);
665 666
}

667 668 669 670 671 672 673 674 675 676
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
 *
677
 */
678
static void xs_udp_data_ready(struct sock *sk, int len)
679
{
680 681
	struct rpc_task *task;
	struct rpc_xprt *xprt;
682
	struct rpc_rqst *rovr;
683
	struct sk_buff *skb;
684
	int err, repsize, copied;
685 686
	u32 _xid;
	__be32 *xp;
687 688

	read_lock(&sk->sk_callback_lock);
689
	dprintk("RPC:       xs_udp_data_ready...\n");
690
	if (!(xprt = xprt_from_sock(sk)))
691 692 693 694 695 696 697 698 699 700
		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) {
701
		dprintk("RPC:       impossible RPC reply size %d!\n", repsize);
702 703 704 705 706 707 708 709 710 711
		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 已提交
712
	spin_lock(&xprt->transport_lock);
713 714 715 716 717 718 719 720 721 722 723 724 725 726 727
	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);

728 729 730
	xprt_adjust_cwnd(task, copied);
	xprt_update_rtt(task);
	xprt_complete_rqst(task, copied);
731 732

 out_unlock:
C
Chuck Lever 已提交
733
	spin_unlock(&xprt->transport_lock);
734 735 736 737 738 739
 dropit:
	skb_free_datagram(sk, skb);
 out:
	read_unlock(&sk->sk_callback_lock);
}

740
static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
741
{
742
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
743 744 745
	size_t len, used;
	char *p;

746 747
	p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
	len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
748
	used = xdr_skb_read_bits(desc, p, len);
749
	transport->tcp_offset += used;
750 751
	if (used != len)
		return;
752

753 754
	transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
	if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
755
		transport->tcp_flags |= TCP_RCV_LAST_FRAG;
756
	else
757
		transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
758
	transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
759

760
	transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
761
	transport->tcp_offset = 0;
762

763
	/* Sanity check of the record length */
764
	if (unlikely(transport->tcp_reclen < 4)) {
765
		dprintk("RPC:       invalid TCP record fragment length\n");
766
		xprt_disconnect(xprt);
767
		return;
768
	}
769
	dprintk("RPC:       reading TCP record fragment of length %d\n",
770
			transport->tcp_reclen);
771 772
}

773
static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
774
{
775
	if (transport->tcp_offset == transport->tcp_reclen) {
776
		transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
777
		transport->tcp_offset = 0;
778 779 780
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
			transport->tcp_flags |= TCP_RCV_COPY_XID;
781
			transport->tcp_copied = 0;
782 783 784 785
		}
	}
}

786
static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
787 788 789 790
{
	size_t len, used;
	char *p;

791
	len = sizeof(transport->tcp_xid) - transport->tcp_offset;
792
	dprintk("RPC:       reading XID (%Zu bytes)\n", len);
793
	p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
794
	used = xdr_skb_read_bits(desc, p, len);
795
	transport->tcp_offset += used;
796 797
	if (used != len)
		return;
798 799
	transport->tcp_flags &= ~TCP_RCV_COPY_XID;
	transport->tcp_flags |= TCP_RCV_COPY_DATA;
800
	transport->tcp_copied = 4;
801
	dprintk("RPC:       reading reply for XID %08x\n",
802 803
			ntohl(transport->tcp_xid));
	xs_tcp_check_fraghdr(transport);
804 805
}

806
static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
807
{
808
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
809 810 811 812 813 814
	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 已提交
815
	spin_lock(&xprt->transport_lock);
816
	req = xprt_lookup_rqst(xprt, transport->tcp_xid);
817
	if (!req) {
818
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
819
		dprintk("RPC:       XID %08x request not found!\n",
820
				ntohl(transport->tcp_xid));
C
Chuck Lever 已提交
821
		spin_unlock(&xprt->transport_lock);
822 823 824 825 826
		return;
	}

	rcvbuf = &req->rq_private_buf;
	len = desc->count;
827
	if (len > transport->tcp_reclen - transport->tcp_offset) {
828
		struct xdr_skb_reader my_desc;
829

830
		len = transport->tcp_reclen - transport->tcp_offset;
831 832
		memcpy(&my_desc, desc, sizeof(my_desc));
		my_desc.count = len;
833
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
834
					  &my_desc, xdr_skb_read_bits);
835 836 837
		desc->count -= r;
		desc->offset += r;
	} else
838
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
839
					  desc, xdr_skb_read_bits);
840 841

	if (r > 0) {
842 843
		transport->tcp_copied += r;
		transport->tcp_offset += r;
844 845 846 847 848
	}
	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
849
		 * is turn off TCP_RCV_COPY_DATA, so the request
850 851 852 853 854
		 * will not receive any additional updates,
		 * and time out.
		 * Any remaining data from this record will
		 * be discarded.
		 */
855
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
856
		dprintk("RPC:       XID %08x truncated request\n",
857
				ntohl(transport->tcp_xid));
858 859 860 861
		dprintk("RPC:       xprt = %p, tcp_copied = %lu, "
				"tcp_offset = %u, tcp_reclen = %u\n",
				xprt, transport->tcp_copied,
				transport->tcp_offset, transport->tcp_reclen);
862 863 864
		goto out;
	}

865
	dprintk("RPC:       XID %08x read %Zd bytes\n",
866
			ntohl(transport->tcp_xid), r);
867 868 869
	dprintk("RPC:       xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
			"tcp_reclen = %u\n", xprt, transport->tcp_copied,
			transport->tcp_offset, transport->tcp_reclen);
870 871

	if (transport->tcp_copied == req->rq_private_buf.buflen)
872
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
873
	else if (transport->tcp_offset == transport->tcp_reclen) {
874 875
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
876 877 878
	}

out:
879
	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
880
		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
C
Chuck Lever 已提交
881
	spin_unlock(&xprt->transport_lock);
882
	xs_tcp_check_fraghdr(transport);
883 884
}

885
static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
886 887 888
{
	size_t len;

889
	len = transport->tcp_reclen - transport->tcp_offset;
890 891 892 893
	if (len > desc->count)
		len = desc->count;
	desc->count -= len;
	desc->offset += len;
894
	transport->tcp_offset += len;
895
	dprintk("RPC:       discarded %Zu bytes\n", len);
896
	xs_tcp_check_fraghdr(transport);
897 898
}

899
static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
900 901
{
	struct rpc_xprt *xprt = rd_desc->arg.data;
902
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
903
	struct xdr_skb_reader desc = {
904 905 906
		.skb	= skb,
		.offset	= offset,
		.count	= len,
907
	};
908

909
	dprintk("RPC:       xs_tcp_data_recv started\n");
910 911 912
	do {
		/* Read in a new fragment marker if necessary */
		/* Can we ever really expect to get completely empty fragments? */
913
		if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
914
			xs_tcp_read_fraghdr(xprt, &desc);
915 916 917
			continue;
		}
		/* Read in the xid if necessary */
918
		if (transport->tcp_flags & TCP_RCV_COPY_XID) {
919
			xs_tcp_read_xid(transport, &desc);
920 921 922
			continue;
		}
		/* Read in the request data */
923
		if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
924
			xs_tcp_read_request(xprt, &desc);
925 926 927
			continue;
		}
		/* Skip over any trailing bytes on short reads */
928
		xs_tcp_read_discard(transport, &desc);
929
	} while (desc.count);
930
	dprintk("RPC:       xs_tcp_data_recv done\n");
931 932 933
	return len - desc.count;
}

934 935 936 937 938 939 940
/**
 * 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)
941 942 943 944
{
	struct rpc_xprt *xprt;
	read_descriptor_t rd_desc;

945 946
	dprintk("RPC:       xs_tcp_data_ready...\n");

947
	read_lock(&sk->sk_callback_lock);
948
	if (!(xprt = xprt_from_sock(sk)))
949 950 951 952
		goto out;
	if (xprt->shutdown)
		goto out;

953
	/* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
954 955
	rd_desc.arg.data = xprt;
	rd_desc.count = 65536;
956
	tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
957 958 959 960
out:
	read_unlock(&sk->sk_callback_lock);
}

961 962 963 964 965 966
/**
 * 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)
967
{
968
	struct rpc_xprt *xprt;
969 970 971 972

	read_lock(&sk->sk_callback_lock);
	if (!(xprt = xprt_from_sock(sk)))
		goto out;
973 974 975 976 977
	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));
978 979 980

	switch (sk->sk_state) {
	case TCP_ESTABLISHED:
C
Chuck Lever 已提交
981
		spin_lock_bh(&xprt->transport_lock);
982
		if (!xprt_test_and_set_connected(xprt)) {
983 984 985
			struct sock_xprt *transport = container_of(xprt,
					struct sock_xprt, xprt);

986
			/* Reset TCP record info */
987 988 989
			transport->tcp_offset = 0;
			transport->tcp_reclen = 0;
			transport->tcp_copied = 0;
990 991
			transport->tcp_flags =
				TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
992

993
			xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
994
			xprt_wake_pending_tasks(xprt, 0);
995
		}
C
Chuck Lever 已提交
996
		spin_unlock_bh(&xprt->transport_lock);
997 998 999 1000
		break;
	case TCP_SYN_SENT:
	case TCP_SYN_RECV:
		break;
1001 1002 1003 1004 1005
	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)
			schedule_work(&xprt->task_cleanup);
1006 1007 1008 1009 1010 1011 1012
	default:
		xprt_disconnect(xprt);
	}
 out:
	read_unlock(&sk->sk_callback_lock);
}

1013
/**
1014 1015
 * xs_udp_write_space - callback invoked when socket buffer space
 *                             becomes available
1016 1017
 * @sk: socket whose state has changed
 *
1018 1019
 * Called when more output buffer space is available for this socket.
 * We try not to wake our writers until they can make "significant"
1020
 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1021 1022
 * with a bunch of small requests.
 */
1023
static void xs_udp_write_space(struct sock *sk)
1024 1025 1026
{
	read_lock(&sk->sk_callback_lock);

1027 1028 1029 1030 1031 1032
	/* 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)))
1033
			goto out;
1034 1035 1036
		if (unlikely(!(xprt = xprt_from_sock(sk))))
			goto out;
		if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
1037
			goto out;
1038 1039

		xprt_write_space(xprt);
1040 1041
	}

1042 1043 1044
 out:
	read_unlock(&sk->sk_callback_lock);
}
1045

1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075
/**
 * 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:
1076 1077 1078
	read_unlock(&sk->sk_callback_lock);
}

1079
static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1080
{
1081 1082
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct sock *sk = transport->inet;
1083

1084
	if (transport->rcvsize) {
1085
		sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1086
		sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1087
	}
1088
	if (transport->sndsize) {
1089
		sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1090
		sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1091 1092 1093 1094
		sk->sk_write_space(sk);
	}
}

1095
/**
1096
 * xs_udp_set_buffer_size - set send and receive limits
1097
 * @xprt: generic transport
1098 1099
 * @sndsize: requested size of send buffer, in bytes
 * @rcvsize: requested size of receive buffer, in bytes
1100
 *
1101
 * Set socket send and receive buffer size limits.
1102
 */
1103
static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1104
{
1105 1106 1107
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

	transport->sndsize = 0;
1108
	if (sndsize)
1109 1110
		transport->sndsize = sndsize + 1024;
	transport->rcvsize = 0;
1111
	if (rcvsize)
1112
		transport->rcvsize = rcvsize + 1024;
1113 1114

	xs_udp_do_set_buffer_size(xprt);
1115 1116
}

1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127
/**
 * 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);
}

1128 1129 1130 1131 1132 1133 1134
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;
}

1135 1136 1137 1138 1139 1140 1141 1142
/**
 * 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)
{
1143 1144
	struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr;

1145
	dprintk("RPC:       setting port for xprt %p to %u\n", xprt, port);
1146 1147

	sap->sin_port = htons(port);
1148 1149
}

1150
static int xs_bindresvport(struct sock_xprt *transport, struct socket *sock)
1151 1152 1153 1154
{
	struct sockaddr_in myaddr = {
		.sin_family = AF_INET,
	};
1155
	int err;
1156
	unsigned short port = transport->port;
1157 1158 1159

	do {
		myaddr.sin_port = htons(port);
1160
		err = kernel_bind(sock, (struct sockaddr *) &myaddr,
1161 1162
						sizeof(myaddr));
		if (err == 0) {
1163
			transport->port = port;
1164
			dprintk("RPC:       xs_bindresvport bound to port %u\n",
1165
					port);
1166 1167
			return 0;
		}
1168 1169 1170 1171
		if (port <= xprt_min_resvport)
			port = xprt_max_resvport;
		else
			port--;
1172
	} while (err == -EADDRINUSE && port != transport->port);
1173

1174
	dprintk("RPC:       can't bind to reserved port (%d).\n", -err);
1175 1176 1177
	return err;
}

1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206
#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

1207 1208
/**
 * xs_udp_connect_worker - set up a UDP socket
1209
 * @work: RPC transport to connect
1210 1211 1212
 *
 * Invoked by a work queue tasklet.
 */
1213
static void xs_udp_connect_worker(struct work_struct *work)
1214
{
1215 1216
	struct sock_xprt *transport =
		container_of(work, struct sock_xprt, connect_worker.work);
1217
	struct rpc_xprt *xprt = &transport->xprt;
1218
	struct socket *sock = transport->sock;
1219
	int err, status = -EIO;
1220

1221
	if (xprt->shutdown || !xprt_bound(xprt))
1222
		goto out;
1223

1224 1225
	/* Start by resetting any existing state */
	xs_close(xprt);
1226

1227
	if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
1228
		dprintk("RPC:       can't create UDP transport socket (%d).\n", -err);
1229 1230
		goto out;
	}
1231
	xs_reclassify_socket(sock);
1232

1233
	if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
1234 1235 1236
		sock_release(sock);
		goto out;
	}
1237

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

1241
	if (!transport->inet) {
1242
		struct sock *sk = sock->sk;
1243

1244
		write_lock_bh(&sk->sk_callback_lock);
1245

1246
		sk->sk_user_data = xprt;
1247 1248 1249
		transport->old_data_ready = sk->sk_data_ready;
		transport->old_state_change = sk->sk_state_change;
		transport->old_write_space = sk->sk_write_space;
1250
		sk->sk_data_ready = xs_udp_data_ready;
1251
		sk->sk_write_space = xs_udp_write_space;
1252
		sk->sk_no_check = UDP_CSUM_NORCV;
1253
		sk->sk_allocation = GFP_ATOMIC;
1254

1255 1256
		xprt_set_connected(xprt);

1257
		/* Reset to new socket */
1258 1259
		transport->sock = sock;
		transport->inet = sk;
1260

1261 1262
		write_unlock_bh(&sk->sk_callback_lock);
	}
1263
	xs_udp_do_set_buffer_size(xprt);
1264 1265 1266 1267
	status = 0;
out:
	xprt_wake_pending_tasks(xprt, status);
	xprt_clear_connecting(xprt);
1268 1269
}

1270 1271 1272 1273 1274 1275 1276
/*
 * 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;
1277
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1278 1279
	struct sockaddr any;

1280
	dprintk("RPC:       disconnecting xprt %p to reuse port\n", xprt);
1281 1282 1283 1284 1285 1286 1287

	/*
	 * 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;
1288
	result = kernel_connect(transport->sock, &any, sizeof(any), 0);
1289
	if (result)
1290
		dprintk("RPC:       AF_UNSPEC connect return code %d\n",
1291 1292 1293
				result);
}

1294
/**
1295
 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint
1296
 * @work: RPC transport to connect
1297 1298
 *
 * Invoked by a work queue tasklet.
1299
 */
1300
static void xs_tcp_connect_worker(struct work_struct *work)
1301
{
1302 1303
	struct sock_xprt *transport =
		container_of(work, struct sock_xprt, connect_worker.work);
1304
	struct rpc_xprt *xprt = &transport->xprt;
1305
	struct socket *sock = transport->sock;
1306
	int err, status = -EIO;
1307

1308
	if (xprt->shutdown || !xprt_bound(xprt))
1309 1310
		goto out;

1311
	if (!sock) {
1312 1313
		/* start from scratch */
		if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
1314 1315
			dprintk("RPC:       can't create TCP transport "
					"socket (%d).\n", -err);
1316 1317
			goto out;
		}
1318
		xs_reclassify_socket(sock);
1319

1320
		if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
1321 1322 1323 1324 1325 1326
			sock_release(sock);
			goto out;
		}
	} else
		/* "close" the socket, preserving the local port */
		xs_tcp_reuse_connection(xprt);
1327

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

1331
	if (!transport->inet) {
1332 1333 1334 1335 1336
		struct sock *sk = sock->sk;

		write_lock_bh(&sk->sk_callback_lock);

		sk->sk_user_data = xprt;
1337 1338 1339
		transport->old_data_ready = sk->sk_data_ready;
		transport->old_state_change = sk->sk_state_change;
		transport->old_write_space = sk->sk_write_space;
1340 1341 1342
		sk->sk_data_ready = xs_tcp_data_ready;
		sk->sk_state_change = xs_tcp_state_change;
		sk->sk_write_space = xs_tcp_write_space;
1343
		sk->sk_allocation = GFP_ATOMIC;
1344 1345 1346 1347 1348 1349

		/* 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;
1350 1351 1352 1353

		xprt_clear_connected(xprt);

		/* Reset to new socket */
1354 1355
		transport->sock = sock;
		transport->inet = sk;
1356 1357 1358 1359 1360

		write_unlock_bh(&sk->sk_callback_lock);
	}

	/* Tell the socket layer to start connecting... */
1361 1362
	xprt->stat.connect_count++;
	xprt->stat.connect_start = jiffies;
1363
	status = kernel_connect(sock, (struct sockaddr *) &xprt->addr,
1364
			xprt->addrlen, O_NONBLOCK);
1365 1366 1367
	dprintk("RPC:       %p connect status %d connected %d sock state %d\n",
			xprt, -status, xprt_connected(xprt),
			sock->sk->sk_state);
1368 1369 1370 1371 1372
	if (status < 0) {
		switch (status) {
			case -EINPROGRESS:
			case -EALREADY:
				goto out_clear;
1373 1374 1375 1376 1377 1378 1379 1380
			case -ECONNREFUSED:
			case -ECONNRESET:
				/* retry with existing socket, after a delay */
				break;
			default:
				/* get rid of existing socket, and retry */
				xs_close(xprt);
				break;
1381 1382 1383
		}
	}
out:
1384
	xprt_wake_pending_tasks(xprt, status);
1385
out_clear:
1386
	xprt_clear_connecting(xprt);
1387 1388
}

1389 1390 1391 1392 1393
/**
 * 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.
1394 1395 1396 1397 1398 1399 1400
 *
 * 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).
1401 1402
 */
static void xs_connect(struct rpc_task *task)
1403 1404
{
	struct rpc_xprt *xprt = task->tk_xprt;
1405
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1406

1407 1408 1409
	if (xprt_test_and_set_connecting(xprt))
		return;

1410
	if (transport->sock != NULL) {
1411 1412
		dprintk("RPC:       xs_connect delayed xprt %p for %lu "
				"seconds\n",
1413
				xprt, xprt->reestablish_timeout / HZ);
1414
		schedule_delayed_work(&transport->connect_worker,
1415 1416 1417 1418
					xprt->reestablish_timeout);
		xprt->reestablish_timeout <<= 1;
		if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
			xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
1419
	} else {
1420
		dprintk("RPC:       xs_connect scheduled xprt %p\n", xprt);
1421
		schedule_delayed_work(&transport->connect_worker, 0);
1422 1423 1424 1425

		/* flush_scheduled_work can sleep... */
		if (!RPC_IS_ASYNC(task))
			flush_scheduled_work();
1426 1427 1428
	}
}

1429 1430 1431 1432 1433 1434 1435 1436
/**
 * 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)
{
1437 1438
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

1439
	seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
1440
			transport->port,
1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456
			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)
{
1457
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1458 1459 1460 1461 1462 1463
	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",
1464
			transport->port,
1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475
			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);
}

1476
static struct rpc_xprt_ops xs_udp_ops = {
1477
	.set_buffer_size	= xs_udp_set_buffer_size,
1478
	.reserve_xprt		= xprt_reserve_xprt_cong,
1479
	.release_xprt		= xprt_release_xprt_cong,
1480
	.rpcbind		= rpc_getport,
1481
	.set_port		= xs_set_port,
1482
	.connect		= xs_connect,
1483 1484
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1485
	.send_request		= xs_udp_send_request,
1486
	.set_retrans_timeout	= xprt_set_retrans_timeout_rtt,
1487
	.timer			= xs_udp_timer,
1488
	.release_request	= xprt_release_rqst_cong,
1489 1490
	.close			= xs_close,
	.destroy		= xs_destroy,
1491
	.print_stats		= xs_udp_print_stats,
1492 1493 1494
};

static struct rpc_xprt_ops xs_tcp_ops = {
1495
	.reserve_xprt		= xprt_reserve_xprt,
1496
	.release_xprt		= xs_tcp_release_xprt,
1497
	.rpcbind		= rpc_getport,
1498
	.set_port		= xs_set_port,
1499
	.connect		= xs_connect,
1500 1501
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1502
	.send_request		= xs_tcp_send_request,
1503
	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
1504 1505
	.close			= xs_close,
	.destroy		= xs_destroy,
1506
	.print_stats		= xs_tcp_print_stats,
1507 1508
};

1509 1510 1511
static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size)
{
	struct rpc_xprt *xprt;
1512
	struct sock_xprt *new;
1513 1514

	if (addrlen > sizeof(xprt->addr)) {
1515
		dprintk("RPC:       xs_setup_xprt: address too large\n");
1516 1517 1518
		return ERR_PTR(-EBADF);
	}

1519 1520
	new = kzalloc(sizeof(*new), GFP_KERNEL);
	if (new == NULL) {
1521 1522
		dprintk("RPC:       xs_setup_xprt: couldn't allocate "
				"rpc_xprt\n");
1523 1524
		return ERR_PTR(-ENOMEM);
	}
1525
	xprt = &new->xprt;
1526 1527 1528 1529 1530

	xprt->max_reqs = slot_table_size;
	xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL);
	if (xprt->slot == NULL) {
		kfree(xprt);
1531 1532
		dprintk("RPC:       xs_setup_xprt: couldn't allocate slot "
				"table\n");
1533 1534 1535 1536 1537
		return ERR_PTR(-ENOMEM);
	}

	memcpy(&xprt->addr, addr, addrlen);
	xprt->addrlen = addrlen;
1538
	new->port = xs_get_random_port();
1539 1540 1541 1542

	return xprt;
}

1543 1544
/**
 * xs_setup_udp - Set up transport to use a UDP socket
1545 1546
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1547 1548 1549
 * @to:   timeout parameters
 *
 */
1550
struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1551
{
1552
	struct rpc_xprt *xprt;
1553
	struct sock_xprt *transport;
1554

1555 1556 1557
	xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1558
	transport = container_of(xprt, struct sock_xprt, xprt);
1559

1560
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1561 1562 1563
		xprt_set_bound(xprt);

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

1568
	INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_connect_worker);
1569 1570 1571 1572
	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;
1573

1574
	xprt->ops = &xs_udp_ops;
1575 1576 1577 1578

	if (to)
		xprt->timeout = *to;
	else
1579
		xprt_set_timeout(&xprt->timeout, 5, 5 * HZ);
1580

1581
	xs_format_peer_addresses(xprt);
1582
	dprintk("RPC:       set up transport to address %s\n",
1583
			xprt->address_strings[RPC_DISPLAY_ALL]);
1584

1585
	return xprt;
1586 1587
}

1588 1589
/**
 * xs_setup_tcp - Set up transport to use a TCP socket
1590 1591
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1592 1593 1594
 * @to: timeout parameters
 *
 */
1595
struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1596
{
1597
	struct rpc_xprt *xprt;
1598
	struct sock_xprt *transport;
1599

1600 1601 1602
	xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1603
	transport = container_of(xprt, struct sock_xprt, xprt);
1604

1605
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1606 1607 1608
		xprt_set_bound(xprt);

	xprt->prot = IPPROTO_TCP;
1609 1610
	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
1611

1612
	INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker);
1613 1614 1615 1616
	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;
1617

1618
	xprt->ops = &xs_tcp_ops;
1619 1620 1621 1622

	if (to)
		xprt->timeout = *to;
	else
1623
		xprt_set_timeout(&xprt->timeout, 2, 60 * HZ);
1624

1625
	xs_format_peer_addresses(xprt);
1626
	dprintk("RPC:       set up transport to address %s\n",
1627
			xprt->address_strings[RPC_DISPLAY_ALL]);
1628

1629
	return xprt;
1630
}
1631 1632

/**
1633
 * init_socket_xprt - set up xprtsock's sysctls
1634 1635 1636 1637
 *
 */
int init_socket_xprt(void)
{
1638 1639 1640 1641 1642 1643 1644 1645 1646 1647
#ifdef RPC_DEBUG
	if (!sunrpc_table_header) {
		sunrpc_table_header = register_sysctl_table(sunrpc_table, 1);
#ifdef CONFIG_PROC_FS
		if (sunrpc_table[0].de)
			sunrpc_table[0].de->owner = THIS_MODULE;
#endif
	}
#endif

1648 1649 1650 1651
	return 0;
}

/**
1652
 * cleanup_socket_xprt - remove xprtsock's sysctls
1653 1654 1655 1656
 *
 */
void cleanup_socket_xprt(void)
{
1657 1658 1659 1660 1661 1662
#ifdef RPC_DEBUG
	if (sunrpc_table_header) {
		unregister_sysctl_table(sunrpc_table_header);
		sunrpc_table_header = NULL;
	}
#endif
1663
}