xprtsock.c 39.7 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
/*
 * How many times to try sending a request on a socket before waiting
 * for the socket buffer to clear.
 */
#define XS_SENDMSG_RETRY	(10U)

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

98 99
#ifdef RPC_DEBUG
# undef  RPC_DEBUG_DATA
100
# define RPCDBG_FACILITY	RPCDBG_TRANS
101 102 103
#endif

#ifdef RPC_DEBUG_DATA
104
static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
105
{
106 107
	u8 *buf = (u8 *) packet;
	int j;
108 109 110 111 112 113 114 115 116 117 118 119 120 121

	dprintk("RPC:      %s\n", msg);
	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
122
static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
123 124 125 126 127
{
	/* NOP */
}
#endif

128 129
struct sock_xprt {
	struct rpc_xprt		xprt;
130 131 132 133 134 135

	/*
	 * Network layer
	 */
	struct socket *		sock;
	struct sock *		inet;
136 137 138 139 140 141 142 143 144 145 146 147

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

	u32			tcp_offset,
				tcp_reclen;

	unsigned long		tcp_copied,
				tcp_flags;
148 149 150 151 152 153

	/*
	 * Connection of transports
	 */
	struct work_struct	connect_worker;
	unsigned short		port;
154 155 156 157 158 159

	/*
	 * UDP socket buffer size parameters
	 */
	size_t			rcvsize,
				sndsize;
160 161 162 163 164 165 166

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

169 170 171 172 173 174 175 176
/*
 * 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)

177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217
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]);
}

218 219
#define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)

T
Trond Myklebust 已提交
220
static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
221 222 223 224
{
	struct msghdr msg = {
		.msg_name	= addr,
		.msg_namelen	= addrlen,
T
Trond Myklebust 已提交
225 226 227 228 229
		.msg_flags	= XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
	};
	struct kvec iov = {
		.iov_base	= vec->iov_base + base,
		.iov_len	= vec->iov_len - base,
230 231
	};

T
Trond Myklebust 已提交
232
	if (iov.iov_len != 0)
233 234 235 236
		return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
	return kernel_sendmsg(sock, &msg, NULL, 0, 0);
}

T
Trond Myklebust 已提交
237
static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
238
{
T
Trond Myklebust 已提交
239 240 241 242 243 244 245 246 247 248 249
	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;
250

T
Trond Myklebust 已提交
251 252 253 254 255 256 257 258 259 260 261 262 263 264 265
		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;
266 267
}

268 269 270 271 272 273 274 275
/**
 * 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
 *
276
 */
T
Trond Myklebust 已提交
277
static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
278
{
T
Trond Myklebust 已提交
279 280
	unsigned int remainder = xdr->len - base;
	int err, sent = 0;
281

282 283 284 285
	if (unlikely(!sock))
		return -ENOTCONN;

	clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
T
Trond Myklebust 已提交
286 287 288 289
	if (base != 0) {
		addr = NULL;
		addrlen = 0;
	}
290

T
Trond Myklebust 已提交
291 292 293 294 295
	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)
296
			goto out;
T
Trond Myklebust 已提交
297
		sent += err;
298 299
		base = 0;
	} else
T
Trond Myklebust 已提交
300
		base -= xdr->head[0].iov_len;
301

T
Trond Myklebust 已提交
302 303 304 305 306
	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)
307
			goto out;
T
Trond Myklebust 已提交
308
		sent += err;
309
		base = 0;
T
Trond Myklebust 已提交
310 311 312 313 314 315
	} 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);
316
out:
T
Trond Myklebust 已提交
317 318 319 320 321
	if (sent == 0)
		return err;
	if (err > 0)
		sent += err;
	return sent;
322 323
}

324
/**
325 326
 * xs_nospace - place task on wait queue if transmit was incomplete
 * @task: task to put to sleep
327
 *
328
 */
