xprtsock.c 39.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 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190
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]);
}

191 192
#define XS_SENDMSG_FLAGS	(MSG_DONTWAIT | MSG_NOSIGNAL)

T
Trond Myklebust 已提交
193
static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
194 195 196 197
{
	struct msghdr msg = {
		.msg_name	= addr,
		.msg_namelen	= addrlen,
T
Trond Myklebust 已提交
198 199 200 201 202
		.msg_flags	= XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
	};
	struct kvec iov = {
		.iov_base	= vec->iov_base + base,
		.iov_len	= vec->iov_len - base,
203 204
	};

T
Trond Myklebust 已提交
205
	if (iov.iov_len != 0)
206 207 208 209
		return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
	return kernel_sendmsg(sock, &msg, NULL, 0, 0);
}

T
Trond Myklebust 已提交
210
static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
211
{
T
Trond Myklebust 已提交
212 213 214 215 216 217 218 219 220 221 222
	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;
223

T
Trond Myklebust 已提交
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238
		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;
239 240
}

241 242 243 244 245 246 247 248
/**
 * 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
 *
249
 */
T
Trond Myklebust 已提交
250
static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
251
{
T
Trond Myklebust 已提交
252 253
	unsigned int remainder = xdr->len - base;
	int err, sent = 0;
254

255 256 257 258
	if (unlikely(!sock))
		return -ENOTCONN;

	clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
T
Trond Myklebust 已提交
259 260 261 262
	if (base != 0) {
		addr = NULL;
		addrlen = 0;
	}
263

T
Trond Myklebust 已提交
264 265 266 267 268
	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)
269
			goto out;
T
Trond Myklebust 已提交
270
		sent += err;
271 272
		base = 0;
	} else
T
Trond Myklebust 已提交
273
		base -= xdr->head[0].iov_len;
274

T
Trond Myklebust 已提交
275 276 277 278 279
	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)
280
			goto out;
T
Trond Myklebust 已提交
281
		sent += err;
282
		base = 0;
T
Trond Myklebust 已提交
283 284 285 286 287 288
	} 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);
289
out:
T
Trond Myklebust 已提交
290 291 292 293 294
	if (sent == 0)
		return err;
	if (err > 0)
		sent += err;
	return sent;
295 296
}

297
/**
298 299
 * xs_nospace - place task on wait queue if transmit was incomplete
 * @task: task to put to sleep
300
 *
301
 */
302
static void xs_nospace(struct rpc_task *task)
303
{
304 305
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
306
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
307

308 309 310 311
	dprintk("RPC: %4d xmit incomplete (%u left of %u)\n",
			task->tk_pid, req->rq_slen - req->rq_bytes_sent,
			req->rq_slen);

312
	if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
313 314 315 316 317 318
		/* 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;
319
		else if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342
			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;
343
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
344 345
	struct xdr_buf *xdr = &req->rq_snd_buf;
	int status;
346

347
	xs_pktdump("packet data:",
348 349 350
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);

351
	req->rq_xtime = jiffies;
352 353 354 355
	status = xs_sendpages(transport->sock,
			      (struct sockaddr *) &xprt->addr,
			      xprt->addrlen, xdr,
			      req->rq_bytes_sent);
356

357 358
	dprintk("RPC:      xs_udp_send_request(%u) = %d\n",
			xdr->len - req->rq_bytes_sent, status);
359

360 361
	if (likely(status >= (int) req->rq_slen))
		return 0;
362

363 364 365
	/* Still some bytes left; set up for a retry later. */
	if (status > 0)
		status = -EAGAIN;
366

367 368 369
	switch (status) {
	case -ENETUNREACH:
	case -EPIPE:
370 371
	case -ECONNREFUSED:
		/* When the server has died, an ICMP port unreachable message
372
		 * prompts ECONNREFUSED. */
373
		break;
374 375
	case -EAGAIN:
		xs_nospace(task);
376 377
		break;
	default:
378 379
		dprintk("RPC:      sendmsg returned unrecognized error %d\n",
			-status);
380
		break;
381
	}
382 383

	return status;
384 385
}

