xprtsock.c 40.0 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 167 168 169
/*
 * 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)

170 171 172 173 174 175 176 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
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]);
}

211 212
#define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)

T
Trond Myklebust 已提交
213
static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
214 215 216 217
{
	struct msghdr msg = {
		.msg_name	= addr,
		.msg_namelen	= addrlen,
T
Trond Myklebust 已提交
218 219 220 221 222
		.msg_flags	= XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
	};
	struct kvec iov = {
		.iov_base	= vec->iov_base + base,
		.iov_len	= vec->iov_len - base,
223 224
	};

T
Trond Myklebust 已提交
225
	if (iov.iov_len != 0)
226 227 228 229
		return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
	return kernel_sendmsg(sock, &msg, NULL, 0, 0);
}

T
Trond Myklebust 已提交
230
static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
231
{
T
Trond Myklebust 已提交
232 233 234 235 236 237 238 239 240 241 242
	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;
243

T
Trond Myklebust 已提交
244 245 246 247 248 249 250 251 252 253 254 255 256 257 258
		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;
259 260
}

261 262 263 264 265 266 267 268
/**
 * 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
 *
269
 */
T
Trond Myklebust 已提交
270
static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
271
{
T
Trond Myklebust 已提交
272 273
	unsigned int remainder = xdr->len - base;
	int err, sent = 0;
274

275 276 277 278
	if (unlikely(!sock))
		return -ENOTCONN;

	clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
T
Trond Myklebust 已提交
279 280 281 282
	if (base != 0) {
		addr = NULL;
		addrlen = 0;
	}
283

T
Trond Myklebust 已提交
284 285 286 287 288
	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)
289
			goto out;
T
Trond Myklebust 已提交
290
		sent += err;
291 292
		base = 0;
	} else
T
Trond Myklebust 已提交
293
		base -= xdr->head[0].iov_len;
294

T
Trond Myklebust 已提交
295 296 297 298 299
	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)
300
			goto out;
T
Trond Myklebust 已提交
301
		sent += err;
302
		base = 0;
T
Trond Myklebust 已提交
303 304 305 306 307 308
	} 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);
309
out:
T
Trond Myklebust 已提交
310 311 312 313 314
	if (sent == 0)
		return err;
	if (err > 0)
		sent += err;
	return sent;
315 316
}

317
/**
318 319
 * xs_nospace - place task on wait queue if transmit was incomplete
 * @task: task to put to sleep
320
 *
321
 */
322
static void xs_nospace(struct rpc_task *task)
323
{
324 325
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
326
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
327

328 329 330 331
	dprintk("RPC: %4d xmit incomplete (%u left of %u)\n",
			task->tk_pid, req->rq_slen - req->rq_bytes_sent,
			req->rq_slen);

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

367
	xs_pktdump("packet data:",
368 369 370
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);

371
	req->rq_xtime = jiffies;
372 373 374 375
	status = xs_sendpages(transport->sock,
			      (struct sockaddr *) &xprt->addr,
			      xprt->addrlen, xdr,
			      req->rq_bytes_sent);
376

377 378
	dprintk("RPC:      xs_udp_send_request(%u) = %d\n",
			xdr->len - req->rq_bytes_sent, status);
379

380 381
	if (likely(status >= (int) req->rq_slen))
		return 0;
382

383 384 385
	/* Still some bytes left; set up for a retry later. */
	if (status > 0)
		status = -EAGAIN;
386

387 388 389
	switch (status) {
	case -ENETUNREACH:
	case -EPIPE:
390 391
	case -ECONNREFUSED:
		/* When the server has died, an ICMP port unreachable message
392
		 * prompts ECONNREFUSED. */
393
		break;
394 395
	case -EAGAIN:
		xs_nospace(task);
396 397
		break;
	default:
398 399
		dprintk("RPC:      sendmsg returned unrecognized error %d\n",
			-status);
400
		break;
401
	}
402 403

	return status;
404 405
}

406 407 408 409 410 411 412
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);
}

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

435
	xs_encode_tcp_record_marker(&req->rq_snd_buf);
436

437 438 439
	xs_pktdump("packet data:",
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);
440 441 442

	/* Continue transmitting the packet/record. We must be careful
	 * to cope with writespace callbacks arriving _after_ we have
443
	 * called sendmsg(). */