329
static void xs_nospace(struct rpc_task *task)
330
{
331 332
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
333
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
334

335 336 337 338
	dprintk("RPC: %4d xmit incomplete (%u left of %u)\n",
			task->tk_pid, req->rq_slen - req->rq_bytes_sent,
			req->rq_slen);

339
	if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
340 341 342 343 344 345
		/* 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;
346
		else if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369
			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;
370
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
371 372
	struct xdr_buf *xdr = &req->rq_snd_buf;
	int status;
373

374
	xs_pktdump("packet data:",
375 376 377
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);

378
	req->rq_xtime = jiffies;
379 380 381 382
	status = xs_sendpages(transport->sock,
			      (struct sockaddr *) &xprt->addr,
			      xprt->addrlen, xdr,
			      req->rq_bytes_sent);
383

384 385
	dprintk("RPC:      xs_udp_send_request(%u) = %d\n",
			xdr->len - req->rq_bytes_sent, status);
386

387 388
	if (likely(status >= (int) req->rq_slen))
		return 0;
389

390 391 392
	/* Still some bytes left; set up for a retry later. */
	if (status > 0)
		status = -EAGAIN;
393

394 395 396
	switch (status) {
	case -ENETUNREACH:
	case -EPIPE:
397 398
	case -ECONNREFUSED:
		/* When the server has died, an ICMP port unreachable message
399
		 * prompts ECONNREFUSED. */
400
		break;
401 402
	case -EAGAIN:
		xs_nospace(task);
403 404
		break;
	default:
405 406
		dprintk("RPC:      sendmsg returned unrecognized error %d\n",
			-status);
407
		break;
408
	}
409 410

	return status;
411 412
}

413 414 415 416 417 418 419
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);
}

420
/**
421
 * xs_tcp_send_request - write an RPC request to a TCP socket
422 423 424
 * @task: address of RPC task that manages the state of an RPC request
 *
 * Return values:
425 426 427 428 429
 *        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
430 431
 *
 * XXX: In the case of soft timeouts, should we eventually give up
432
 *	if sendmsg is not able to make progress?
433
 */
434
static int xs_tcp_send_request(struct rpc_task *task)
435 436 437
{
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
438
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
439
	struct xdr_buf *xdr = &req->rq_snd_buf;
440 441
	int status, retry = 0;

442
	xs_encode_tcp_record_marker(&req->rq_snd_buf);
443

444 445 446
	xs_pktdump("packet data:",
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);
447 448 449

	/* Continue transmitting the packet/record. We must be careful
	 * to cope with writespace callbacks arriving _after_ we have
450
	 * called sendmsg(). */
451 452
	while (1) {
		req->rq_xtime = jiffies;
453 454
		status = xs_sendpages(transport->sock,
					NULL, 0, xdr, req->rq_bytes_sent);
455

456 457
		dprintk("RPC:      xs_tcp_send_request(%u) = %d\n",
				xdr->len - req->rq_bytes_sent, status);
458

459
		if (unlikely(status < 0))
460 461
			break;

462 463 464
		/* If we've sent the entire packet, immediately
		 * reset the count of bytes sent. */
		req->rq_bytes_sent += status;
465
		task->tk_bytes_sent += status;
466 467 468 469
		if (likely(req->rq_bytes_sent >= req->rq_slen)) {
			req->rq_bytes_sent = 0;
			return 0;
		}
470 471

		status = -EAGAIN;
472
		if (retry++ > XS_SENDMSG_RETRY)
473 474 475
			break;
	}

476 477 478 479 480 481 482 483 484 485 486 487 488
	switch (status) {
	case -EAGAIN:
		xs_nospace(task);
		break;
	case -ECONNREFUSED:
	case -ECONNRESET:
	case -ENOTCONN:
	case -EPIPE:
		status = -ENOTCONN;
		break;
	default:
		dprintk("RPC:      sendmsg returned unrecognized error %d\n",
			-status);
489
		xprt_disconnect(xprt);
490
		break;
491
	}
492

493 494 495
	return status;
}

496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522
/**
 * 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);
}

523 524 525 526
/**
 * xs_close - close a socket
 * @xprt: transport
 *
527 528
 * This is used when all requests are complete; ie, no DRC state remains
 * on the server we want to save.
529
 */
530
static void xs_close(struct rpc_xprt *xprt)
531
{
532 533 534
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct socket *sock = transport->sock;
	struct sock *sk = transport->inet;
535 536

	if (!sk)
537
		goto clear_close_wait;
538

539 540
	dprintk("RPC:      xs_close xprt %p\n", xprt);

541
	write_lock_bh(&sk->sk_callback_lock);
542 543
	transport->inet = NULL;
	transport->sock = NULL;
544

545
	sk->sk_user_data = NULL;
546 547 548
	sk->sk_data_ready = transport->old_data_ready;
	sk->sk_state_change = transport->old_state_change;
	sk->sk_write_space = transport->old_write_space;
549 550
	write_unlock_bh(&sk->sk_callback_lock);

551
	sk->sk_no_check = 0;
552 553

	sock_release(sock);
554 555 556 557
clear_close_wait:
	smp_mb__before_clear_bit();
	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
	smp_mb__after_clear_bit();
558 559
}

