xprtsock.c 39.8 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 160 161 162 163
/*
 * 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)

164 165 166 167 168 169 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
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]);
}

205 206
#define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)

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

T
Trond Myklebust 已提交
219
	if (iov.iov_len != 0)
220 221 222 223
		return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
	return kernel_sendmsg(sock, &msg, NULL, 0, 0);
}

T
Trond Myklebust 已提交
224
static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
225
{
T
Trond Myklebust 已提交
226 227 228 229 230 231 232 233 234 235 236
	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;
237

T
Trond Myklebust 已提交
238 239 240 241 242 243 244 245 246 247 248 249 250 251 252
		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;
253 254
}

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

269 270 271 272
	if (unlikely(!sock))
		return -ENOTCONN;

	clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
T
Trond Myklebust 已提交
273 274 275 276
	if (base != 0) {
		addr = NULL;
		addrlen = 0;
	}
277

T
Trond Myklebust 已提交
278 279 280 281 282
	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)
283
			goto out;
T
Trond Myklebust 已提交
284
		sent += err;
285 286
		base = 0;
	} else
T
Trond Myklebust 已提交
287
		base -= xdr->head[0].iov_len;
288

T
Trond Myklebust 已提交
289 290 291 292 293
	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)
294
			goto out;
T
Trond Myklebust 已提交
295
		sent += err;
296
		base = 0;
T
Trond Myklebust 已提交
297 298 299 300 301 302
	} 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);
303
out:
T
Trond Myklebust 已提交
304 305 306 307 308
	if (sent == 0)
		return err;
	if (err > 0)
		sent += err;
	return sent;
309 310
}

311
/**
312 313
 * xs_nospace - place task on wait queue if transmit was incomplete
 * @task: task to put to sleep
314
 *
315
 */
316
static void xs_nospace(struct rpc_task *task)
317
{
318 319
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
320
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
321

322 323 324 325
	dprintk("RPC: %4d xmit incomplete (%u left of %u)\n",
			task->tk_pid, req->rq_slen - req->rq_bytes_sent,
			req->rq_slen);

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

361
	xs_pktdump("packet data:",
362 363 364
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);

365
	req->rq_xtime = jiffies;
366 367 368 369
	status = xs_sendpages(transport->sock,
			      (struct sockaddr *) &xprt->addr,
			      xprt->addrlen, xdr,
			      req->rq_bytes_sent);
370

371 372
	dprintk("RPC:      xs_udp_send_request(%u) = %d\n",
			xdr->len - req->rq_bytes_sent, status);
373

374 375
	if (likely(status >= (int) req->rq_slen))
		return 0;
376

377 378 379
	/* Still some bytes left; set up for a retry later. */
	if (status > 0)
		status = -EAGAIN;
380

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

	return status;
398 399
}

400 401 402 403 404 405 406
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);
}

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

429
	xs_encode_tcp_record_marker(&req->rq_snd_buf);
430

431 432 433
	xs_pktdump("packet data:",
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);
434 435 436

	/* Continue transmitting the packet/record. We must be careful
	 * to cope with writespace callbacks arriving _after_ we have
437
	 * called sendmsg(). */
438 439
	while (1) {
		req->rq_xtime = jiffies;
440 441
		status = xs_sendpages(transport->sock,
					NULL, 0, xdr, req->rq_bytes_sent);
442

443 444
		dprintk("RPC:      xs_tcp_send_request(%u) = %d\n",
				xdr->len - req->rq_bytes_sent, status);
445

446
		if (unlikely(status < 0))
447 448
			break;

449 450 451
		/* If we've sent the entire packet, immediately
		 * reset the count of bytes sent. */
		req->rq_bytes_sent += status;
452
		task->tk_bytes_sent += status;
453 454 455 456
		if (likely(req->rq_bytes_sent >= req->rq_slen)) {
			req->rq_bytes_sent = 0;
			return 0;
		}
457 458

		status = -EAGAIN;
459
		if (retry++ > XS_SENDMSG_RETRY)
460 461 462
			break;
	}

463 464 465 466 467 468 469 470 471 472 473 474 475
	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);
476
		xprt_disconnect(xprt);
477
		break;
478
	}
479

480 481 482
	return status;
}

483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509
/**
 * 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);
}

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

	if (!sk)
524
		goto clear_close_wait;
525

526 527
	dprintk("RPC:      xs_close xprt %p\n", xprt);

528
	write_lock_bh(&sk->sk_callback_lock);
529 530
	transport->inet = NULL;
	transport->sock = NULL;
531

532 533
	sk->sk_user_data = NULL;
	sk->sk_data_ready = xprt->old_data_ready;
534
	sk->sk_state_change = xprt->old_state_change;
535
	sk->sk_write_space = xprt->old_write_space;
536 537
	write_unlock_bh(&sk->sk_callback_lock);

538
	sk->sk_no_check = 0;
539 540

	sock_release(sock);
541 542 543 544
clear_close_wait:
	smp_mb__before_clear_bit();
	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
	smp_mb__after_clear_bit();
545 546
}

547 548 549 550 551 552
/**
 * xs_destroy - prepare to shutdown a transport
 * @xprt: doomed transport
 *
 */