386 387 388 389 390 391 392
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);
}

393
/**
394
 * xs_tcp_send_request - write an RPC request to a TCP socket
395 396 397
 * @task: address of RPC task that manages the state of an RPC request
 *
 * Return values:
398 399 400 401 402
 *        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
403 404
 *
 * XXX: In the case of soft timeouts, should we eventually give up
405
 *	if sendmsg is not able to make progress?
406
 */
407
static int xs_tcp_send_request(struct rpc_task *task)
408 409 410
{
	struct rpc_rqst *req = task->tk_rqstp;
	struct rpc_xprt *xprt = req->rq_xprt;
411
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
412
	struct xdr_buf *xdr = &req->rq_snd_buf;
413 414
	int status, retry = 0;

415
	xs_encode_tcp_record_marker(&req->rq_snd_buf);
416

417 418 419
	xs_pktdump("packet data:",
				req->rq_svec->iov_base,
				req->rq_svec->iov_len);
420 421 422

	/* Continue transmitting the packet/record. We must be careful
	 * to cope with writespace callbacks arriving _after_ we have
423
	 * called sendmsg(). */
424 425
	while (1) {
		req->rq_xtime = jiffies;
426 427
		status = xs_sendpages(transport->sock,
					NULL, 0, xdr, req->rq_bytes_sent);
428

429 430
		dprintk("RPC:      xs_tcp_send_request(%u) = %d\n",
				xdr->len - req->rq_bytes_sent, status);
431

432
		if (unlikely(status < 0))
433 434
			break;

435 436 437
		/* If we've sent the entire packet, immediately
		 * reset the count of bytes sent. */
		req->rq_bytes_sent += status;
438
		task->tk_bytes_sent += status;
439 440 441 442
		if (likely(req->rq_bytes_sent >= req->rq_slen)) {
			req->rq_bytes_sent = 0;
			return 0;
		}
443 444

		status = -EAGAIN;
445
		if (retry++ > XS_SENDMSG_RETRY)
446 447 448
			break;
	}

449 450 451 452 453 454 455 456 457 458 459 460 461
	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);
462
		xprt_disconnect(xprt);
463
		break;
464
	}
465

466 467 468
	return status;
}

469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495
/**
 * 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);
}

496 497 498 499
/**
 * xs_close - close a socket
 * @xprt: transport
 *
500 501
 * This is used when all requests are complete; ie, no DRC state remains
 * on the server we want to save.
502
 */
503
static void xs_close(struct rpc_xprt *xprt)
504
{
505 506 507
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct socket *sock = transport->sock;
	struct sock *sk = transport->inet;
508 509

	if (!sk)
510
		goto clear_close_wait;
511

512 513
	dprintk("RPC:      xs_close xprt %p\n", xprt);

514
	write_lock_bh(&sk->sk_callback_lock);
515 516
	transport->inet = NULL;
	transport->sock = NULL;
517

518 519
	sk->sk_user_data = NULL;
	sk->sk_data_ready = xprt->old_data_ready;
520
	sk->sk_state_change = xprt->old_state_change;
521
	sk->sk_write_space = xprt->old_write_space;
522 523
	write_unlock_bh(&sk->sk_callback_lock);

524
	sk->sk_no_check = 0;
525 526

	sock_release(sock);
527 528 529 530
clear_close_wait:
	smp_mb__before_clear_bit();
	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
	smp_mb__after_clear_bit();
531 532
}

533 534 535 536 537 538
/**
 * xs_destroy - prepare to shutdown a transport
 * @xprt: doomed transport
 *
 */