560 561 562 563 564 565
/**
 * xs_destroy - prepare to shutdown a transport
 * @xprt: doomed transport
 *
 */
static void xs_destroy(struct rpc_xprt *xprt)
566
{
567 568
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

569 570
	dprintk("RPC:      xs_destroy xprt %p\n", xprt);

571
	cancel_delayed_work(&transport->connect_worker);
572 573 574
	flush_scheduled_work();

	xprt_disconnect(xprt);
575
	xs_close(xprt);
576
	xs_free_peer_addresses(xprt);
577
	kfree(xprt->slot);
578
	kfree(xprt);
579 580
}

581 582 583 584 585 586 587 588 589 590
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
 *
591
 */
592
static void xs_udp_data_ready(struct sock *sk, int len)
593
{
594 595
	struct rpc_task *task;
	struct rpc_xprt *xprt;
596
	struct rpc_rqst *rovr;
597
	struct sk_buff *skb;
598
	int err, repsize, copied;
599 600
	u32 _xid;
	__be32 *xp;
601 602

	read_lock(&sk->sk_callback_lock);
603 604
	dprintk("RPC:      xs_udp_data_ready...\n");
	if (!(xprt = xprt_from_sock(sk)))
605 606 607 608 609 610 611 612 613 614
		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) {
615
		dprintk("RPC:      impossible RPC reply size %d!\n", repsize);
616 617 618 619 620 621 622 623 624 625
		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 已提交
626
	spin_lock(&xprt->transport_lock);
627 628 629 630 631 632 633 634 635 636 637 638 639 640 641
	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);

642 643 644
	xprt_adjust_cwnd(task, copied);
	xprt_update_rtt(task);
	xprt_complete_rqst(task, copied);
645 646

 out_unlock:
C
Chuck Lever 已提交
647
	spin_unlock(&xprt->transport_lock);
648 649 650 651 652 653
 dropit:
	skb_free_datagram(sk, skb);
 out:
	read_unlock(&sk->sk_callback_lock);
}

654
static inline size_t xs_tcp_copy_data(skb_reader_t *desc, void *p, size_t len)
655 656 657 658 659 660 661 662 663 664 665 666 667 668 669
{
	if (len > desc->count)
		len = desc->count;
	if (skb_copy_bits(desc->skb, desc->offset, p, len)) {
		dprintk("RPC:      failed to copy %zu bytes from skb. %zu bytes remain\n",
				len, desc->count);
		return 0;
	}
	desc->offset += len;
	desc->count -= len;
	dprintk("RPC:      copied %zu bytes from skb. %zu bytes remain\n",
			len, desc->count);
	return len;
}

670
static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc)
671
{
672
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
673 674 675
	size_t len, used;
	char *p;

676 677
	p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
	len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
678
	used = xs_tcp_copy_data(desc, p, len);
679
	transport->tcp_offset += used;
680 681
	if (used != len)
		return;
682

683 684
	transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
	if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
685
		transport->tcp_flags |= TCP_RCV_LAST_FRAG;
686
	else
687
		transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
688
	transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
689

690
	transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
691
	transport->tcp_offset = 0;
692

693
	/* Sanity check of the record length */
694
	if (unlikely(transport->tcp_reclen < 4)) {
695
		dprintk("RPC:      invalid TCP record fragment length\n");
696
		xprt_disconnect(xprt);
697
		return;
698 699
	}
	dprintk("RPC:      reading TCP record fragment of length %d\n",
700
			transport->tcp_reclen);
701 702
}

703
static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
704
{
705
	if (transport->tcp_offset == transport->tcp_reclen) {
706
		transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
707
		transport->tcp_offset = 0;
708 709 710
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
			transport->tcp_flags |= TCP_RCV_COPY_XID;
711
			transport->tcp_copied = 0;
712 713 714 715
		}
	}
}