static void xs_destroy(struct rpc_xprt *xprt)
553
{
554 555
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

556 557
	dprintk("RPC:      xs_destroy xprt %p\n", xprt);

558
	cancel_delayed_work(&transport->connect_worker);
559 560 561
	flush_scheduled_work();

	xprt_disconnect(xprt);
562
	xs_close(xprt);
563
	xs_free_peer_addresses(xprt);
564
	kfree(xprt->slot);
565
	kfree(xprt);
566 567
}

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

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

629 630 631
	xprt_adjust_cwnd(task, copied);
	xprt_update_rtt(task);
	xprt_complete_rqst(task, copied);
632 633

 out_unlock:
C
Chuck Lever 已提交
634
	spin_unlock(&xprt->transport_lock);
635 636 637 638 639 640
 dropit:
	skb_free_datagram(sk, skb);
 out:
	read_unlock(&sk->sk_callback_lock);
}

641
static inline size_t xs_tcp_copy_data(skb_reader_t *desc, void *p, size_t len)
642 643 644 645 646 647 648 649 650 651 652 653 654 655 656
{
	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;
}

657
static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc)
658
{
659
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
660 661 662
	size_t len, used;
	char *p;

663 664
	p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
	len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
665
	used = xs_tcp_copy_data(desc, p, len);
666
	transport->tcp_offset += used;
667 668
	if (used != len)
		return;
669

670 671
	transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
	if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
672
		transport->tcp_flags |= TCP_RCV_LAST_FRAG;
673
	else
674
		transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
675
	transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
676

677
	transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
678
	transport->tcp_offset = 0;
679

680
	/* Sanity check of the record length */
681
	if (unlikely(transport->tcp_reclen < 4)) {
682
		dprintk("RPC:      invalid TCP record fragment length\n");
683
		xprt_disconnect(xprt);
684
		return;
685 686
	}
	dprintk("RPC:      reading TCP record fragment of length %d\n",
687
			transport->tcp_reclen);
688 689
}

690
static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
691
{
692
	if (transport->tcp_offset == transport->tcp_reclen) {
693
		transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
694
		transport->tcp_offset = 0;
695 696 697
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
			transport->tcp_flags |= TCP_RCV_COPY_XID;
698
			transport->tcp_copied = 0;
699 700 701 702
		}
	}
}

703
static inline void xs_tcp_read_xid(struct sock_xprt *transport, skb_reader_t *desc)
704 705 706 707
{
	size_t len, used;
	char *p;

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

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

	rcvbuf = &req->rq_private_buf;
	len = desc->count;
744
	if (len > transport->tcp_reclen - transport->tcp_offset) {
745 746
		skb_reader_t my_desc;

747
		len = transport->tcp_reclen - transport->tcp_offset;
748 749
		memcpy(&my_desc, desc, sizeof(my_desc));
		my_desc.count = len;
750
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
751
					  &my_desc, xs_tcp_copy_data);
752 753 754
		desc->count -= r;
		desc->offset += r;
	} else
755
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
756
					  desc, xs_tcp_copy_data);
757 758

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

	dprintk("RPC:      XID %08x read %Zd bytes\n",
782
			ntohl(transport->tcp_xid), r);
783
	dprintk("RPC:      xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
784 785 786 787
			xprt, transport->tcp_copied, transport->tcp_offset,
				transport->tcp_reclen);

	if (transport->tcp_copied == req->rq_private_buf.buflen)
788
		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
789
	else if (transport->tcp_offset == transport->tcp_reclen) {
790 791
		if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
792 793 794
	}

out:
795
	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
796
		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
C
Chuck Lever 已提交
797
	spin_unlock(&xprt->transport_lock);
798
	xs_tcp_check_fraghdr(transport);
799 800
}

801
static inline void xs_tcp_read_discard(struct sock_xprt *transport, skb_reader_t *desc)
802 803 804
{
	size_t len;

805
	len = transport->tcp_reclen - transport->tcp_offset;
806 807 808 809
	if (len > desc->count)
		len = desc->count;
	desc->count -= len;
	desc->offset += len;
810
	transport->tcp_offset += len;
811
	dprintk("RPC:      discarded %Zu bytes\n", len);
812
	xs_tcp_check_fraghdr(transport);
813 814
}