static void xs_destroy(struct rpc_xprt *xprt)
539
{
540 541
	dprintk("RPC:      xs_destroy xprt %p\n", xprt);

542
	cancel_delayed_work(&xprt->connect_worker);
543 544 545
	flush_scheduled_work();

	xprt_disconnect(xprt);
546
	xs_close(xprt);
547
	xs_free_peer_addresses(xprt);
548
	kfree(xprt->slot);
549
	kfree(xprt);
550 551
}

552 553 554 555 556 557 558 559 560 561
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
 *
562
 */
563
static void xs_udp_data_ready(struct sock *sk, int len)
564
{
565 566
	struct rpc_task *task;
	struct rpc_xprt *xprt;
567
	struct rpc_rqst *rovr;
568
	struct sk_buff *skb;
569
	int err, repsize, copied;
570 571
	u32 _xid;
	__be32 *xp;
572 573

	read_lock(&sk->sk_callback_lock);
574 575
	dprintk("RPC:      xs_udp_data_ready...\n");
	if (!(xprt = xprt_from_sock(sk)))
576 577 578 579 580 581 582 583 584 585
		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) {
586
		dprintk("RPC:      impossible RPC reply size %d!\n", repsize);
587 588 589 590 591 592 593 594 595 596
		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 已提交
597
	spin_lock(&xprt->transport_lock);
598 599 600 601 602 603 604 605 606 607 608 609 610 611 612
	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);

613 614 615
	xprt_adjust_cwnd(task, copied);
	xprt_update_rtt(task);
	xprt_complete_rqst(task, copied);
616 617

 out_unlock:
C
Chuck Lever 已提交
618
	spin_unlock(&xprt->transport_lock);
619 620 621 622 623 624
 dropit:
	skb_free_datagram(sk, skb);
 out:
	read_unlock(&sk->sk_callback_lock);
}

625
static inline size_t xs_tcp_copy_data(skb_reader_t *desc, void *p, size_t len)
626 627 628 629 630 631 632 633 634 635 636 637 638 639 640
{
	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;
}

641
static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc)
642
{
643
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
644 645 646
	size_t len, used;
	char *p;

647 648
	p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
	len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
649
	used = xs_tcp_copy_data(desc, p, len);
650
	transport->tcp_offset += used;
651 652
	if (used != len)
		return;
653

654 655 656
	transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
	if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
		transport->tcp_flags |= XPRT_LAST_FRAG;
657
	else
658 659
		transport->tcp_flags &= ~XPRT_LAST_FRAG;
	transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
660

661 662
	transport->tcp_flags &= ~XPRT_COPY_RECM;
	transport->tcp_offset = 0;
663

664
	/* Sanity check of the record length */
665
	if (unlikely(transport->tcp_reclen < 4)) {
666
		dprintk("RPC:      invalid TCP record fragment length\n");
667
		xprt_disconnect(xprt);
668
		return;
669 670
	}
	dprintk("RPC:      reading TCP record fragment of length %d\n",
671
			transport->tcp_reclen);
672 673
}

674
static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
675
{
676 677 678 679 680 681 682
	if (transport->tcp_offset == transport->tcp_reclen) {
		transport->tcp_flags |= XPRT_COPY_RECM;
		transport->tcp_offset = 0;
		if (transport->tcp_flags & XPRT_LAST_FRAG) {
			transport->tcp_flags &= ~XPRT_COPY_DATA;
			transport->tcp_flags |= XPRT_COPY_XID;
			transport->tcp_copied = 0;
683 684 685 686
		}
	}
}

687
static inline void xs_tcp_read_xid(struct sock_xprt *transport, skb_reader_t *desc)
688 689 690 691
{
	size_t len, used;
	char *p;

692
	len = sizeof(transport->tcp_xid) - transport->tcp_offset;
693
	dprintk("RPC:      reading XID (%Zu bytes)\n", len);
694
	p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
695
	used = xs_tcp_copy_data(desc, p, len);
696
	transport->tcp_offset += used;
697 698
	if (used != len)
		return;
699 700 701
	transport->tcp_flags &= ~XPRT_COPY_XID;
	transport->tcp_flags |= XPRT_COPY_DATA;
	transport->tcp_copied = 4;
702
	dprintk("RPC:      reading reply for XID %08x\n",
703 704
			ntohl(transport->tcp_xid));
	xs_tcp_check_fraghdr(transport);
705 706
}