716
static inline void xs_tcp_read_xid(struct sock_xprt *transport, skb_reader_t *desc)
717 718 719 720
{
	size_t len, used;
	char *p;

721
	len = sizeof(transport->tcp_xid) - transport->tcp_offset;
722
	dprintk("RPC:      reading XID (%Zu bytes)\n", len);
723
	p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
724
	used = xs_tcp_copy_data(desc, p, len);
725
	transport->tcp_offset += used;
726 727
	if (used != len)
		return;
728 729
	transport->tcp_flags &= ~TCP_RCV_COPY_XID;
	transport->tcp_flags |= TCP_RCV_COPY_DATA;
730
	transport->tcp_copied = 4;
731
	dprintk("RPC:      reading reply for XID %08x\n",
732 733
			ntohl(transport->tcp_xid));
	xs_tcp_check_fraghdr(transport);
734 735
}

736
static inline void xs_tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc)
737
{
738
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
739 740 741 742 743 744
	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 已提交
745
	spin_lock(&xprt->transport_lock);
746
	req = xprt_lookup_rqst(xprt, transport->tcp_xid);
747
	if (!req) {
748
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
749
		dprintk("RPC:      XID %08x request not found!\n",
750
				ntohl(transport->tcp_xid));
C
Chuck Lever 已提交
751
		spin_unlock(&xprt->transport_lock);
752 753 754 755 756
		return;
	}

	rcvbuf = &req->rq_private_buf;
	len = desc->count;
757
	if (len > transport->tcp_reclen - transport->tcp_offset) {
758 759
		skb_reader_t my_desc;

760
		len = transport->tcp_reclen - transport->tcp_offset;
761 762
		memcpy(&my_desc, desc, sizeof(my_desc));
		my_desc.count = len;
763
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
764
					  &my_desc, xs_tcp_copy_data);
765 766 767
		desc->count -= r;
		desc->offset += r;
	} else
768
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
769
					  desc, xs_tcp_copy_data);
770 771

	if (r > 0) {
772 773
		transport->tcp_copied += r;
		transport->tcp_offset += r;
774 775 776 777 778
	}
	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
779
		 * is turn off TCP_RCV_COPY_DATA, so the request
780 781 782 783 784
		 * will not receive any additional updates,
		 * and time out.
		 * Any remaining data from this record will
		 * be discarded.
		 */
785
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
786
		dprintk("RPC:      XID %08x truncated request\n",
787
				ntohl(transport->tcp_xid));
788
		dprintk("RPC:      xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
789 790
				xprt, transport->tcp_copied, transport->tcp_offset,
					transport->tcp_reclen);
791 792 793 794
		goto out;
	}

	dprintk("RPC:      XID %08x read %Zd bytes\n",
795
			ntohl(transport->tcp_xid), r);
796
	dprintk("RPC:      xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
797 798 799 800
			xprt, transport->tcp_copied, transport->tcp_offset,
				transport->tcp_reclen);

	if (transport->tcp_copied == req->rq_private_buf.buflen)
801
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
802
	else if (transport->tcp_offset == transport->tcp_reclen) {
803 804
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
805 806 807
	}

out:
808
	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
809
		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
C
Chuck Lever 已提交
810
	spin_unlock(&xprt->transport_lock);
811
	xs_tcp_check_fraghdr(transport);
812 813
}

814
static inline void xs_tcp_read_discard(struct sock_xprt *transport, skb_reader_t *desc)
815 816 817
{
	size_t len;

818
	len = transport->tcp_reclen - transport->tcp_offset;
819 820 821 822
	if (len > desc->count)
		len = desc->count;
	desc->count -= len;
	desc->offset += len;
823
	transport->tcp_offset += len;
824
	dprintk("RPC:      discarded %Zu bytes\n", len);
825
	xs_tcp_check_fraghdr(transport);
826 827
}

828
static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
829 830
{
	struct rpc_xprt *xprt = rd_desc->arg.data;
831
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
832 833 834 835
	skb_reader_t desc = {
		.skb	= skb,
		.offset	= offset,
		.count	= len,
836
	};
837

838
	dprintk("RPC:      xs_tcp_data_recv started\n");
839 840 841
	do {
		/* Read in a new fragment marker if necessary */
		/* Can we ever really expect to get completely empty fragments? */
842
		if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
843
			xs_tcp_read_fraghdr(xprt, &desc);
844 845 846
			continue;
		}
		/* Read in the xid if necessary */
847
		if (transport->tcp_flags & TCP_RCV_COPY_XID) {
848
			xs_tcp_read_xid(transport, &desc);
849 850 851
			continue;
		}
		/* Read in the request data */
852
		if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
853
			xs_tcp_read_request(xprt, &desc);
854 855 856
			continue;
		}
		/* Skip over any trailing bytes on short reads */