815
static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
816 817
{
	struct rpc_xprt *xprt = rd_desc->arg.data;
818
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
819 820 821 822
	skb_reader_t desc = {
		.skb	= skb,
		.offset	= offset,
		.count	= len,
823
	};
824

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

850 851 852 853 854 855 856
/**
 * 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)
857 858 859 860 861
{
	struct rpc_xprt *xprt;
	read_descriptor_t rd_desc;

	read_lock(&sk->sk_callback_lock);
862 863
	dprintk("RPC:      xs_tcp_data_ready...\n");
	if (!(xprt = xprt_from_sock(sk)))
864 865 866 867
		goto out;
	if (xprt->shutdown)
		goto out;

868
	/* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
869 870
	rd_desc.arg.data = xprt;
	rd_desc.count = 65536;
871
	tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
872 873 874 875
out:
	read_unlock(&sk->sk_callback_lock);
}

876 877 878 879 880 881
/**
 * 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)
882
{
883
	struct rpc_xprt *xprt;
884 885 886 887

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

901
			/* Reset TCP record info */
902 903 904
			transport->tcp_offset = 0;
			transport->tcp_reclen = 0;
			transport->tcp_copied = 0;
905 906
			transport->tcp_flags =
				TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
907

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

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

942 943 944 945 946 947
	/* 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)))
948
			goto out;
949 950 951
		if (unlikely(!(xprt = xprt_from_sock(sk))))
			goto out;
		if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
952
			goto out;
953 954

		xprt_write_space(xprt);
955 956
	}

957 958 959
 out:
	read_unlock(&sk->sk_callback_lock);
}
960

961 962 963 964 965 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
/**
 * 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:
991 992 993
	read_unlock(&sk->sk_callback_lock);
}

994
static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
995
{
996 997
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct sock *sk = transport->inet;
998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009

	if (xprt->rcvsize) {
		sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
		sk->sk_rcvbuf = xprt->rcvsize * xprt->max_reqs *  2;
	}
	if (xprt->sndsize) {
		sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
		sk->sk_sndbuf = xprt->sndsize * xprt->max_reqs * 2;
		sk->sk_write_space(sk);
	}
}

1010
/**
1011
 * xs_udp_set_buffer_size - set send and receive limits
1012
 * @xprt: generic transport
1013 1014
 * @sndsize: requested size of send buffer, in bytes
 * @rcvsize: requested size of receive buffer, in bytes
1015
 *
1016
 * Set socket send and receive buffer size limits.
1017
 */
1018
static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1019
{
1020 1021 1022 1023 1024 1025 1026 1027
	xprt->sndsize = 0;
	if (sndsize)
		xprt->sndsize = sndsize + 1024;
	xprt->rcvsize = 0;
	if (rcvsize)
		xprt->rcvsize = rcvsize + 1024;

	xs_udp_do_set_buffer_size(xprt);
1028 1029
}

1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040
/**
 * 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);
}

1041 1042 1043 1044 1045 1046 1047
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;
}

1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060
/**
 * 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";
}

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

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

	sap->sin_port = htons(port);
1074 1075
}

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

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

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

1104 1105 1106 1107 1108 1109 1110
/**
 * 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)
1111
{
1112 1113
	struct sock_xprt *transport = (struct sock_xprt *)args;
	struct rpc_xprt *xprt = &transport->xprt;
1114
	struct socket *sock = transport->sock;
1115
	int err, status = -EIO;
1116

1117
	if (xprt->shutdown || !xprt_bound(xprt))
1118
		goto out;
1119

1120 1121
	/* Start by resetting any existing state */
	xs_close(xprt);
1122

1123 1124 1125 1126
	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;
	}
1127

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

1133 1134 1135
	dprintk("RPC:      worker connecting xprt %p to address: %s\n",
			xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

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

1139
		write_lock_bh(&sk->sk_callback_lock);
1140

1141 1142 1143 1144
		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;
1145
		sk->sk_data_ready = xs_udp_data_ready;
1146
		sk->sk_write_space = xs_udp_write_space;
1147
		sk->sk_no_check = UDP_CSUM_NORCV;
1148
		sk->sk_allocation = GFP_ATOMIC;
1149

1150 1151
		xprt_set_connected(xprt);

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

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

1165 1166 1167 1168 1169 1170 1171
/*
 * 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;
1172
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1173 1174 1175 1176 1177 1178 1179 1180 1181 1182
	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;
1183
	result = kernel_connect(transport->sock, &any, sizeof(any), 0);
1184 1185 1186 1187 1188
	if (result)
		dprintk("RPC:      AF_UNSPEC connect return code %d\n",
				result);
}

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

1202
	if (xprt->shutdown || !xprt_bound(xprt))
1203 1204
		goto out;

1205
	if (!sock) {
1206 1207 1208 1209 1210
		/* 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;
		}
1211

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

1220 1221 1222
	dprintk("RPC:      worker connecting xprt %p to address: %s\n",
			xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1223
	if (!transport->inet) {
1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234
		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;
1235
		sk->sk_allocation = GFP_ATOMIC;
1236 1237 1238 1239 1240 1241

		/* 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;
1242 1243 1244 1245

		xprt_clear_connected(xprt);

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

		write_unlock_bh(&sk->sk_callback_lock);
	}

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

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

1298 1299 1300
	if (xprt_test_and_set_connecting(xprt))
		return;

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

		/* flush_scheduled_work can sleep... */
		if (!RPC_IS_ASYNC(task))
			flush_scheduled_work();
1316 1317 1318
	}
}

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

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

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

static struct rpc_xprt_ops xs_tcp_ops = {
1386
	.print_addr		= xs_print_peer_address,
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 1473 1474
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
			xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

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 1517 1518
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
			xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1519
	return xprt;
1520
}