Linux Kernel  3.7.1
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Groups Pages
xprtsock.c
Go to the documentation of this file.
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
7  * TCP send fixes (C) 1998 Red Hat
8  * TCP NFS related read + write fixes
9  * (C) 1999 Dave Airlie, University of Limerick, Ireland <[email protected]>
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 <[email protected]>
14  *
15  * IP socket transport implementation, (C) 2005 Chuck Lever <[email protected]>
16  *
17  * IPv6 support contributed by Gilles Quillard, Bull Open Source, 2005.
18  * <[email protected]>
19  */
20 
21 #include <linux/types.h>
22 #include <linux/string.h>
23 #include <linux/slab.h>
24 #include <linux/module.h>
25 #include <linux/capability.h>
26 #include <linux/pagemap.h>
27 #include <linux/errno.h>
28 #include <linux/socket.h>
29 #include <linux/in.h>
30 #include <linux/net.h>
31 #include <linux/mm.h>
32 #include <linux/un.h>
33 #include <linux/udp.h>
34 #include <linux/tcp.h>
35 #include <linux/sunrpc/clnt.h>
36 #include <linux/sunrpc/sched.h>
37 #include <linux/sunrpc/svcsock.h>
38 #include <linux/sunrpc/xprtsock.h>
39 #include <linux/file.h>
40 #ifdef CONFIG_SUNRPC_BACKCHANNEL
41 #include <linux/sunrpc/bc_xprt.h>
42 #endif
43 
44 #include <net/sock.h>
45 #include <net/checksum.h>
46 #include <net/udp.h>
47 #include <net/tcp.h>
48 
49 #include "sunrpc.h"
50 
51 static void xs_close(struct rpc_xprt *xprt);
52 
53 /*
54  * xprtsock tunables
55  */
56 static unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
57 static unsigned int xprt_tcp_slot_table_entries = RPC_MIN_SLOT_TABLE;
58 static unsigned int xprt_max_tcp_slot_table_entries = RPC_MAX_SLOT_TABLE;
59 
60 static unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
61 static unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
62 
63 #define XS_TCP_LINGER_TO (15U * HZ)
64 static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
65 
66 /*
67  * We can register our own files under /proc/sys/sunrpc by
68  * calling register_sysctl_table() again. The files in that
69  * directory become the union of all files registered there.
70  *
71  * We simply need to make sure that we don't collide with
72  * someone else's file names!
73  */
74 
75 #ifdef RPC_DEBUG
76 
77 static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
78 static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
79 static unsigned int max_tcp_slot_table_limit = RPC_MAX_SLOT_TABLE_LIMIT;
80 static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
81 static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
82 
83 static struct ctl_table_header *sunrpc_table_header;
84 
85 /*
86  * FIXME: changing the UDP slot table size should also resize the UDP
87  * socket buffers for existing UDP transports
88  */
89 static ctl_table xs_tunables_table[] = {
90  {
91  .procname = "udp_slot_table_entries",
92  .data = &xprt_udp_slot_table_entries,
93  .maxlen = sizeof(unsigned int),
94  .mode = 0644,
95  .proc_handler = proc_dointvec_minmax,
96  .extra1 = &min_slot_table_size,
97  .extra2 = &max_slot_table_size
98  },
99  {
100  .procname = "tcp_slot_table_entries",
101  .data = &xprt_tcp_slot_table_entries,
102  .maxlen = sizeof(unsigned int),
103  .mode = 0644,
104  .proc_handler = proc_dointvec_minmax,
105  .extra1 = &min_slot_table_size,
106  .extra2 = &max_slot_table_size
107  },
108  {
109  .procname = "tcp_max_slot_table_entries",
110  .data = &xprt_max_tcp_slot_table_entries,
111  .maxlen = sizeof(unsigned int),
112  .mode = 0644,
113  .proc_handler = proc_dointvec_minmax,
114  .extra1 = &min_slot_table_size,
115  .extra2 = &max_tcp_slot_table_limit
116  },
117  {
118  .procname = "min_resvport",
119  .data = &xprt_min_resvport,
120  .maxlen = sizeof(unsigned int),
121  .mode = 0644,
122  .proc_handler = proc_dointvec_minmax,
123  .extra1 = &xprt_min_resvport_limit,
124  .extra2 = &xprt_max_resvport_limit
125  },
126  {
127  .procname = "max_resvport",
128  .data = &xprt_max_resvport,
129  .maxlen = sizeof(unsigned int),
130  .mode = 0644,
131  .proc_handler = proc_dointvec_minmax,
132  .extra1 = &xprt_min_resvport_limit,
133  .extra2 = &xprt_max_resvport_limit
134  },
135  {
136  .procname = "tcp_fin_timeout",
137  .data = &xs_tcp_fin_timeout,
138  .maxlen = sizeof(xs_tcp_fin_timeout),
139  .mode = 0644,
140  .proc_handler = proc_dointvec_jiffies,
141  },
142  { },
143 };
144 
145 static ctl_table sunrpc_table[] = {
146  {
147  .procname = "sunrpc",
148  .mode = 0555,
149  .child = xs_tunables_table
150  },
151  { },
152 };
153 
154 #endif
155 
156 /*
157  * Wait duration for a reply from the RPC portmapper.
158  */
159 #define XS_BIND_TO (60U * HZ)
160 
161 /*
162  * Delay if a UDP socket connect error occurs. This is most likely some
163  * kind of resource problem on the local host.
164  */
165 #define XS_UDP_REEST_TO (2U * HZ)
166 
167 /*
168  * The reestablish timeout allows clients to delay for a bit before attempting
169  * to reconnect to a server that just dropped our connection.
170  *
171  * We implement an exponential backoff when trying to reestablish a TCP
172  * transport connection with the server. Some servers like to drop a TCP
173  * connection when they are overworked, so we start with a short timeout and
174  * increase over time if the server is down or not responding.
175  */
176 #define XS_TCP_INIT_REEST_TO (3U * HZ)
177 #define XS_TCP_MAX_REEST_TO (5U * 60 * HZ)
178 
179 /*
180  * TCP idle timeout; client drops the transport socket if it is idle
181  * for this long. Note that we also timeout UDP sockets to prevent
182  * holding port numbers when there is no RPC traffic.
183  */
184 #define XS_IDLE_DISC_TO (5U * 60 * HZ)
185 
186 #ifdef RPC_DEBUG
187 # undef RPC_DEBUG_DATA
188 # define RPCDBG_FACILITY RPCDBG_TRANS
189 #endif
190 
191 #ifdef RPC_DEBUG_DATA
192 static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
193 {
194  u8 *buf = (u8 *) packet;
195  int j;
196 
197  dprintk("RPC: %s\n", msg);
198  for (j = 0; j < count && j < 128; j += 4) {
199  if (!(j & 31)) {
200  if (j)
201  dprintk("\n");
202  dprintk("0x%04x ", j);
203  }
204  dprintk("%02x%02x%02x%02x ",
205  buf[j], buf[j+1], buf[j+2], buf[j+3]);
206  }
207  dprintk("\n");
208 }
209 #else
210 static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
211 {
212  /* NOP */
213 }
214 #endif
215 
216 struct sock_xprt {
217  struct rpc_xprt xprt;
218 
219  /*
220  * Network layer
221  */
222  struct socket * sock;
223  struct sock * inet;
224 
225  /*
226  * State of TCP reply receive
227  */
228  __be32 tcp_fraghdr,
229  tcp_xid,
230  tcp_calldir;
231 
232  u32 tcp_offset,
233  tcp_reclen;
234 
235  unsigned long tcp_copied,
236  tcp_flags;
237 
238  /*
239  * Connection of transports
240  */
241  struct delayed_work connect_worker;
242  struct sockaddr_storage srcaddr;
243  unsigned short srcport;
244 
245  /*
246  * UDP socket buffer size parameters
247  */
248  size_t rcvsize,
249  sndsize;
250 
251  /*
252  * Saved socket callback addresses
253  */
254  void (*old_data_ready)(struct sock *, int);
255  void (*old_state_change)(struct sock *);
256  void (*old_write_space)(struct sock *);
257 };
258 
259 /*
260  * TCP receive state flags
261  */
262 #define TCP_RCV_LAST_FRAG (1UL << 0)
263 #define TCP_RCV_COPY_FRAGHDR (1UL << 1)
264 #define TCP_RCV_COPY_XID (1UL << 2)
265 #define TCP_RCV_COPY_DATA (1UL << 3)
266 #define TCP_RCV_READ_CALLDIR (1UL << 4)
267 #define TCP_RCV_COPY_CALLDIR (1UL << 5)
268 
269 /*
270  * TCP RPC flags
271  */
272 #define TCP_RPC_REPLY (1UL << 6)
273 
274 static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
275 {
276  return (struct sockaddr *) &xprt->addr;
277 }
278 
279 static inline struct sockaddr_un *xs_addr_un(struct rpc_xprt *xprt)
280 {
281  return (struct sockaddr_un *) &xprt->addr;
282 }
283 
284 static inline struct sockaddr_in *xs_addr_in(struct rpc_xprt *xprt)
285 {
286  return (struct sockaddr_in *) &xprt->addr;
287 }
288 
289 static inline struct sockaddr_in6 *xs_addr_in6(struct rpc_xprt *xprt)
290 {
291  return (struct sockaddr_in6 *) &xprt->addr;
292 }
293 
294 static void xs_format_common_peer_addresses(struct rpc_xprt *xprt)
295 {
296  struct sockaddr *sap = xs_addr(xprt);
297  struct sockaddr_in6 *sin6;
298  struct sockaddr_in *sin;
299  struct sockaddr_un *sun;
300  char buf[128];
301 
302  switch (sap->sa_family) {
303  case AF_LOCAL:
304  sun = xs_addr_un(xprt);
305  strlcpy(buf, sun->sun_path, sizeof(buf));
306  xprt->address_strings[RPC_DISPLAY_ADDR] =
307  kstrdup(buf, GFP_KERNEL);
308  break;
309  case AF_INET:
310  (void)rpc_ntop(sap, buf, sizeof(buf));
311  xprt->address_strings[RPC_DISPLAY_ADDR] =
312  kstrdup(buf, GFP_KERNEL);
313  sin = xs_addr_in(xprt);
314  snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
315  break;
316  case AF_INET6:
317  (void)rpc_ntop(sap, buf, sizeof(buf));
318  xprt->address_strings[RPC_DISPLAY_ADDR] =
319  kstrdup(buf, GFP_KERNEL);
320  sin6 = xs_addr_in6(xprt);
321  snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
322  break;
323  default:
324  BUG();
325  }
326 
327  xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
328 }
329 
330 static void xs_format_common_peer_ports(struct rpc_xprt *xprt)
331 {
332  struct sockaddr *sap = xs_addr(xprt);
333  char buf[128];
334 
335  snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
336  xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
337 
338  snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
339  xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
340 }
341 
342 static void xs_format_peer_addresses(struct rpc_xprt *xprt,
343  const char *protocol,
344  const char *netid)
345 {
346  xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
347  xprt->address_strings[RPC_DISPLAY_NETID] = netid;
348  xs_format_common_peer_addresses(xprt);
349  xs_format_common_peer_ports(xprt);
350 }
351 
352 static void xs_update_peer_port(struct rpc_xprt *xprt)
353 {
354  kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]);
355  kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
356 
357  xs_format_common_peer_ports(xprt);
358 }
359 
360 static void xs_free_peer_addresses(struct rpc_xprt *xprt)
361 {
362  unsigned int i;
363 
364  for (i = 0; i < RPC_DISPLAY_MAX; i++)
365  switch (i) {
366  case RPC_DISPLAY_PROTO:
367  case RPC_DISPLAY_NETID:
368  continue;
369  default:
370  kfree(xprt->address_strings[i]);
371  }
372 }
373 
374 #define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL)
375 
376 static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
377 {
378  struct msghdr msg = {
379  .msg_name = addr,
380  .msg_namelen = addrlen,
381  .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
382  };
383  struct kvec iov = {
384  .iov_base = vec->iov_base + base,
385  .iov_len = vec->iov_len - base,
386  };
387 
388  if (iov.iov_len != 0)
389  return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
390  return kernel_sendmsg(sock, &msg, NULL, 0, 0);
391 }
392 
393 static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
394 {
395  struct page **ppage;
396  unsigned int remainder;
397  int err, sent = 0;
398 
399  remainder = xdr->page_len - base;
400  base += xdr->page_base;
401  ppage = xdr->pages + (base >> PAGE_SHIFT);
402  base &= ~PAGE_MASK;
403  for(;;) {
404  unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
405  int flags = XS_SENDMSG_FLAGS;
406 
407  remainder -= len;
408  if (remainder != 0 || more)
409  flags |= MSG_MORE;
410  err = sock->ops->sendpage(sock, *ppage, base, len, flags);
411  if (remainder == 0 || err != len)
412  break;
413  sent += err;
414  ppage++;
415  base = 0;
416  }
417  if (sent == 0)
418  return err;
419  if (err > 0)
420  sent += err;
421  return sent;
422 }
423 
433 static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
434 {
435  unsigned int remainder = xdr->len - base;
436  int err, sent = 0;
437 
438  if (unlikely(!sock))
439  return -ENOTSOCK;
440 
441  clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
442  if (base != 0) {
443  addr = NULL;
444  addrlen = 0;
445  }
446 
447  if (base < xdr->head[0].iov_len || addr != NULL) {
448  unsigned int len = xdr->head[0].iov_len - base;
449  remainder -= len;
450  err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
451  if (remainder == 0 || err != len)
452  goto out;
453  sent += err;
454  base = 0;
455  } else
456  base -= xdr->head[0].iov_len;
457 
458  if (base < xdr->page_len) {
459  unsigned int len = xdr->page_len - base;
460  remainder -= len;
461  err = xs_send_pagedata(sock, xdr, base, remainder != 0);
462  if (remainder == 0 || err != len)
463  goto out;
464  sent += err;
465  base = 0;
466  } else
467  base -= xdr->page_len;
468 
469  if (base >= xdr->tail[0].iov_len)
470  return sent;
471  err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
472 out:
473  if (sent == 0)
474  return err;
475  if (err > 0)
476  sent += err;
477  return sent;
478 }
479 
480 static void xs_nospace_callback(struct rpc_task *task)
481 {
482  struct sock_xprt *transport = container_of(task->tk_rqstp->rq_xprt, struct sock_xprt, xprt);
483 
484  transport->inet->sk_write_pending--;
485  clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
486 }
487 
493 static int xs_nospace(struct rpc_task *task)
494 {
495  struct rpc_rqst *req = task->tk_rqstp;
496  struct rpc_xprt *xprt = req->rq_xprt;
497  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
498  int ret = -EAGAIN;
499 
500  dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
501  task->tk_pid, req->rq_slen - req->rq_bytes_sent,
502  req->rq_slen);
503 
504  /* Protect against races with write_space */
505  spin_lock_bh(&xprt->transport_lock);
506 
507  /* Don't race with disconnect */
508  if (xprt_connected(xprt)) {
509  if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
510  /*
511  * Notify TCP that we're limited by the application
512  * window size
513  */
514  set_bit(SOCK_NOSPACE, &transport->sock->flags);
515  transport->inet->sk_write_pending++;
516  /* ...and wait for more buffer space */
517  xprt_wait_for_buffer_space(task, xs_nospace_callback);
518  }
519  } else {
520  clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
521  ret = -ENOTCONN;
522  }
523 
524  spin_unlock_bh(&xprt->transport_lock);
525  return ret;
526 }
527 
528 /*
529  * Construct a stream transport record marker in @buf.
530  */
531 static inline void xs_encode_stream_record_marker(struct xdr_buf *buf)
532 {
533  u32 reclen = buf->len - sizeof(rpc_fraghdr);
534  rpc_fraghdr *base = buf->head[0].iov_base;
535  *base = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT | reclen);
536 }
537 
549 static int xs_local_send_request(struct rpc_task *task)
550 {
551  struct rpc_rqst *req = task->tk_rqstp;
552  struct rpc_xprt *xprt = req->rq_xprt;
553  struct sock_xprt *transport =
554  container_of(xprt, struct sock_xprt, xprt);
555  struct xdr_buf *xdr = &req->rq_snd_buf;
556  int status;
557 
558  xs_encode_stream_record_marker(&req->rq_snd_buf);
559 
560  xs_pktdump("packet data:",
561  req->rq_svec->iov_base, req->rq_svec->iov_len);
562 
563  status = xs_sendpages(transport->sock, NULL, 0,
564  xdr, req->rq_bytes_sent);
565  dprintk("RPC: %s(%u) = %d\n",
566  __func__, xdr->len - req->rq_bytes_sent, status);
567  if (likely(status >= 0)) {
568  req->rq_bytes_sent += status;
569  req->rq_xmit_bytes_sent += status;
570  if (likely(req->rq_bytes_sent >= req->rq_slen)) {
571  req->rq_bytes_sent = 0;
572  return 0;
573  }
574  status = -EAGAIN;
575  }
576 
577  switch (status) {
578  case -EAGAIN:
579  status = xs_nospace(task);
580  break;
581  default:
582  dprintk("RPC: sendmsg returned unrecognized error %d\n",
583  -status);
584  case -EPIPE:
585  xs_close(xprt);
586  status = -ENOTCONN;
587  }
588 
589  return status;
590 }
591 
603 static int xs_udp_send_request(struct rpc_task *task)
604 {
605  struct rpc_rqst *req = task->tk_rqstp;
606  struct rpc_xprt *xprt = req->rq_xprt;
607  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
608  struct xdr_buf *xdr = &req->rq_snd_buf;
609  int status;
610 
611  xs_pktdump("packet data:",
612  req->rq_svec->iov_base,
613  req->rq_svec->iov_len);
614 
615  if (!xprt_bound(xprt))
616  return -ENOTCONN;
617  status = xs_sendpages(transport->sock,
618  xs_addr(xprt),
619  xprt->addrlen, xdr,
620  req->rq_bytes_sent);
621 
622  dprintk("RPC: xs_udp_send_request(%u) = %d\n",
623  xdr->len - req->rq_bytes_sent, status);
624 
625  if (status >= 0) {
626  req->rq_xmit_bytes_sent += status;
627  if (status >= req->rq_slen)
628  return 0;
629  /* Still some bytes left; set up for a retry later. */
630  status = -EAGAIN;
631  }
632 
633  switch (status) {
634  case -ENOTSOCK:
635  status = -ENOTCONN;
636  /* Should we call xs_close() here? */
637  break;
638  case -EAGAIN:
639  status = xs_nospace(task);
640  break;
641  default:
642  dprintk("RPC: sendmsg returned unrecognized error %d\n",
643  -status);
644  case -ENETUNREACH:
645  case -EPIPE:
646  case -ECONNREFUSED:
647  /* When the server has died, an ICMP port unreachable message
648  * prompts ECONNREFUSED. */
649  clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
650  }
651 
652  return status;
653 }
654 
662 static void xs_tcp_shutdown(struct rpc_xprt *xprt)
663 {
664  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
665  struct socket *sock = transport->sock;
666 
667  if (sock != NULL)
668  kernel_sock_shutdown(sock, SHUT_WR);
669 }
670 
685 static int xs_tcp_send_request(struct rpc_task *task)
686 {
687  struct rpc_rqst *req = task->tk_rqstp;
688  struct rpc_xprt *xprt = req->rq_xprt;
689  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
690  struct xdr_buf *xdr = &req->rq_snd_buf;
691  int status;
692 
693  xs_encode_stream_record_marker(&req->rq_snd_buf);
694 
695  xs_pktdump("packet data:",
696  req->rq_svec->iov_base,
697  req->rq_svec->iov_len);
698 
699  /* Continue transmitting the packet/record. We must be careful
700  * to cope with writespace callbacks arriving _after_ we have
701  * called sendmsg(). */
702  while (1) {
703  status = xs_sendpages(transport->sock,
704  NULL, 0, xdr, req->rq_bytes_sent);
705 
706  dprintk("RPC: xs_tcp_send_request(%u) = %d\n",
707  xdr->len - req->rq_bytes_sent, status);
708 
709  if (unlikely(status < 0))
710  break;
711 
712  /* If we've sent the entire packet, immediately
713  * reset the count of bytes sent. */
714  req->rq_bytes_sent += status;
715  req->rq_xmit_bytes_sent += status;
716  if (likely(req->rq_bytes_sent >= req->rq_slen)) {
717  req->rq_bytes_sent = 0;
718  return 0;
719  }
720 
721  if (status != 0)
722  continue;
723  status = -EAGAIN;
724  break;
725  }
726 
727  switch (status) {
728  case -ENOTSOCK:
729  status = -ENOTCONN;
730  /* Should we call xs_close() here? */
731  break;
732  case -EAGAIN:
733  status = xs_nospace(task);
734  break;
735  default:
736  dprintk("RPC: sendmsg returned unrecognized error %d\n",
737  -status);
738  case -ECONNRESET:
739  xs_tcp_shutdown(xprt);
740  case -ECONNREFUSED:
741  case -ENOTCONN:
742  case -EPIPE:
743  clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
744  }
745 
746  return status;
747 }
748 
758 static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
759 {
760  struct rpc_rqst *req;
761 
762  if (task != xprt->snd_task)
763  return;
764  if (task == NULL)
765  goto out_release;
766  req = task->tk_rqstp;
767  if (req == NULL)
768  goto out_release;
769  if (req->rq_bytes_sent == 0)
770  goto out_release;
771  if (req->rq_bytes_sent == req->rq_snd_buf.len)
772  goto out_release;
773  set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state);
774 out_release:
775  xprt_release_xprt(xprt, task);
776 }
777 
778 static void xs_save_old_callbacks(struct sock_xprt *transport, struct sock *sk)
779 {
780  transport->old_data_ready = sk->sk_data_ready;
781  transport->old_state_change = sk->sk_state_change;
782  transport->old_write_space = sk->sk_write_space;
783 }
784 
785 static void xs_restore_old_callbacks(struct sock_xprt *transport, struct sock *sk)
786 {
787  sk->sk_data_ready = transport->old_data_ready;
788  sk->sk_state_change = transport->old_state_change;
789  sk->sk_write_space = transport->old_write_space;
790 }
791 
792 static void xs_reset_transport(struct sock_xprt *transport)
793 {
794  struct socket *sock = transport->sock;
795  struct sock *sk = transport->inet;
796 
797  if (sk == NULL)
798  return;
799 
800  transport->srcport = 0;
801 
802  write_lock_bh(&sk->sk_callback_lock);
803  transport->inet = NULL;
804  transport->sock = NULL;
805 
806  sk->sk_user_data = NULL;
807 
808  xs_restore_old_callbacks(transport, sk);
809  write_unlock_bh(&sk->sk_callback_lock);
810 
811  sk->sk_no_check = 0;
812 
813  sock_release(sock);
814 }
815 
826 static void xs_close(struct rpc_xprt *xprt)
827 {
828  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
829 
830  dprintk("RPC: xs_close xprt %p\n", xprt);
831 
832  xs_reset_transport(transport);
833  xprt->reestablish_timeout = 0;
834 
835  smp_mb__before_clear_bit();
836  clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
837  clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
838  clear_bit(XPRT_CLOSING, &xprt->state);
839  smp_mb__after_clear_bit();
840  xprt_disconnect_done(xprt);
841 }
842 
843 static void xs_tcp_close(struct rpc_xprt *xprt)
844 {
845  if (test_and_clear_bit(XPRT_CONNECTION_CLOSE, &xprt->state))
846  xs_close(xprt);
847  else
848  xs_tcp_shutdown(xprt);
849 }
850 
856 static void xs_destroy(struct rpc_xprt *xprt)
857 {
858  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
859 
860  dprintk("RPC: xs_destroy xprt %p\n", xprt);
861 
862  cancel_delayed_work_sync(&transport->connect_worker);
863 
864  xs_close(xprt);
865  xs_free_peer_addresses(xprt);
866  xprt_free(xprt);
867  module_put(THIS_MODULE);
868 }
869 
870 static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
871 {
872  return (struct rpc_xprt *) sk->sk_user_data;
873 }
874 
875 static int xs_local_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb)
876 {
877  struct xdr_skb_reader desc = {
878  .skb = skb,
879  .offset = sizeof(rpc_fraghdr),
880  .count = skb->len - sizeof(rpc_fraghdr),
881  };
882 
883  if (xdr_partial_copy_from_skb(xdr, 0, &desc, xdr_skb_read_bits) < 0)
884  return -1;
885  if (desc.count)
886  return -1;
887  return 0;
888 }
889 
897 static void xs_local_data_ready(struct sock *sk, int len)
898 {
899  struct rpc_task *task;
900  struct rpc_xprt *xprt;
901  struct rpc_rqst *rovr;
902  struct sk_buff *skb;
903  int err, repsize, copied;
904  u32 _xid;
905  __be32 *xp;
906 
907  read_lock_bh(&sk->sk_callback_lock);
908  dprintk("RPC: %s...\n", __func__);
909  xprt = xprt_from_sock(sk);
910  if (xprt == NULL)
911  goto out;
912 
913  skb = skb_recv_datagram(sk, 0, 1, &err);
914  if (skb == NULL)
915  goto out;
916 
917  repsize = skb->len - sizeof(rpc_fraghdr);
918  if (repsize < 4) {
919  dprintk("RPC: impossible RPC reply size %d\n", repsize);
920  goto dropit;
921  }
922 
923  /* Copy the XID from the skb... */
924  xp = skb_header_pointer(skb, sizeof(rpc_fraghdr), sizeof(_xid), &_xid);
925  if (xp == NULL)
926  goto dropit;
927 
928  /* Look up and lock the request corresponding to the given XID */
929  spin_lock(&xprt->transport_lock);
930  rovr = xprt_lookup_rqst(xprt, *xp);
931  if (!rovr)
932  goto out_unlock;
933  task = rovr->rq_task;
934 
935  copied = rovr->rq_private_buf.buflen;
936  if (copied > repsize)
937  copied = repsize;
938 
939  if (xs_local_copy_to_xdr(&rovr->rq_private_buf, skb)) {
940  dprintk("RPC: sk_buff copy failed\n");
941  goto out_unlock;
942  }
943 
944  xprt_complete_rqst(task, copied);
945 
946  out_unlock:
947  spin_unlock(&xprt->transport_lock);
948  dropit:
949  skb_free_datagram(sk, skb);
950  out:
951  read_unlock_bh(&sk->sk_callback_lock);
952 }
953 
960 static void xs_udp_data_ready(struct sock *sk, int len)
961 {
962  struct rpc_task *task;
963  struct rpc_xprt *xprt;
964  struct rpc_rqst *rovr;
965  struct sk_buff *skb;
966  int err, repsize, copied;
967  u32 _xid;
968  __be32 *xp;
969 
970  read_lock_bh(&sk->sk_callback_lock);
971  dprintk("RPC: xs_udp_data_ready...\n");
972  if (!(xprt = xprt_from_sock(sk)))
973  goto out;
974 
975  if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
976  goto out;
977 
978  repsize = skb->len - sizeof(struct udphdr);
979  if (repsize < 4) {
980  dprintk("RPC: impossible RPC reply size %d!\n", repsize);
981  goto dropit;
982  }
983 
984  /* Copy the XID from the skb... */
985  xp = skb_header_pointer(skb, sizeof(struct udphdr),
986  sizeof(_xid), &_xid);
987  if (xp == NULL)
988  goto dropit;
989 
990  /* Look up and lock the request corresponding to the given XID */
991  spin_lock(&xprt->transport_lock);
992  rovr = xprt_lookup_rqst(xprt, *xp);
993  if (!rovr)
994  goto out_unlock;
995  task = rovr->rq_task;
996 
997  if ((copied = rovr->rq_private_buf.buflen) > repsize)
998  copied = repsize;
999 
1000  /* Suck it into the iovec, verify checksum if not done by hw. */
1001  if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
1002  UDPX_INC_STATS_BH(sk, UDP_MIB_INERRORS);
1003  goto out_unlock;
1004  }
1005 
1006  UDPX_INC_STATS_BH(sk, UDP_MIB_INDATAGRAMS);
1007 
1008  xprt_adjust_cwnd(task, copied);
1009  xprt_complete_rqst(task, copied);
1010 
1011  out_unlock:
1012  spin_unlock(&xprt->transport_lock);
1013  dropit:
1014  skb_free_datagram(sk, skb);
1015  out:
1016  read_unlock_bh(&sk->sk_callback_lock);
1017 }
1018 
1019 /*
1020  * Helper function to force a TCP close if the server is sending
1021  * junk and/or it has put us in CLOSE_WAIT
1022  */
1023 static void xs_tcp_force_close(struct rpc_xprt *xprt)
1024 {
1025  set_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
1026  xprt_force_disconnect(xprt);
1027 }
1028 
1029 static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
1030 {
1031  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1032  size_t len, used;
1033  char *p;
1034 
1035  p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
1036  len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
1037  used = xdr_skb_read_bits(desc, p, len);
1038  transport->tcp_offset += used;
1039  if (used != len)
1040  return;
1041 
1042  transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
1043  if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
1044  transport->tcp_flags |= TCP_RCV_LAST_FRAG;
1045  else
1046  transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
1047  transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
1048 
1049  transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
1050  transport->tcp_offset = 0;
1051 
1052  /* Sanity check of the record length */
1053  if (unlikely(transport->tcp_reclen < 8)) {
1054  dprintk("RPC: invalid TCP record fragment length\n");
1055  xs_tcp_force_close(xprt);
1056  return;
1057  }
1058  dprintk("RPC: reading TCP record fragment of length %d\n",
1059  transport->tcp_reclen);
1060 }
1061 
1062 static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
1063 {
1064  if (transport->tcp_offset == transport->tcp_reclen) {
1065  transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
1066  transport->tcp_offset = 0;
1067  if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
1068  transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1069  transport->tcp_flags |= TCP_RCV_COPY_XID;
1070  transport->tcp_copied = 0;
1071  }
1072  }
1073 }
1074 
1075 static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
1076 {
1077  size_t len, used;
1078  char *p;
1079 
1080  len = sizeof(transport->tcp_xid) - transport->tcp_offset;
1081  dprintk("RPC: reading XID (%Zu bytes)\n", len);
1082  p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
1083  used = xdr_skb_read_bits(desc, p, len);
1084  transport->tcp_offset += used;
1085  if (used != len)
1086  return;
1087  transport->tcp_flags &= ~TCP_RCV_COPY_XID;
1088  transport->tcp_flags |= TCP_RCV_READ_CALLDIR;
1089  transport->tcp_copied = 4;
1090  dprintk("RPC: reading %s XID %08x\n",
1091  (transport->tcp_flags & TCP_RPC_REPLY) ? "reply for"
1092  : "request with",
1093  ntohl(transport->tcp_xid));
1094  xs_tcp_check_fraghdr(transport);
1095 }
1096 
1097 static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
1098  struct xdr_skb_reader *desc)
1099 {
1100  size_t len, used;
1101  u32 offset;
1102  char *p;
1103 
1104  /*
1105  * We want transport->tcp_offset to be 8 at the end of this routine
1106  * (4 bytes for the xid and 4 bytes for the call/reply flag).
1107  * When this function is called for the first time,
1108  * transport->tcp_offset is 4 (after having already read the xid).
1109  */
1110  offset = transport->tcp_offset - sizeof(transport->tcp_xid);
1111  len = sizeof(transport->tcp_calldir) - offset;
1112  dprintk("RPC: reading CALL/REPLY flag (%Zu bytes)\n", len);
1113  p = ((char *) &transport->tcp_calldir) + offset;
1114  used = xdr_skb_read_bits(desc, p, len);
1115  transport->tcp_offset += used;
1116  if (used != len)
1117  return;
1118  transport->tcp_flags &= ~TCP_RCV_READ_CALLDIR;
1119  /*
1120  * We don't yet have the XDR buffer, so we will write the calldir
1121  * out after we get the buffer from the 'struct rpc_rqst'
1122  */
1123  switch (ntohl(transport->tcp_calldir)) {
1124  case RPC_REPLY:
1125  transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
1126  transport->tcp_flags |= TCP_RCV_COPY_DATA;
1127  transport->tcp_flags |= TCP_RPC_REPLY;
1128  break;
1129  case RPC_CALL:
1130  transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
1131  transport->tcp_flags |= TCP_RCV_COPY_DATA;
1132  transport->tcp_flags &= ~TCP_RPC_REPLY;
1133  break;
1134  default:
1135  dprintk("RPC: invalid request message type\n");
1136  xs_tcp_force_close(&transport->xprt);
1137  }
1138  xs_tcp_check_fraghdr(transport);
1139 }
1140 
1141 static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
1142  struct xdr_skb_reader *desc,
1143  struct rpc_rqst *req)
1144 {
1145  struct sock_xprt *transport =
1146  container_of(xprt, struct sock_xprt, xprt);
1147  struct xdr_buf *rcvbuf;
1148  size_t len;
1149  ssize_t r;
1150 
1151  rcvbuf = &req->rq_private_buf;
1152 
1153  if (transport->tcp_flags & TCP_RCV_COPY_CALLDIR) {
1154  /*
1155  * Save the RPC direction in the XDR buffer
1156  */
1157  memcpy(rcvbuf->head[0].iov_base + transport->tcp_copied,
1158  &transport->tcp_calldir,
1159  sizeof(transport->tcp_calldir));
1160  transport->tcp_copied += sizeof(transport->tcp_calldir);
1161  transport->tcp_flags &= ~TCP_RCV_COPY_CALLDIR;
1162  }
1163 
1164  len = desc->count;
1165  if (len > transport->tcp_reclen - transport->tcp_offset) {
1166  struct xdr_skb_reader my_desc;
1167 
1168  len = transport->tcp_reclen - transport->tcp_offset;
1169  memcpy(&my_desc, desc, sizeof(my_desc));
1170  my_desc.count = len;
1171  r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
1172  &my_desc, xdr_skb_read_bits);
1173  desc->count -= r;
1174  desc->offset += r;
1175  } else
1176  r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
1177  desc, xdr_skb_read_bits);
1178 
1179  if (r > 0) {
1180  transport->tcp_copied += r;
1181  transport->tcp_offset += r;
1182  }
1183  if (r != len) {
1184  /* Error when copying to the receive buffer,
1185  * usually because we weren't able to allocate
1186  * additional buffer pages. All we can do now
1187  * is turn off TCP_RCV_COPY_DATA, so the request
1188  * will not receive any additional updates,
1189  * and time out.
1190  * Any remaining data from this record will
1191  * be discarded.
1192  */
1193  transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1194  dprintk("RPC: XID %08x truncated request\n",
1195  ntohl(transport->tcp_xid));
1196  dprintk("RPC: xprt = %p, tcp_copied = %lu, "
1197  "tcp_offset = %u, tcp_reclen = %u\n",
1198  xprt, transport->tcp_copied,
1199  transport->tcp_offset, transport->tcp_reclen);
1200  return;
1201  }
1202 
1203  dprintk("RPC: XID %08x read %Zd bytes\n",
1204  ntohl(transport->tcp_xid), r);
1205  dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
1206  "tcp_reclen = %u\n", xprt, transport->tcp_copied,
1207  transport->tcp_offset, transport->tcp_reclen);
1208 
1209  if (transport->tcp_copied == req->rq_private_buf.buflen)
1210  transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1211  else if (transport->tcp_offset == transport->tcp_reclen) {
1212  if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
1213  transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1214  }
1215 }
1216 
1217 /*
1218  * Finds the request corresponding to the RPC xid and invokes the common
1219  * tcp read code to read the data.
1220  */
1221 static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
1222  struct xdr_skb_reader *desc)
1223 {
1224  struct sock_xprt *transport =
1225  container_of(xprt, struct sock_xprt, xprt);
1226  struct rpc_rqst *req;
1227 
1228  dprintk("RPC: read reply XID %08x\n", ntohl(transport->tcp_xid));
1229 
1230  /* Find and lock the request corresponding to this xid */
1231  spin_lock(&xprt->transport_lock);
1232  req = xprt_lookup_rqst(xprt, transport->tcp_xid);
1233  if (!req) {
1234  dprintk("RPC: XID %08x request not found!\n",
1235  ntohl(transport->tcp_xid));
1236  spin_unlock(&xprt->transport_lock);
1237  return -1;
1238  }
1239 
1240  xs_tcp_read_common(xprt, desc, req);
1241 
1242  if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
1243  xprt_complete_rqst(req->rq_task, transport->tcp_copied);
1244 
1245  spin_unlock(&xprt->transport_lock);
1246  return 0;
1247 }
1248 
1249 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1250 /*
1251  * Obtains an rpc_rqst previously allocated and invokes the common
1252  * tcp read code to read the data. The result is placed in the callback
1253  * queue.
1254  * If we're unable to obtain the rpc_rqst we schedule the closing of the
1255  * connection and return -1.
1256  */
1257 static inline int xs_tcp_read_callback(struct rpc_xprt *xprt,
1258  struct xdr_skb_reader *desc)
1259 {
1260  struct sock_xprt *transport =
1261  container_of(xprt, struct sock_xprt, xprt);
1262  struct rpc_rqst *req;
1263 
1264  req = xprt_alloc_bc_request(xprt);
1265  if (req == NULL) {
1266  printk(KERN_WARNING "Callback slot table overflowed\n");
1267  xprt_force_disconnect(xprt);
1268  return -1;
1269  }
1270 
1271  req->rq_xid = transport->tcp_xid;
1272  dprintk("RPC: read callback XID %08x\n", ntohl(req->rq_xid));
1273  xs_tcp_read_common(xprt, desc, req);
1274 
1275  if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) {
1276  struct svc_serv *bc_serv = xprt->bc_serv;
1277 
1278  /*
1279  * Add callback request to callback list. The callback
1280  * service sleeps on the sv_cb_waitq waiting for new
1281  * requests. Wake it up after adding enqueing the
1282  * request.
1283  */
1284  dprintk("RPC: add callback request to list\n");
1285  spin_lock(&bc_serv->sv_cb_lock);
1286  list_add(&req->rq_bc_list, &bc_serv->sv_cb_list);
1287  spin_unlock(&bc_serv->sv_cb_lock);
1288  wake_up(&bc_serv->sv_cb_waitq);
1289  }
1290 
1291  req->rq_private_buf.len = transport->tcp_copied;
1292 
1293  return 0;
1294 }
1295 
1296 static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1297  struct xdr_skb_reader *desc)
1298 {
1299  struct sock_xprt *transport =
1300  container_of(xprt, struct sock_xprt, xprt);
1301 
1302  return (transport->tcp_flags & TCP_RPC_REPLY) ?
1303  xs_tcp_read_reply(xprt, desc) :
1304  xs_tcp_read_callback(xprt, desc);
1305 }
1306 #else
1307 static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1308  struct xdr_skb_reader *desc)
1309 {
1310  return xs_tcp_read_reply(xprt, desc);
1311 }
1312 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
1313 
1314 /*
1315  * Read data off the transport. This can be either an RPC_CALL or an
1316  * RPC_REPLY. Relay the processing to helper functions.
1317  */
1318 static void xs_tcp_read_data(struct rpc_xprt *xprt,
1319  struct xdr_skb_reader *desc)
1320 {
1321  struct sock_xprt *transport =
1322  container_of(xprt, struct sock_xprt, xprt);
1323 
1324  if (_xs_tcp_read_data(xprt, desc) == 0)
1325  xs_tcp_check_fraghdr(transport);
1326  else {
1327  /*
1328  * The transport_lock protects the request handling.
1329  * There's no need to hold it to update the tcp_flags.
1330  */
1331  transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1332  }
1333 }
1334 
1335 static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
1336 {
1337  size_t len;
1338 
1339  len = transport->tcp_reclen - transport->tcp_offset;
1340  if (len > desc->count)
1341  len = desc->count;
1342  desc->count -= len;
1343  desc->offset += len;
1344  transport->tcp_offset += len;
1345  dprintk("RPC: discarded %Zu bytes\n", len);
1346  xs_tcp_check_fraghdr(transport);
1347 }
1348 
1349 static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
1350 {
1351  struct rpc_xprt *xprt = rd_desc->arg.data;
1352  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1353  struct xdr_skb_reader desc = {
1354  .skb = skb,
1355  .offset = offset,
1356  .count = len,
1357  };
1358 
1359  dprintk("RPC: xs_tcp_data_recv started\n");
1360  do {
1361  /* Read in a new fragment marker if necessary */
1362  /* Can we ever really expect to get completely empty fragments? */
1363  if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
1364  xs_tcp_read_fraghdr(xprt, &desc);
1365  continue;
1366  }
1367  /* Read in the xid if necessary */
1368  if (transport->tcp_flags & TCP_RCV_COPY_XID) {
1369  xs_tcp_read_xid(transport, &desc);
1370  continue;
1371  }
1372  /* Read in the call/reply flag */
1373  if (transport->tcp_flags & TCP_RCV_READ_CALLDIR) {
1374  xs_tcp_read_calldir(transport, &desc);
1375  continue;
1376  }
1377  /* Read in the request data */
1378  if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
1379  xs_tcp_read_data(xprt, &desc);
1380  continue;
1381  }
1382  /* Skip over any trailing bytes on short reads */
1383  xs_tcp_read_discard(transport, &desc);
1384  } while (desc.count);
1385  dprintk("RPC: xs_tcp_data_recv done\n");
1386  return len - desc.count;
1387 }
1388 
1395 static void xs_tcp_data_ready(struct sock *sk, int bytes)
1396 {
1397  struct rpc_xprt *xprt;
1398  read_descriptor_t rd_desc;
1399  int read;
1400 
1401  dprintk("RPC: xs_tcp_data_ready...\n");
1402 
1403  read_lock_bh(&sk->sk_callback_lock);
1404  if (!(xprt = xprt_from_sock(sk)))
1405  goto out;
1406  /* Any data means we had a useful conversation, so
1407  * the we don't need to delay the next reconnect
1408  */
1409  if (xprt->reestablish_timeout)
1410  xprt->reestablish_timeout = 0;
1411 
1412  /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
1413  rd_desc.arg.data = xprt;
1414  do {
1415  rd_desc.count = 65536;
1416  read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
1417  } while (read > 0);
1418 out:
1419  read_unlock_bh(&sk->sk_callback_lock);
1420 }
1421 
1422 /*
1423  * Do the equivalent of linger/linger2 handling for dealing with
1424  * broken servers that don't close the socket in a timely
1425  * fashion
1426  */
1427 static void xs_tcp_schedule_linger_timeout(struct rpc_xprt *xprt,
1428  unsigned long timeout)
1429 {
1430  struct sock_xprt *transport;
1431 
1432  if (xprt_test_and_set_connecting(xprt))
1433  return;
1434  set_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1435  transport = container_of(xprt, struct sock_xprt, xprt);
1436  queue_delayed_work(rpciod_workqueue, &transport->connect_worker,
1437  timeout);
1438 }
1439 
1440 static void xs_tcp_cancel_linger_timeout(struct rpc_xprt *xprt)
1441 {
1442  struct sock_xprt *transport;
1443 
1444  transport = container_of(xprt, struct sock_xprt, xprt);
1445 
1446  if (!test_bit(XPRT_CONNECTION_ABORT, &xprt->state) ||
1447  !cancel_delayed_work(&transport->connect_worker))
1448  return;
1449  clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1450  xprt_clear_connecting(xprt);
1451 }
1452 
1453 static void xs_sock_reset_connection_flags(struct rpc_xprt *xprt)
1454 {
1455  smp_mb__before_clear_bit();
1456  clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1457  clear_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
1458  clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1459  clear_bit(XPRT_CLOSING, &xprt->state);
1460  smp_mb__after_clear_bit();
1461 }
1462 
1463 static void xs_sock_mark_closed(struct rpc_xprt *xprt)
1464 {
1465  xs_sock_reset_connection_flags(xprt);
1466  /* Mark transport as closed and wake up all pending tasks */
1467  xprt_disconnect_done(xprt);
1468 }
1469 
1475 static void xs_tcp_state_change(struct sock *sk)
1476 {
1477  struct rpc_xprt *xprt;
1478 
1479  read_lock_bh(&sk->sk_callback_lock);
1480  if (!(xprt = xprt_from_sock(sk)))
1481  goto out;
1482  dprintk("RPC: xs_tcp_state_change client %p...\n", xprt);
1483  dprintk("RPC: state %x conn %d dead %d zapped %d sk_shutdown %d\n",
1484  sk->sk_state, xprt_connected(xprt),
1485  sock_flag(sk, SOCK_DEAD),
1486  sock_flag(sk, SOCK_ZAPPED),
1487  sk->sk_shutdown);
1488 
1489  switch (sk->sk_state) {
1490  case TCP_ESTABLISHED:
1491  spin_lock(&xprt->transport_lock);
1492  if (!xprt_test_and_set_connected(xprt)) {
1493  struct sock_xprt *transport = container_of(xprt,
1494  struct sock_xprt, xprt);
1495 
1496  /* Reset TCP record info */
1497  transport->tcp_offset = 0;
1498  transport->tcp_reclen = 0;
1499  transport->tcp_copied = 0;
1500  transport->tcp_flags =
1501  TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
1502 
1503  xprt_wake_pending_tasks(xprt, -EAGAIN);
1504  }
1505  spin_unlock(&xprt->transport_lock);
1506  break;
1507  case TCP_FIN_WAIT1:
1508  /* The client initiated a shutdown of the socket */
1509  xprt->connect_cookie++;
1510  xprt->reestablish_timeout = 0;
1511  set_bit(XPRT_CLOSING, &xprt->state);
1512  smp_mb__before_clear_bit();
1513  clear_bit(XPRT_CONNECTED, &xprt->state);
1514  clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1515  smp_mb__after_clear_bit();
1516  xs_tcp_schedule_linger_timeout(xprt, xs_tcp_fin_timeout);
1517  break;
1518  case TCP_CLOSE_WAIT:
1519  /* The server initiated a shutdown of the socket */
1520  xprt->connect_cookie++;
1521  clear_bit(XPRT_CONNECTED, &xprt->state);
1522  xs_tcp_force_close(xprt);
1523  case TCP_CLOSING:
1524  /*
1525  * If the server closed down the connection, make sure that
1526  * we back off before reconnecting
1527  */
1528  if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
1529  xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
1530  break;
1531  case TCP_LAST_ACK:
1532  set_bit(XPRT_CLOSING, &xprt->state);
1533  xs_tcp_schedule_linger_timeout(xprt, xs_tcp_fin_timeout);
1534  smp_mb__before_clear_bit();
1535  clear_bit(XPRT_CONNECTED, &xprt->state);
1536  smp_mb__after_clear_bit();
1537  break;
1538  case TCP_CLOSE:
1539  xs_tcp_cancel_linger_timeout(xprt);
1540  xs_sock_mark_closed(xprt);
1541  }
1542  out:
1543  read_unlock_bh(&sk->sk_callback_lock);
1544 }
1545 
1546 static void xs_write_space(struct sock *sk)
1547 {
1548  struct socket *sock;
1549  struct rpc_xprt *xprt;
1550 
1551  if (unlikely(!(sock = sk->sk_socket)))
1552  return;
1553  clear_bit(SOCK_NOSPACE, &sock->flags);
1554 
1555  if (unlikely(!(xprt = xprt_from_sock(sk))))
1556  return;
1557  if (test_and_clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags) == 0)
1558  return;
1559 
1560  xprt_write_space(xprt);
1561 }
1562 
1573 static void xs_udp_write_space(struct sock *sk)
1574 {
1575  read_lock_bh(&sk->sk_callback_lock);
1576 
1577  /* from net/core/sock.c:sock_def_write_space */
1578  if (sock_writeable(sk))
1579  xs_write_space(sk);
1580 
1581  read_unlock_bh(&sk->sk_callback_lock);
1582 }
1583 
1594 static void xs_tcp_write_space(struct sock *sk)
1595 {
1596  read_lock_bh(&sk->sk_callback_lock);
1597 
1598  /* from net/core/stream.c:sk_stream_write_space */
1599  if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk))
1600  xs_write_space(sk);
1601 
1602  read_unlock_bh(&sk->sk_callback_lock);
1603 }
1604 
1605 static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1606 {
1607  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1608  struct sock *sk = transport->inet;
1609 
1610  if (transport->rcvsize) {
1611  sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1612  sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1613  }
1614  if (transport->sndsize) {
1615  sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1616  sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1617  sk->sk_write_space(sk);
1618  }
1619 }
1620 
1629 static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1630 {
1631  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1632 
1633  transport->sndsize = 0;
1634  if (sndsize)
1635  transport->sndsize = sndsize + 1024;
1636  transport->rcvsize = 0;
1637  if (rcvsize)
1638  transport->rcvsize = rcvsize + 1024;
1639 
1640  xs_udp_do_set_buffer_size(xprt);
1641 }
1642 
1649 static void xs_udp_timer(struct rpc_task *task)
1650 {
1651  xprt_adjust_cwnd(task, -ETIMEDOUT);
1652 }
1653 
1654 static unsigned short xs_get_random_port(void)
1655 {
1656  unsigned short range = xprt_max_resvport - xprt_min_resvport;
1657  unsigned short rand = (unsigned short) net_random() % range;
1658  return rand + xprt_min_resvport;
1659 }
1660 
1667 static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
1668 {
1669  dprintk("RPC: setting port for xprt %p to %u\n", xprt, port);
1670 
1671  rpc_set_port(xs_addr(xprt), port);
1672  xs_update_peer_port(xprt);
1673 }
1674 
1675 static unsigned short xs_get_srcport(struct sock_xprt *transport)
1676 {
1677  unsigned short port = transport->srcport;
1678 
1679  if (port == 0 && transport->xprt.resvport)
1680  port = xs_get_random_port();
1681  return port;
1682 }
1683 
1684 static unsigned short xs_next_srcport(struct sock_xprt *transport, unsigned short port)
1685 {
1686  if (transport->srcport != 0)
1687  transport->srcport = 0;
1688  if (!transport->xprt.resvport)
1689  return 0;
1690  if (port <= xprt_min_resvport || port > xprt_max_resvport)
1691  return xprt_max_resvport;
1692  return --port;
1693 }
1694 static int xs_bind(struct sock_xprt *transport, struct socket *sock)
1695 {
1696  struct sockaddr_storage myaddr;
1697  int err, nloop = 0;
1698  unsigned short port = xs_get_srcport(transport);
1699  unsigned short last;
1700 
1701  memcpy(&myaddr, &transport->srcaddr, transport->xprt.addrlen);
1702  do {
1703  rpc_set_port((struct sockaddr *)&myaddr, port);
1704  err = kernel_bind(sock, (struct sockaddr *)&myaddr,
1705  transport->xprt.addrlen);
1706  if (port == 0)
1707  break;
1708  if (err == 0) {
1709  transport->srcport = port;
1710  break;
1711  }
1712  last = port;
1713  port = xs_next_srcport(transport, port);
1714  if (port > last)
1715  nloop++;
1716  } while (err == -EADDRINUSE && nloop != 2);
1717 
1718  if (myaddr.ss_family == AF_INET)
1719  dprintk("RPC: %s %pI4:%u: %s (%d)\n", __func__,
1720  &((struct sockaddr_in *)&myaddr)->sin_addr,
1721  port, err ? "failed" : "ok", err);
1722  else
1723  dprintk("RPC: %s %pI6:%u: %s (%d)\n", __func__,
1724  &((struct sockaddr_in6 *)&myaddr)->sin6_addr,
1725  port, err ? "failed" : "ok", err);
1726  return err;
1727 }
1728 
1729 /*
1730  * We don't support autobind on AF_LOCAL sockets
1731  */
1732 static void xs_local_rpcbind(struct rpc_task *task)
1733 {
1734  xprt_set_bound(task->tk_xprt);
1735 }
1736 
1737 static void xs_local_set_port(struct rpc_xprt *xprt, unsigned short port)
1738 {
1739 }
1740 
1741 #ifdef CONFIG_DEBUG_LOCK_ALLOC
1742 static struct lock_class_key xs_key[2];
1743 static struct lock_class_key xs_slock_key[2];
1744 
1745 static inline void xs_reclassify_socketu(struct socket *sock)
1746 {
1747  struct sock *sk = sock->sk;
1748 
1749  BUG_ON(sock_owned_by_user(sk));
1750  sock_lock_init_class_and_name(sk, "slock-AF_LOCAL-RPC",
1751  &xs_slock_key[1], "sk_lock-AF_LOCAL-RPC", &xs_key[1]);
1752 }
1753 
1754 static inline void xs_reclassify_socket4(struct socket *sock)
1755 {
1756  struct sock *sk = sock->sk;
1757 
1758  BUG_ON(sock_owned_by_user(sk));
1759  sock_lock_init_class_and_name(sk, "slock-AF_INET-RPC",
1760  &xs_slock_key[0], "sk_lock-AF_INET-RPC", &xs_key[0]);
1761 }
1762 
1763 static inline void xs_reclassify_socket6(struct socket *sock)
1764 {
1765  struct sock *sk = sock->sk;
1766 
1767  BUG_ON(sock_owned_by_user(sk));
1768  sock_lock_init_class_and_name(sk, "slock-AF_INET6-RPC",
1769  &xs_slock_key[1], "sk_lock-AF_INET6-RPC", &xs_key[1]);
1770 }
1771 
1772 static inline void xs_reclassify_socket(int family, struct socket *sock)
1773 {
1774  switch (family) {
1775  case AF_LOCAL:
1776  xs_reclassify_socketu(sock);
1777  break;
1778  case AF_INET:
1779  xs_reclassify_socket4(sock);
1780  break;
1781  case AF_INET6:
1782  xs_reclassify_socket6(sock);
1783  break;
1784  }
1785 }
1786 #else
1787 static inline void xs_reclassify_socketu(struct socket *sock)
1788 {
1789 }
1790 
1791 static inline void xs_reclassify_socket4(struct socket *sock)
1792 {
1793 }
1794 
1795 static inline void xs_reclassify_socket6(struct socket *sock)
1796 {
1797 }
1798 
1799 static inline void xs_reclassify_socket(int family, struct socket *sock)
1800 {
1801 }
1802 #endif
1803 
1804 static struct socket *xs_create_sock(struct rpc_xprt *xprt,
1805  struct sock_xprt *transport, int family, int type, int protocol)
1806 {
1807  struct socket *sock;
1808  int err;
1809 
1810  err = __sock_create(xprt->xprt_net, family, type, protocol, &sock, 1);
1811  if (err < 0) {
1812  dprintk("RPC: can't create %d transport socket (%d).\n",
1813  protocol, -err);
1814  goto out;
1815  }
1816  xs_reclassify_socket(family, sock);
1817 
1818  err = xs_bind(transport, sock);
1819  if (err) {
1820  sock_release(sock);
1821  goto out;
1822  }
1823 
1824  return sock;
1825 out:
1826  return ERR_PTR(err);
1827 }
1828 
1829 static int xs_local_finish_connecting(struct rpc_xprt *xprt,
1830  struct socket *sock)
1831 {
1832  struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
1833  xprt);
1834 
1835  if (!transport->inet) {
1836  struct sock *sk = sock->sk;
1837 
1838  write_lock_bh(&sk->sk_callback_lock);
1839 
1840  xs_save_old_callbacks(transport, sk);
1841 
1842  sk->sk_user_data = xprt;
1843  sk->sk_data_ready = xs_local_data_ready;
1844  sk->sk_write_space = xs_udp_write_space;
1845  sk->sk_allocation = GFP_ATOMIC;
1846 
1847  xprt_clear_connected(xprt);
1848 
1849  /* Reset to new socket */
1850  transport->sock = sock;
1851  transport->inet = sk;
1852 
1853  write_unlock_bh(&sk->sk_callback_lock);
1854  }
1855 
1856  /* Tell the socket layer to start connecting... */
1857  xprt->stat.connect_count++;
1858  xprt->stat.connect_start = jiffies;
1859  return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, 0);
1860 }
1861 
1870 static void xs_local_setup_socket(struct work_struct *work)
1871 {
1872  struct sock_xprt *transport =
1873  container_of(work, struct sock_xprt, connect_worker.work);
1874  struct rpc_xprt *xprt = &transport->xprt;
1875  struct socket *sock;
1876  int status = -EIO;
1877 
1878  current->flags |= PF_FSTRANS;
1879 
1880  clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
1881  status = __sock_create(xprt->xprt_net, AF_LOCAL,
1882  SOCK_STREAM, 0, &sock, 1);
1883  if (status < 0) {
1884  dprintk("RPC: can't create AF_LOCAL "
1885  "transport socket (%d).\n", -status);
1886  goto out;
1887  }
1888  xs_reclassify_socketu(sock);
1889 
1890  dprintk("RPC: worker connecting xprt %p via AF_LOCAL to %s\n",
1891  xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1892 
1893  status = xs_local_finish_connecting(xprt, sock);
1894  switch (status) {
1895  case 0:
1896  dprintk("RPC: xprt %p connected to %s\n",
1897  xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1898  xprt_set_connected(xprt);
1899  break;
1900  case -ENOENT:
1901  dprintk("RPC: xprt %p: socket %s does not exist\n",
1902  xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
1903  break;
1904  default:
1905  printk(KERN_ERR "%s: unhandled error (%d) connecting to %s\n",
1906  __func__, -status,
1907  xprt->address_strings[RPC_DISPLAY_ADDR]);
1908  }
1909 
1910 out:
1911  xprt_clear_connecting(xprt);
1912  xprt_wake_pending_tasks(xprt, status);
1913  current->flags &= ~PF_FSTRANS;
1914 }
1915 
1916 #ifdef CONFIG_SUNRPC_SWAP
1917 static void xs_set_memalloc(struct rpc_xprt *xprt)
1918 {
1919  struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
1920  xprt);
1921 
1922  if (xprt->swapper)
1923  sk_set_memalloc(transport->inet);
1924 }
1925 
1932 int xs_swapper(struct rpc_xprt *xprt, int enable)
1933 {
1934  struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
1935  xprt);
1936  int err = 0;
1937 
1938  if (enable) {
1939  xprt->swapper++;
1940  xs_set_memalloc(xprt);
1941  } else if (xprt->swapper) {
1942  xprt->swapper--;
1943  sk_clear_memalloc(transport->inet);
1944  }
1945 
1946  return err;
1947 }
1948 EXPORT_SYMBOL_GPL(xs_swapper);
1949 #else
1950 static void xs_set_memalloc(struct rpc_xprt *xprt)
1951 {
1952 }
1953 #endif
1954 
1955 static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
1956 {
1957  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1958 
1959  if (!transport->inet) {
1960  struct sock *sk = sock->sk;
1961 
1962  write_lock_bh(&sk->sk_callback_lock);
1963 
1964  xs_save_old_callbacks(transport, sk);
1965 
1966  sk->sk_user_data = xprt;
1967  sk->sk_data_ready = xs_udp_data_ready;
1968  sk->sk_write_space = xs_udp_write_space;
1969  sk->sk_no_check = UDP_CSUM_NORCV;
1970  sk->sk_allocation = GFP_ATOMIC;
1971 
1972  xprt_set_connected(xprt);
1973 
1974  /* Reset to new socket */
1975  transport->sock = sock;
1976  transport->inet = sk;
1977 
1978  xs_set_memalloc(xprt);
1979 
1980  write_unlock_bh(&sk->sk_callback_lock);
1981  }
1982  xs_udp_do_set_buffer_size(xprt);
1983 }
1984 
1985 static void xs_udp_setup_socket(struct work_struct *work)
1986 {
1987  struct sock_xprt *transport =
1988  container_of(work, struct sock_xprt, connect_worker.work);
1989  struct rpc_xprt *xprt = &transport->xprt;
1990  struct socket *sock = transport->sock;
1991  int status = -EIO;
1992 
1993  current->flags |= PF_FSTRANS;
1994 
1995  /* Start by resetting any existing state */
1996  xs_reset_transport(transport);
1997  sock = xs_create_sock(xprt, transport,
1998  xs_addr(xprt)->sa_family, SOCK_DGRAM, IPPROTO_UDP);
1999  if (IS_ERR(sock))
2000  goto out;
2001 
2002  dprintk("RPC: worker connecting xprt %p via %s to "
2003  "%s (port %s)\n", xprt,
2004  xprt->address_strings[RPC_DISPLAY_PROTO],
2005  xprt->address_strings[RPC_DISPLAY_ADDR],
2006  xprt->address_strings[RPC_DISPLAY_PORT]);
2007 
2008  xs_udp_finish_connecting(xprt, sock);
2009  status = 0;
2010 out:
2011  xprt_clear_connecting(xprt);
2012  xprt_wake_pending_tasks(xprt, status);
2013  current->flags &= ~PF_FSTRANS;
2014 }
2015 
2016 /*
2017  * We need to preserve the port number so the reply cache on the server can
2018  * find our cached RPC replies when we get around to reconnecting.
2019  */
2020 static void xs_abort_connection(struct sock_xprt *transport)
2021 {
2022  int result;
2023  struct sockaddr any;
2024 
2025  dprintk("RPC: disconnecting xprt %p to reuse port\n", transport);
2026 
2027  /*
2028  * Disconnect the transport socket by doing a connect operation
2029  * with AF_UNSPEC. This should return immediately...
2030  */
2031  memset(&any, 0, sizeof(any));
2032  any.sa_family = AF_UNSPEC;
2033  result = kernel_connect(transport->sock, &any, sizeof(any), 0);
2034  if (!result)
2035  xs_sock_reset_connection_flags(&transport->xprt);
2036  dprintk("RPC: AF_UNSPEC connect return code %d\n", result);
2037 }
2038 
2039 static void xs_tcp_reuse_connection(struct sock_xprt *transport)
2040 {
2041  unsigned int state = transport->inet->sk_state;
2042 
2043  if (state == TCP_CLOSE && transport->sock->state == SS_UNCONNECTED) {
2044  /* we don't need to abort the connection if the socket
2045  * hasn't undergone a shutdown
2046  */
2047  if (transport->inet->sk_shutdown == 0)
2048  return;
2049  dprintk("RPC: %s: TCP_CLOSEd and sk_shutdown set to %d\n",
2050  __func__, transport->inet->sk_shutdown);
2051  }
2052  if ((1 << state) & (TCPF_ESTABLISHED|TCPF_SYN_SENT)) {
2053  /* we don't need to abort the connection if the socket
2054  * hasn't undergone a shutdown
2055  */
2056  if (transport->inet->sk_shutdown == 0)
2057  return;
2058  dprintk("RPC: %s: ESTABLISHED/SYN_SENT "
2059  "sk_shutdown set to %d\n",
2060  __func__, transport->inet->sk_shutdown);
2061  }
2062  xs_abort_connection(transport);
2063 }
2064 
2065 static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
2066 {
2067  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2068  int ret = -ENOTCONN;
2069 
2070  if (!transport->inet) {
2071  struct sock *sk = sock->sk;
2072 
2073  write_lock_bh(&sk->sk_callback_lock);
2074 
2075  xs_save_old_callbacks(transport, sk);
2076 
2077  sk->sk_user_data = xprt;
2078  sk->sk_data_ready = xs_tcp_data_ready;
2079  sk->sk_state_change = xs_tcp_state_change;
2080  sk->sk_write_space = xs_tcp_write_space;
2081  sk->sk_allocation = GFP_ATOMIC;
2082 
2083  /* socket options */
2084  sk->sk_userlocks |= SOCK_BINDPORT_LOCK;
2085  sock_reset_flag(sk, SOCK_LINGER);
2086  tcp_sk(sk)->linger2 = 0;
2087  tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
2088 
2089  xprt_clear_connected(xprt);
2090 
2091  /* Reset to new socket */
2092  transport->sock = sock;
2093  transport->inet = sk;
2094 
2095  write_unlock_bh(&sk->sk_callback_lock);
2096  }
2097 
2098  if (!xprt_bound(xprt))
2099  goto out;
2100 
2101  xs_set_memalloc(xprt);
2102 
2103  /* Tell the socket layer to start connecting... */
2104  xprt->stat.connect_count++;
2105  xprt->stat.connect_start = jiffies;
2106  ret = kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
2107  switch (ret) {
2108  case 0:
2109  case -EINPROGRESS:
2110  /* SYN_SENT! */
2111  xprt->connect_cookie++;
2112  if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2113  xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2114  }
2115 out:
2116  return ret;
2117 }
2118 
2127 static void xs_tcp_setup_socket(struct work_struct *work)
2128 {
2129  struct sock_xprt *transport =
2130  container_of(work, struct sock_xprt, connect_worker.work);
2131  struct socket *sock = transport->sock;
2132  struct rpc_xprt *xprt = &transport->xprt;
2133  int status = -EIO;
2134 
2135  current->flags |= PF_FSTRANS;
2136 
2137  if (!sock) {
2138  clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
2139  sock = xs_create_sock(xprt, transport,
2140  xs_addr(xprt)->sa_family, SOCK_STREAM, IPPROTO_TCP);
2141  if (IS_ERR(sock)) {
2142  status = PTR_ERR(sock);
2143  goto out;
2144  }
2145  } else {
2146  int abort_and_exit;
2147 
2148  abort_and_exit = test_and_clear_bit(XPRT_CONNECTION_ABORT,
2149  &xprt->state);
2150  /* "close" the socket, preserving the local port */
2151  xs_tcp_reuse_connection(transport);
2152 
2153  if (abort_and_exit)
2154  goto out_eagain;
2155  }
2156 
2157  dprintk("RPC: worker connecting xprt %p via %s to "
2158  "%s (port %s)\n", xprt,
2159  xprt->address_strings[RPC_DISPLAY_PROTO],
2160  xprt->address_strings[RPC_DISPLAY_ADDR],
2161  xprt->address_strings[RPC_DISPLAY_PORT]);
2162 
2163  status = xs_tcp_finish_connecting(xprt, sock);
2164  dprintk("RPC: %p connect status %d connected %d sock state %d\n",
2165  xprt, -status, xprt_connected(xprt),
2166  sock->sk->sk_state);
2167  switch (status) {
2168  default:
2169  printk("%s: connect returned unhandled error %d\n",
2170  __func__, status);
2171  case -EADDRNOTAVAIL:
2172  /* We're probably in TIME_WAIT. Get rid of existing socket,
2173  * and retry
2174  */
2175  xs_tcp_force_close(xprt);
2176  break;
2177  case -ECONNREFUSED:
2178  case -ECONNRESET:
2179  case -ENETUNREACH:
2180  /* retry with existing socket, after a delay */
2181  case 0:
2182  case -EINPROGRESS:
2183  case -EALREADY:
2184  xprt_clear_connecting(xprt);
2185  current->flags &= ~PF_FSTRANS;
2186  return;
2187  case -EINVAL:
2188  /* Happens, for instance, if the user specified a link
2189  * local IPv6 address without a scope-id.
2190  */
2191  goto out;
2192  }
2193 out_eagain:
2194  status = -EAGAIN;
2195 out:
2196  xprt_clear_connecting(xprt);
2197  xprt_wake_pending_tasks(xprt, status);
2198  current->flags &= ~PF_FSTRANS;
2199 }
2200 
2214 static void xs_connect(struct rpc_task *task)
2215 {
2216  struct rpc_xprt *xprt = task->tk_xprt;
2217  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2218 
2219  if (transport->sock != NULL && !RPC_IS_SOFTCONN(task)) {
2220  dprintk("RPC: xs_connect delayed xprt %p for %lu "
2221  "seconds\n",
2222  xprt, xprt->reestablish_timeout / HZ);
2223  queue_delayed_work(rpciod_workqueue,
2224  &transport->connect_worker,
2225  xprt->reestablish_timeout);
2226  xprt->reestablish_timeout <<= 1;
2227  if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2228  xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2229  if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
2230  xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
2231  } else {
2232  dprintk("RPC: xs_connect scheduled xprt %p\n", xprt);
2233  queue_delayed_work(rpciod_workqueue,
2234  &transport->connect_worker, 0);
2235  }
2236 }
2237 
2244 static void xs_local_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2245 {
2246  long idle_time = 0;
2247 
2248  if (xprt_connected(xprt))
2249  idle_time = (long)(jiffies - xprt->last_used) / HZ;
2250 
2251  seq_printf(seq, "\txprt:\tlocal %lu %lu %lu %ld %lu %lu %lu "
2252  "%llu %llu %lu %llu %llu\n",
2253  xprt->stat.bind_count,
2254  xprt->stat.connect_count,
2255  xprt->stat.connect_time,
2256  idle_time,
2257  xprt->stat.sends,
2258  xprt->stat.recvs,
2259  xprt->stat.bad_xids,
2260  xprt->stat.req_u,
2261  xprt->stat.bklog_u,
2262  xprt->stat.max_slots,
2263  xprt->stat.sending_u,
2264  xprt->stat.pending_u);
2265 }
2266 
2273 static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2274 {
2275  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2276 
2277  seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %llu %llu "
2278  "%lu %llu %llu\n",
2279  transport->srcport,
2280  xprt->stat.bind_count,
2281  xprt->stat.sends,
2282  xprt->stat.recvs,
2283  xprt->stat.bad_xids,
2284  xprt->stat.req_u,
2285  xprt->stat.bklog_u,
2286  xprt->stat.max_slots,
2287  xprt->stat.sending_u,
2288  xprt->stat.pending_u);
2289 }
2290 
2297 static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2298 {
2299  struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2300  long idle_time = 0;
2301 
2302  if (xprt_connected(xprt))
2303  idle_time = (long)(jiffies - xprt->last_used) / HZ;
2304 
2305  seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu "
2306  "%llu %llu %lu %llu %llu\n",
2307  transport->srcport,
2308  xprt->stat.bind_count,
2309  xprt->stat.connect_count,
2310  xprt->stat.connect_time,
2311  idle_time,
2312  xprt->stat.sends,
2313  xprt->stat.recvs,
2314  xprt->stat.bad_xids,
2315  xprt->stat.req_u,
2316  xprt->stat.bklog_u,
2317  xprt->stat.max_slots,
2318  xprt->stat.sending_u,
2319  xprt->stat.pending_u);
2320 }
2321 
2322 /*
2323  * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
2324  * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
2325  * to use the server side send routines.
2326  */
2327 static void *bc_malloc(struct rpc_task *task, size_t size)
2328 {
2329  struct page *page;
2330  struct rpc_buffer *buf;
2331 
2332  BUG_ON(size > PAGE_SIZE - sizeof(struct rpc_buffer));
2333  page = alloc_page(GFP_KERNEL);
2334 
2335  if (!page)
2336  return NULL;
2337 
2338  buf = page_address(page);
2339  buf->len = PAGE_SIZE;
2340 
2341  return buf->data;
2342 }
2343 
2344 /*
2345  * Free the space allocated in the bc_alloc routine
2346  */
2347 static void bc_free(void *buffer)
2348 {
2349  struct rpc_buffer *buf;
2350 
2351  if (!buffer)
2352  return;
2353 
2354  buf = container_of(buffer, struct rpc_buffer, data);
2355  free_page((unsigned long)buf);
2356 }
2357 
2358 /*
2359  * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
2360  * held. Borrows heavily from svc_tcp_sendto and xs_tcp_send_request.
2361  */
2362 static int bc_sendto(struct rpc_rqst *req)
2363 {
2364  int len;
2365  struct xdr_buf *xbufp = &req->rq_snd_buf;
2366  struct rpc_xprt *xprt = req->rq_xprt;
2367  struct sock_xprt *transport =
2368  container_of(xprt, struct sock_xprt, xprt);
2369  struct socket *sock = transport->sock;
2370  unsigned long headoff;
2371  unsigned long tailoff;
2372 
2373  xs_encode_stream_record_marker(xbufp);
2374 
2375  tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
2376  headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
2377  len = svc_send_common(sock, xbufp,
2378  virt_to_page(xbufp->head[0].iov_base), headoff,
2379  xbufp->tail[0].iov_base, tailoff);
2380 
2381  if (len != xbufp->len) {
2382  printk(KERN_NOTICE "Error sending entire callback!\n");
2383  len = -EAGAIN;
2384  }
2385 
2386  return len;
2387 }
2388 
2389 /*
2390  * The send routine. Borrows from svc_send
2391  */
2392 static int bc_send_request(struct rpc_task *task)
2393 {
2394  struct rpc_rqst *req = task->tk_rqstp;
2395  struct svc_xprt *xprt;
2396  struct svc_sock *svsk;
2397  u32 len;
2398 
2399  dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
2400  /*
2401  * Get the server socket associated with this callback xprt
2402  */
2403  xprt = req->rq_xprt->bc_xprt;
2404  svsk = container_of(xprt, struct svc_sock, sk_xprt);
2405 
2406  /*
2407  * Grab the mutex to serialize data as the connection is shared
2408  * with the fore channel
2409  */
2410  if (!mutex_trylock(&xprt->xpt_mutex)) {
2411  rpc_sleep_on(&xprt->xpt_bc_pending, task, NULL);
2412  if (!mutex_trylock(&xprt->xpt_mutex))
2413  return -EAGAIN;
2414  rpc_wake_up_queued_task(&xprt->xpt_bc_pending, task);
2415  }
2416  if (test_bit(XPT_DEAD, &xprt->xpt_flags))
2417  len = -ENOTCONN;
2418  else
2419  len = bc_sendto(req);
2420  mutex_unlock(&xprt->xpt_mutex);
2421 
2422  if (len > 0)
2423  len = 0;
2424 
2425  return len;
2426 }
2427 
2428 /*
2429  * The close routine. Since this is client initiated, we do nothing
2430  */
2431 
2432 static void bc_close(struct rpc_xprt *xprt)
2433 {
2434 }
2435 
2436 /*
2437  * The xprt destroy routine. Again, because this connection is client
2438  * initiated, we do nothing
2439  */
2440 
2441 static void bc_destroy(struct rpc_xprt *xprt)
2442 {
2443 }
2444 
2445 static struct rpc_xprt_ops xs_local_ops = {
2446  .reserve_xprt = xprt_reserve_xprt,
2447  .release_xprt = xs_tcp_release_xprt,
2448  .alloc_slot = xprt_alloc_slot,
2449  .rpcbind = xs_local_rpcbind,
2450  .set_port = xs_local_set_port,
2451  .connect = xs_connect,
2452  .buf_alloc = rpc_malloc,
2453  .buf_free = rpc_free,
2454  .send_request = xs_local_send_request,
2455  .set_retrans_timeout = xprt_set_retrans_timeout_def,
2456  .close = xs_close,
2457  .destroy = xs_destroy,
2458  .print_stats = xs_local_print_stats,
2459 };
2460 
2461 static struct rpc_xprt_ops xs_udp_ops = {
2462  .set_buffer_size = xs_udp_set_buffer_size,
2463  .reserve_xprt = xprt_reserve_xprt_cong,
2464  .release_xprt = xprt_release_xprt_cong,
2465  .alloc_slot = xprt_alloc_slot,
2466  .rpcbind = rpcb_getport_async,
2467  .set_port = xs_set_port,
2468  .connect = xs_connect,
2469  .buf_alloc = rpc_malloc,
2470  .buf_free = rpc_free,
2471  .send_request = xs_udp_send_request,
2472  .set_retrans_timeout = xprt_set_retrans_timeout_rtt,
2473  .timer = xs_udp_timer,
2474  .release_request = xprt_release_rqst_cong,
2475  .close = xs_close,
2476  .destroy = xs_destroy,
2477  .print_stats = xs_udp_print_stats,
2478 };
2479 
2480 static struct rpc_xprt_ops xs_tcp_ops = {
2481  .reserve_xprt = xprt_reserve_xprt,
2482  .release_xprt = xs_tcp_release_xprt,
2483  .alloc_slot = xprt_lock_and_alloc_slot,
2484  .rpcbind = rpcb_getport_async,
2485  .set_port = xs_set_port,
2486  .connect = xs_connect,
2487  .buf_alloc = rpc_malloc,
2488  .buf_free = rpc_free,
2489  .send_request = xs_tcp_send_request,
2490  .set_retrans_timeout = xprt_set_retrans_timeout_def,
2491  .close = xs_tcp_close,
2492  .destroy = xs_destroy,
2493  .print_stats = xs_tcp_print_stats,
2494 };
2495 
2496 /*
2497  * The rpc_xprt_ops for the server backchannel
2498  */
2499 
2500 static struct rpc_xprt_ops bc_tcp_ops = {
2501  .reserve_xprt = xprt_reserve_xprt,
2502  .release_xprt = xprt_release_xprt,
2503  .alloc_slot = xprt_alloc_slot,
2504  .rpcbind = xs_local_rpcbind,
2505  .buf_alloc = bc_malloc,
2506  .buf_free = bc_free,
2507  .send_request = bc_send_request,
2508  .set_retrans_timeout = xprt_set_retrans_timeout_def,
2509  .close = bc_close,
2510  .destroy = bc_destroy,
2511  .print_stats = xs_tcp_print_stats,
2512 };
2513 
2514 static int xs_init_anyaddr(const int family, struct sockaddr *sap)
2515 {
2516  static const struct sockaddr_in sin = {
2517  .sin_family = AF_INET,
2518  .sin_addr.s_addr = htonl(INADDR_ANY),
2519  };
2520  static const struct sockaddr_in6 sin6 = {
2521  .sin6_family = AF_INET6,
2522  .sin6_addr = IN6ADDR_ANY_INIT,
2523  };
2524 
2525  switch (family) {
2526  case AF_LOCAL:
2527  break;
2528  case AF_INET:
2529  memcpy(sap, &sin, sizeof(sin));
2530  break;
2531  case AF_INET6:
2532  memcpy(sap, &sin6, sizeof(sin6));
2533  break;
2534  default:
2535  dprintk("RPC: %s: Bad address family\n", __func__);
2536  return -EAFNOSUPPORT;
2537  }
2538  return 0;
2539 }
2540 
2541 static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
2542  unsigned int slot_table_size,
2543  unsigned int max_slot_table_size)
2544 {
2545  struct rpc_xprt *xprt;
2546  struct sock_xprt *new;
2547 
2548  if (args->addrlen > sizeof(xprt->addr)) {
2549  dprintk("RPC: xs_setup_xprt: address too large\n");
2550  return ERR_PTR(-EBADF);
2551  }
2552 
2553  xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
2554  max_slot_table_size);
2555  if (xprt == NULL) {
2556  dprintk("RPC: xs_setup_xprt: couldn't allocate "
2557  "rpc_xprt\n");
2558  return ERR_PTR(-ENOMEM);
2559  }
2560 
2561  new = container_of(xprt, struct sock_xprt, xprt);
2562  memcpy(&xprt->addr, args->dstaddr, args->addrlen);
2563  xprt->addrlen = args->addrlen;
2564  if (args->srcaddr)
2565  memcpy(&new->srcaddr, args->srcaddr, args->addrlen);
2566  else {
2567  int err;
2568  err = xs_init_anyaddr(args->dstaddr->sa_family,
2569  (struct sockaddr *)&new->srcaddr);
2570  if (err != 0) {
2571  xprt_free(xprt);
2572  return ERR_PTR(err);
2573  }
2574  }
2575 
2576  return xprt;
2577 }
2578 
2579 static const struct rpc_timeout xs_local_default_timeout = {
2580  .to_initval = 10 * HZ,
2581  .to_maxval = 10 * HZ,
2582  .to_retries = 2,
2583 };
2584 
2591 static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
2592 {
2593  struct sockaddr_un *sun = (struct sockaddr_un *)args->dstaddr;
2594  struct sock_xprt *transport;
2595  struct rpc_xprt *xprt;
2596  struct rpc_xprt *ret;
2597 
2598  xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
2599  xprt_max_tcp_slot_table_entries);
2600  if (IS_ERR(xprt))
2601  return xprt;
2602  transport = container_of(xprt, struct sock_xprt, xprt);
2603 
2604  xprt->prot = 0;
2605  xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2606  xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2607 
2608  xprt->bind_timeout = XS_BIND_TO;
2609  xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2610  xprt->idle_timeout = XS_IDLE_DISC_TO;
2611 
2612  xprt->ops = &xs_local_ops;
2613  xprt->timeout = &xs_local_default_timeout;
2614 
2615  switch (sun->sun_family) {
2616  case AF_LOCAL:
2617  if (sun->sun_path[0] != '/') {
2618  dprintk("RPC: bad AF_LOCAL address: %s\n",
2619  sun->sun_path);
2620  ret = ERR_PTR(-EINVAL);
2621  goto out_err;
2622  }
2623  xprt_set_bound(xprt);
2624  INIT_DELAYED_WORK(&transport->connect_worker,
2625  xs_local_setup_socket);
2626  xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
2627  break;
2628  default:
2629  ret = ERR_PTR(-EAFNOSUPPORT);
2630  goto out_err;
2631  }
2632 
2633  dprintk("RPC: set up xprt to %s via AF_LOCAL\n",
2634  xprt->address_strings[RPC_DISPLAY_ADDR]);
2635 
2636  if (try_module_get(THIS_MODULE))
2637  return xprt;
2638  ret = ERR_PTR(-EINVAL);
2639 out_err:
2640  xprt_free(xprt);
2641  return ret;
2642 }
2643 
2644 static const struct rpc_timeout xs_udp_default_timeout = {
2645  .to_initval = 5 * HZ,
2646  .to_maxval = 30 * HZ,
2647  .to_increment = 5 * HZ,
2648  .to_retries = 5,
2649 };
2650 
2656 static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
2657 {
2658  struct sockaddr *addr = args->dstaddr;
2659  struct rpc_xprt *xprt;
2660  struct sock_xprt *transport;
2661  struct rpc_xprt *ret;
2662 
2663  xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
2664  xprt_udp_slot_table_entries);
2665  if (IS_ERR(xprt))
2666  return xprt;
2667  transport = container_of(xprt, struct sock_xprt, xprt);
2668 
2669  xprt->prot = IPPROTO_UDP;
2670  xprt->tsh_size = 0;
2671  /* XXX: header size can vary due to auth type, IPv6, etc. */
2672  xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
2673 
2674  xprt->bind_timeout = XS_BIND_TO;
2675  xprt->reestablish_timeout = XS_UDP_REEST_TO;
2676  xprt->idle_timeout = XS_IDLE_DISC_TO;
2677 
2678  xprt->ops = &xs_udp_ops;
2679 
2680  xprt->timeout = &xs_udp_default_timeout;
2681 
2682  switch (addr->sa_family) {
2683  case AF_INET:
2684  if (((struct sockaddr_in *)addr)->sin_port != htons(0))
2685  xprt_set_bound(xprt);
2686 
2687  INIT_DELAYED_WORK(&transport->connect_worker,
2688  xs_udp_setup_socket);
2689  xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
2690  break;
2691  case AF_INET6:
2692  if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
2693  xprt_set_bound(xprt);
2694 
2695  INIT_DELAYED_WORK(&transport->connect_worker,
2696  xs_udp_setup_socket);
2697  xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
2698  break;
2699  default:
2700  ret = ERR_PTR(-EAFNOSUPPORT);
2701  goto out_err;
2702  }
2703 
2704  if (xprt_bound(xprt))
2705  dprintk("RPC: set up xprt to %s (port %s) via %s\n",
2706  xprt->address_strings[RPC_DISPLAY_ADDR],
2707  xprt->address_strings[RPC_DISPLAY_PORT],
2708  xprt->address_strings[RPC_DISPLAY_PROTO]);
2709  else
2710  dprintk("RPC: set up xprt to %s (autobind) via %s\n",
2711  xprt->address_strings[RPC_DISPLAY_ADDR],
2712  xprt->address_strings[RPC_DISPLAY_PROTO]);
2713 
2714  if (try_module_get(THIS_MODULE))
2715  return xprt;
2716  ret = ERR_PTR(-EINVAL);
2717 out_err:
2718  xprt_free(xprt);
2719  return ret;
2720 }
2721 
2722 static const struct rpc_timeout xs_tcp_default_timeout = {
2723  .to_initval = 60 * HZ,
2724  .to_maxval = 60 * HZ,
2725  .to_retries = 2,
2726 };
2727 
2733 static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
2734 {
2735  struct sockaddr *addr = args->dstaddr;
2736  struct rpc_xprt *xprt;
2737  struct sock_xprt *transport;
2738  struct rpc_xprt *ret;
2739 
2740  xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
2741  xprt_max_tcp_slot_table_entries);
2742  if (IS_ERR(xprt))
2743  return xprt;
2744  transport = container_of(xprt, struct sock_xprt, xprt);
2745 
2746  xprt->prot = IPPROTO_TCP;
2747  xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2748  xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2749 
2750  xprt->bind_timeout = XS_BIND_TO;
2751  xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2752  xprt->idle_timeout = XS_IDLE_DISC_TO;
2753 
2754  xprt->ops = &xs_tcp_ops;
2755  xprt->timeout = &xs_tcp_default_timeout;
2756 
2757  switch (addr->sa_family) {
2758  case AF_INET:
2759  if (((struct sockaddr_in *)addr)->sin_port != htons(0))
2760  xprt_set_bound(xprt);
2761 
2762  INIT_DELAYED_WORK(&transport->connect_worker,
2763  xs_tcp_setup_socket);
2764  xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
2765  break;
2766  case AF_INET6:
2767  if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
2768  xprt_set_bound(xprt);
2769 
2770  INIT_DELAYED_WORK(&transport->connect_worker,
2771  xs_tcp_setup_socket);
2772  xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
2773  break;
2774  default:
2775  ret = ERR_PTR(-EAFNOSUPPORT);
2776  goto out_err;
2777  }
2778 
2779  if (xprt_bound(xprt))
2780  dprintk("RPC: set up xprt to %s (port %s) via %s\n",
2781  xprt->address_strings[RPC_DISPLAY_ADDR],
2782  xprt->address_strings[RPC_DISPLAY_PORT],
2783  xprt->address_strings[RPC_DISPLAY_PROTO]);
2784  else
2785  dprintk("RPC: set up xprt to %s (autobind) via %s\n",
2786  xprt->address_strings[RPC_DISPLAY_ADDR],
2787  xprt->address_strings[RPC_DISPLAY_PROTO]);
2788 
2789 
2790  if (try_module_get(THIS_MODULE))
2791  return xprt;
2792  ret = ERR_PTR(-EINVAL);
2793 out_err:
2794  xprt_free(xprt);
2795  return ret;
2796 }
2797 
2803 static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
2804 {
2805  struct sockaddr *addr = args->dstaddr;
2806  struct rpc_xprt *xprt;
2807  struct sock_xprt *transport;
2808  struct svc_sock *bc_sock;
2809  struct rpc_xprt *ret;
2810 
2811  if (args->bc_xprt->xpt_bc_xprt) {
2812  /*
2813  * This server connection already has a backchannel
2814  * export; we can't create a new one, as we wouldn't be
2815  * able to match replies based on xid any more. So,
2816  * reuse the already-existing one:
2817  */
2818  return args->bc_xprt->xpt_bc_xprt;
2819  }
2820  xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
2821  xprt_tcp_slot_table_entries);
2822  if (IS_ERR(xprt))
2823  return xprt;
2824  transport = container_of(xprt, struct sock_xprt, xprt);
2825 
2826  xprt->prot = IPPROTO_TCP;
2827  xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2828  xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2829  xprt->timeout = &xs_tcp_default_timeout;
2830 
2831  /* backchannel */
2832  xprt_set_bound(xprt);
2833  xprt->bind_timeout = 0;
2834  xprt->reestablish_timeout = 0;
2835  xprt->idle_timeout = 0;
2836 
2837  xprt->ops = &bc_tcp_ops;
2838 
2839  switch (addr->sa_family) {
2840  case AF_INET:
2841  xs_format_peer_addresses(xprt, "tcp",
2842  RPCBIND_NETID_TCP);
2843  break;
2844  case AF_INET6:
2845  xs_format_peer_addresses(xprt, "tcp",
2846  RPCBIND_NETID_TCP6);
2847  break;
2848  default:
2849  ret = ERR_PTR(-EAFNOSUPPORT);
2850  goto out_err;
2851  }
2852 
2853  dprintk("RPC: set up xprt to %s (port %s) via %s\n",
2854  xprt->address_strings[RPC_DISPLAY_ADDR],
2855  xprt->address_strings[RPC_DISPLAY_PORT],
2856  xprt->address_strings[RPC_DISPLAY_PROTO]);
2857 
2858  /*
2859  * Once we've associated a backchannel xprt with a connection,
2860  * we want to keep it around as long as long as the connection
2861  * lasts, in case we need to start using it for a backchannel
2862  * again; this reference won't be dropped until bc_xprt is
2863  * destroyed.
2864  */
2865  xprt_get(xprt);
2866  args->bc_xprt->xpt_bc_xprt = xprt;
2867  xprt->bc_xprt = args->bc_xprt;
2868  bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
2869  transport->sock = bc_sock->sk_sock;
2870  transport->inet = bc_sock->sk_sk;
2871 
2872  /*
2873  * Since we don't want connections for the backchannel, we set
2874  * the xprt status to connected
2875  */
2876  xprt_set_connected(xprt);
2877 
2878 
2879  if (try_module_get(THIS_MODULE))
2880  return xprt;
2881  xprt_put(xprt);
2882  ret = ERR_PTR(-EINVAL);
2883 out_err:
2884  xprt_free(xprt);
2885  return ret;
2886 }
2887 
2888 static struct xprt_class xs_local_transport = {
2889  .list = LIST_HEAD_INIT(xs_local_transport.list),
2890  .name = "named UNIX socket",
2891  .owner = THIS_MODULE,
2892  .ident = XPRT_TRANSPORT_LOCAL,
2893  .setup = xs_setup_local,
2894 };
2895 
2896 static struct xprt_class xs_udp_transport = {
2897  .list = LIST_HEAD_INIT(xs_udp_transport.list),
2898  .name = "udp",
2899  .owner = THIS_MODULE,
2900  .ident = XPRT_TRANSPORT_UDP,
2901  .setup = xs_setup_udp,
2902 };
2903 
2904 static struct xprt_class xs_tcp_transport = {
2905  .list = LIST_HEAD_INIT(xs_tcp_transport.list),
2906  .name = "tcp",
2907  .owner = THIS_MODULE,
2908  .ident = XPRT_TRANSPORT_TCP,
2909  .setup = xs_setup_tcp,
2910 };
2911 
2912 static struct xprt_class xs_bc_tcp_transport = {
2913  .list = LIST_HEAD_INIT(xs_bc_tcp_transport.list),
2914  .name = "tcp NFSv4.1 backchannel",
2915  .owner = THIS_MODULE,
2916  .ident = XPRT_TRANSPORT_BC_TCP,
2917  .setup = xs_setup_bc_tcp,
2918 };
2919 
2924 int init_socket_xprt(void)
2925 {
2926 #ifdef RPC_DEBUG
2927  if (!sunrpc_table_header)
2928  sunrpc_table_header = register_sysctl_table(sunrpc_table);
2929 #endif
2930 
2931  xprt_register_transport(&xs_local_transport);
2932  xprt_register_transport(&xs_udp_transport);
2933  xprt_register_transport(&xs_tcp_transport);
2934  xprt_register_transport(&xs_bc_tcp_transport);
2935 
2936  return 0;
2937 }
2938 
2943 void cleanup_socket_xprt(void)
2944 {
2945 #ifdef RPC_DEBUG
2946  if (sunrpc_table_header) {
2947  unregister_sysctl_table(sunrpc_table_header);
2948  sunrpc_table_header = NULL;
2949  }
2950 #endif
2951 
2952  xprt_unregister_transport(&xs_local_transport);
2953  xprt_unregister_transport(&xs_udp_transport);
2954  xprt_unregister_transport(&xs_tcp_transport);
2955  xprt_unregister_transport(&xs_bc_tcp_transport);
2956 }
2957 
2958 static int param_set_uint_minmax(const char *val,
2959  const struct kernel_param *kp,
2960  unsigned int min, unsigned int max)
2961 {
2962  unsigned long num;
2963  int ret;
2964 
2965  if (!val)
2966  return -EINVAL;
2967  ret = strict_strtoul(val, 0, &num);
2968  if (ret == -EINVAL || num < min || num > max)
2969  return -EINVAL;
2970  *((unsigned int *)kp->arg) = num;
2971  return 0;
2972 }
2973 
2974 static int param_set_portnr(const char *val, const struct kernel_param *kp)
2975 {
2976  return param_set_uint_minmax(val, kp,
2977  RPC_MIN_RESVPORT,
2978  RPC_MAX_RESVPORT);
2979 }
2980 
2981 static struct kernel_param_ops param_ops_portnr = {
2982  .set = param_set_portnr,
2983  .get = param_get_uint,
2984 };
2985 
2986 #define param_check_portnr(name, p) \
2987  __param_check(name, p, unsigned int);
2988 
2989 module_param_named(min_resvport, xprt_min_resvport, portnr, 0644);
2990 module_param_named(max_resvport, xprt_max_resvport, portnr, 0644);
2991 
2992 static int param_set_slot_table_size(const char *val,
2993  const struct kernel_param *kp)
2994 {
2995  return param_set_uint_minmax(val, kp,
2996  RPC_MIN_SLOT_TABLE,
2997  RPC_MAX_SLOT_TABLE);
2998 }
2999 
3000 static struct kernel_param_ops param_ops_slot_table_size = {
3001  .set = param_set_slot_table_size,
3002  .get = param_get_uint,
3003 };
3004 
3005 #define param_check_slot_table_size(name, p) \
3006  __param_check(name, p, unsigned int);
3007 
3008 static int param_set_max_slot_table_size(const char *val,
3009  const struct kernel_param *kp)
3010 {
3011  return param_set_uint_minmax(val, kp,
3012  RPC_MIN_SLOT_TABLE,
3013  RPC_MAX_SLOT_TABLE_LIMIT);
3014 }
3015 
3016 static struct kernel_param_ops param_ops_max_slot_table_size = {
3017  .set = param_set_max_slot_table_size,
3018  .get = param_get_uint,
3019 };
3020 
3021 #define param_check_max_slot_table_size(name, p) \
3022  __param_check(name, p, unsigned int);
3023 
3024 module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
3025  slot_table_size, 0644);
3026 module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
3027  max_slot_table_size, 0644);
3028 module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
3029  slot_table_size, 0644);
3030 
Generated on Thu Jan 10 2013 15:02:21 for Linux Kernel by   doxygen 1.8.2