857
		xs_tcp_read_discard(transport, &desc);
858
	} while (desc.count);
859
	dprintk("RPC:      xs_tcp_data_recv done\n");
860 861 862
	return len - desc.count;
}

863 864 865 866 867 868 869
/**
 * 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)
870 871 872 873 874
{
	struct rpc_xprt *xprt;
	read_descriptor_t rd_desc;

	read_lock(&sk->sk_callback_lock);
875 876
	dprintk("RPC:      xs_tcp_data_ready...\n");
	if (!(xprt = xprt_from_sock(sk)))
877 878 879 880
		goto out;
	if (xprt->shutdown)
		goto out;

881
	/* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
882 883
	rd_desc.arg.data = xprt;
	rd_desc.count = 65536;
884
	tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
885 886 887 888
out:
	read_unlock(&sk->sk_callback_lock);
}

889 890 891 892 893 894
/**
 * 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)
895
{
896
	struct rpc_xprt *xprt;
897 898 899 900

	read_lock(&sk->sk_callback_lock);
	if (!(xprt = xprt_from_sock(sk)))
		goto out;
901
	dprintk("RPC:      xs_tcp_state_change client %p...\n", xprt);
902 903 904 905 906 907 908
	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));

	switch (sk->sk_state) {
	case TCP_ESTABLISHED:
C
Chuck Lever 已提交
909
		spin_lock_bh(&xprt->transport_lock);
910
		if (!xprt_test_and_set_connected(xprt)) {
911 912 913
			struct sock_xprt *transport = container_of(xprt,
					struct sock_xprt, xprt);

914
			/* Reset TCP record info */
915 916 917
			transport->tcp_offset = 0;
			transport->tcp_reclen = 0;
			transport->tcp_copied = 0;
918 919
			transport->tcp_flags =
				TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
920

921
			xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
922
			xprt_wake_pending_tasks(xprt, 0);
923
		}
C
Chuck Lever 已提交
924
		spin_unlock_bh(&xprt->transport_lock);
925 926 927 928
		break;
	case TCP_SYN_SENT:
	case TCP_SYN_RECV:
		break;
929 930 931 932 933
	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);
934 935 936 937 938 939 940
	default:
		xprt_disconnect(xprt);
	}
 out:
	read_unlock(&sk->sk_callback_lock);
}

941
/**
942 943
 * xs_udp_write_space - callback invoked when socket buffer space
 *                             becomes available
944 945
 * @sk: socket whose state has changed
 *
946 947
 * Called when more output buffer space is available for this socket.
 * We try not to wake our writers until they can make "significant"
948
 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
949 950
 * with a bunch of small requests.
 */