707
static inline void xs_tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc)
708
{
709
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
710 711 712 713 714 715
	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 已提交
716
	spin_lock(&xprt->transport_lock);
717
	req = xprt_lookup_rqst(xprt, transport->tcp_xid);
718
	if (!req) {
719
		transport->tcp_flags &= ~XPRT_COPY_DATA;
720
		dprintk("RPC:      XID %08x request not found!\n",
721
				ntohl(transport->tcp_xid));
C
Chuck Lever 已提交
722
		spin_unlock(&xprt->transport_lock);
723 724 725 726 727
		return;
	}

	rcvbuf = &req->rq_private_buf;
	len = desc->count;
728
	if (len > transport->tcp_reclen - transport->tcp_offset) {
729 730
		skb_reader_t my_desc;

731
		len = transport->tcp_reclen - transport->tcp_offset;
732 733
		memcpy(&my_desc, desc, sizeof(my_desc));
		my_desc.count = len;
734
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
735
					  &my_desc, xs_tcp_copy_data);
736 737 738
		desc->count -= r;
		desc->offset += r;
	} else
739
		r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
740
					  desc, xs_tcp_copy_data);
741 742

	if (r > 0) {
743 744
		transport->tcp_copied += r;
		transport->tcp_offset += r;
745 746 747 748 749 750 751 752 753 754 755
	}
	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
		 * is turn off XPRT_COPY_DATA, so the request
		 * will not receive any additional updates,
		 * and time out.
		 * Any remaining data from this record will
		 * be discarded.
		 */
756
		transport->tcp_flags &= ~XPRT_COPY_DATA;
757
		dprintk("RPC:      XID %08x truncated request\n",
758
				ntohl(transport->tcp_xid));
759
		dprintk("RPC:      xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
760 761
				xprt, transport->tcp_copied, transport->tcp_offset,
					transport->tcp_reclen);
762 763 764 765
		goto out;
	}

	dprintk("RPC:      XID %08x read %Zd bytes\n",
766
			ntohl(transport->tcp_xid), r);
767
	dprintk("RPC:      xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
768 769 770 771 772 773 774 775
			xprt, transport->tcp_copied, transport->tcp_offset,
				transport->tcp_reclen);

	if (transport->tcp_copied == req->rq_private_buf.buflen)
		transport->tcp_flags &= ~XPRT_COPY_DATA;
	else if (transport->tcp_offset == transport->tcp_reclen) {
		if (transport->tcp_flags & XPRT_LAST_FRAG)
			transport->tcp_flags &= ~XPRT_COPY_DATA;
776 777 778
	}

out:
779 780
	if (!(transport->tcp_flags & XPRT_COPY_DATA))
		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
C
Chuck Lever 已提交
781
	spin_unlock(&xprt->transport_lock);
782
	xs_tcp_check_fraghdr(transport);
783 784
}

785
static inline void xs_tcp_read_discard(struct sock_xprt *transport, skb_reader_t *desc)
786 787 788
{
	size_t len;

789
	len = transport->tcp_reclen - transport->tcp_offset;
790 791 792 793
	if (len > desc->count)
		len = desc->count;
	desc->count -= len;
	desc->offset += len;
794
	transport->tcp_offset += len;
795
	dprintk("RPC:      discarded %Zu bytes\n", len);
796
	xs_tcp_check_fraghdr(transport);
797 798
}