444 445
	while (1) {
		req->rq_xtime = jiffies;
446 447
		status = xs_sendpages(transport->sock,
					NULL, 0, xdr, req->rq_bytes_sent);
448

449 450
		dprintk("RPC:      xs_tcp_send_request(%u) = %d\n",
				xdr->len - req->rq_bytes_sent, status);
451

452
		if (unlikely(status < 0))
453 454
			break;

455 456 457
		/* If we've sent the entire packet, immediately
		 * reset the count of bytes sent. */
		req->rq_bytes_sent += status;
458
		task->tk_bytes_sent += status;
459 460 461 462
		if (likely(req->rq_bytes_sent >= req->rq_slen)) {
			req->rq_bytes_sent = 0;
			return 0;
		}
463 464

		status = -EAGAIN;
465
		if (retry++ > XS_SENDMSG_RETRY)
466 467 468
			break;
	}

469 470 471 472 473 474 475 476 477 478 479 480 481
	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);
482
		xprt_disconnect(xprt);
483
		break;
484
	}
485

486 487 488
	return status;
}

489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515
/**
 * 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);
}

516 517 518 519
/**
 * xs_close - close a socket
 * @xprt: transport
 *
520 521
 * This is used when all requests are complete; ie, no DRC state remains
 * on the server we want to save.
522
 */
523
static void xs_close(struct rpc_xprt *xprt)
524
{
525 526 527
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct socket *sock = transport->sock;
	struct sock *sk = transport->inet;
528 529

	if (!sk)
530
		goto clear_close_wait;
531

532 533
	dprintk("RPC:      xs_close xprt %p\n", xprt);

534
	write_lock_bh(&sk->sk_callback_lock);
535 536
	transport->inet = NULL;
	transport->sock = NULL;
537

538 539
	sk->sk_user_data = NULL;
	sk->sk_data_ready = xprt->old_data_ready;
540
	sk->sk_state_change = xprt->old_state_change;
541
	sk->sk_write_space = xprt->old_write_space;
542 543
	write_unlock_bh(&sk->sk_callback_lock);

544
	sk->sk_no_check = 0;
545 546

	sock_release(sock);
547 548 549 550
clear_close_wait:
	smp_mb__before_clear_bit();
	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
	smp_mb__after_clear_bit();
551 552
}

553 554 555 556 557 558
/**
 * xs_destroy - prepare to shutdown a transport
 * @xprt: doomed transport
 *
 */
static void xs_destroy(struct rpc_xprt *xprt)
559
{
560 561
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

562 563
	dprintk("RPC:      xs_destroy xprt %p\n", xprt);

564
	cancel_delayed_work(&transport->connect_worker);
565 566 567
	flush_scheduled_work();

	xprt_disconnect(xprt);
568
	xs_close(xprt);
569
	xs_free_peer_addresses(xprt);
570
	kfree(xprt->slot);
571
	kfree(xprt);
572 573
}

574 575 576 577 578 579 580 581 582 583
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
 *
584
 */
585
static void xs_udp_data_ready(struct sock *sk, int len)
586
{
587 588
	struct rpc_task *task;
	struct rpc_xprt *xprt;
589
	struct rpc_rqst *rovr;
590
	struct sk_buff *skb;
591
	int err, repsize, copied;
592 593
	u32 _xid;
	__be32 *xp;
594 595

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

635 636 637
	xprt_adjust_cwnd(task, copied);
	xprt_update_rtt(task);
	xprt_complete_rqst(task, copied);
638 639

 out_unlock:
C
Chuck Lever 已提交
640
	spin_unlock(&xprt->transport_lock);
641 642 643 644 645 646
 dropit:
	skb_free_datagram(sk, skb);
 out:
	read_unlock(&sk->sk_callback_lock);
}

647
static inline size_t xs_tcp_copy_data(skb_reader_t *desc, void *p, size_t len)
648 649 650 651 652 653 654 655 656 657 658 659 660 661 662
{
	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;
}