951
static void xs_udp_write_space(struct sock *sk)
952 953 954
{
	read_lock(&sk->sk_callback_lock);

955 956 957 958 959 960
	/* 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)))
961
			goto out;
962 963 964
		if (unlikely(!(xprt = xprt_from_sock(sk))))
			goto out;
		if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
965
			goto out;
966 967

		xprt_write_space(xprt);
968 969
	}

970 971 972
 out:
	read_unlock(&sk->sk_callback_lock);
}
973

974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003
/**
 * 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:
1004 1005 1006
	read_unlock(&sk->sk_callback_lock);
}

1007
static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1008
{
1009 1010
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct sock *sk = transport->inet;
1011

1012
	if (transport->rcvsize) {
1013
		sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1014
		sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1015
	}
1016
	if (transport->sndsize) {
1017
		sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1018
		sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1019 1020 1021 1022
		sk->sk_write_space(sk);
	}
}

1023
/**
1024
 * xs_udp_set_buffer_size - set send and receive limits
1025
 * @xprt: generic transport
1026 1027
 * @sndsize: requested size of send buffer, in bytes
 * @rcvsize: requested size of receive buffer, in bytes
1028
 *
1029
 * Set socket send and receive buffer size limits.
1030
 */
1031
static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1032
{
1033 1034 1035
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

	transport->sndsize = 0;
1036
	if (sndsize)
1037 1038
		transport->sndsize = sndsize + 1024;
	transport->rcvsize = 0;
1039
	if (rcvsize)
1040
		transport->rcvsize = rcvsize + 1024;
1041 1042

	xs_udp_do_set_buffer_size(xprt);
1043 1044
}

1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055
/**
 * 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);
}

1056 1057 1058 1059 1060 1061 1062
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;
}

1063 1064 1065 1066 1067 1068 1069 1070
/**
 * 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)
{
1071 1072
	struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr;

1073
	dprintk("RPC:      setting port for xprt %p to %u\n", xprt, port);
1074 1075

	sap->sin_port = htons(port);
1076 1077
}

1078
static int xs_bindresvport(struct sock_xprt *transport, struct socket *sock)
1079 1080 1081 1082
{
	struct sockaddr_in myaddr = {
		.sin_family = AF_INET,
	};
1083
	int err;
1084
	unsigned short port = transport->port;
1085 1086 1087

	do {
		myaddr.sin_port = htons(port);
1088
		err = kernel_bind(sock, (struct sockaddr *) &myaddr,
1089 1090
						sizeof(myaddr));
		if (err == 0) {
1091
			transport->port = port;
1092 1093
			dprintk("RPC:      xs_bindresvport bound to port %u\n",
					port);
1094 1095
			return 0;
		}
1096 1097 1098 1099
		if (port <= xprt_min_resvport)
			port = xprt_max_resvport;
		else
			port--;
1100
	} while (err == -EADDRINUSE && port != transport->port);
1101

1102
	dprintk("RPC:      can't bind to reserved port (%d).\n", -err);
1103 1104 1105
	return err;
}

1106 1107 1108 1109 1110 1111 1112
/**
 * xs_udp_connect_worker - set up a UDP socket
 * @args: RPC transport to connect
 *
 * Invoked by a work queue tasklet.
 */
static void xs_udp_connect_worker(void *args)
1113
{
1114 1115
	struct sock_xprt *transport = (struct sock_xprt *)args;
	struct rpc_xprt *xprt = &transport->xprt;
1116
	struct socket *sock = transport->sock;
1117
	int err, status = -EIO;
1118

1119
	if (xprt->shutdown || !xprt_bound(xprt))
1120
		goto out;
1121

1122 1123
	/* Start by resetting any existing state */
	xs_close(xprt);
1124

1125 1126 1127 1128
	if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
		dprintk("RPC:      can't create UDP transport socket (%d).\n", -err);
		goto out;
	}
1129

1130
	if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
1131 1132 1133
		sock_release(sock);
		goto out;
	}
1134

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

1138
	if (!transport->inet) {
1139
		struct sock *sk = sock->sk;
1140

1141
		write_lock_bh(&sk->sk_callback_lock);
1142

1143
		sk->sk_user_data = xprt;
1144 1145 1146
		transport->old_data_ready = sk->sk_data_ready;
		transport->old_state_change = sk->sk_state_change;
		transport->old_write_space = sk->sk_write_space;
1147
		sk->sk_data_ready = xs_udp_data_ready;
1148
		sk->sk_write_space = xs_udp_write_space;
1149
		sk->sk_no_check = UDP_CSUM_NORCV;
1150
		sk->sk_allocation = GFP_ATOMIC;
1151

1152 1153
		xprt_set_connected(xprt);

1154
		/* Reset to new socket */
1155 1156
		transport->sock = sock;
		transport->inet = sk;
1157

1158 1159
		write_unlock_bh(&sk->sk_callback_lock);
	}
1160
	xs_udp_do_set_buffer_size(xprt);
1161 1162 1163 1164
	status = 0;
out:
	xprt_wake_pending_tasks(xprt, status);
	xprt_clear_connecting(xprt);
1165 1166
}

1167 1168 1169 1170 1171 1172 1173
/*
 * 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;
1174
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1175 1176 1177 1178 1179 1180 1181 1182 1183 1184
	struct sockaddr any;

	dprintk("RPC:      disconnecting xprt %p to reuse port\n", xprt);

	/*
	 * 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;
1185
	result = kernel_connect(transport->sock, &any, sizeof(any), 0);
1186 1187 1188 1189 1190
	if (result)
		dprintk("RPC:      AF_UNSPEC connect return code %d\n",
				result);
}

1191
/**
1192
 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint
1193 1194 1195
 * @args: RPC transport to connect
 *
 * Invoked by a work queue tasklet.
1196
 */