799
static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
800 801
{
	struct rpc_xprt *xprt = rd_desc->arg.data;
802
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
803 804 805 806
	skb_reader_t desc = {
		.skb	= skb,
		.offset	= offset,
		.count	= len,
807
	};
808

809
	dprintk("RPC:      xs_tcp_data_recv started\n");
810 811 812
	do {
		/* Read in a new fragment marker if necessary */
		/* Can we ever really expect to get completely empty fragments? */
813
		if (transport->tcp_flags & XPRT_COPY_RECM) {
814
			xs_tcp_read_fraghdr(xprt, &desc);
815 816 817
			continue;
		}
		/* Read in the xid if necessary */
818 819
		if (transport->tcp_flags & XPRT_COPY_XID) {
			xs_tcp_read_xid(transport, &desc);
820 821 822
			continue;
		}
		/* Read in the request data */
823
		if (transport->tcp_flags & XPRT_COPY_DATA) {
824
			xs_tcp_read_request(xprt, &desc);
825 826 827
			continue;
		}
		/* Skip over any trailing bytes on short reads */
828
		xs_tcp_read_discard(transport, &desc);
829
	} while (desc.count);
830
	dprintk("RPC:      xs_tcp_data_recv done\n");
831 832 833
	return len - desc.count;
}

834 835 836 837 838 839 840
/**
 * 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)
841 842 843 844 845
{
	struct rpc_xprt *xprt;
	read_descriptor_t rd_desc;

	read_lock(&sk->sk_callback_lock);
846 847
	dprintk("RPC:      xs_tcp_data_ready...\n");
	if (!(xprt = xprt_from_sock(sk)))
848 849 850 851
		goto out;
	if (xprt->shutdown)
		goto out;

852
	/* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
853 854
	rd_desc.arg.data = xprt;
	rd_desc.count = 65536;
855
	tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
856 857 858 859
out:
	read_unlock(&sk->sk_callback_lock);
}

860 861 862 863 864 865
/**
 * 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)
866
{
867
	struct rpc_xprt *xprt;
868 869 870 871

	read_lock(&sk->sk_callback_lock);
	if (!(xprt = xprt_from_sock(sk)))
		goto out;
872
	dprintk("RPC:      xs_tcp_state_change client %p...\n", xprt);
873 874 875 876 877 878 879
	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 已提交
880
		spin_lock_bh(&xprt->transport_lock);
881
		if (!xprt_test_and_set_connected(xprt)) {
882 883 884
			struct sock_xprt *transport = container_of(xprt,
					struct sock_xprt, xprt);

885
			/* Reset TCP record info */
886 887 888 889 890
			transport->tcp_offset = 0;
			transport->tcp_reclen = 0;
			transport->tcp_copied = 0;
			transport->tcp_flags = XPRT_COPY_RECM | XPRT_COPY_XID;

891
			xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
892
			xprt_wake_pending_tasks(xprt, 0);
893
		}
C
Chuck Lever 已提交
894
		spin_unlock_bh(&xprt->transport_lock);
895 896 897 898
		break;
	case TCP_SYN_SENT:
	case TCP_SYN_RECV:
		break;
899 900 901 902 903
	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);
904 905 906 907 908 909 910
	default:
		xprt_disconnect(xprt);
	}
 out:
	read_unlock(&sk->sk_callback_lock);
}

911
/**
912 913
 * xs_udp_write_space - callback invoked when socket buffer space
 *                             becomes available
914 915
 * @sk: socket whose state has changed
 *
916 917
 * Called when more output buffer space is available for this socket.
 * We try not to wake our writers until they can make "significant"
918
 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
919 920
 * with a bunch of small requests.
 */
921
static void xs_udp_write_space(struct sock *sk)
922 923 924
{
	read_lock(&sk->sk_callback_lock);

925 926 927 928 929 930
	/* 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)))
931
			goto out;
932 933 934
		if (unlikely(!(xprt = xprt_from_sock(sk))))
			goto out;
		if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
935
			goto out;
936 937

		xprt_write_space(xprt);
938 939
	}

940 941 942
 out:
	read_unlock(&sk->sk_callback_lock);
}
943

944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973
/**
 * 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:
974 975 976
	read_unlock(&sk->sk_callback_lock);
}

977
static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
978
{
979 980
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct sock *sk = transport->inet;
981 982 983 984 985 986 987 988 989 990 991 992

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

993
/**
994
 * xs_udp_set_buffer_size - set send and receive limits
995
 * @xprt: generic transport
996 997
 * @sndsize: requested size of send buffer, in bytes
 * @rcvsize: requested size of receive buffer, in bytes
998
 *
999
 * Set socket send and receive buffer size limits.
1000
 */
1001
static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1002
{
1003 1004 1005 1006 1007 1008 1009 1010
	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);
1011 1012
}

