SUNRPC: skb_read_bits is the same as xs_tcp_copy_data
[GitHub/mt8127/android_kernel_alcatel_ttab.git] / net / sunrpc / xprtsock.c
CommitLineData
a246b010
CL
1/*
2 * linux/net/sunrpc/xprtsock.c
3 *
4 * Client-side transport implementation for sockets.
5 *
6 * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com>
7 * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com>
8 * TCP NFS related read + write fixes
9 * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
10 *
11 * Rewrite of larges part of the code in order to stabilize TCP stuff.
12 * Fix behaviour when socket buffer is full.
13 * (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
55aa4f58
CL
14 *
15 * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com>
a246b010
CL
16 */
17
18#include <linux/types.h>
19#include <linux/slab.h>
20#include <linux/capability.h>
21#include <linux/sched.h>
22#include <linux/pagemap.h>
23#include <linux/errno.h>
24#include <linux/socket.h>
25#include <linux/in.h>
26#include <linux/net.h>
27#include <linux/mm.h>
28#include <linux/udp.h>
29#include <linux/tcp.h>
30#include <linux/sunrpc/clnt.h>
02107148 31#include <linux/sunrpc/sched.h>
a246b010
CL
32#include <linux/file.h>
33
34#include <net/sock.h>
35#include <net/checksum.h>
36#include <net/udp.h>
37#include <net/tcp.h>
38
c556b754
CL
39/*
40 * xprtsock tunables
41 */
42unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
43unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;
44
45unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
46unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
47
262965f5
CL
48/*
49 * How many times to try sending a request on a socket before waiting
50 * for the socket buffer to clear.
51 */
52#define XS_SENDMSG_RETRY (10U)
53
03bf4b70
CL
54/*
55 * Time out for an RPC UDP socket connect. UDP socket connects are
56 * synchronous, but we set a timeout anyway in case of resource
57 * exhaustion on the local host.
58 */
59#define XS_UDP_CONN_TO (5U * HZ)
60
61/*
62 * Wait duration for an RPC TCP connection to be established. Solaris
63 * NFS over TCP uses 60 seconds, for example, which is in line with how
64 * long a server takes to reboot.
65 */
66#define XS_TCP_CONN_TO (60U * HZ)
67
68/*
69 * Wait duration for a reply from the RPC portmapper.
70 */
71#define XS_BIND_TO (60U * HZ)
72
73/*
74 * Delay if a UDP socket connect error occurs. This is most likely some
75 * kind of resource problem on the local host.
76 */
77#define XS_UDP_REEST_TO (2U * HZ)
78
79/*
80 * The reestablish timeout allows clients to delay for a bit before attempting
81 * to reconnect to a server that just dropped our connection.
82 *
83 * We implement an exponential backoff when trying to reestablish a TCP
84 * transport connection with the server. Some servers like to drop a TCP
85 * connection when they are overworked, so we start with a short timeout and
86 * increase over time if the server is down or not responding.
87 */
88#define XS_TCP_INIT_REEST_TO (3U * HZ)
89#define XS_TCP_MAX_REEST_TO (5U * 60 * HZ)
90
91/*
92 * TCP idle timeout; client drops the transport socket if it is idle
93 * for this long. Note that we also timeout UDP sockets to prevent
94 * holding port numbers when there is no RPC traffic.
95 */
96#define XS_IDLE_DISC_TO (5U * 60 * HZ)
97
a246b010
CL
98#ifdef RPC_DEBUG
99# undef RPC_DEBUG_DATA
9903cd1c 100# define RPCDBG_FACILITY RPCDBG_TRANS
a246b010
CL
101#endif
102
a246b010 103#ifdef RPC_DEBUG_DATA
9903cd1c 104static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
a246b010 105{
9903cd1c
CL
106 u8 *buf = (u8 *) packet;
107 int j;
a246b010
CL
108
109 dprintk("RPC: %s\n", msg);
110 for (j = 0; j < count && j < 128; j += 4) {
111 if (!(j & 31)) {
112 if (j)
113 dprintk("\n");
114 dprintk("0x%04x ", j);
115 }
116 dprintk("%02x%02x%02x%02x ",
117 buf[j], buf[j+1], buf[j+2], buf[j+3]);
118 }
119 dprintk("\n");
120}
121#else
9903cd1c 122static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
a246b010
CL
123{
124 /* NOP */
125}
126#endif
127
ffc2e518
CL
128struct sock_xprt {
129 struct rpc_xprt xprt;
ee0ac0c2
CL
130
131 /*
132 * Network layer
133 */
134 struct socket * sock;
135 struct sock * inet;
51971139
CL
136
137 /*
138 * State of TCP reply receive
139 */
140 __be32 tcp_fraghdr,
141 tcp_xid;
142
143 u32 tcp_offset,
144 tcp_reclen;
145
146 unsigned long tcp_copied,
147 tcp_flags;
c8475461
CL
148
149 /*
150 * Connection of transports
151 */
152 struct work_struct connect_worker;
153 unsigned short port;
7c6e066e
CL
154
155 /*
156 * UDP socket buffer size parameters
157 */
158 size_t rcvsize,
159 sndsize;
314dfd79
CL
160
161 /*
162 * Saved socket callback addresses
163 */
164 void (*old_data_ready)(struct sock *, int);
165 void (*old_state_change)(struct sock *);
166 void (*old_write_space)(struct sock *);
ffc2e518
CL
167};
168
e136d092
CL
169/*
170 * TCP receive state flags
171 */
172#define TCP_RCV_LAST_FRAG (1UL << 0)
173#define TCP_RCV_COPY_FRAGHDR (1UL << 1)
174#define TCP_RCV_COPY_XID (1UL << 2)
175#define TCP_RCV_COPY_DATA (1UL << 3)
176
edb267a6
CL
177static void xs_format_peer_addresses(struct rpc_xprt *xprt)
178{
179 struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr;
180 char *buf;
181
182 buf = kzalloc(20, GFP_KERNEL);
183 if (buf) {
184 snprintf(buf, 20, "%u.%u.%u.%u",
185 NIPQUAD(addr->sin_addr.s_addr));
186 }
187 xprt->address_strings[RPC_DISPLAY_ADDR] = buf;
188
189 buf = kzalloc(8, GFP_KERNEL);
190 if (buf) {
191 snprintf(buf, 8, "%u",
192 ntohs(addr->sin_port));
193 }
194 xprt->address_strings[RPC_DISPLAY_PORT] = buf;
195
196 if (xprt->prot == IPPROTO_UDP)
197 xprt->address_strings[RPC_DISPLAY_PROTO] = "udp";
198 else
199 xprt->address_strings[RPC_DISPLAY_PROTO] = "tcp";
200
201 buf = kzalloc(48, GFP_KERNEL);
202 if (buf) {
203 snprintf(buf, 48, "addr=%u.%u.%u.%u port=%u proto=%s",
204 NIPQUAD(addr->sin_addr.s_addr),
205 ntohs(addr->sin_port),
206 xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
207 }
208 xprt->address_strings[RPC_DISPLAY_ALL] = buf;
209}
210
211static void xs_free_peer_addresses(struct rpc_xprt *xprt)
212{
213 kfree(xprt->address_strings[RPC_DISPLAY_ADDR]);
214 kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
215 kfree(xprt->address_strings[RPC_DISPLAY_ALL]);
216}
217
b4b5cc85
CL
218#define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL)
219
24c5684b 220static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
b4b5cc85 221{
b4b5cc85
CL
222 struct msghdr msg = {
223 .msg_name = addr,
224 .msg_namelen = addrlen,
24c5684b
TM
225 .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
226 };
227 struct kvec iov = {
228 .iov_base = vec->iov_base + base,
229 .iov_len = vec->iov_len - base,
b4b5cc85
CL
230 };
231
24c5684b 232 if (iov.iov_len != 0)
b4b5cc85
CL
233 return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
234 return kernel_sendmsg(sock, &msg, NULL, 0, 0);
235}
236
24c5684b 237static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
b4b5cc85 238{
24c5684b
TM
239 struct page **ppage;
240 unsigned int remainder;
241 int err, sent = 0;
242
243 remainder = xdr->page_len - base;
244 base += xdr->page_base;
245 ppage = xdr->pages + (base >> PAGE_SHIFT);
246 base &= ~PAGE_MASK;
247 for(;;) {
248 unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
249 int flags = XS_SENDMSG_FLAGS;
b4b5cc85 250
24c5684b
TM
251 remainder -= len;
252 if (remainder != 0 || more)
253 flags |= MSG_MORE;
254 err = sock->ops->sendpage(sock, *ppage, base, len, flags);
255 if (remainder == 0 || err != len)
256 break;
257 sent += err;
258 ppage++;
259 base = 0;
260 }
261 if (sent == 0)
262 return err;
263 if (err > 0)
264 sent += err;
265 return sent;
b4b5cc85
CL
266}
267
9903cd1c
CL
268/**
269 * xs_sendpages - write pages directly to a socket
270 * @sock: socket to send on
271 * @addr: UDP only -- address of destination
272 * @addrlen: UDP only -- length of destination address
273 * @xdr: buffer containing this request
274 * @base: starting position in the buffer
275 *
a246b010 276 */
24c5684b 277static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
a246b010 278{
24c5684b
TM
279 unsigned int remainder = xdr->len - base;
280 int err, sent = 0;
a246b010 281
262965f5
CL
282 if (unlikely(!sock))
283 return -ENOTCONN;
284
285 clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
24c5684b
TM
286 if (base != 0) {
287 addr = NULL;
288 addrlen = 0;
289 }
262965f5 290
24c5684b
TM
291 if (base < xdr->head[0].iov_len || addr != NULL) {
292 unsigned int len = xdr->head[0].iov_len - base;
293 remainder -= len;
294 err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
295 if (remainder == 0 || err != len)
a246b010 296 goto out;
24c5684b 297 sent += err;
a246b010
CL
298 base = 0;
299 } else
24c5684b 300 base -= xdr->head[0].iov_len;
a246b010 301
24c5684b
TM
302 if (base < xdr->page_len) {
303 unsigned int len = xdr->page_len - base;
304 remainder -= len;
305 err = xs_send_pagedata(sock, xdr, base, remainder != 0);
306 if (remainder == 0 || err != len)
a246b010 307 goto out;
24c5684b 308 sent += err;
a246b010 309 base = 0;
24c5684b
TM
310 } else
311 base -= xdr->page_len;
312
313 if (base >= xdr->tail[0].iov_len)
314 return sent;
315 err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
a246b010 316out:
24c5684b
TM
317 if (sent == 0)
318 return err;
319 if (err > 0)
320 sent += err;
321 return sent;
a246b010
CL
322}
323
9903cd1c 324/**
262965f5
CL
325 * xs_nospace - place task on wait queue if transmit was incomplete
326 * @task: task to put to sleep
9903cd1c 327 *
a246b010 328 */
262965f5 329static void xs_nospace(struct rpc_task *task)
a246b010 330{
262965f5
CL
331 struct rpc_rqst *req = task->tk_rqstp;
332 struct rpc_xprt *xprt = req->rq_xprt;
ee0ac0c2 333 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
a246b010 334
262965f5
CL
335 dprintk("RPC: %4d xmit incomplete (%u left of %u)\n",
336 task->tk_pid, req->rq_slen - req->rq_bytes_sent,
337 req->rq_slen);
338
ee0ac0c2 339 if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
262965f5
CL
340 /* Protect against races with write_space */
341 spin_lock_bh(&xprt->transport_lock);
342
343 /* Don't race with disconnect */
344 if (!xprt_connected(xprt))
345 task->tk_status = -ENOTCONN;
ee0ac0c2 346 else if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
262965f5
CL
347 xprt_wait_for_buffer_space(task);
348
349 spin_unlock_bh(&xprt->transport_lock);
350 } else
351 /* Keep holding the socket if it is blocked */
352 rpc_delay(task, HZ>>4);
353}
354
355/**
356 * xs_udp_send_request - write an RPC request to a UDP socket
357 * @task: address of RPC task that manages the state of an RPC request
358 *
359 * Return values:
360 * 0: The request has been sent
361 * EAGAIN: The socket was blocked, please call again later to
362 * complete the request
363 * ENOTCONN: Caller needs to invoke connect logic then call again
364 * other: Some other error occured, the request was not sent
365 */
366static int xs_udp_send_request(struct rpc_task *task)
367{
368 struct rpc_rqst *req = task->tk_rqstp;
369 struct rpc_xprt *xprt = req->rq_xprt;
ee0ac0c2 370 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
262965f5
CL
371 struct xdr_buf *xdr = &req->rq_snd_buf;
372 int status;
a246b010 373
9903cd1c 374 xs_pktdump("packet data:",
a246b010
CL
375 req->rq_svec->iov_base,
376 req->rq_svec->iov_len);
377
262965f5 378 req->rq_xtime = jiffies;
ee0ac0c2
CL
379 status = xs_sendpages(transport->sock,
380 (struct sockaddr *) &xprt->addr,
381 xprt->addrlen, xdr,
382 req->rq_bytes_sent);
a246b010 383
262965f5
CL
384 dprintk("RPC: xs_udp_send_request(%u) = %d\n",
385 xdr->len - req->rq_bytes_sent, status);
a246b010 386
262965f5
CL
387 if (likely(status >= (int) req->rq_slen))
388 return 0;
a246b010 389
262965f5
CL
390 /* Still some bytes left; set up for a retry later. */
391 if (status > 0)
392 status = -EAGAIN;
a246b010 393
262965f5
CL
394 switch (status) {
395 case -ENETUNREACH:
396 case -EPIPE:
a246b010
CL
397 case -ECONNREFUSED:
398 /* When the server has died, an ICMP port unreachable message
9903cd1c 399 * prompts ECONNREFUSED. */
a246b010 400 break;
262965f5
CL
401 case -EAGAIN:
402 xs_nospace(task);
a246b010
CL
403 break;
404 default:
262965f5
CL
405 dprintk("RPC: sendmsg returned unrecognized error %d\n",
406 -status);
9903cd1c 407 break;
a246b010 408 }
262965f5
CL
409
410 return status;
a246b010
CL
411}
412
808012fb
CL
413static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf)
414{
415 u32 reclen = buf->len - sizeof(rpc_fraghdr);
416 rpc_fraghdr *base = buf->head[0].iov_base;
417 *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen);
418}
419
9903cd1c 420/**
262965f5 421 * xs_tcp_send_request - write an RPC request to a TCP socket
9903cd1c
CL
422 * @task: address of RPC task that manages the state of an RPC request
423 *
424 * Return values:
262965f5
CL
425 * 0: The request has been sent
426 * EAGAIN: The socket was blocked, please call again later to
427 * complete the request
428 * ENOTCONN: Caller needs to invoke connect logic then call again
429 * other: Some other error occured, the request was not sent
9903cd1c
CL
430 *
431 * XXX: In the case of soft timeouts, should we eventually give up
262965f5 432 * if sendmsg is not able to make progress?
9903cd1c 433 */
262965f5 434static int xs_tcp_send_request(struct rpc_task *task)
a246b010
CL
435{
436 struct rpc_rqst *req = task->tk_rqstp;
437 struct rpc_xprt *xprt = req->rq_xprt;
ee0ac0c2 438 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
262965f5 439 struct xdr_buf *xdr = &req->rq_snd_buf;
a246b010
CL
440 int status, retry = 0;
441
808012fb 442 xs_encode_tcp_record_marker(&req->rq_snd_buf);
a246b010 443
262965f5
CL
444 xs_pktdump("packet data:",
445 req->rq_svec->iov_base,
446 req->rq_svec->iov_len);
a246b010
CL
447
448 /* Continue transmitting the packet/record. We must be careful
449 * to cope with writespace callbacks arriving _after_ we have
262965f5 450 * called sendmsg(). */
a246b010
CL
451 while (1) {
452 req->rq_xtime = jiffies;
ee0ac0c2
CL
453 status = xs_sendpages(transport->sock,
454 NULL, 0, xdr, req->rq_bytes_sent);
a246b010 455
262965f5
CL
456 dprintk("RPC: xs_tcp_send_request(%u) = %d\n",
457 xdr->len - req->rq_bytes_sent, status);
a246b010 458
262965f5 459 if (unlikely(status < 0))
a246b010 460 break;
a246b010 461
262965f5
CL
462 /* If we've sent the entire packet, immediately
463 * reset the count of bytes sent. */
464 req->rq_bytes_sent += status;
ef759a2e 465 task->tk_bytes_sent += status;
262965f5
CL
466 if (likely(req->rq_bytes_sent >= req->rq_slen)) {
467 req->rq_bytes_sent = 0;
468 return 0;
469 }
a246b010
CL
470
471 status = -EAGAIN;
262965f5 472 if (retry++ > XS_SENDMSG_RETRY)
a246b010
CL
473 break;
474 }
475
262965f5
CL
476 switch (status) {
477 case -EAGAIN:
478 xs_nospace(task);
479 break;
480 case -ECONNREFUSED:
481 case -ECONNRESET:
482 case -ENOTCONN:
483 case -EPIPE:
484 status = -ENOTCONN;
485 break;
486 default:
487 dprintk("RPC: sendmsg returned unrecognized error %d\n",
488 -status);
43118c29 489 xprt_disconnect(xprt);
262965f5 490 break;
a246b010 491 }
262965f5 492
a246b010
CL
493 return status;
494}
495
e0ab53de
TM
496/**
497 * xs_tcp_release_xprt - clean up after a tcp transmission
498 * @xprt: transport
499 * @task: rpc task
500 *
501 * This cleans up if an error causes us to abort the transmission of a request.
502 * In this case, the socket may need to be reset in order to avoid confusing
503 * the server.
504 */
505static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
506{
507 struct rpc_rqst *req;
508
509 if (task != xprt->snd_task)
510 return;
511 if (task == NULL)
512 goto out_release;
513 req = task->tk_rqstp;
514 if (req->rq_bytes_sent == 0)
515 goto out_release;
516 if (req->rq_bytes_sent == req->rq_snd_buf.len)
517 goto out_release;
518 set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state);
519out_release:
520 xprt_release_xprt(xprt, task);
521}
522
9903cd1c
CL
523/**
524 * xs_close - close a socket
525 * @xprt: transport
526 *
3167e12c
CL
527 * This is used when all requests are complete; ie, no DRC state remains
528 * on the server we want to save.
a246b010 529 */
9903cd1c 530static void xs_close(struct rpc_xprt *xprt)
a246b010 531{
ee0ac0c2
CL
532 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
533 struct socket *sock = transport->sock;
534 struct sock *sk = transport->inet;
a246b010
CL
535
536 if (!sk)
632e3bdc 537 goto clear_close_wait;
a246b010 538
9903cd1c
CL
539 dprintk("RPC: xs_close xprt %p\n", xprt);
540
a246b010 541 write_lock_bh(&sk->sk_callback_lock);
ee0ac0c2
CL
542 transport->inet = NULL;
543 transport->sock = NULL;
a246b010 544
9903cd1c 545 sk->sk_user_data = NULL;
314dfd79
CL
546 sk->sk_data_ready = transport->old_data_ready;
547 sk->sk_state_change = transport->old_state_change;
548 sk->sk_write_space = transport->old_write_space;
a246b010
CL
549 write_unlock_bh(&sk->sk_callback_lock);
550
9903cd1c 551 sk->sk_no_check = 0;
a246b010
CL
552
553 sock_release(sock);
632e3bdc
TM
554clear_close_wait:
555 smp_mb__before_clear_bit();
556 clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
557 smp_mb__after_clear_bit();
a246b010
CL
558}
559
9903cd1c
CL
560/**
561 * xs_destroy - prepare to shutdown a transport
562 * @xprt: doomed transport
563 *
564 */
565static void xs_destroy(struct rpc_xprt *xprt)
a246b010 566{
c8475461
CL
567 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
568
9903cd1c
CL
569 dprintk("RPC: xs_destroy xprt %p\n", xprt);
570
c8475461 571 cancel_delayed_work(&transport->connect_worker);
a246b010
CL
572 flush_scheduled_work();
573
574 xprt_disconnect(xprt);
9903cd1c 575 xs_close(xprt);
edb267a6 576 xs_free_peer_addresses(xprt);
a246b010 577 kfree(xprt->slot);
c8541ecd 578 kfree(xprt);
a246b010
CL
579}
580
9903cd1c
CL
581static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
582{
583 return (struct rpc_xprt *) sk->sk_user_data;
584}
585
586/**
587 * xs_udp_data_ready - "data ready" callback for UDP sockets
588 * @sk: socket with data to read
589 * @len: how much data to read
590 *
a246b010 591 */
9903cd1c 592static void xs_udp_data_ready(struct sock *sk, int len)
a246b010 593{
9903cd1c
CL
594 struct rpc_task *task;
595 struct rpc_xprt *xprt;
a246b010 596 struct rpc_rqst *rovr;
9903cd1c 597 struct sk_buff *skb;
a246b010 598 int err, repsize, copied;
d8ed029d
AD
599 u32 _xid;
600 __be32 *xp;
a246b010
CL
601
602 read_lock(&sk->sk_callback_lock);
9903cd1c
CL
603 dprintk("RPC: xs_udp_data_ready...\n");
604 if (!(xprt = xprt_from_sock(sk)))
a246b010 605 goto out;
a246b010
CL
606
607 if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
608 goto out;
609
610 if (xprt->shutdown)
611 goto dropit;
612
613 repsize = skb->len - sizeof(struct udphdr);
614 if (repsize < 4) {
9903cd1c 615 dprintk("RPC: impossible RPC reply size %d!\n", repsize);
a246b010
CL
616 goto dropit;
617 }
618
619 /* Copy the XID from the skb... */
620 xp = skb_header_pointer(skb, sizeof(struct udphdr),
621 sizeof(_xid), &_xid);
622 if (xp == NULL)
623 goto dropit;
624
625 /* Look up and lock the request corresponding to the given XID */
4a0f8c04 626 spin_lock(&xprt->transport_lock);
a246b010
CL
627 rovr = xprt_lookup_rqst(xprt, *xp);
628 if (!rovr)
629 goto out_unlock;
630 task = rovr->rq_task;
631
a246b010
CL
632 if ((copied = rovr->rq_private_buf.buflen) > repsize)
633 copied = repsize;
634
635 /* Suck it into the iovec, verify checksum if not done by hw. */
636 if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb))
637 goto out_unlock;
638
639 /* Something worked... */
640 dst_confirm(skb->dst);
641
1570c1e4
CL
642 xprt_adjust_cwnd(task, copied);
643 xprt_update_rtt(task);
644 xprt_complete_rqst(task, copied);
a246b010
CL
645
646 out_unlock:
4a0f8c04 647 spin_unlock(&xprt->transport_lock);
a246b010
CL
648 dropit:
649 skb_free_datagram(sk, skb);
650 out:
651 read_unlock(&sk->sk_callback_lock);
652}
653
9903cd1c 654static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, skb_reader_t *desc)
a246b010 655{
51971139 656 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
a246b010
CL
657 size_t len, used;
658 char *p;
659
51971139
CL
660 p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
661 len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
9d292316 662 used = xdr_skb_read_bits(desc, p, len);
51971139 663 transport->tcp_offset += used;
a246b010
CL
664 if (used != len)
665 return;
808012fb 666
51971139
CL
667 transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
668 if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
e136d092 669 transport->tcp_flags |= TCP_RCV_LAST_FRAG;
a246b010 670 else
e136d092 671 transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
51971139 672 transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
808012fb 673
e136d092 674 transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
51971139 675 transport->tcp_offset = 0;
808012fb 676
a246b010 677 /* Sanity check of the record length */
51971139 678 if (unlikely(transport->tcp_reclen < 4)) {
9903cd1c 679 dprintk("RPC: invalid TCP record fragment length\n");
a246b010 680 xprt_disconnect(xprt);
9903cd1c 681 return;
a246b010
CL
682 }
683 dprintk("RPC: reading TCP record fragment of length %d\n",
51971139 684 transport->tcp_reclen);
a246b010
CL
685}
686
51971139 687static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
a246b010 688{
51971139 689 if (transport->tcp_offset == transport->tcp_reclen) {
e136d092 690 transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
51971139 691 transport->tcp_offset = 0;
e136d092
CL
692 if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
693 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
694 transport->tcp_flags |= TCP_RCV_COPY_XID;
51971139 695 transport->tcp_copied = 0;
a246b010
CL
696 }
697 }
698}
699
51971139 700static inline void xs_tcp_read_xid(struct sock_xprt *transport, skb_reader_t *desc)
a246b010
CL
701{
702 size_t len, used;
703 char *p;
704
51971139 705 len = sizeof(transport->tcp_xid) - transport->tcp_offset;
a246b010 706 dprintk("RPC: reading XID (%Zu bytes)\n", len);
51971139 707 p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
9d292316 708 used = xdr_skb_read_bits(desc, p, len);
51971139 709 transport->tcp_offset += used;
a246b010
CL
710 if (used != len)
711 return;
e136d092
CL
712 transport->tcp_flags &= ~TCP_RCV_COPY_XID;
713 transport->tcp_flags |= TCP_RCV_COPY_DATA;
51971139 714 transport->tcp_copied = 4;
a246b010 715 dprintk("RPC: reading reply for XID %08x\n",
51971139
CL
716 ntohl(transport->tcp_xid));
717 xs_tcp_check_fraghdr(transport);
a246b010
CL
718}
719
9903cd1c 720static inline void xs_tcp_read_request(struct rpc_xprt *xprt, skb_reader_t *desc)
a246b010 721{
51971139 722 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
a246b010
CL
723 struct rpc_rqst *req;
724 struct xdr_buf *rcvbuf;
725 size_t len;
726 ssize_t r;
727
728 /* Find and lock the request corresponding to this xid */
4a0f8c04 729 spin_lock(&xprt->transport_lock);
51971139 730 req = xprt_lookup_rqst(xprt, transport->tcp_xid);
a246b010 731 if (!req) {
e136d092 732 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
a246b010 733 dprintk("RPC: XID %08x request not found!\n",
51971139 734 ntohl(transport->tcp_xid));
4a0f8c04 735 spin_unlock(&xprt->transport_lock);
a246b010
CL
736 return;
737 }
738
739 rcvbuf = &req->rq_private_buf;
740 len = desc->count;
51971139 741 if (len > transport->tcp_reclen - transport->tcp_offset) {
a246b010
CL
742 skb_reader_t my_desc;
743
51971139 744 len = transport->tcp_reclen - transport->tcp_offset;
a246b010
CL
745 memcpy(&my_desc, desc, sizeof(my_desc));
746 my_desc.count = len;
51971139 747 r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
9d292316 748 &my_desc, xdr_skb_read_bits);
a246b010
CL
749 desc->count -= r;
750 desc->offset += r;
751 } else
51971139 752 r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
9d292316 753 desc, xdr_skb_read_bits);
a246b010
CL
754
755 if (r > 0) {
51971139
CL
756 transport->tcp_copied += r;
757 transport->tcp_offset += r;
a246b010
CL
758 }
759 if (r != len) {
760 /* Error when copying to the receive buffer,
761 * usually because we weren't able to allocate
762 * additional buffer pages. All we can do now
e136d092 763 * is turn off TCP_RCV_COPY_DATA, so the request
a246b010
CL
764 * will not receive any additional updates,
765 * and time out.
766 * Any remaining data from this record will
767 * be discarded.
768 */
e136d092 769 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
a246b010 770 dprintk("RPC: XID %08x truncated request\n",
51971139 771 ntohl(transport->tcp_xid));
a246b010 772 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
51971139
CL
773 xprt, transport->tcp_copied, transport->tcp_offset,
774 transport->tcp_reclen);
a246b010
CL
775 goto out;
776 }
777
778 dprintk("RPC: XID %08x read %Zd bytes\n",
51971139 779 ntohl(transport->tcp_xid), r);
a246b010 780 dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, tcp_reclen = %u\n",
51971139
CL
781 xprt, transport->tcp_copied, transport->tcp_offset,
782 transport->tcp_reclen);
783
784 if (transport->tcp_copied == req->rq_private_buf.buflen)
e136d092 785 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
51971139 786 else if (transport->tcp_offset == transport->tcp_reclen) {
e136d092
CL
787 if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
788 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
a246b010
CL
789 }
790
791out:
e136d092 792 if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
51971139 793 xprt_complete_rqst(req->rq_task, transport->tcp_copied);
4a0f8c04 794 spin_unlock(&xprt->transport_lock);
51971139 795 xs_tcp_check_fraghdr(transport);
a246b010
CL
796}
797
51971139 798static inline void xs_tcp_read_discard(struct sock_xprt *transport, skb_reader_t *desc)
a246b010
CL
799{
800 size_t len;
801
51971139 802 len = transport->tcp_reclen - transport->tcp_offset;
a246b010
CL
803 if (len > desc->count)
804 len = desc->count;
805 desc->count -= len;
806 desc->offset += len;
51971139 807 transport->tcp_offset += len;
a246b010 808 dprintk("RPC: discarded %Zu bytes\n", len);
51971139 809 xs_tcp_check_fraghdr(transport);
a246b010
CL
810}
811
9903cd1c 812static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
a246b010
CL
813{
814 struct rpc_xprt *xprt = rd_desc->arg.data;
51971139 815 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
a246b010
CL
816 skb_reader_t desc = {
817 .skb = skb,
818 .offset = offset,
819 .count = len,
9903cd1c 820 };
a246b010 821
9903cd1c 822 dprintk("RPC: xs_tcp_data_recv started\n");
a246b010
CL
823 do {
824 /* Read in a new fragment marker if necessary */
825 /* Can we ever really expect to get completely empty fragments? */
e136d092 826 if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
9903cd1c 827 xs_tcp_read_fraghdr(xprt, &desc);
a246b010
CL
828 continue;
829 }
830 /* Read in the xid if necessary */
e136d092 831 if (transport->tcp_flags & TCP_RCV_COPY_XID) {
51971139 832 xs_tcp_read_xid(transport, &desc);
a246b010
CL
833 continue;
834 }
835 /* Read in the request data */
e136d092 836 if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
9903cd1c 837 xs_tcp_read_request(xprt, &desc);
a246b010
CL
838 continue;
839 }
840 /* Skip over any trailing bytes on short reads */
51971139 841 xs_tcp_read_discard(transport, &desc);
a246b010 842 } while (desc.count);
9903cd1c 843 dprintk("RPC: xs_tcp_data_recv done\n");
a246b010
CL
844 return len - desc.count;
845}
846
9903cd1c
CL
847/**
848 * xs_tcp_data_ready - "data ready" callback for TCP sockets
849 * @sk: socket with data to read
850 * @bytes: how much data to read
851 *
852 */
853static void xs_tcp_data_ready(struct sock *sk, int bytes)
a246b010
CL
854{
855 struct rpc_xprt *xprt;
856 read_descriptor_t rd_desc;
857
858 read_lock(&sk->sk_callback_lock);
9903cd1c
CL
859 dprintk("RPC: xs_tcp_data_ready...\n");
860 if (!(xprt = xprt_from_sock(sk)))
a246b010 861 goto out;
a246b010
CL
862 if (xprt->shutdown)
863 goto out;
864
9903cd1c 865 /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
a246b010
CL
866 rd_desc.arg.data = xprt;
867 rd_desc.count = 65536;
9903cd1c 868 tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
a246b010
CL
869out:
870 read_unlock(&sk->sk_callback_lock);
871}
872
9903cd1c
CL
873/**
874 * xs_tcp_state_change - callback to handle TCP socket state changes
875 * @sk: socket whose state has changed
876 *
877 */
878static void xs_tcp_state_change(struct sock *sk)
a246b010 879{
9903cd1c 880 struct rpc_xprt *xprt;
a246b010
CL
881
882 read_lock(&sk->sk_callback_lock);
883 if (!(xprt = xprt_from_sock(sk)))
884 goto out;
9903cd1c 885 dprintk("RPC: xs_tcp_state_change client %p...\n", xprt);
a246b010
CL
886 dprintk("RPC: state %x conn %d dead %d zapped %d\n",
887 sk->sk_state, xprt_connected(xprt),
888 sock_flag(sk, SOCK_DEAD),
889 sock_flag(sk, SOCK_ZAPPED));
890
891 switch (sk->sk_state) {
892 case TCP_ESTABLISHED:
4a0f8c04 893 spin_lock_bh(&xprt->transport_lock);
a246b010 894 if (!xprt_test_and_set_connected(xprt)) {
51971139
CL
895 struct sock_xprt *transport = container_of(xprt,
896 struct sock_xprt, xprt);
897
a246b010 898 /* Reset TCP record info */
51971139
CL
899 transport->tcp_offset = 0;
900 transport->tcp_reclen = 0;
901 transport->tcp_copied = 0;
e136d092
CL
902 transport->tcp_flags =
903 TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
51971139 904
03bf4b70 905 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
44fbac22 906 xprt_wake_pending_tasks(xprt, 0);
a246b010 907 }
4a0f8c04 908 spin_unlock_bh(&xprt->transport_lock);
a246b010
CL
909 break;
910 case TCP_SYN_SENT:
911 case TCP_SYN_RECV:
912 break;
632e3bdc
TM
913 case TCP_CLOSE_WAIT:
914 /* Try to schedule an autoclose RPC calls */
915 set_bit(XPRT_CLOSE_WAIT, &xprt->state);
916 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
917 schedule_work(&xprt->task_cleanup);
a246b010
CL
918 default:
919 xprt_disconnect(xprt);
a246b010
CL
920 }
921 out:
922 read_unlock(&sk->sk_callback_lock);
923}
924
9903cd1c 925/**
c7b2cae8
CL
926 * xs_udp_write_space - callback invoked when socket buffer space
927 * becomes available
9903cd1c
CL
928 * @sk: socket whose state has changed
929 *
a246b010
CL
930 * Called when more output buffer space is available for this socket.
931 * We try not to wake our writers until they can make "significant"
c7b2cae8 932 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
a246b010
CL
933 * with a bunch of small requests.
934 */
c7b2cae8 935static void xs_udp_write_space(struct sock *sk)
a246b010 936{
a246b010 937 read_lock(&sk->sk_callback_lock);
a246b010 938
c7b2cae8
CL
939 /* from net/core/sock.c:sock_def_write_space */
940 if (sock_writeable(sk)) {
941 struct socket *sock;
942 struct rpc_xprt *xprt;
943
944 if (unlikely(!(sock = sk->sk_socket)))
a246b010 945 goto out;
c7b2cae8
CL
946 if (unlikely(!(xprt = xprt_from_sock(sk))))
947 goto out;
948 if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
a246b010 949 goto out;
c7b2cae8
CL
950
951 xprt_write_space(xprt);
a246b010
CL
952 }
953
c7b2cae8
CL
954 out:
955 read_unlock(&sk->sk_callback_lock);
956}
a246b010 957
c7b2cae8
CL
958/**
959 * xs_tcp_write_space - callback invoked when socket buffer space
960 * becomes available
961 * @sk: socket whose state has changed
962 *
963 * Called when more output buffer space is available for this socket.
964 * We try not to wake our writers until they can make "significant"
965 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
966 * with a bunch of small requests.
967 */
968static void xs_tcp_write_space(struct sock *sk)
969{
970 read_lock(&sk->sk_callback_lock);
971
972 /* from net/core/stream.c:sk_stream_write_space */
973 if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
974 struct socket *sock;
975 struct rpc_xprt *xprt;
976
977 if (unlikely(!(sock = sk->sk_socket)))
978 goto out;
979 if (unlikely(!(xprt = xprt_from_sock(sk))))
980 goto out;
981 if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
982 goto out;
983
984 xprt_write_space(xprt);
985 }
986
987 out:
a246b010
CL
988 read_unlock(&sk->sk_callback_lock);
989}
990
470056c2 991static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
a246b010 992{
ee0ac0c2
CL
993 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
994 struct sock *sk = transport->inet;
a246b010 995
7c6e066e 996 if (transport->rcvsize) {
a246b010 997 sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
7c6e066e 998 sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
a246b010 999 }
7c6e066e 1000 if (transport->sndsize) {
a246b010 1001 sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
7c6e066e 1002 sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
a246b010
CL
1003 sk->sk_write_space(sk);
1004 }
1005}
1006
43118c29 1007/**
470056c2 1008 * xs_udp_set_buffer_size - set send and receive limits
43118c29 1009 * @xprt: generic transport
470056c2
CL
1010 * @sndsize: requested size of send buffer, in bytes
1011 * @rcvsize: requested size of receive buffer, in bytes
43118c29 1012 *
470056c2 1013 * Set socket send and receive buffer size limits.
43118c29 1014 */
470056c2 1015static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
43118c29 1016{
7c6e066e
CL
1017 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1018
1019 transport->sndsize = 0;
470056c2 1020 if (sndsize)
7c6e066e
CL
1021 transport->sndsize = sndsize + 1024;
1022 transport->rcvsize = 0;
470056c2 1023 if (rcvsize)
7c6e066e 1024 transport->rcvsize = rcvsize + 1024;
470056c2
CL
1025
1026 xs_udp_do_set_buffer_size(xprt);
43118c29
CL
1027}
1028
46c0ee8b
CL
1029/**
1030 * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
1031 * @task: task that timed out
1032 *
1033 * Adjust the congestion window after a retransmit timeout has occurred.
1034 */
1035static void xs_udp_timer(struct rpc_task *task)
1036{
1037 xprt_adjust_cwnd(task, -ETIMEDOUT);
1038}
1039
b85d8806
CL
1040static unsigned short xs_get_random_port(void)
1041{
1042 unsigned short range = xprt_max_resvport - xprt_min_resvport;
1043 unsigned short rand = (unsigned short) net_random() % range;
1044 return rand + xprt_min_resvport;
1045}
1046
92200412
CL
1047/**
1048 * xs_set_port - reset the port number in the remote endpoint address
1049 * @xprt: generic transport
1050 * @port: new port number
1051 *
1052 */
1053static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
1054{
c4efcb1d
CL
1055 struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr;
1056
92200412 1057 dprintk("RPC: setting port for xprt %p to %u\n", xprt, port);
c4efcb1d
CL
1058
1059 sap->sin_port = htons(port);
92200412
CL
1060}
1061
c8475461 1062static int xs_bindresvport(struct sock_xprt *transport, struct socket *sock)
a246b010
CL
1063{
1064 struct sockaddr_in myaddr = {
1065 .sin_family = AF_INET,
1066 };
529b33c6 1067 int err;
c8475461 1068 unsigned short port = transport->port;
a246b010 1069
a246b010
CL
1070 do {
1071 myaddr.sin_port = htons(port);
e6242e92 1072 err = kernel_bind(sock, (struct sockaddr *) &myaddr,
a246b010
CL
1073 sizeof(myaddr));
1074 if (err == 0) {
c8475461 1075 transport->port = port;
9903cd1c
CL
1076 dprintk("RPC: xs_bindresvport bound to port %u\n",
1077 port);
a246b010
CL
1078 return 0;
1079 }
529b33c6
CL
1080 if (port <= xprt_min_resvport)
1081 port = xprt_max_resvport;
1082 else
1083 port--;
c8475461 1084 } while (err == -EADDRINUSE && port != transport->port);
a246b010 1085
9903cd1c 1086 dprintk("RPC: can't bind to reserved port (%d).\n", -err);
a246b010
CL
1087 return err;
1088}
1089
b0d93ad5
CL
1090/**
1091 * xs_udp_connect_worker - set up a UDP socket
1092 * @args: RPC transport to connect
1093 *
1094 * Invoked by a work queue tasklet.
1095 */
1096static void xs_udp_connect_worker(void *args)
a246b010 1097{
c8475461
CL
1098 struct sock_xprt *transport = (struct sock_xprt *)args;
1099 struct rpc_xprt *xprt = &transport->xprt;
ee0ac0c2 1100 struct socket *sock = transport->sock;
b0d93ad5 1101 int err, status = -EIO;
9903cd1c 1102
ec739ef0 1103 if (xprt->shutdown || !xprt_bound(xprt))
b0d93ad5 1104 goto out;
9903cd1c 1105
b0d93ad5
CL
1106 /* Start by resetting any existing state */
1107 xs_close(xprt);
9903cd1c 1108
b0d93ad5
CL
1109 if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
1110 dprintk("RPC: can't create UDP transport socket (%d).\n", -err);
1111 goto out;
1112 }
9903cd1c 1113
c8475461 1114 if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
b0d93ad5
CL
1115 sock_release(sock);
1116 goto out;
1117 }
9903cd1c 1118
edb267a6 1119 dprintk("RPC: worker connecting xprt %p to address: %s\n",
7559c7a2 1120 xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
edb267a6 1121
ee0ac0c2 1122 if (!transport->inet) {
b0d93ad5 1123 struct sock *sk = sock->sk;
a246b010 1124
b0d93ad5 1125 write_lock_bh(&sk->sk_callback_lock);
a246b010 1126
b0d93ad5 1127 sk->sk_user_data = xprt;
314dfd79
CL
1128 transport->old_data_ready = sk->sk_data_ready;
1129 transport->old_state_change = sk->sk_state_change;
1130 transport->old_write_space = sk->sk_write_space;
9903cd1c 1131 sk->sk_data_ready = xs_udp_data_ready;
c7b2cae8 1132 sk->sk_write_space = xs_udp_write_space;
a246b010 1133 sk->sk_no_check = UDP_CSUM_NORCV;
b079fa7b 1134 sk->sk_allocation = GFP_ATOMIC;
b0d93ad5 1135
a246b010 1136 xprt_set_connected(xprt);
a246b010 1137
b0d93ad5 1138 /* Reset to new socket */
ee0ac0c2
CL
1139 transport->sock = sock;
1140 transport->inet = sk;
a246b010 1141
b0d93ad5
CL
1142 write_unlock_bh(&sk->sk_callback_lock);
1143 }
470056c2 1144 xs_udp_do_set_buffer_size(xprt);
b0d93ad5
CL
1145 status = 0;
1146out:
1147 xprt_wake_pending_tasks(xprt, status);
1148 xprt_clear_connecting(xprt);
a246b010
CL
1149}
1150
3167e12c
CL
1151/*
1152 * We need to preserve the port number so the reply cache on the server can
1153 * find our cached RPC replies when we get around to reconnecting.
1154 */
1155static void xs_tcp_reuse_connection(struct rpc_xprt *xprt)
1156{
1157 int result;
ee0ac0c2 1158 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
3167e12c
CL
1159 struct sockaddr any;
1160
1161 dprintk("RPC: disconnecting xprt %p to reuse port\n", xprt);
1162
1163 /*
1164 * Disconnect the transport socket by doing a connect operation
1165 * with AF_UNSPEC. This should return immediately...
1166 */
1167 memset(&any, 0, sizeof(any));
1168 any.sa_family = AF_UNSPEC;
ee0ac0c2 1169 result = kernel_connect(transport->sock, &any, sizeof(any), 0);
3167e12c
CL
1170 if (result)
1171 dprintk("RPC: AF_UNSPEC connect return code %d\n",
1172 result);
1173}
1174
9903cd1c 1175/**
b0d93ad5 1176 * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint
9903cd1c
CL
1177 * @args: RPC transport to connect
1178 *
1179 * Invoked by a work queue tasklet.
a246b010 1180 */
b0d93ad5 1181static void xs_tcp_connect_worker(void *args)
a246b010 1182{
c8475461
CL
1183 struct sock_xprt *transport = (struct sock_xprt *)args;
1184 struct rpc_xprt *xprt = &transport->xprt;
ee0ac0c2 1185 struct socket *sock = transport->sock;
b0d93ad5 1186 int err, status = -EIO;
a246b010 1187
ec739ef0 1188 if (xprt->shutdown || !xprt_bound(xprt))
a246b010
CL
1189 goto out;
1190
ee0ac0c2 1191 if (!sock) {
3167e12c
CL
1192 /* start from scratch */
1193 if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
1194 dprintk("RPC: can't create TCP transport socket (%d).\n", -err);
1195 goto out;
1196 }
a246b010 1197
c8475461 1198 if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
3167e12c
CL
1199 sock_release(sock);
1200 goto out;
1201 }
1202 } else
1203 /* "close" the socket, preserving the local port */
1204 xs_tcp_reuse_connection(xprt);
a246b010 1205
edb267a6 1206 dprintk("RPC: worker connecting xprt %p to address: %s\n",
7559c7a2 1207 xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
edb267a6 1208
ee0ac0c2 1209 if (!transport->inet) {
b0d93ad5
CL
1210 struct sock *sk = sock->sk;
1211
1212 write_lock_bh(&sk->sk_callback_lock);
1213
1214 sk->sk_user_data = xprt;
314dfd79
CL
1215 transport->old_data_ready = sk->sk_data_ready;
1216 transport->old_state_change = sk->sk_state_change;
1217 transport->old_write_space = sk->sk_write_space;
b0d93ad5
CL
1218 sk->sk_data_ready = xs_tcp_data_ready;
1219 sk->sk_state_change = xs_tcp_state_change;
1220 sk->sk_write_space = xs_tcp_write_space;
b079fa7b 1221 sk->sk_allocation = GFP_ATOMIC;
3167e12c
CL
1222
1223 /* socket options */
1224 sk->sk_userlocks |= SOCK_BINDPORT_LOCK;
1225 sock_reset_flag(sk, SOCK_LINGER);
1226 tcp_sk(sk)->linger2 = 0;
1227 tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
b0d93ad5
CL
1228
1229 xprt_clear_connected(xprt);
1230
1231 /* Reset to new socket */
ee0ac0c2
CL
1232 transport->sock = sock;
1233 transport->inet = sk;
b0d93ad5
CL
1234
1235 write_unlock_bh(&sk->sk_callback_lock);
1236 }
1237
1238 /* Tell the socket layer to start connecting... */
262ca07d
CL
1239 xprt->stat.connect_count++;
1240 xprt->stat.connect_start = jiffies;
e6242e92 1241 status = kernel_connect(sock, (struct sockaddr *) &xprt->addr,
c4efcb1d 1242 xprt->addrlen, O_NONBLOCK);
a246b010
CL
1243 dprintk("RPC: %p connect status %d connected %d sock state %d\n",
1244 xprt, -status, xprt_connected(xprt), sock->sk->sk_state);
1245 if (status < 0) {
1246 switch (status) {
1247 case -EINPROGRESS:
1248 case -EALREADY:
1249 goto out_clear;
3167e12c
CL
1250 case -ECONNREFUSED:
1251 case -ECONNRESET:
1252 /* retry with existing socket, after a delay */
1253 break;
1254 default:
1255 /* get rid of existing socket, and retry */
1256 xs_close(xprt);
1257 break;
a246b010
CL
1258 }
1259 }
1260out:
44fbac22 1261 xprt_wake_pending_tasks(xprt, status);
a246b010 1262out_clear:
2226feb6 1263 xprt_clear_connecting(xprt);
a246b010
CL
1264}
1265
9903cd1c
CL
1266/**
1267 * xs_connect - connect a socket to a remote endpoint
1268 * @task: address of RPC task that manages state of connect request
1269 *
1270 * TCP: If the remote end dropped the connection, delay reconnecting.
03bf4b70
CL
1271 *
1272 * UDP socket connects are synchronous, but we use a work queue anyway
1273 * to guarantee that even unprivileged user processes can set up a
1274 * socket on a privileged port.
1275 *
1276 * If a UDP socket connect fails, the delay behavior here prevents
1277 * retry floods (hard mounts).
9903cd1c
CL
1278 */
1279static void xs_connect(struct rpc_task *task)
a246b010
CL
1280{
1281 struct rpc_xprt *xprt = task->tk_xprt;
ee0ac0c2 1282 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
a246b010 1283
b0d93ad5
CL
1284 if (xprt_test_and_set_connecting(xprt))
1285 return;
1286
ee0ac0c2 1287 if (transport->sock != NULL) {
03bf4b70
CL
1288 dprintk("RPC: xs_connect delayed xprt %p for %lu seconds\n",
1289 xprt, xprt->reestablish_timeout / HZ);
c8475461 1290 schedule_delayed_work(&transport->connect_worker,
03bf4b70
CL
1291 xprt->reestablish_timeout);
1292 xprt->reestablish_timeout <<= 1;
1293 if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
1294 xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
b0d93ad5
CL
1295 } else {
1296 dprintk("RPC: xs_connect scheduled xprt %p\n", xprt);
c8475461 1297 schedule_work(&transport->connect_worker);
b0d93ad5
CL
1298
1299 /* flush_scheduled_work can sleep... */
1300 if (!RPC_IS_ASYNC(task))
1301 flush_scheduled_work();
a246b010
CL
1302 }
1303}
1304
262ca07d
CL
1305/**
1306 * xs_udp_print_stats - display UDP socket-specifc stats
1307 * @xprt: rpc_xprt struct containing statistics
1308 * @seq: output file
1309 *
1310 */
1311static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
1312{
c8475461
CL
1313 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1314
262ca07d 1315 seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
c8475461 1316 transport->port,
262ca07d
CL
1317 xprt->stat.bind_count,
1318 xprt->stat.sends,
1319 xprt->stat.recvs,
1320 xprt->stat.bad_xids,
1321 xprt->stat.req_u,
1322 xprt->stat.bklog_u);
1323}
1324
1325/**
1326 * xs_tcp_print_stats - display TCP socket-specifc stats
1327 * @xprt: rpc_xprt struct containing statistics
1328 * @seq: output file
1329 *
1330 */
1331static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
1332{
c8475461 1333 struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
262ca07d
CL
1334 long idle_time = 0;
1335
1336 if (xprt_connected(xprt))
1337 idle_time = (long)(jiffies - xprt->last_used) / HZ;
1338
1339 seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n",
c8475461 1340 transport->port,
262ca07d
CL
1341 xprt->stat.bind_count,
1342 xprt->stat.connect_count,
1343 xprt->stat.connect_time,
1344 idle_time,
1345 xprt->stat.sends,
1346 xprt->stat.recvs,
1347 xprt->stat.bad_xids,
1348 xprt->stat.req_u,
1349 xprt->stat.bklog_u);
1350}
1351
262965f5 1352static struct rpc_xprt_ops xs_udp_ops = {
43118c29 1353 .set_buffer_size = xs_udp_set_buffer_size,
12a80469 1354 .reserve_xprt = xprt_reserve_xprt_cong,
49e9a890 1355 .release_xprt = xprt_release_xprt_cong,
bbf7c1dd 1356 .rpcbind = rpc_getport,
92200412 1357 .set_port = xs_set_port,
262965f5 1358 .connect = xs_connect,
02107148
CL
1359 .buf_alloc = rpc_malloc,
1360 .buf_free = rpc_free,
262965f5 1361 .send_request = xs_udp_send_request,
fe3aca29 1362 .set_retrans_timeout = xprt_set_retrans_timeout_rtt,
46c0ee8b 1363 .timer = xs_udp_timer,
a58dd398 1364 .release_request = xprt_release_rqst_cong,
262965f5
CL
1365 .close = xs_close,
1366 .destroy = xs_destroy,
262ca07d 1367 .print_stats = xs_udp_print_stats,
262965f5
CL
1368};
1369
1370static struct rpc_xprt_ops xs_tcp_ops = {
12a80469 1371 .reserve_xprt = xprt_reserve_xprt,
e0ab53de 1372 .release_xprt = xs_tcp_release_xprt,
bbf7c1dd 1373 .rpcbind = rpc_getport,
92200412 1374 .set_port = xs_set_port,
9903cd1c 1375 .connect = xs_connect,
02107148
CL
1376 .buf_alloc = rpc_malloc,
1377 .buf_free = rpc_free,
262965f5 1378 .send_request = xs_tcp_send_request,
fe3aca29 1379 .set_retrans_timeout = xprt_set_retrans_timeout_def,
9903cd1c
CL
1380 .close = xs_close,
1381 .destroy = xs_destroy,
262ca07d 1382 .print_stats = xs_tcp_print_stats,
a246b010
CL
1383};
1384
c8541ecd
CL
1385static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size)
1386{
1387 struct rpc_xprt *xprt;
ffc2e518 1388 struct sock_xprt *new;
c8541ecd
CL
1389
1390 if (addrlen > sizeof(xprt->addr)) {
1391 dprintk("RPC: xs_setup_xprt: address too large\n");
1392 return ERR_PTR(-EBADF);
1393 }
1394
ffc2e518
CL
1395 new = kzalloc(sizeof(*new), GFP_KERNEL);
1396 if (new == NULL) {
c8541ecd
CL
1397 dprintk("RPC: xs_setup_xprt: couldn't allocate rpc_xprt\n");
1398 return ERR_PTR(-ENOMEM);
1399 }
ffc2e518 1400 xprt = &new->xprt;
c8541ecd
CL
1401
1402 xprt->max_reqs = slot_table_size;
1403 xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL);
1404 if (xprt->slot == NULL) {
1405 kfree(xprt);
1406 dprintk("RPC: xs_setup_xprt: couldn't allocate slot table\n");
1407 return ERR_PTR(-ENOMEM);
1408 }
1409
1410 memcpy(&xprt->addr, addr, addrlen);
1411 xprt->addrlen = addrlen;
c8475461 1412 new->port = xs_get_random_port();
c8541ecd
CL
1413
1414 return xprt;
1415}
1416
9903cd1c
CL
1417/**
1418 * xs_setup_udp - Set up transport to use a UDP socket
c8541ecd
CL
1419 * @addr: address of remote server
1420 * @addrlen: length of address in bytes
9903cd1c
CL
1421 * @to: timeout parameters
1422 *
1423 */
c8541ecd 1424struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
a246b010 1425{
c8541ecd 1426 struct rpc_xprt *xprt;
c8475461 1427 struct sock_xprt *transport;
a246b010 1428
c8541ecd
CL
1429 xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries);
1430 if (IS_ERR(xprt))
1431 return xprt;
c8475461 1432 transport = container_of(xprt, struct sock_xprt, xprt);
a246b010 1433
c8541ecd 1434 if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
ec739ef0 1435 xprt_set_bound(xprt);
ec739ef0
CL
1436
1437 xprt->prot = IPPROTO_UDP;
808012fb 1438 xprt->tsh_size = 0;
a246b010
CL
1439 /* XXX: header size can vary due to auth type, IPv6, etc. */
1440 xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
1441
c8475461 1442 INIT_WORK(&transport->connect_worker, xs_udp_connect_worker, transport);
03bf4b70
CL
1443 xprt->bind_timeout = XS_BIND_TO;
1444 xprt->connect_timeout = XS_UDP_CONN_TO;
1445 xprt->reestablish_timeout = XS_UDP_REEST_TO;
1446 xprt->idle_timeout = XS_IDLE_DISC_TO;
a246b010 1447
262965f5 1448 xprt->ops = &xs_udp_ops;
a246b010
CL
1449
1450 if (to)
1451 xprt->timeout = *to;
1452 else
9903cd1c 1453 xprt_set_timeout(&xprt->timeout, 5, 5 * HZ);
a246b010 1454
edb267a6
CL
1455 xs_format_peer_addresses(xprt);
1456 dprintk("RPC: set up transport to address %s\n",
7559c7a2 1457 xprt->address_strings[RPC_DISPLAY_ALL]);
edb267a6 1458
c8541ecd 1459 return xprt;
a246b010
CL
1460}
1461
9903cd1c
CL
1462/**
1463 * xs_setup_tcp - Set up transport to use a TCP socket
c8541ecd
CL
1464 * @addr: address of remote server
1465 * @addrlen: length of address in bytes
9903cd1c
CL
1466 * @to: timeout parameters
1467 *
1468 */
c8541ecd 1469struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
a246b010 1470{
c8541ecd 1471 struct rpc_xprt *xprt;
c8475461 1472 struct sock_xprt *transport;
a246b010 1473
c8541ecd
CL
1474 xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries);
1475 if (IS_ERR(xprt))
1476 return xprt;
c8475461 1477 transport = container_of(xprt, struct sock_xprt, xprt);
a246b010 1478
c8541ecd 1479 if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
ec739ef0 1480 xprt_set_bound(xprt);
ec739ef0
CL
1481
1482 xprt->prot = IPPROTO_TCP;
808012fb 1483 xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
808012fb 1484 xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
a246b010 1485
c8475461 1486 INIT_WORK(&transport->connect_worker, xs_tcp_connect_worker, transport);
03bf4b70
CL
1487 xprt->bind_timeout = XS_BIND_TO;
1488 xprt->connect_timeout = XS_TCP_CONN_TO;
1489 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
1490 xprt->idle_timeout = XS_IDLE_DISC_TO;
a246b010 1491
262965f5 1492 xprt->ops = &xs_tcp_ops;
a246b010
CL
1493
1494 if (to)
1495 xprt->timeout = *to;
1496 else
9903cd1c 1497 xprt_set_timeout(&xprt->timeout, 2, 60 * HZ);
a246b010 1498
edb267a6
CL
1499 xs_format_peer_addresses(xprt);
1500 dprintk("RPC: set up transport to address %s\n",
7559c7a2 1501 xprt->address_strings[RPC_DISPLAY_ALL]);
edb267a6 1502
c8541ecd 1503 return xprt;
a246b010 1504}