1197
static void xs_tcp_connect_worker(void *args)
1198
{
1199 1200
	struct sock_xprt *transport = (struct sock_xprt *)args;
	struct rpc_xprt *xprt = &transport->xprt;
1201
	struct socket *sock = transport->sock;
1202
	int err, status = -EIO;
1203

1204
	if (xprt->shutdown || !xprt_bound(xprt))
1205 1206
		goto out;

1207
	if (!sock) {
1208 1209 1210 1211 1212
		/* start from scratch */
		if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
			dprintk("RPC:      can't create TCP transport socket (%d).\n", -err);
			goto out;
		}
1213

1214
		if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
1215 1216 1217 1218 1219 1220
			sock_release(sock);
			goto out;
		}
	} else
		/* "close" the socket, preserving the local port */
		xs_tcp_reuse_connection(xprt);
1221

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

1225
	if (!transport->inet) {
1226 1227 1228 1229 1230
		struct sock *sk = sock->sk;

		write_lock_bh(&sk->sk_callback_lock);

		sk->sk_user_data = xprt;
1231 1232 1233
		transport->old_data_ready = sk->sk_data_ready;
		transport->old_state_change = sk->sk_state_change;
		transport->old_write_space = sk->sk_write_space;
1234 1235 1236
		sk->sk_data_ready = xs_tcp_data_ready;
		sk->sk_state_change = xs_tcp_state_change;
		sk->sk_write_space = xs_tcp_write_space;
1237
		sk->sk_allocation = GFP_ATOMIC;
1238 1239 1240 1241 1242 1243

		/* 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;
1244 1245 1246 1247

		xprt_clear_connected(xprt);

		/* Reset to new socket */
1248 1249
		transport->sock = sock;
		transport->inet = sk;
1250 1251 1252 1253 1254

		write_unlock_bh(&sk->sk_callback_lock);
	}

	/* Tell the socket layer to start connecting... */
1255 1256
	xprt->stat.connect_count++;
	xprt->stat.connect_start = jiffies;
1257
	status = kernel_connect(sock, (struct sockaddr *) &xprt->addr,
1258
			xprt->addrlen, O_NONBLOCK);
1259 1260 1261 1262 1263 1264 1265
	dprintk("RPC: %p  connect status %d connected %d sock state %d\n",
			xprt, -status, xprt_connected(xprt), sock->sk->sk_state);
	if (status < 0) {
		switch (status) {
			case -EINPROGRESS:
			case -EALREADY:
				goto out_clear;
1266 1267 1268 1269 1270 1271 1272 1273
			case -ECONNREFUSED:
			case -ECONNRESET:
				/* retry with existing socket, after a delay */
				break;
			default:
				/* get rid of existing socket, and retry */
				xs_close(xprt);
				break;
1274 1275 1276
		}
	}
out:
1277
	xprt_wake_pending_tasks(xprt, status);
1278
out_clear:
1279
	xprt_clear_connecting(xprt);
1280 1281
}

1282 1283 1284 1285 1286
/**
 * 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.
1287 1288 1289 1290 1291 1292 1293
 *
 * 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).
1294 1295
 */
static void xs_connect(struct rpc_task *task)
1296 1297
{
	struct rpc_xprt *xprt = task->tk_xprt;
1298
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1299

1300 1301 1302
	if (xprt_test_and_set_connecting(xprt))
		return;

1303
	if (transport->sock != NULL) {
1304 1305
		dprintk("RPC:      xs_connect delayed xprt %p for %lu seconds\n",
				xprt, xprt->reestablish_timeout / HZ);
1306
		schedule_delayed_work(&transport->connect_worker,
1307 1308 1309 1310
					xprt->reestablish_timeout);
		xprt->reestablish_timeout <<= 1;
		if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
			xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
1311 1312
	} else {
		dprintk("RPC:      xs_connect scheduled xprt %p\n", xprt);
1313
		schedule_work(&transport->connect_worker);
1314 1315 1316 1317

		/* flush_scheduled_work can sleep... */
		if (!RPC_IS_ASYNC(task))
			flush_scheduled_work();
1318 1319 1320
	}
}