1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023
/**
 * 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);
}

1024 1025 1026 1027 1028 1029 1030
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;
}

1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043
/**
 * 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";
}

1044 1045 1046 1047 1048 1049 1050 1051
/**
 * 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)
{
1052 1053
	struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr;

1054
	dprintk("RPC:      setting port for xprt %p to %u\n", xprt, port);
1055 1056

	sap->sin_port = htons(port);
1057 1058
}

1059
static int xs_bindresvport(struct rpc_xprt *xprt, struct socket *sock)
1060 1061 1062 1063
{
	struct sockaddr_in myaddr = {
		.sin_family = AF_INET,
	};
1064 1065
	int err;
	unsigned short port = xprt->port;
1066 1067 1068

	do {
		myaddr.sin_port = htons(port);
1069
		err = kernel_bind(sock, (struct sockaddr *) &myaddr,
1070 1071 1072
						sizeof(myaddr));
		if (err == 0) {
			xprt->port = port;
1073 1074
			dprintk("RPC:      xs_bindresvport bound to port %u\n",
					port);
1075 1076
			return 0;
		}
1077 1078 1079 1080
		if (port <= xprt_min_resvport)
			port = xprt_max_resvport;
		else
			port--;
1081 1082
	} while (err == -EADDRINUSE && port != xprt->port);

1083
	dprintk("RPC:      can't bind to reserved port (%d).\n", -err);
1084 1085 1086
	return err;
}

1087 1088 1089 1090 1091 1092 1093
/**
 * 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)
1094
{
1095
	struct rpc_xprt *xprt = (struct rpc_xprt *) args;
1096 1097
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct socket *sock = transport->sock;
1098
	int err, status = -EIO;
1099

1100
	if (xprt->shutdown || !xprt_bound(xprt))
1101
		goto out;
1102

1103 1104
	/* Start by resetting any existing state */
	xs_close(xprt);
1105

1106 1107 1108 1109
	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;
	}
1110

1111 1112 1113 1114
	if (xprt->resvport && xs_bindresvport(xprt, sock) < 0) {
		sock_release(sock);
		goto out;
	}
1115

1116 1117 1118
	dprintk("RPC:      worker connecting xprt %p to address: %s\n",
			xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1119
	if (!transport->inet) {
1120
		struct sock *sk = sock->sk;
1121

1122
		write_lock_bh(&sk->sk_callback_lock);
1123

1124 1125 1126 1127
		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;
1128
		sk->sk_data_ready = xs_udp_data_ready;
1129
		sk->sk_write_space = xs_udp_write_space;
1130
		sk->sk_no_check = UDP_CSUM_NORCV;
1131
		sk->sk_allocation = GFP_ATOMIC;
1132

1133 1134
		xprt_set_connected(xprt);

1135
		/* Reset to new socket */
1136 1137
		transport->sock = sock;
		transport->inet = sk;
1138

1139 1140
		write_unlock_bh(&sk->sk_callback_lock);
	}
1141
	xs_udp_do_set_buffer_size(xprt);
1142 1143 1144 1145
	status = 0;
out:
	xprt_wake_pending_tasks(xprt, status);
	xprt_clear_connecting(xprt);
1146 1147
}