663
static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc)
664
{
665
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
666 667 668
	size_t len, used;
	char *p;

669 670
	p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
	len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
671
	used = xs_tcp_copy_data(desc, p, len);
672
	transport->tcp_offset += used;
673 674
	if (used != len)
		return;
675

676 677
	transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
	if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
678
		transport->tcp_flags |= TCP_RCV_LAST_FRAG;
679
	else
680
		transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
681
	transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
682

683
	transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
684
	transport->tcp_offset = 0;
685

686
	/* Sanity check of the record length */
687
	if (unlikely(transport->tcp_reclen < 4)) {
688
		dprintk("RPC:      invalid TCP record fragment length\n");
689
		xprt_disconnect(xprt);
690
		return;
691 692
	}
	dprintk("RPC:      reading TCP record fragment of length %d\n",
693
			transport->tcp_reclen);
694 695
}

696
static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
697
{
698
	if (transport->tcp_offset == transport->tcp_reclen) {
699
		transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
700
		transport->tcp_offset = 0;
701 702 703
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
			transport->tcp_flags |= TCP_RCV_COPY_XID;
704
			transport->tcp_copied = 0;
705 706 707 708
		}
	}
}

709
static inline void xs_tcp_read_xid(struct sock_xprt *transport, skb_reader_t *desc)
710 711 712 713
{
	size_t len, used;
	char *p;

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

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

	rcvbuf = &req->rq_private_buf;
	len = desc->count;
750
	if (len > transport->tcp_reclen - transport->tcp_offset) {
751 752
		skb_reader_t my_desc;

753
		len = transport->tcp_reclen - transport->tcp_offset;
754 755
		memcpy(&my_desc, desc, sizeof(my_desc));
		my_desc.count = len;
756
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
757
					  &my_desc, xs_tcp_copy_data);
758 759 760
		desc->count -= r;
		desc->offset += r;
	} else
761
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
762
					  desc, xs_tcp_copy_data);
763 764

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

	dprintk("RPC:      XID %08x read %Zd bytes\n",
788
			ntohl(transport->tcp_xid), r);
789
	dprintk("RPC:      xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
790 791 792 793
			xprt, transport->tcp_copied, transport->tcp_offset,
				transport->tcp_reclen);

	if (transport->tcp_copied == req->rq_private_buf.buflen)
794
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
795
	else if (transport->tcp_offset == transport->tcp_reclen) {
796 797
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
798 799 800
	}

out:
801
	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
802
		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
C
Chuck Lever 已提交
803
	spin_unlock(&xprt->transport_lock);
804
	xs_tcp_check_fraghdr(transport);
805 806
}

807
static inline void xs_tcp_read_discard(struct sock_xprt *transport, skb_reader_t *desc)
808 809 810
{
	size_t len;

811
	len = transport->tcp_reclen - transport->tcp_offset;
812 813 814 815
	if (len > desc->count)
		len = desc->count;
	desc->count -= len;
	desc->offset += len;
816
	transport->tcp_offset += len;
817
	dprintk("RPC:      discarded %Zu bytes\n", len);
818
	xs_tcp_check_fraghdr(transport);
819 820
}

821
static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
822 823
{
	struct rpc_xprt *xprt = rd_desc->arg.data;
824
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
825 826 827 828
	skb_reader_t desc = {
		.skb	= skb,
		.offset	= offset,
		.count	= len,
829
	};
830

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

856 857 858 859 860 861 862
/**
 * 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)
863 864 865 866 867
{
	struct rpc_xprt *xprt;
	read_descriptor_t rd_desc;

	read_lock(&sk->sk_callback_lock);
868 869
	dprintk("RPC:      xs_tcp_data_ready...\n");
	if (!(xprt = xprt_from_sock(sk)))
870 871 872 873
		goto out;
	if (xprt->shutdown)
		goto out;

874
	/* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
875 876
	rd_desc.arg.data = xprt;
	rd_desc.count = 65536;
877
	tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
878 879 880 881
out:
	read_unlock(&sk->sk_callback_lock);
}

882 883 884 885 886 887
/**
 * 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)
888
{
889
	struct rpc_xprt *xprt;
890 891 892 893

	read_lock(&sk->sk_callback_lock);
	if (!(xprt = xprt_from_sock(sk)))
		goto out;
894
	dprintk("RPC:      xs_tcp_state_change client %p...\n", xprt);
895 896 897 898 899 900 901
	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 已提交
902
		spin_lock_bh(&xprt->transport_lock);
903
		if (!xprt_test_and_set_connected(xprt)) {
904 905 906
			struct sock_xprt *transport = container_of(xprt,
					struct sock_xprt, xprt);

907
			/* Reset TCP record info */
908 909 910
			transport->tcp_offset = 0;
			transport->tcp_reclen = 0;
			transport->tcp_copied = 0;
911 912
			transport->tcp_flags =
				TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
913

914
			xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
915
			xprt_wake_pending_tasks(xprt, 0);
916
		}