1321 1322 1323 1324 1325 1326 1327 1328
/**
 * 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)
{
1329 1330
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

1331
	seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
1332
			transport->port,
1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348
			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)
{
1349
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1350 1351 1352 1353 1354 1355
	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",
1356
			transport->port,
1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367
			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);
}

1368
static struct rpc_xprt_ops xs_udp_ops = {
1369
	.set_buffer_size	= xs_udp_set_buffer_size,
1370
	.reserve_xprt		= xprt_reserve_xprt_cong,
1371
	.release_xprt		= xprt_release_xprt_cong,
1372
	.rpcbind		= rpc_getport,
1373
	.set_port		= xs_set_port,
1374
	.connect		= xs_connect,
1375 1376
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1377
	.send_request		= xs_udp_send_request,
1378
	.set_retrans_timeout	= xprt_set_retrans_timeout_rtt,
1379
	.timer			= xs_udp_timer,
1380
	.release_request	= xprt_release_rqst_cong,
1381 1382
	.close			= xs_close,
	.destroy		= xs_destroy,
1383
	.print_stats		= xs_udp_print_stats,
1384 1385 1386
};

static struct rpc_xprt_ops xs_tcp_ops = {
1387
	.reserve_xprt		= xprt_reserve_xprt,
1388
	.release_xprt		= xs_tcp_release_xprt,
1389
	.rpcbind		= rpc_getport,
1390
	.set_port		= xs_set_port,
1391
	.connect		= xs_connect,
1392 1393
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1394
	.send_request		= xs_tcp_send_request,
1395
	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
1396 1397
	.close			= xs_close,
	.destroy		= xs_destroy,
1398
	.print_stats		= xs_tcp_print_stats,
1399 1400
};

1401 1402 1403
static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size)
{
	struct rpc_xprt *xprt;
1404
	struct sock_xprt *new;
1405 1406 1407 1408 1409 1410

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

1411 1412
	new = kzalloc(sizeof(*new), GFP_KERNEL);
	if (new == NULL) {
1413 1414 1415
		dprintk("RPC:      xs_setup_xprt: couldn't allocate rpc_xprt\n");
		return ERR_PTR(-ENOMEM);
	}
1416
	xprt = &new->xprt;
1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427

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

	memcpy(&xprt->addr, addr, addrlen);
	xprt->addrlen = addrlen;
1428
	new->port = xs_get_random_port();
1429 1430 1431 1432

	return xprt;
}

1433 1434
/**
 * xs_setup_udp - Set up transport to use a UDP socket
1435 1436
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1437 1438 1439
 * @to:   timeout parameters
 *
 */
1440
struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1441
{
1442
	struct rpc_xprt *xprt;
1443
	struct sock_xprt *transport;
1444

1445 1446 1447
	xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1448
	transport = container_of(xprt, struct sock_xprt, xprt);
1449

1450
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1451 1452 1453
		xprt_set_bound(xprt);

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

1458
	INIT_WORK(&transport->connect_worker, xs_udp_connect_worker, transport);
1459 1460 1461 1462
	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;
1463

1464
	xprt->ops = &xs_udp_ops;
1465 1466 1467 1468

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

1471 1472
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
1473
			xprt->address_strings[RPC_DISPLAY_ALL]);
1474

1475
	return xprt;
1476 1477
}

1478 1479
/**
 * xs_setup_tcp - Set up transport to use a TCP socket
1480 1481
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1482 1483 1484
 * @to: timeout parameters
 *
 */
1485
struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1486
{
1487
	struct rpc_xprt *xprt;
1488
	struct sock_xprt *transport;
1489

1490 1491 1492
	xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1493
	transport = container_of(xprt, struct sock_xprt, xprt);
1494

1495
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1496 1497 1498
		xprt_set_bound(xprt);

	xprt->prot = IPPROTO_TCP;
1499 1500
	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
1501

1502
	INIT_WORK(&transport->connect_worker, xs_tcp_connect_worker, transport);
1503 1504 1505 1506
	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;
1507

1508
	xprt->ops = &xs_tcp_ops;
1509 1510 1511 1512

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

1515 1516
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
1517
			xprt->address_strings[RPC_DISPLAY_ALL]);
1518

1519
	return xprt;
1520
}