1148 1149 1150 1151 1152 1153 1154
/*
 * 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;
1155
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1156 1157 1158 1159 1160 1161 1162 1163 1164 1165
	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;
1166
	result = kernel_connect(transport->sock, &any, sizeof(any), 0);
1167 1168 1169 1170 1171
	if (result)
		dprintk("RPC:      AF_UNSPEC connect return code %d\n",
				result);
}

1172
/**
1173
 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint
1174 1175 1176
 * @args: RPC transport to connect
 *
 * Invoked by a work queue tasklet.
1177
 */
1178
static void xs_tcp_connect_worker(void *args)
1179 1180
{
	struct rpc_xprt *xprt = (struct rpc_xprt *)args;
1181 1182
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
	struct socket *sock = transport->sock;
1183
	int err, status = -EIO;
1184

1185
	if (xprt->shutdown || !xprt_bound(xprt))
1186 1187
		goto out;

1188
	if (!sock) {
1189 1190 1191 1192 1193
		/* 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;
		}
1194

1195 1196 1197 1198 1199 1200 1201
		if (xprt->resvport && xs_bindresvport(xprt, sock) < 0) {
			sock_release(sock);
			goto out;
		}
	} else
		/* "close" the socket, preserving the local port */
		xs_tcp_reuse_connection(xprt);
1202

1203 1204 1205
	dprintk("RPC:      worker connecting xprt %p to address: %s\n",
			xprt, xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1206
	if (!transport->inet) {
1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217
		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;
1218
		sk->sk_allocation = GFP_ATOMIC;
1219 1220 1221 1222 1223 1224

		/* 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;
1225 1226 1227 1228

		xprt_clear_connected(xprt);

		/* Reset to new socket */
1229 1230
		transport->sock = sock;
		transport->inet = sk;
1231 1232 1233 1234 1235

		write_unlock_bh(&sk->sk_callback_lock);
	}

	/* Tell the socket layer to start connecting... */
1236 1237
	xprt->stat.connect_count++;
	xprt->stat.connect_start = jiffies;
1238
	status = kernel_connect(sock, (struct sockaddr *) &xprt->addr,
1239
			xprt->addrlen, O_NONBLOCK);
1240 1241 1242 1243 1244 1245 1246
	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;
1247 1248 1249 1250 1251 1252 1253 1254
			case -ECONNREFUSED:
			case -ECONNRESET:
				/* retry with existing socket, after a delay */
				break;
			default:
				/* get rid of existing socket, and retry */
				xs_close(xprt);
				break;
1255 1256 1257
		}
	}
out:
1258
	xprt_wake_pending_tasks(xprt, status);
1259
out_clear:
1260
	xprt_clear_connecting(xprt);
1261 1262
}

1263 1264 1265 1266 1267
/**
 * 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.
1268 1269 1270 1271 1272 1273 1274
 *
 * 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).
1275 1276
 */
static void xs_connect(struct rpc_task *task)
1277 1278
{
	struct rpc_xprt *xprt = task->tk_xprt;
1279
	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1280

1281 1282 1283
	if (xprt_test_and_set_connecting(xprt))
		return;

1284
	if (transport->sock != NULL) {
1285 1286
		dprintk("RPC:      xs_connect delayed xprt %p for %lu seconds\n",
				xprt, xprt->reestablish_timeout / HZ);
1287
		schedule_delayed_work(&xprt->connect_worker,
1288 1289 1290 1291
					xprt->reestablish_timeout);
		xprt->reestablish_timeout <<= 1;
		if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
			xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
1292 1293 1294 1295 1296 1297 1298
	} else {
		dprintk("RPC:      xs_connect scheduled xprt %p\n", xprt);
		schedule_work(&xprt->connect_worker);

		/* flush_scheduled_work can sleep... */
		if (!RPC_IS_ASYNC(task))
			flush_scheduled_work();
1299 1300 1301
	}
}