C
Chuck Lever 已提交
917
		spin_unlock_bh(&xprt->transport_lock);
918 919 920 921
		break;
	case TCP_SYN_SENT:
	case TCP_SYN_RECV:
		break;
922 923 924 925 926
	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);
927 928 929 930 931 932 933
	default:
		xprt_disconnect(xprt);
	}
 out:
	read_unlock(&sk->sk_callback_lock);
}

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

948 949 950 951 952 953
	/* 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)))
954
			goto out;
955 956 957
		if (unlikely(!(xprt = xprt_from_sock(sk))))
			goto out;
		if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
958
			goto out;
959 960

		xprt_write_space(xprt);
961 962
	}

963 964 965
 out:
	read_unlock(&sk->sk_callback_lock);
}
966

967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996
/**
 * 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:
997 998 999
	read_unlock(&sk->sk_callback_lock);
}

1000
static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1001
{
1002 1003
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct sock *sk = transport->inet;
1004

1005
	if (transport->rcvsize) {
1006
		sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1007
		sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1008
	}
1009
	if (transport->sndsize) {
1010
		sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1011
		sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1012 1013 1014 1015
		sk->sk_write_space(sk);
	}
}

1016
/**
1017
 * xs_udp_set_buffer_size - set send and receive limits
1018
 * @xprt: generic transport
1019 1020
 * @sndsize: requested size of send buffer, in bytes
 * @rcvsize: requested size of receive buffer, in bytes
1021
 *
1022
 * Set socket send and receive buffer size limits.
1023
 */
1024
static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1025
{
1026 1027 1028
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

	transport->sndsize = 0;
1029
	if (sndsize)
1030 1031
		transport->sndsize = sndsize + 1024;
	transport->rcvsize = 0;
1032
	if (rcvsize)
1033
		transport->rcvsize = rcvsize + 1024;
1034 1035

	xs_udp_do_set_buffer_size(xprt);
1036 1037
}

1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048
/**
 * 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);
}

1049 1050 1051 1052 1053 1054 1055
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;
}

1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068
/**
 * xs_print_peer_address - format an IPv4 address for printing
 * @xprt: generic transport
 * @format: flags field indicating which parts of the address to render
 */
static char *xs_print_peer_address(struct rpc_xprt *xprt, enum rpc_display_format_t format)
{
	if (xprt->address_strings[format] != NULL)
		return xprt->address_strings[format];
	else
		return "unprintable";
}

1069 1070 1071 1072 1073 1074 1075 1076
/**
 * 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)
{
1077 1078
	struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr;

1079
	dprintk("RPC:      setting port for xprt %p to %u\n", xprt, port);
1080 1081

	sap->sin_port = htons(port);
1082 1083
}

1084
static int xs_bindresvport(struct sock_xprt *transport, struct socket *sock)
1085 1086 1087 1088
{
	struct sockaddr_in myaddr = {
		.sin_family = AF_INET,
	};
1089
	int err;
1090
	unsigned short port = transport->port;
1091 1092 1093

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

1108
	dprintk("RPC:      can't bind to reserved port (%d).\n", -err);
1109 1110 1111
	return err;
}

1112 1113 1114 1115 1116 1117 1118
/**
 * 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)
1119
{
1120 1121
	struct sock_xprt *transport = (struct sock_xprt *)args;
	struct rpc_xprt *xprt = &transport->xprt;
1122
	struct socket *sock = transport->sock;
1123
	int err, status = -EIO;
1124

1125
	if (xprt->shutdown || !xprt_bound(xprt))
1126
		goto out;
1127

1128 1129
	/* Start by resetting any existing state */
	xs_close(xprt);
1130

1131 1132 1133 1134
	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;
	}
