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

#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
	unsigned short		port;
239 240 241 242 243 244

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

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

254 255 256 257 258 259 260 261
/*
 * 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)

262 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
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]);
}

303 304
#define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	return status;
496 497
}

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

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

527
	xs_encode_tcp_record_marker(&req->rq_snd_buf);
528

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

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

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

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

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

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

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

578 579 580
	return status;
}

581 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
/**
 * 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);
}

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

	if (!sk)
622
		goto clear_close_wait;
623

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

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

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

636
	sk->sk_no_check = 0;
637 638

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

		xprt_write_space(xprt);
1039 1040
	}

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

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
/**
 * 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:
1075 1076 1077
	read_unlock(&sk->sk_callback_lock);
}

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

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

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

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

	xs_udp_do_set_buffer_size(xprt);
1114 1115
}

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

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

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

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

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

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

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

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

1177 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
#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

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

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

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

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

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

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

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

1243
		write_lock_bh(&sk->sk_callback_lock);
1244

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

1254 1255
		xprt_set_connected(xprt);

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

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

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

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

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

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

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

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

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

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

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

		write_lock_bh(&sk->sk_callback_lock);

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

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

		xprt_clear_connected(xprt);

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

		write_unlock_bh(&sk->sk_callback_lock);
	}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	return xprt;
}

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

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

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

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

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

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

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

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

1584
	return xprt;
1585 1586
}

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

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

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

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

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

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

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

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

1628
	return xprt;
1629
}
1630 1631

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

1647 1648 1649 1650
	return 0;
}

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