1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345
/**
 * 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)
{
	seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
			xprt->port,
			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)
{
	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",
			xprt->port,
			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);
}

1346
static struct rpc_xprt_ops xs_udp_ops = {
1347
	.set_buffer_size	= xs_udp_set_buffer_size,
1348
	.print_addr		= xs_print_peer_address,
1349
	.reserve_xprt		= xprt_reserve_xprt_cong,
1350
	.release_xprt		= xprt_release_xprt_cong,
1351
	.rpcbind		= rpc_getport,
1352
	.set_port		= xs_set_port,
1353
	.connect		= xs_connect,
1354 1355
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1356
	.send_request		= xs_udp_send_request,
1357
	.set_retrans_timeout	= xprt_set_retrans_timeout_rtt,
1358
	.timer			= xs_udp_timer,
1359
	.release_request	= xprt_release_rqst_cong,
1360 1361
	.close			= xs_close,
	.destroy		= xs_destroy,
1362
	.print_stats		= xs_udp_print_stats,
1363 1364 1365
};

static struct rpc_xprt_ops xs_tcp_ops = {
1366
	.print_addr		= xs_print_peer_address,
1367
	.reserve_xprt		= xprt_reserve_xprt,
1368
	.release_xprt		= xs_tcp_release_xprt,
1369
	.rpcbind		= rpc_getport,
1370
	.set_port		= xs_set_port,
1371
	.connect		= xs_connect,
1372 1373
	.buf_alloc		= rpc_malloc,
	.buf_free		= rpc_free,
1374
	.send_request		= xs_tcp_send_request,
1375
	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
1376 1377
	.close			= xs_close,
	.destroy		= xs_destroy,
1378
	.print_stats		= xs_tcp_print_stats,
1379 1380
};

1381 1382 1383
static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size)
{
	struct rpc_xprt *xprt;
1384
	struct sock_xprt *new;
1385 1386 1387 1388 1389 1390

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

1391 1392
	new = kzalloc(sizeof(*new), GFP_KERNEL);
	if (new == NULL) {
1393 1394 1395
		dprintk("RPC:      xs_setup_xprt: couldn't allocate rpc_xprt\n");
		return ERR_PTR(-ENOMEM);
	}
1396
	xprt = &new->xprt;
1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412

	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;
	xprt->port = xs_get_random_port();

	return xprt;
}

1413 1414
/**
 * xs_setup_udp - Set up transport to use a UDP socket
1415 1416
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1417 1418 1419
 * @to:   timeout parameters
 *
 */
1420
struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1421
{
1422
	struct rpc_xprt *xprt;
1423

1424 1425 1426
	xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1427

1428
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1429 1430 1431
		xprt_set_bound(xprt);

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

1436
	INIT_WORK(&xprt->connect_worker, xs_udp_connect_worker, xprt);
1437 1438 1439 1440
	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;
1441

1442
	xprt->ops = &xs_udp_ops;
1443 1444 1445 1446

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

1449 1450 1451 1452
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
			xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1453
	return xprt;
1454 1455
}

1456 1457
/**
 * xs_setup_tcp - Set up transport to use a TCP socket
1458 1459
 * @addr: address of remote server
 * @addrlen: length of address in bytes
1460 1461 1462
 * @to: timeout parameters
 *
 */
1463
struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
1464
{
1465
	struct rpc_xprt *xprt;
1466

1467 1468 1469
	xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries);
	if (IS_ERR(xprt))
		return xprt;
1470

1471
	if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
1472 1473 1474
		xprt_set_bound(xprt);

	xprt->prot = IPPROTO_TCP;
1475 1476
	xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
	xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
1477

1478
	INIT_WORK(&xprt->connect_worker, xs_tcp_connect_worker, xprt);
1479 1480 1481 1482
	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;
1483

1484
	xprt->ops = &xs_tcp_ops;
1485 1486 1487 1488

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

1491 1492 1493 1494
	xs_format_peer_addresses(xprt);
	dprintk("RPC:      set up transport to address %s\n",
			xs_print_peer_address(xprt, RPC_DISPLAY_ALL));

1495
	return xprt;
1496
}