1135

1136
	if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
1137 1138 1139
		sock_release(sock);
		goto out;
	}
1140

1141 1142 1143
	dprintk("RPC:      worker connecting xprt %p to address: %s\n",
			xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1144
	if (!transport->inet) {
1145
		struct sock *sk = sock->sk;
1146

1147
		write_lock_bh(&sk->sk_callback_lock);
1148

1149 1150 1151 1152
		sk->sk_user_data = xprt;
		xprt->old_data_ready = sk->sk_data_ready;
		xprt->old_state_change = sk->sk_state_change;
		xprt->old_write_space = sk->sk_write_space;
1153
		sk->sk_data_ready = xs_udp_data_ready;
1154
		sk->sk_write_space = xs_udp_write_space;
1155
		sk->sk_no_check = UDP_CSUM_NORCV;
1156
		sk->sk_allocation = GFP_ATOMIC;
1157

1158 1159
		xprt_set_connected(xprt);

1160
		/* Reset to new socket */
1161 1162
		transport->sock = sock;
		transport->inet = sk;
1163

1164 1165
		write_unlock_bh(&sk->sk_callback_lock);
	}
1166
	xs_udp_do_set_buffer_size(xprt);
1167 1168 1169 1170
	status = 0;
out:
	xprt_wake_pending_tasks(xprt, status);
	xprt_clear_connecting(xprt);
1171 1172
}

1173 1174 1175 1176 1177 1178 1179
/*
 * 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;
1180
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1181 1182 1183 1184 1185 1186 1187 1188 1189 1190
	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;
1191
	result = kernel_connect(transport->sock, &any, sizeof(any), 0);
1192 1193 1194 1195 1196
	if (result)
		dprintk("RPC:      AF_UNSPEC connect return code %d\n",
				result);
}

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

1210
	if (xprt->shutdown || !xprt_bound(xprt))
1211 1212
		goto out;

1213
	if (!sock) {
1214 1215 1216 1217 1218
		/* 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;
		}
1219

1220
		if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
1221 1222 1223 1224 1225 1226
			sock_release(sock);
			goto out;
		}
	} else
		/* "close" the socket, preserving the local port */
		xs_tcp_reuse_connection(xprt);
1227

1228 1229 1230
	dprintk("RPC:      worker connecting xprt %p to address: %s\n",
			xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1231
	if (!transport->inet) {
1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242
		struct sock *sk = sock->sk;

		write_lock_bh(&sk->sk_callback_lock);

		sk->sk_user_data = xprt;
		xprt->old_data_ready = sk->sk_data_ready;
		xprt->old_state_change = sk->sk_state_change;
		xprt->old_write_space = sk->sk_write_space;
		sk->sk_data_ready = xs_tcp_data_ready;
		sk->sk_state_change = xs_tcp_state_change;
		sk->sk_write_space = xs_tcp_write_space;
1243
		sk->sk_allocation = GFP_ATOMIC;
1244 1245 1246 1247 1248 1249

		/* 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;
1250 1251 1252 1253

		xprt_clear_connected(xprt);

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

		write_unlock_bh(&sk->sk_callback_lock);
	}

	/* Tell the socket layer to start connecting... */
1261 1262
	xprt->stat.connect_count++;
	xprt->stat.connect_start = jiffies;
1263
	status = kernel_connect(sock, (struct sockaddr *) &xprt->addr,
1264
			xprt->addrlen, O_NONBLOCK);
1265 1266 1267 1268 1269 1270 1271
	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;
1272 1273 1274 1275 1276 1277 1278 1279
			case -ECONNREFUSED:
			case -ECONNRESET:
				/* retry with existing socket, after a delay */
				break;
			default:
				/* get rid of existing socket, and retry */
				xs_close(xprt);
				break;
1280 1281 1282
		}
	}
out:
1283
	xprt_wake_pending_tasks(xprt, status);
1284
out_clear:
1285
	xprt_clear_connecting(xprt);
1286 1287
}

1288 1289 1290 1291 1292
/**
 * 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.
1293 1294 1295 1296 1297 1298 1299
 *
 * 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).
1300 1301
 */
static void xs_connect(struct rpc_task *task)
1302 1303
{
	struct rpc_xprt *xprt = task->tk_xprt;
1304
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1305

1306 1307 1308
	if (xprt_test_and_set_connecting(xprt))
		return;

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

		/* flush_scheduled_work can sleep... */
		if (!RPC_IS_ASYNC(task))
			flush_scheduled_work();
1324 1325 1326
	}
}

1327 1328 1329 1330 1331 1332 1333 1334
/**
 * 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)
{
1335 1336
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

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

1374
static struct rpc_xprt_ops xs_udp_ops = {
1375
	.set_buffer_size	= xs_udp_set_buffer_size,
1376
	.print_addr		= xs_print_peer_address,
1377
	.reserve_xprt		= xprt_reserve_xprt_cong,
1378
	.release_xprt		= xprt_release_xprt_cong,
1379
	.rpcbind		= rpc_getport,
1380
	.set_port		= xs_set_port,
1381
	.connect		= xs_connect,
1382 1383
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1384
	.send_request		= xs_udp_send_request,
1385
	.set_retrans_timeout	= xprt_set_retrans_timeout_rtt,
1386
	.timer			= xs_udp_timer,
1387
	.release_request	= xprt_release_rqst_cong,
1388 1389
	.close			= xs_close,
	.destroy		= xs_destroy,
1390
	.print_stats		= xs_udp_print_stats,
1391 1392 1393
};

static struct rpc_xprt_ops xs_tcp_ops = {
1394
	.print_addr		= xs_print_peer_address,
1395
	.reserve_xprt		= xprt_reserve_xprt,
1396
	.release_xprt		= xs_tcp_release_xprt,
1397
	.rpcbind		= rpc_getport,
1398
	.set_port		= xs_set_port,
1399
	.connect		= xs_connect,
1400 1401
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1402
	.send_request		= xs_tcp_send_request,
1403
	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
1404 1405
	.close			= xs_close,
	.destroy		= xs_destroy,
1406
	.print_stats		= xs_tcp_print_stats,
1407 1408
};

1409 1410 1411
static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size)
{
	struct rpc_xprt *xprt;
1412
	struct sock_xprt *new;
1413 1414 1415 1416 1417 1418

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

1419 1420
	new = kzalloc(sizeof(*new), GFP_KERNEL);
	if (new == NULL) {
1421 1422 1423
		dprintk("RPC:      xs_setup_xprt: couldn't allocate rpc_xprt\n");
		return ERR_PTR(-ENOMEM);
	}
1424
	xprt = &new->xprt;
1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435

	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;
1436
	new->port = xs_get_random_port();
1437 1438 1439 1440

	return xprt;
}

1441 1442
/**
 * xs_setup_udp - Set up transport to use a UDP socket
1443 1444
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1445 1446 1447
 * @to:   timeout parameters
 *
 */
1448
struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1449
{
1450
	struct rpc_xprt *xprt;
1451
	struct sock_xprt *transport;
1452

1453 1454 1455
	xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1456
	transport = container_of(xprt, struct sock_xprt, xprt);
1457

1458
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1459 1460 1461
		xprt_set_bound(xprt);

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

1466
	INIT_WORK(&transport->connect_worker, xs_udp_connect_worker, transport);
1467 1468 1469 1470
	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;
1471

1472
	xprt->ops = &xs_udp_ops;
1473 1474 1475 1476

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

1479 1480 1481 1482
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
			xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1483
	return xprt;
1484 1485
}

1486 1487
/**
 * xs_setup_tcp - Set up transport to use a TCP socket
1488 1489
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1490 1491 1492
 * @to: timeout parameters
 *
 */
1493
struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1494
{
1495
	struct rpc_xprt *xprt;
1496
	struct sock_xprt *transport;
1497

1498 1499 1500
	xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1501
	transport = container_of(xprt, struct sock_xprt, xprt);
1502

1503
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1504 1505 1506
		xprt_set_bound(xprt);

	xprt->prot = IPPROTO_TCP;
1507 1508
	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
1509

1510
	INIT_WORK(&transport->connect_worker, xs_tcp_connect_worker, transport);
1511 1512 1513 1514
	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;
1515

1516
	xprt->ops = &xs_tcp_ops;
1517 1518 1519 1520

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

1523 1524 1525 1526
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
			xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1527
	return xprt;
1528
}