~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/net/sunrpc/xprtsock.c

Version: ~ [ linux-5.9 ] ~ [ linux-5.8.14 ] ~ [ linux-5.7.19 ] ~ [ linux-5.6.19 ] ~ [ linux-5.5.19 ] ~ [ linux-5.4.70 ] ~ [ linux-5.3.18 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.150 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.200 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.238 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.238 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.85 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

  1 // SPDX-License-Identifier: GPL-2.0
  2 /*
  3  * linux/net/sunrpc/xprtsock.c
  4  *
  5  * Client-side transport implementation for sockets.
  6  *
  7  * TCP callback races fixes (C) 1998 Red Hat
  8  * TCP send fixes (C) 1998 Red Hat
  9  * TCP NFS related read + write fixes
 10  *  (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
 11  *
 12  * Rewrite of larges part of the code in order to stabilize TCP stuff.
 13  * Fix behaviour when socket buffer is full.
 14  *  (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
 15  *
 16  * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com>
 17  *
 18  * IPv6 support contributed by Gilles Quillard, Bull Open Source, 2005.
 19  *   <gilles.quillard@bull.net>
 20  */
 21 
 22 #include <linux/types.h>
 23 #include <linux/string.h>
 24 #include <linux/slab.h>
 25 #include <linux/module.h>
 26 #include <linux/capability.h>
 27 #include <linux/pagemap.h>
 28 #include <linux/errno.h>
 29 #include <linux/socket.h>
 30 #include <linux/in.h>
 31 #include <linux/net.h>
 32 #include <linux/mm.h>
 33 #include <linux/un.h>
 34 #include <linux/udp.h>
 35 #include <linux/tcp.h>
 36 #include <linux/sunrpc/clnt.h>
 37 #include <linux/sunrpc/addr.h>
 38 #include <linux/sunrpc/sched.h>
 39 #include <linux/sunrpc/svcsock.h>
 40 #include <linux/sunrpc/xprtsock.h>
 41 #include <linux/file.h>
 42 #ifdef CONFIG_SUNRPC_BACKCHANNEL
 43 #include <linux/sunrpc/bc_xprt.h>
 44 #endif
 45 
 46 #include <net/sock.h>
 47 #include <net/checksum.h>
 48 #include <net/udp.h>
 49 #include <net/tcp.h>
 50 
 51 #include <trace/events/sunrpc.h>
 52 
 53 #include "sunrpc.h"
 54 
 55 static void xs_close(struct rpc_xprt *xprt);
 56 static void xs_tcp_set_socket_timeouts(struct rpc_xprt *xprt,
 57                 struct socket *sock);
 58 
 59 /*
 60  * xprtsock tunables
 61  */
 62 static unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
 63 static unsigned int xprt_tcp_slot_table_entries = RPC_MIN_SLOT_TABLE;
 64 static unsigned int xprt_max_tcp_slot_table_entries = RPC_MAX_SLOT_TABLE;
 65 
 66 static unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
 67 static unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
 68 
 69 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 70 
 71 #define XS_TCP_LINGER_TO        (15U * HZ)
 72 static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
 73 
 74 /*
 75  * We can register our own files under /proc/sys/sunrpc by
 76  * calling register_sysctl_table() again.  The files in that
 77  * directory become the union of all files registered there.
 78  *
 79  * We simply need to make sure that we don't collide with
 80  * someone else's file names!
 81  */
 82 
 83 static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
 84 static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
 85 static unsigned int max_tcp_slot_table_limit = RPC_MAX_SLOT_TABLE_LIMIT;
 86 static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
 87 static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
 88 
 89 static struct ctl_table_header *sunrpc_table_header;
 90 
 91 /*
 92  * FIXME: changing the UDP slot table size should also resize the UDP
 93  *        socket buffers for existing UDP transports
 94  */
 95 static struct ctl_table xs_tunables_table[] = {
 96         {
 97                 .procname       = "udp_slot_table_entries",
 98                 .data           = &xprt_udp_slot_table_entries,
 99                 .maxlen         = sizeof(unsigned int),
100                 .mode           = 0644,
101                 .proc_handler   = proc_dointvec_minmax,
102                 .extra1         = &min_slot_table_size,
103                 .extra2         = &max_slot_table_size
104         },
105         {
106                 .procname       = "tcp_slot_table_entries",
107                 .data           = &xprt_tcp_slot_table_entries,
108                 .maxlen         = sizeof(unsigned int),
109                 .mode           = 0644,
110                 .proc_handler   = proc_dointvec_minmax,
111                 .extra1         = &min_slot_table_size,
112                 .extra2         = &max_slot_table_size
113         },
114         {
115                 .procname       = "tcp_max_slot_table_entries",
116                 .data           = &xprt_max_tcp_slot_table_entries,
117                 .maxlen         = sizeof(unsigned int),
118                 .mode           = 0644,
119                 .proc_handler   = proc_dointvec_minmax,
120                 .extra1         = &min_slot_table_size,
121                 .extra2         = &max_tcp_slot_table_limit
122         },
123         {
124                 .procname       = "min_resvport",
125                 .data           = &xprt_min_resvport,
126                 .maxlen         = sizeof(unsigned int),
127                 .mode           = 0644,
128                 .proc_handler   = proc_dointvec_minmax,
129                 .extra1         = &xprt_min_resvport_limit,
130                 .extra2         = &xprt_max_resvport
131         },
132         {
133                 .procname       = "max_resvport",
134                 .data           = &xprt_max_resvport,
135                 .maxlen         = sizeof(unsigned int),
136                 .mode           = 0644,
137                 .proc_handler   = proc_dointvec_minmax,
138                 .extra1         = &xprt_min_resvport,
139                 .extra2         = &xprt_max_resvport_limit
140         },
141         {
142                 .procname       = "tcp_fin_timeout",
143                 .data           = &xs_tcp_fin_timeout,
144                 .maxlen         = sizeof(xs_tcp_fin_timeout),
145                 .mode           = 0644,
146                 .proc_handler   = proc_dointvec_jiffies,
147         },
148         { },
149 };
150 
151 static struct ctl_table sunrpc_table[] = {
152         {
153                 .procname       = "sunrpc",
154                 .mode           = 0555,
155                 .child          = xs_tunables_table
156         },
157         { },
158 };
159 
160 #endif
161 
162 /*
163  * Wait duration for a reply from the RPC portmapper.
164  */
165 #define XS_BIND_TO              (60U * HZ)
166 
167 /*
168  * Delay if a UDP socket connect error occurs.  This is most likely some
169  * kind of resource problem on the local host.
170  */
171 #define XS_UDP_REEST_TO         (2U * HZ)
172 
173 /*
174  * The reestablish timeout allows clients to delay for a bit before attempting
175  * to reconnect to a server that just dropped our connection.
176  *
177  * We implement an exponential backoff when trying to reestablish a TCP
178  * transport connection with the server.  Some servers like to drop a TCP
179  * connection when they are overworked, so we start with a short timeout and
180  * increase over time if the server is down or not responding.
181  */
182 #define XS_TCP_INIT_REEST_TO    (3U * HZ)
183 
184 /*
185  * TCP idle timeout; client drops the transport socket if it is idle
186  * for this long.  Note that we also timeout UDP sockets to prevent
187  * holding port numbers when there is no RPC traffic.
188  */
189 #define XS_IDLE_DISC_TO         (5U * 60 * HZ)
190 
191 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
192 # undef  RPC_DEBUG_DATA
193 # define RPCDBG_FACILITY        RPCDBG_TRANS
194 #endif
195 
196 #ifdef RPC_DEBUG_DATA
197 static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
198 {
199         u8 *buf = (u8 *) packet;
200         int j;
201 
202         dprintk("RPC:       %s\n", msg);
203         for (j = 0; j < count && j < 128; j += 4) {
204                 if (!(j & 31)) {
205                         if (j)
206                                 dprintk("\n");
207                         dprintk("0x%04x ", j);
208                 }
209                 dprintk("%02x%02x%02x%02x ",
210                         buf[j], buf[j+1], buf[j+2], buf[j+3]);
211         }
212         dprintk("\n");
213 }
214 #else
215 static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
216 {
217         /* NOP */
218 }
219 #endif
220 
221 static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
222 {
223         return (struct rpc_xprt *) sk->sk_user_data;
224 }
225 
226 static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
227 {
228         return (struct sockaddr *) &xprt->addr;
229 }
230 
231 static inline struct sockaddr_un *xs_addr_un(struct rpc_xprt *xprt)
232 {
233         return (struct sockaddr_un *) &xprt->addr;
234 }
235 
236 static inline struct sockaddr_in *xs_addr_in(struct rpc_xprt *xprt)
237 {
238         return (struct sockaddr_in *) &xprt->addr;
239 }
240 
241 static inline struct sockaddr_in6 *xs_addr_in6(struct rpc_xprt *xprt)
242 {
243         return (struct sockaddr_in6 *) &xprt->addr;
244 }
245 
246 static void xs_format_common_peer_addresses(struct rpc_xprt *xprt)
247 {
248         struct sockaddr *sap = xs_addr(xprt);
249         struct sockaddr_in6 *sin6;
250         struct sockaddr_in *sin;
251         struct sockaddr_un *sun;
252         char buf[128];
253 
254         switch (sap->sa_family) {
255         case AF_LOCAL:
256                 sun = xs_addr_un(xprt);
257                 strlcpy(buf, sun->sun_path, sizeof(buf));
258                 xprt->address_strings[RPC_DISPLAY_ADDR] =
259                                                 kstrdup(buf, GFP_KERNEL);
260                 break;
261         case AF_INET:
262                 (void)rpc_ntop(sap, buf, sizeof(buf));
263                 xprt->address_strings[RPC_DISPLAY_ADDR] =
264                                                 kstrdup(buf, GFP_KERNEL);
265                 sin = xs_addr_in(xprt);
266                 snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
267                 break;
268         case AF_INET6:
269                 (void)rpc_ntop(sap, buf, sizeof(buf));
270                 xprt->address_strings[RPC_DISPLAY_ADDR] =
271                                                 kstrdup(buf, GFP_KERNEL);
272                 sin6 = xs_addr_in6(xprt);
273                 snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
274                 break;
275         default:
276                 BUG();
277         }
278 
279         xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
280 }
281 
282 static void xs_format_common_peer_ports(struct rpc_xprt *xprt)
283 {
284         struct sockaddr *sap = xs_addr(xprt);
285         char buf[128];
286 
287         snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
288         xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
289 
290         snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
291         xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
292 }
293 
294 static void xs_format_peer_addresses(struct rpc_xprt *xprt,
295                                      const char *protocol,
296                                      const char *netid)
297 {
298         xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
299         xprt->address_strings[RPC_DISPLAY_NETID] = netid;
300         xs_format_common_peer_addresses(xprt);
301         xs_format_common_peer_ports(xprt);
302 }
303 
304 static void xs_update_peer_port(struct rpc_xprt *xprt)
305 {
306         kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]);
307         kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
308 
309         xs_format_common_peer_ports(xprt);
310 }
311 
312 static void xs_free_peer_addresses(struct rpc_xprt *xprt)
313 {
314         unsigned int i;
315 
316         for (i = 0; i < RPC_DISPLAY_MAX; i++)
317                 switch (i) {
318                 case RPC_DISPLAY_PROTO:
319                 case RPC_DISPLAY_NETID:
320                         continue;
321                 default:
322                         kfree(xprt->address_strings[i]);
323                 }
324 }
325 
326 #define XS_SENDMSG_FLAGS        (MSG_DONTWAIT | MSG_NOSIGNAL)
327 
328 static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
329 {
330         struct msghdr msg = {
331                 .msg_name       = addr,
332                 .msg_namelen    = addrlen,
333                 .msg_flags      = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
334         };
335         struct kvec iov = {
336                 .iov_base       = vec->iov_base + base,
337                 .iov_len        = vec->iov_len - base,
338         };
339 
340         if (iov.iov_len != 0)
341                 return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
342         return kernel_sendmsg(sock, &msg, NULL, 0, 0);
343 }
344 
345 static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more, bool zerocopy, int *sent_p)
346 {
347         ssize_t (*do_sendpage)(struct socket *sock, struct page *page,
348                         int offset, size_t size, int flags);
349         struct page **ppage;
350         unsigned int remainder;
351         int err;
352 
353         remainder = xdr->page_len - base;
354         base += xdr->page_base;
355         ppage = xdr->pages + (base >> PAGE_SHIFT);
356         base &= ~PAGE_MASK;
357         do_sendpage = sock->ops->sendpage;
358         if (!zerocopy)
359                 do_sendpage = sock_no_sendpage;
360         for(;;) {
361                 unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
362                 int flags = XS_SENDMSG_FLAGS;
363 
364                 remainder -= len;
365                 if (more)
366                         flags |= MSG_MORE;
367                 if (remainder != 0)
368                         flags |= MSG_SENDPAGE_NOTLAST | MSG_MORE;
369                 err = do_sendpage(sock, *ppage, base, len, flags);
370                 if (remainder == 0 || err != len)
371                         break;
372                 *sent_p += err;
373                 ppage++;
374                 base = 0;
375         }
376         if (err > 0) {
377                 *sent_p += err;
378                 err = 0;
379         }
380         return err;
381 }
382 
383 /**
384  * xs_sendpages - write pages directly to a socket
385  * @sock: socket to send on
386  * @addr: UDP only -- address of destination
387  * @addrlen: UDP only -- length of destination address
388  * @xdr: buffer containing this request
389  * @base: starting position in the buffer
390  * @zerocopy: true if it is safe to use sendpage()
391  * @sent_p: return the total number of bytes successfully queued for sending
392  *
393  */
394 static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base, bool zerocopy, int *sent_p)
395 {
396         unsigned int remainder = xdr->len - base;
397         int err = 0;
398         int sent = 0;
399 
400         if (unlikely(!sock))
401                 return -ENOTSOCK;
402 
403         if (base != 0) {
404                 addr = NULL;
405                 addrlen = 0;
406         }
407 
408         if (base < xdr->head[0].iov_len || addr != NULL) {
409                 unsigned int len = xdr->head[0].iov_len - base;
410                 remainder -= len;
411                 err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
412                 if (remainder == 0 || err != len)
413                         goto out;
414                 *sent_p += err;
415                 base = 0;
416         } else
417                 base -= xdr->head[0].iov_len;
418 
419         if (base < xdr->page_len) {
420                 unsigned int len = xdr->page_len - base;
421                 remainder -= len;
422                 err = xs_send_pagedata(sock, xdr, base, remainder != 0, zerocopy, &sent);
423                 *sent_p += sent;
424                 if (remainder == 0 || sent != len)
425                         goto out;
426                 base = 0;
427         } else
428                 base -= xdr->page_len;
429 
430         if (base >= xdr->tail[0].iov_len)
431                 return 0;
432         err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
433 out:
434         if (err > 0) {
435                 *sent_p += err;
436                 err = 0;
437         }
438         return err;
439 }
440 
441 static void xs_nospace_callback(struct rpc_task *task)
442 {
443         struct sock_xprt *transport = container_of(task->tk_rqstp->rq_xprt, struct sock_xprt, xprt);
444 
445         transport->inet->sk_write_pending--;
446 }
447 
448 /**
449  * xs_nospace - place task on wait queue if transmit was incomplete
450  * @task: task to put to sleep
451  *
452  */
453 static int xs_nospace(struct rpc_task *task)
454 {
455         struct rpc_rqst *req = task->tk_rqstp;
456         struct rpc_xprt *xprt = req->rq_xprt;
457         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
458         struct sock *sk = transport->inet;
459         int ret = -EAGAIN;
460 
461         dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
462                         task->tk_pid, req->rq_slen - req->rq_bytes_sent,
463                         req->rq_slen);
464 
465         /* Protect against races with write_space */
466         spin_lock_bh(&xprt->transport_lock);
467 
468         /* Don't race with disconnect */
469         if (xprt_connected(xprt)) {
470                 /* wait for more buffer space */
471                 sk->sk_write_pending++;
472                 xprt_wait_for_buffer_space(task, xs_nospace_callback);
473         } else
474                 ret = -ENOTCONN;
475 
476         spin_unlock_bh(&xprt->transport_lock);
477 
478         /* Race breaker in case memory is freed before above code is called */
479         if (ret == -EAGAIN) {
480                 struct socket_wq *wq;
481 
482                 rcu_read_lock();
483                 wq = rcu_dereference(sk->sk_wq);
484                 set_bit(SOCKWQ_ASYNC_NOSPACE, &wq->flags);
485                 rcu_read_unlock();
486 
487                 sk->sk_write_space(sk);
488         }
489         return ret;
490 }
491 
492 /*
493  * Construct a stream transport record marker in @buf.
494  */
495 static inline void xs_encode_stream_record_marker(struct xdr_buf *buf)
496 {
497         u32 reclen = buf->len - sizeof(rpc_fraghdr);
498         rpc_fraghdr *base = buf->head[0].iov_base;
499         *base = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT | reclen);
500 }
501 
502 /**
503  * xs_local_send_request - write an RPC request to an AF_LOCAL socket
504  * @task: RPC task that manages the state of an RPC request
505  *
506  * Return values:
507  *        0:    The request has been sent
508  *   EAGAIN:    The socket was blocked, please call again later to
509  *              complete the request
510  * ENOTCONN:    Caller needs to invoke connect logic then call again
511  *    other:    Some other error occured, the request was not sent
512  */
513 static int xs_local_send_request(struct rpc_task *task)
514 {
515         struct rpc_rqst *req = task->tk_rqstp;
516         struct rpc_xprt *xprt = req->rq_xprt;
517         struct sock_xprt *transport =
518                                 container_of(xprt, struct sock_xprt, xprt);
519         struct xdr_buf *xdr = &req->rq_snd_buf;
520         int status;
521         int sent = 0;
522 
523         xs_encode_stream_record_marker(&req->rq_snd_buf);
524 
525         xs_pktdump("packet data:",
526                         req->rq_svec->iov_base, req->rq_svec->iov_len);
527 
528         status = xs_sendpages(transport->sock, NULL, 0, xdr, req->rq_bytes_sent,
529                               true, &sent);
530         dprintk("RPC:       %s(%u) = %d\n",
531                         __func__, xdr->len - req->rq_bytes_sent, status);
532 
533         if (status == -EAGAIN && sock_writeable(transport->inet))
534                 status = -ENOBUFS;
535 
536         if (likely(sent > 0) || status == 0) {
537                 req->rq_bytes_sent += sent;
538                 req->rq_xmit_bytes_sent += sent;
539                 if (likely(req->rq_bytes_sent >= req->rq_slen)) {
540                         req->rq_bytes_sent = 0;
541                         return 0;
542                 }
543                 status = -EAGAIN;
544         }
545 
546         switch (status) {
547         case -ENOBUFS:
548                 break;
549         case -EAGAIN:
550                 status = xs_nospace(task);
551                 break;
552         default:
553                 dprintk("RPC:       sendmsg returned unrecognized error %d\n",
554                         -status);
555                 /* fall through */
556         case -EPIPE:
557                 xs_close(xprt);
558                 status = -ENOTCONN;
559         }
560 
561         return status;
562 }
563 
564 /**
565  * xs_udp_send_request - write an RPC request to a UDP socket
566  * @task: address of RPC task that manages the state of an RPC request
567  *
568  * Return values:
569  *        0:    The request has been sent
570  *   EAGAIN:    The socket was blocked, please call again later to
571  *              complete the request
572  * ENOTCONN:    Caller needs to invoke connect logic then call again
573  *    other:    Some other error occurred, the request was not sent
574  */
575 static int xs_udp_send_request(struct rpc_task *task)
576 {
577         struct rpc_rqst *req = task->tk_rqstp;
578         struct rpc_xprt *xprt = req->rq_xprt;
579         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
580         struct xdr_buf *xdr = &req->rq_snd_buf;
581         int sent = 0;
582         int status;
583 
584         xs_pktdump("packet data:",
585                                 req->rq_svec->iov_base,
586                                 req->rq_svec->iov_len);
587 
588         if (!xprt_bound(xprt))
589                 return -ENOTCONN;
590         status = xs_sendpages(transport->sock, xs_addr(xprt), xprt->addrlen,
591                               xdr, req->rq_bytes_sent, true, &sent);
592 
593         dprintk("RPC:       xs_udp_send_request(%u) = %d\n",
594                         xdr->len - req->rq_bytes_sent, status);
595 
596         /* firewall is blocking us, don't return -EAGAIN or we end up looping */
597         if (status == -EPERM)
598                 goto process_status;
599 
600         if (status == -EAGAIN && sock_writeable(transport->inet))
601                 status = -ENOBUFS;
602 
603         if (sent > 0 || status == 0) {
604                 req->rq_xmit_bytes_sent += sent;
605                 if (sent >= req->rq_slen)
606                         return 0;
607                 /* Still some bytes left; set up for a retry later. */
608                 status = -EAGAIN;
609         }
610 
611 process_status:
612         switch (status) {
613         case -ENOTSOCK:
614                 status = -ENOTCONN;
615                 /* Should we call xs_close() here? */
616                 break;
617         case -EAGAIN:
618                 status = xs_nospace(task);
619                 break;
620         case -ENETUNREACH:
621         case -ENOBUFS:
622         case -EPIPE:
623         case -ECONNREFUSED:
624         case -EPERM:
625                 /* When the server has died, an ICMP port unreachable message
626                  * prompts ECONNREFUSED. */
627                 break;
628         default:
629                 dprintk("RPC:       sendmsg returned unrecognized error %d\n",
630                         -status);
631         }
632 
633         return status;
634 }
635 
636 /**
637  * xs_tcp_send_request - write an RPC request to a TCP socket
638  * @task: address of RPC task that manages the state of an RPC request
639  *
640  * Return values:
641  *        0:    The request has been sent
642  *   EAGAIN:    The socket was blocked, please call again later to
643  *              complete the request
644  * ENOTCONN:    Caller needs to invoke connect logic then call again
645  *    other:    Some other error occurred, the request was not sent
646  *
647  * XXX: In the case of soft timeouts, should we eventually give up
648  *      if sendmsg is not able to make progress?
649  */
650 static int xs_tcp_send_request(struct rpc_task *task)
651 {
652         struct rpc_rqst *req = task->tk_rqstp;
653         struct rpc_xprt *xprt = req->rq_xprt;
654         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
655         struct xdr_buf *xdr = &req->rq_snd_buf;
656         bool zerocopy = true;
657         bool vm_wait = false;
658         int status;
659         int sent;
660 
661         xs_encode_stream_record_marker(&req->rq_snd_buf);
662 
663         xs_pktdump("packet data:",
664                                 req->rq_svec->iov_base,
665                                 req->rq_svec->iov_len);
666         /* Don't use zero copy if this is a resend. If the RPC call
667          * completes while the socket holds a reference to the pages,
668          * then we may end up resending corrupted data.
669          */
670         if (task->tk_flags & RPC_TASK_SENT)
671                 zerocopy = false;
672 
673         if (test_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state))
674                 xs_tcp_set_socket_timeouts(xprt, transport->sock);
675 
676         /* Continue transmitting the packet/record. We must be careful
677          * to cope with writespace callbacks arriving _after_ we have
678          * called sendmsg(). */
679         while (1) {
680                 sent = 0;
681                 status = xs_sendpages(transport->sock, NULL, 0, xdr,
682                                       req->rq_bytes_sent, zerocopy, &sent);
683 
684                 dprintk("RPC:       xs_tcp_send_request(%u) = %d\n",
685                                 xdr->len - req->rq_bytes_sent, status);
686 
687                 /* If we've sent the entire packet, immediately
688                  * reset the count of bytes sent. */
689                 req->rq_bytes_sent += sent;
690                 req->rq_xmit_bytes_sent += sent;
691                 if (likely(req->rq_bytes_sent >= req->rq_slen)) {
692                         req->rq_bytes_sent = 0;
693                         return 0;
694                 }
695 
696                 WARN_ON_ONCE(sent == 0 && status == 0);
697 
698                 if (status == -EAGAIN ) {
699                         /*
700                          * Return EAGAIN if we're sure we're hitting the
701                          * socket send buffer limits.
702                          */
703                         if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
704                                 break;
705                         /*
706                          * Did we hit a memory allocation failure?
707                          */
708                         if (sent == 0) {
709                                 status = -ENOBUFS;
710                                 if (vm_wait)
711                                         break;
712                                 /* Retry, knowing now that we're below the
713                                  * socket send buffer limit
714                                  */
715                                 vm_wait = true;
716                         }
717                         continue;
718                 }
719                 if (status < 0)
720                         break;
721                 vm_wait = false;
722         }
723 
724         switch (status) {
725         case -ENOTSOCK:
726                 status = -ENOTCONN;
727                 /* Should we call xs_close() here? */
728                 break;
729         case -EAGAIN:
730                 status = xs_nospace(task);
731                 break;
732         case -ECONNRESET:
733         case -ECONNREFUSED:
734         case -ENOTCONN:
735         case -EADDRINUSE:
736         case -ENOBUFS:
737         case -EPIPE:
738                 break;
739         default:
740                 dprintk("RPC:       sendmsg returned unrecognized error %d\n",
741                         -status);
742         }
743 
744         return status;
745 }
746 
747 /**
748  * xs_tcp_release_xprt - clean up after a tcp transmission
749  * @xprt: transport
750  * @task: rpc task
751  *
752  * This cleans up if an error causes us to abort the transmission of a request.
753  * In this case, the socket may need to be reset in order to avoid confusing
754  * the server.
755  */
756 static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
757 {
758         struct rpc_rqst *req;
759 
760         if (task != xprt->snd_task)
761                 return;
762         if (task == NULL)
763                 goto out_release;
764         req = task->tk_rqstp;
765         if (req == NULL)
766                 goto out_release;
767         if (req->rq_bytes_sent == 0)
768                 goto out_release;
769         if (req->rq_bytes_sent == req->rq_snd_buf.len)
770                 goto out_release;
771         set_bit(XPRT_CLOSE_WAIT, &xprt->state);
772 out_release:
773         xprt_release_xprt(xprt, task);
774 }
775 
776 static void xs_save_old_callbacks(struct sock_xprt *transport, struct sock *sk)
777 {
778         transport->old_data_ready = sk->sk_data_ready;
779         transport->old_state_change = sk->sk_state_change;
780         transport->old_write_space = sk->sk_write_space;
781         transport->old_error_report = sk->sk_error_report;
782 }
783 
784 static void xs_restore_old_callbacks(struct sock_xprt *transport, struct sock *sk)
785 {
786         sk->sk_data_ready = transport->old_data_ready;
787         sk->sk_state_change = transport->old_state_change;
788         sk->sk_write_space = transport->old_write_space;
789         sk->sk_error_report = transport->old_error_report;
790 }
791 
792 static void xs_sock_reset_state_flags(struct rpc_xprt *xprt)
793 {
794         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
795 
796         clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
797 }
798 
799 static void xs_sock_reset_connection_flags(struct rpc_xprt *xprt)
800 {
801         smp_mb__before_atomic();
802         clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
803         clear_bit(XPRT_CLOSING, &xprt->state);
804         xs_sock_reset_state_flags(xprt);
805         smp_mb__after_atomic();
806 }
807 
808 static void xs_sock_mark_closed(struct rpc_xprt *xprt)
809 {
810         xs_sock_reset_connection_flags(xprt);
811         /* Mark transport as closed and wake up all pending tasks */
812         xprt_disconnect_done(xprt);
813 }
814 
815 /**
816  * xs_error_report - callback to handle TCP socket state errors
817  * @sk: socket
818  *
819  * Note: we don't call sock_error() since there may be a rpc_task
820  * using the socket, and so we don't want to clear sk->sk_err.
821  */
822 static void xs_error_report(struct sock *sk)
823 {
824         struct rpc_xprt *xprt;
825         int err;
826 
827         read_lock_bh(&sk->sk_callback_lock);
828         if (!(xprt = xprt_from_sock(sk)))
829                 goto out;
830 
831         err = -sk->sk_err;
832         if (err == 0)
833                 goto out;
834         /* Is this a reset event? */
835         if (sk->sk_state == TCP_CLOSE)
836                 xs_sock_mark_closed(xprt);
837         dprintk("RPC:       xs_error_report client %p, error=%d...\n",
838                         xprt, -err);
839         trace_rpc_socket_error(xprt, sk->sk_socket, err);
840         xprt_wake_pending_tasks(xprt, err);
841  out:
842         read_unlock_bh(&sk->sk_callback_lock);
843 }
844 
845 static void xs_reset_transport(struct sock_xprt *transport)
846 {
847         struct socket *sock = transport->sock;
848         struct sock *sk = transport->inet;
849         struct rpc_xprt *xprt = &transport->xprt;
850 
851         if (sk == NULL)
852                 return;
853 
854         if (atomic_read(&transport->xprt.swapper))
855                 sk_clear_memalloc(sk);
856 
857         kernel_sock_shutdown(sock, SHUT_RDWR);
858 
859         mutex_lock(&transport->recv_mutex);
860         write_lock_bh(&sk->sk_callback_lock);
861         transport->inet = NULL;
862         transport->sock = NULL;
863 
864         sk->sk_user_data = NULL;
865 
866         xs_restore_old_callbacks(transport, sk);
867         xprt_clear_connected(xprt);
868         write_unlock_bh(&sk->sk_callback_lock);
869         xs_sock_reset_connection_flags(xprt);
870         mutex_unlock(&transport->recv_mutex);
871 
872         trace_rpc_socket_close(xprt, sock);
873         sock_release(sock);
874 }
875 
876 /**
877  * xs_close - close a socket
878  * @xprt: transport
879  *
880  * This is used when all requests are complete; ie, no DRC state remains
881  * on the server we want to save.
882  *
883  * The caller _must_ be holding XPRT_LOCKED in order to avoid issues with
884  * xs_reset_transport() zeroing the socket from underneath a writer.
885  */
886 static void xs_close(struct rpc_xprt *xprt)
887 {
888         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
889 
890         dprintk("RPC:       xs_close xprt %p\n", xprt);
891 
892         xs_reset_transport(transport);
893         xprt->reestablish_timeout = 0;
894 
895         xprt_disconnect_done(xprt);
896 }
897 
898 static void xs_inject_disconnect(struct rpc_xprt *xprt)
899 {
900         dprintk("RPC:       injecting transport disconnect on xprt=%p\n",
901                 xprt);
902         xprt_disconnect_done(xprt);
903 }
904 
905 static void xs_xprt_free(struct rpc_xprt *xprt)
906 {
907         xs_free_peer_addresses(xprt);
908         xprt_free(xprt);
909 }
910 
911 /**
912  * xs_destroy - prepare to shutdown a transport
913  * @xprt: doomed transport
914  *
915  */
916 static void xs_destroy(struct rpc_xprt *xprt)
917 {
918         struct sock_xprt *transport = container_of(xprt,
919                         struct sock_xprt, xprt);
920         dprintk("RPC:       xs_destroy xprt %p\n", xprt);
921 
922         cancel_delayed_work_sync(&transport->connect_worker);
923         xs_close(xprt);
924         cancel_work_sync(&transport->recv_worker);
925         xs_xprt_free(xprt);
926         module_put(THIS_MODULE);
927 }
928 
929 static int xs_local_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb)
930 {
931         struct xdr_skb_reader desc = {
932                 .skb            = skb,
933                 .offset         = sizeof(rpc_fraghdr),
934                 .count          = skb->len - sizeof(rpc_fraghdr),
935         };
936 
937         if (xdr_partial_copy_from_skb(xdr, 0, &desc, xdr_skb_read_bits) < 0)
938                 return -1;
939         if (desc.count)
940                 return -1;
941         return 0;
942 }
943 
944 /**
945  * xs_local_data_read_skb
946  * @xprt: transport
947  * @sk: socket
948  * @skb: skbuff
949  *
950  * Currently this assumes we can read the whole reply in a single gulp.
951  */
952 static void xs_local_data_read_skb(struct rpc_xprt *xprt,
953                 struct sock *sk,
954                 struct sk_buff *skb)
955 {
956         struct rpc_task *task;
957         struct rpc_rqst *rovr;
958         int repsize, copied;
959         u32 _xid;
960         __be32 *xp;
961 
962         repsize = skb->len - sizeof(rpc_fraghdr);
963         if (repsize < 4) {
964                 dprintk("RPC:       impossible RPC reply size %d\n", repsize);
965                 return;
966         }
967 
968         /* Copy the XID from the skb... */
969         xp = skb_header_pointer(skb, sizeof(rpc_fraghdr), sizeof(_xid), &_xid);
970         if (xp == NULL)
971                 return;
972 
973         /* Look up and lock the request corresponding to the given XID */
974         spin_lock(&xprt->recv_lock);
975         rovr = xprt_lookup_rqst(xprt, *xp);
976         if (!rovr)
977                 goto out_unlock;
978         xprt_pin_rqst(rovr);
979         spin_unlock(&xprt->recv_lock);
980         task = rovr->rq_task;
981 
982         copied = rovr->rq_private_buf.buflen;
983         if (copied > repsize)
984                 copied = repsize;
985 
986         if (xs_local_copy_to_xdr(&rovr->rq_private_buf, skb)) {
987                 dprintk("RPC:       sk_buff copy failed\n");
988                 spin_lock(&xprt->recv_lock);
989                 goto out_unpin;
990         }
991 
992         spin_lock(&xprt->recv_lock);
993         xprt_complete_rqst(task, copied);
994 out_unpin:
995         xprt_unpin_rqst(rovr);
996  out_unlock:
997         spin_unlock(&xprt->recv_lock);
998 }
999 
1000 static void xs_local_data_receive(struct sock_xprt *transport)
1001 {
1002         struct sk_buff *skb;
1003         struct sock *sk;
1004         int err;
1005 
1006         mutex_lock(&transport->recv_mutex);
1007         sk = transport->inet;
1008         if (sk == NULL)
1009                 goto out;
1010         for (;;) {
1011                 skb = skb_recv_datagram(sk, 0, 1, &err);
1012                 if (skb != NULL) {
1013                         xs_local_data_read_skb(&transport->xprt, sk, skb);
1014                         skb_free_datagram(sk, skb);
1015                         continue;
1016                 }
1017                 if (!test_and_clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
1018                         break;
1019         }
1020 out:
1021         mutex_unlock(&transport->recv_mutex);
1022 }
1023 
1024 static void xs_local_data_receive_workfn(struct work_struct *work)
1025 {
1026         struct sock_xprt *transport =
1027                 container_of(work, struct sock_xprt, recv_worker);
1028         xs_local_data_receive(transport);
1029 }
1030 
1031 /**
1032  * xs_udp_data_read_skb - receive callback for UDP sockets
1033  * @xprt: transport
1034  * @sk: socket
1035  * @skb: skbuff
1036  *
1037  */
1038 static void xs_udp_data_read_skb(struct rpc_xprt *xprt,
1039                 struct sock *sk,
1040                 struct sk_buff *skb)
1041 {
1042         struct rpc_task *task;
1043         struct rpc_rqst *rovr;
1044         int repsize, copied;
1045         u32 _xid;
1046         __be32 *xp;
1047 
1048         repsize = skb->len;
1049         if (repsize < 4) {
1050                 dprintk("RPC:       impossible RPC reply size %d!\n", repsize);
1051                 return;
1052         }
1053 
1054         /* Copy the XID from the skb... */
1055         xp = skb_header_pointer(skb, 0, sizeof(_xid), &_xid);
1056         if (xp == NULL)
1057                 return;
1058 
1059         /* Look up and lock the request corresponding to the given XID */
1060         spin_lock(&xprt->recv_lock);
1061         rovr = xprt_lookup_rqst(xprt, *xp);
1062         if (!rovr)
1063                 goto out_unlock;
1064         xprt_pin_rqst(rovr);
1065         spin_unlock(&xprt->recv_lock);
1066         task = rovr->rq_task;
1067 
1068         if ((copied = rovr->rq_private_buf.buflen) > repsize)
1069                 copied = repsize;
1070 
1071         /* Suck it into the iovec, verify checksum if not done by hw. */
1072         if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
1073                 __UDPX_INC_STATS(sk, UDP_MIB_INERRORS);
1074                 spin_lock(&xprt->recv_lock);
1075                 goto out_unpin;
1076         }
1077 
1078         __UDPX_INC_STATS(sk, UDP_MIB_INDATAGRAMS);
1079 
1080         spin_lock_bh(&xprt->transport_lock);
1081         xprt_adjust_cwnd(xprt, task, copied);
1082         spin_unlock_bh(&xprt->transport_lock);
1083         spin_lock(&xprt->recv_lock);
1084         xprt_complete_rqst(task, copied);
1085 out_unpin:
1086         xprt_unpin_rqst(rovr);
1087  out_unlock:
1088         spin_unlock(&xprt->recv_lock);
1089 }
1090 
1091 static void xs_udp_data_receive(struct sock_xprt *transport)
1092 {
1093         struct sk_buff *skb;
1094         struct sock *sk;
1095         int err;
1096 
1097         mutex_lock(&transport->recv_mutex);
1098         sk = transport->inet;
1099         if (sk == NULL)
1100                 goto out;
1101         for (;;) {
1102                 skb = skb_recv_udp(sk, 0, 1, &err);
1103                 if (skb != NULL) {
1104                         xs_udp_data_read_skb(&transport->xprt, sk, skb);
1105                         consume_skb(skb);
1106                         continue;
1107                 }
1108                 if (!test_and_clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
1109                         break;
1110         }
1111 out:
1112         mutex_unlock(&transport->recv_mutex);
1113 }
1114 
1115 static void xs_udp_data_receive_workfn(struct work_struct *work)
1116 {
1117         struct sock_xprt *transport =
1118                 container_of(work, struct sock_xprt, recv_worker);
1119         xs_udp_data_receive(transport);
1120 }
1121 
1122 /**
1123  * xs_data_ready - "data ready" callback for UDP sockets
1124  * @sk: socket with data to read
1125  *
1126  */
1127 static void xs_data_ready(struct sock *sk)
1128 {
1129         struct rpc_xprt *xprt;
1130 
1131         read_lock_bh(&sk->sk_callback_lock);
1132         dprintk("RPC:       xs_data_ready...\n");
1133         xprt = xprt_from_sock(sk);
1134         if (xprt != NULL) {
1135                 struct sock_xprt *transport = container_of(xprt,
1136                                 struct sock_xprt, xprt);
1137                 transport->old_data_ready(sk);
1138                 /* Any data means we had a useful conversation, so
1139                  * then we don't need to delay the next reconnect
1140                  */
1141                 if (xprt->reestablish_timeout)
1142                         xprt->reestablish_timeout = 0;
1143                 if (!test_and_set_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
1144                         queue_work(xprtiod_workqueue, &transport->recv_worker);
1145         }
1146         read_unlock_bh(&sk->sk_callback_lock);
1147 }
1148 
1149 /*
1150  * Helper function to force a TCP close if the server is sending
1151  * junk and/or it has put us in CLOSE_WAIT
1152  */
1153 static void xs_tcp_force_close(struct rpc_xprt *xprt)
1154 {
1155         xprt_force_disconnect(xprt);
1156 }
1157 
1158 static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
1159 {
1160         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1161         size_t len, used;
1162         char *p;
1163 
1164         p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
1165         len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
1166         used = xdr_skb_read_bits(desc, p, len);
1167         transport->tcp_offset += used;
1168         if (used != len)
1169                 return;
1170 
1171         transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
1172         if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
1173                 transport->tcp_flags |= TCP_RCV_LAST_FRAG;
1174         else
1175                 transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
1176         transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
1177 
1178         transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
1179         transport->tcp_offset = 0;
1180 
1181         /* Sanity check of the record length */
1182         if (unlikely(transport->tcp_reclen < 8)) {
1183                 dprintk("RPC:       invalid TCP record fragment length\n");
1184                 xs_tcp_force_close(xprt);
1185                 return;
1186         }
1187         dprintk("RPC:       reading TCP record fragment of length %d\n",
1188                         transport->tcp_reclen);
1189 }
1190 
1191 static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
1192 {
1193         if (transport->tcp_offset == transport->tcp_reclen) {
1194                 transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
1195                 transport->tcp_offset = 0;
1196                 if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
1197                         transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1198                         transport->tcp_flags |= TCP_RCV_COPY_XID;
1199                         transport->tcp_copied = 0;
1200                 }
1201         }
1202 }
1203 
1204 static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
1205 {
1206         size_t len, used;
1207         char *p;
1208 
1209         len = sizeof(transport->tcp_xid) - transport->tcp_offset;
1210         dprintk("RPC:       reading XID (%zu bytes)\n", len);
1211         p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
1212         used = xdr_skb_read_bits(desc, p, len);
1213         transport->tcp_offset += used;
1214         if (used != len)
1215                 return;
1216         transport->tcp_flags &= ~TCP_RCV_COPY_XID;
1217         transport->tcp_flags |= TCP_RCV_READ_CALLDIR;
1218         transport->tcp_copied = 4;
1219         dprintk("RPC:       reading %s XID %08x\n",
1220                         (transport->tcp_flags & TCP_RPC_REPLY) ? "reply for"
1221                                                               : "request with",
1222                         ntohl(transport->tcp_xid));
1223         xs_tcp_check_fraghdr(transport);
1224 }
1225 
1226 static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
1227                                        struct xdr_skb_reader *desc)
1228 {
1229         size_t len, used;
1230         u32 offset;
1231         char *p;
1232 
1233         /*
1234          * We want transport->tcp_offset to be 8 at the end of this routine
1235          * (4 bytes for the xid and 4 bytes for the call/reply flag).
1236          * When this function is called for the first time,
1237          * transport->tcp_offset is 4 (after having already read the xid).
1238          */
1239         offset = transport->tcp_offset - sizeof(transport->tcp_xid);
1240         len = sizeof(transport->tcp_calldir) - offset;
1241         dprintk("RPC:       reading CALL/REPLY flag (%zu bytes)\n", len);
1242         p = ((char *) &transport->tcp_calldir) + offset;
1243         used = xdr_skb_read_bits(desc, p, len);
1244         transport->tcp_offset += used;
1245         if (used != len)
1246                 return;
1247         transport->tcp_flags &= ~TCP_RCV_READ_CALLDIR;
1248         /*
1249          * We don't yet have the XDR buffer, so we will write the calldir
1250          * out after we get the buffer from the 'struct rpc_rqst'
1251          */
1252         switch (ntohl(transport->tcp_calldir)) {
1253         case RPC_REPLY:
1254                 transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
1255                 transport->tcp_flags |= TCP_RCV_COPY_DATA;
1256                 transport->tcp_flags |= TCP_RPC_REPLY;
1257                 break;
1258         case RPC_CALL:
1259                 transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
1260                 transport->tcp_flags |= TCP_RCV_COPY_DATA;
1261                 transport->tcp_flags &= ~TCP_RPC_REPLY;
1262                 break;
1263         default:
1264                 dprintk("RPC:       invalid request message type\n");
1265                 xs_tcp_force_close(&transport->xprt);
1266         }
1267         xs_tcp_check_fraghdr(transport);
1268 }
1269 
1270 static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
1271                                      struct xdr_skb_reader *desc,
1272                                      struct rpc_rqst *req)
1273 {
1274         struct sock_xprt *transport =
1275                                 container_of(xprt, struct sock_xprt, xprt);
1276         struct xdr_buf *rcvbuf;
1277         size_t len;
1278         ssize_t r;
1279 
1280         rcvbuf = &req->rq_private_buf;
1281 
1282         if (transport->tcp_flags & TCP_RCV_COPY_CALLDIR) {
1283                 /*
1284                  * Save the RPC direction in the XDR buffer
1285                  */
1286                 memcpy(rcvbuf->head[0].iov_base + transport->tcp_copied,
1287                         &transport->tcp_calldir,
1288                         sizeof(transport->tcp_calldir));
1289                 transport->tcp_copied += sizeof(transport->tcp_calldir);
1290                 transport->tcp_flags &= ~TCP_RCV_COPY_CALLDIR;
1291         }
1292 
1293         len = desc->count;
1294         if (len > transport->tcp_reclen - transport->tcp_offset)
1295                 desc->count = transport->tcp_reclen - transport->tcp_offset;
1296         r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
1297                                           desc, xdr_skb_read_bits);
1298 
1299         if (desc->count) {
1300                 /* Error when copying to the receive buffer,
1301                  * usually because we weren't able to allocate
1302                  * additional buffer pages. All we can do now
1303                  * is turn off TCP_RCV_COPY_DATA, so the request
1304                  * will not receive any additional updates,
1305                  * and time out.
1306                  * Any remaining data from this record will
1307                  * be discarded.
1308                  */
1309                 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1310                 dprintk("RPC:       XID %08x truncated request\n",
1311                                 ntohl(transport->tcp_xid));
1312                 dprintk("RPC:       xprt = %p, tcp_copied = %lu, "
1313                                 "tcp_offset = %u, tcp_reclen = %u\n",
1314                                 xprt, transport->tcp_copied,
1315                                 transport->tcp_offset, transport->tcp_reclen);
1316                 return;
1317         }
1318 
1319         transport->tcp_copied += r;
1320         transport->tcp_offset += r;
1321         desc->count = len - r;
1322 
1323         dprintk("RPC:       XID %08x read %zd bytes\n",
1324                         ntohl(transport->tcp_xid), r);
1325         dprintk("RPC:       xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
1326                         "tcp_reclen = %u\n", xprt, transport->tcp_copied,
1327                         transport->tcp_offset, transport->tcp_reclen);
1328 
1329         if (transport->tcp_copied == req->rq_private_buf.buflen)
1330                 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1331         else if (transport->tcp_offset == transport->tcp_reclen) {
1332                 if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
1333                         transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1334         }
1335 }
1336 
1337 /*
1338  * Finds the request corresponding to the RPC xid and invokes the common
1339  * tcp read code to read the data.
1340  */
1341 static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
1342                                     struct xdr_skb_reader *desc)
1343 {
1344         struct sock_xprt *transport =
1345                                 container_of(xprt, struct sock_xprt, xprt);
1346         struct rpc_rqst *req;
1347 
1348         dprintk("RPC:       read reply XID %08x\n", ntohl(transport->tcp_xid));
1349 
1350         /* Find and lock the request corresponding to this xid */
1351         spin_lock(&xprt->recv_lock);
1352         req = xprt_lookup_rqst(xprt, transport->tcp_xid);
1353         if (!req) {
1354                 dprintk("RPC:       XID %08x request not found!\n",
1355                                 ntohl(transport->tcp_xid));
1356                 spin_unlock(&xprt->recv_lock);
1357                 return -1;
1358         }
1359         xprt_pin_rqst(req);
1360         spin_unlock(&xprt->recv_lock);
1361 
1362         xs_tcp_read_common(xprt, desc, req);
1363 
1364         spin_lock(&xprt->recv_lock);
1365         if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
1366                 xprt_complete_rqst(req->rq_task, transport->tcp_copied);
1367         xprt_unpin_rqst(req);
1368         spin_unlock(&xprt->recv_lock);
1369         return 0;
1370 }
1371 
1372 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
1373 /*
1374  * Obtains an rpc_rqst previously allocated and invokes the common
1375  * tcp read code to read the data.  The result is placed in the callback
1376  * queue.
1377  * If we're unable to obtain the rpc_rqst we schedule the closing of the
1378  * connection and return -1.
1379  */
1380 static int xs_tcp_read_callback(struct rpc_xprt *xprt,
1381                                        struct xdr_skb_reader *desc)
1382 {
1383         struct sock_xprt *transport =
1384                                 container_of(xprt, struct sock_xprt, xprt);
1385         struct rpc_rqst *req;
1386 
1387         /* Look up the request corresponding to the given XID */
1388         req = xprt_lookup_bc_request(xprt, transport->tcp_xid);
1389         if (req == NULL) {
1390                 printk(KERN_WARNING "Callback slot table overflowed\n");
1391                 xprt_force_disconnect(xprt);
1392                 return -1;
1393         }
1394 
1395         dprintk("RPC:       read callback  XID %08x\n", ntohl(req->rq_xid));
1396         xs_tcp_read_common(xprt, desc, req);
1397 
1398         if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
1399                 xprt_complete_bc_request(req, transport->tcp_copied);
1400 
1401         return 0;
1402 }
1403 
1404 static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1405                                         struct xdr_skb_reader *desc)
1406 {
1407         struct sock_xprt *transport =
1408                                 container_of(xprt, struct sock_xprt, xprt);
1409 
1410         return (transport->tcp_flags & TCP_RPC_REPLY) ?
1411                 xs_tcp_read_reply(xprt, desc) :
1412                 xs_tcp_read_callback(xprt, desc);
1413 }
1414 
1415 static int xs_tcp_bc_up(struct svc_serv *serv, struct net *net)
1416 {
1417         int ret;
1418 
1419         ret = svc_create_xprt(serv, "tcp-bc", net, PF_INET, 0,
1420                               SVC_SOCK_ANONYMOUS);
1421         if (ret < 0)
1422                 return ret;
1423         return 0;
1424 }
1425 
1426 static size_t xs_tcp_bc_maxpayload(struct rpc_xprt *xprt)
1427 {
1428         return PAGE_SIZE;
1429 }
1430 #else
1431 static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
1432                                         struct xdr_skb_reader *desc)
1433 {
1434         return xs_tcp_read_reply(xprt, desc);
1435 }
1436 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
1437 
1438 /*
1439  * Read data off the transport.  This can be either an RPC_CALL or an
1440  * RPC_REPLY.  Relay the processing to helper functions.
1441  */
1442 static void xs_tcp_read_data(struct rpc_xprt *xprt,
1443                                     struct xdr_skb_reader *desc)
1444 {
1445         struct sock_xprt *transport =
1446                                 container_of(xprt, struct sock_xprt, xprt);
1447 
1448         if (_xs_tcp_read_data(xprt, desc) == 0)
1449                 xs_tcp_check_fraghdr(transport);
1450         else {
1451                 /*
1452                  * The transport_lock protects the request handling.
1453                  * There's no need to hold it to update the tcp_flags.
1454                  */
1455                 transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1456         }
1457 }
1458 
1459 static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
1460 {
1461         size_t len;
1462 
1463         len = transport->tcp_reclen - transport->tcp_offset;
1464         if (len > desc->count)
1465                 len = desc->count;
1466         desc->count -= len;
1467         desc->offset += len;
1468         transport->tcp_offset += len;
1469         dprintk("RPC:       discarded %zu bytes\n", len);
1470         xs_tcp_check_fraghdr(transport);
1471 }
1472 
1473 static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
1474 {
1475         struct rpc_xprt *xprt = rd_desc->arg.data;
1476         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1477         struct xdr_skb_reader desc = {
1478                 .skb    = skb,
1479                 .offset = offset,
1480                 .count  = len,
1481         };
1482 
1483         dprintk("RPC:       xs_tcp_data_recv started\n");
1484         do {
1485                 trace_xs_tcp_data_recv(transport);
1486                 /* Read in a new fragment marker if necessary */
1487                 /* Can we ever really expect to get completely empty fragments? */
1488                 if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
1489                         xs_tcp_read_fraghdr(xprt, &desc);
1490                         continue;
1491                 }
1492                 /* Read in the xid if necessary */
1493                 if (transport->tcp_flags & TCP_RCV_COPY_XID) {
1494                         xs_tcp_read_xid(transport, &desc);
1495                         continue;
1496                 }
1497                 /* Read in the call/reply flag */
1498                 if (transport->tcp_flags & TCP_RCV_READ_CALLDIR) {
1499                         xs_tcp_read_calldir(transport, &desc);
1500                         continue;
1501                 }
1502                 /* Read in the request data */
1503                 if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
1504                         xs_tcp_read_data(xprt, &desc);
1505                         continue;
1506                 }
1507                 /* Skip over any trailing bytes on short reads */
1508                 xs_tcp_read_discard(transport, &desc);
1509         } while (desc.count);
1510         trace_xs_tcp_data_recv(transport);
1511         dprintk("RPC:       xs_tcp_data_recv done\n");
1512         return len - desc.count;
1513 }
1514 
1515 static void xs_tcp_data_receive(struct sock_xprt *transport)
1516 {
1517         struct rpc_xprt *xprt = &transport->xprt;
1518         struct sock *sk;
1519         read_descriptor_t rd_desc = {
1520                 .count = 2*1024*1024,
1521                 .arg.data = xprt,
1522         };
1523         unsigned long total = 0;
1524         int loop;
1525         int read = 0;
1526 
1527         mutex_lock(&transport->recv_mutex);
1528         sk = transport->inet;
1529         if (sk == NULL)
1530                 goto out;
1531 
1532         /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
1533         for (loop = 0; loop < 64; loop++) {
1534                 lock_sock(sk);
1535                 read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
1536                 if (read <= 0) {
1537                         clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
1538                         release_sock(sk);
1539                         break;
1540                 }
1541                 release_sock(sk);
1542                 total += read;
1543                 rd_desc.count = 65536;
1544         }
1545         if (test_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
1546                 queue_work(xprtiod_workqueue, &transport->recv_worker);
1547 out:
1548         mutex_unlock(&transport->recv_mutex);
1549         trace_xs_tcp_data_ready(xprt, read, total);
1550 }
1551 
1552 static void xs_tcp_data_receive_workfn(struct work_struct *work)
1553 {
1554         struct sock_xprt *transport =
1555                 container_of(work, struct sock_xprt, recv_worker);
1556         xs_tcp_data_receive(transport);
1557 }
1558 
1559 /**
1560  * xs_tcp_state_change - callback to handle TCP socket state changes
1561  * @sk: socket whose state has changed
1562  *
1563  */
1564 static void xs_tcp_state_change(struct sock *sk)
1565 {
1566         struct rpc_xprt *xprt;
1567         struct sock_xprt *transport;
1568 
1569         read_lock_bh(&sk->sk_callback_lock);
1570         if (!(xprt = xprt_from_sock(sk)))
1571                 goto out;
1572         dprintk("RPC:       xs_tcp_state_change client %p...\n", xprt);
1573         dprintk("RPC:       state %x conn %d dead %d zapped %d sk_shutdown %d\n",
1574                         sk->sk_state, xprt_connected(xprt),
1575                         sock_flag(sk, SOCK_DEAD),
1576                         sock_flag(sk, SOCK_ZAPPED),
1577                         sk->sk_shutdown);
1578 
1579         transport = container_of(xprt, struct sock_xprt, xprt);
1580         trace_rpc_socket_state_change(xprt, sk->sk_socket);
1581         switch (sk->sk_state) {
1582         case TCP_ESTABLISHED:
1583                 spin_lock(&xprt->transport_lock);
1584                 if (!xprt_test_and_set_connected(xprt)) {
1585 
1586                         /* Reset TCP record info */
1587                         transport->tcp_offset = 0;
1588                         transport->tcp_reclen = 0;
1589                         transport->tcp_copied = 0;
1590                         transport->tcp_flags =
1591                                 TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
1592                         xprt->connect_cookie++;
1593                         clear_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
1594                         xprt_clear_connecting(xprt);
1595 
1596                         xprt_wake_pending_tasks(xprt, -EAGAIN);
1597                 }
1598                 spin_unlock(&xprt->transport_lock);
1599                 break;
1600         case TCP_FIN_WAIT1:
1601                 /* The client initiated a shutdown of the socket */
1602                 xprt->connect_cookie++;
1603                 xprt->reestablish_timeout = 0;
1604                 set_bit(XPRT_CLOSING, &xprt->state);
1605                 smp_mb__before_atomic();
1606                 clear_bit(XPRT_CONNECTED, &xprt->state);
1607                 clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1608                 smp_mb__after_atomic();
1609                 break;
1610         case TCP_CLOSE_WAIT:
1611                 /* The server initiated a shutdown of the socket */
1612                 xprt->connect_cookie++;
1613                 clear_bit(XPRT_CONNECTED, &xprt->state);
1614                 xs_tcp_force_close(xprt);
1615                 /* fall through */
1616         case TCP_CLOSING:
1617                 /*
1618                  * If the server closed down the connection, make sure that
1619                  * we back off before reconnecting
1620                  */
1621                 if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
1622                         xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
1623                 break;
1624         case TCP_LAST_ACK:
1625                 set_bit(XPRT_CLOSING, &xprt->state);
1626                 smp_mb__before_atomic();
1627                 clear_bit(XPRT_CONNECTED, &xprt->state);
1628                 smp_mb__after_atomic();
1629                 break;
1630         case TCP_CLOSE:
1631                 if (test_and_clear_bit(XPRT_SOCK_CONNECTING,
1632                                         &transport->sock_state))
1633                         xprt_clear_connecting(xprt);
1634                 if (sk->sk_err)
1635                         xprt_wake_pending_tasks(xprt, -sk->sk_err);
1636                 xs_sock_mark_closed(xprt);
1637         }
1638  out:
1639         read_unlock_bh(&sk->sk_callback_lock);
1640 }
1641 
1642 static void xs_write_space(struct sock *sk)
1643 {
1644         struct socket_wq *wq;
1645         struct rpc_xprt *xprt;
1646 
1647         if (!sk->sk_socket)
1648                 return;
1649         clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
1650 
1651         if (unlikely(!(xprt = xprt_from_sock(sk))))
1652                 return;
1653         rcu_read_lock();
1654         wq = rcu_dereference(sk->sk_wq);
1655         if (!wq || test_and_clear_bit(SOCKWQ_ASYNC_NOSPACE, &wq->flags) == 0)
1656                 goto out;
1657 
1658         xprt_write_space(xprt);
1659 out:
1660         rcu_read_unlock();
1661 }
1662 
1663 /**
1664  * xs_udp_write_space - callback invoked when socket buffer space
1665  *                             becomes available
1666  * @sk: socket whose state has changed
1667  *
1668  * Called when more output buffer space is available for this socket.
1669  * We try not to wake our writers until they can make "significant"
1670  * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1671  * with a bunch of small requests.
1672  */
1673 static void xs_udp_write_space(struct sock *sk)
1674 {
1675         read_lock_bh(&sk->sk_callback_lock);
1676 
1677         /* from net/core/sock.c:sock_def_write_space */
1678         if (sock_writeable(sk))
1679                 xs_write_space(sk);
1680 
1681         read_unlock_bh(&sk->sk_callback_lock);
1682 }
1683 
1684 /**
1685  * xs_tcp_write_space - callback invoked when socket buffer space
1686  *                             becomes available
1687  * @sk: socket whose state has changed
1688  *
1689  * Called when more output buffer space is available for this socket.
1690  * We try not to wake our writers until they can make "significant"
1691  * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1692  * with a bunch of small requests.
1693  */
1694 static void xs_tcp_write_space(struct sock *sk)
1695 {
1696         read_lock_bh(&sk->sk_callback_lock);
1697 
1698         /* from net/core/stream.c:sk_stream_write_space */
1699         if (sk_stream_is_writeable(sk))
1700                 xs_write_space(sk);
1701 
1702         read_unlock_bh(&sk->sk_callback_lock);
1703 }
1704 
1705 static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1706 {
1707         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1708         struct sock *sk = transport->inet;
1709 
1710         if (transport->rcvsize) {
1711                 sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1712                 sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1713         }
1714         if (transport->sndsize) {
1715                 sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1716                 sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1717                 sk->sk_write_space(sk);
1718         }
1719 }
1720 
1721 /**
1722  * xs_udp_set_buffer_size - set send and receive limits
1723  * @xprt: generic transport
1724  * @sndsize: requested size of send buffer, in bytes
1725  * @rcvsize: requested size of receive buffer, in bytes
1726  *
1727  * Set socket send and receive buffer size limits.
1728  */
1729 static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1730 {
1731         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1732 
1733         transport->sndsize = 0;
1734         if (sndsize)
1735                 transport->sndsize = sndsize + 1024;
1736         transport->rcvsize = 0;
1737         if (rcvsize)
1738                 transport->rcvsize = rcvsize + 1024;
1739 
1740         xs_udp_do_set_buffer_size(xprt);
1741 }
1742 
1743 /**
1744  * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
1745  * @task: task that timed out
1746  *
1747  * Adjust the congestion window after a retransmit timeout has occurred.
1748  */
1749 static void xs_udp_timer(struct rpc_xprt *xprt, struct rpc_task *task)
1750 {
1751         spin_lock_bh(&xprt->transport_lock);
1752         xprt_adjust_cwnd(xprt, task, -ETIMEDOUT);
1753         spin_unlock_bh(&xprt->transport_lock);
1754 }
1755 
1756 static unsigned short xs_get_random_port(void)
1757 {
1758         unsigned short range = xprt_max_resvport - xprt_min_resvport + 1;
1759         unsigned short rand = (unsigned short) prandom_u32() % range;
1760         return rand + xprt_min_resvport;
1761 }
1762 
1763 /**
1764  * xs_set_reuseaddr_port - set the socket's port and address reuse options
1765  * @sock: socket
1766  *
1767  * Note that this function has to be called on all sockets that share the
1768  * same port, and it must be called before binding.
1769  */
1770 static void xs_sock_set_reuseport(struct socket *sock)
1771 {
1772         int opt = 1;
1773 
1774         kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEPORT,
1775                         (char *)&opt, sizeof(opt));
1776 }
1777 
1778 static unsigned short xs_sock_getport(struct socket *sock)
1779 {
1780         struct sockaddr_storage buf;
1781         int buflen;
1782         unsigned short port = 0;
1783 
1784         if (kernel_getsockname(sock, (struct sockaddr *)&buf, &buflen) < 0)
1785                 goto out;
1786         switch (buf.ss_family) {
1787         case AF_INET6:
1788                 port = ntohs(((struct sockaddr_in6 *)&buf)->sin6_port);
1789                 break;
1790         case AF_INET:
1791                 port = ntohs(((struct sockaddr_in *)&buf)->sin_port);
1792         }
1793 out:
1794         return port;
1795 }
1796 
1797 /**
1798  * xs_set_port - reset the port number in the remote endpoint address
1799  * @xprt: generic transport
1800  * @port: new port number
1801  *
1802  */
1803 static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
1804 {
1805         dprintk("RPC:       setting port for xprt %p to %u\n", xprt, port);
1806 
1807         rpc_set_port(xs_addr(xprt), port);
1808         xs_update_peer_port(xprt);
1809 }
1810 
1811 static void xs_set_srcport(struct sock_xprt *transport, struct socket *sock)
1812 {
1813         if (transport->srcport == 0)
1814                 transport->srcport = xs_sock_getport(sock);
1815 }
1816 
1817 static unsigned short xs_get_srcport(struct sock_xprt *transport)
1818 {
1819         unsigned short port = transport->srcport;
1820 
1821         if (port == 0 && transport->xprt.resvport)
1822                 port = xs_get_random_port();
1823         return port;
1824 }
1825 
1826 static unsigned short xs_next_srcport(struct sock_xprt *transport, unsigned short port)
1827 {
1828         if (transport->srcport != 0)
1829                 transport->srcport = 0;
1830         if (!transport->xprt.resvport)
1831                 return 0;
1832         if (port <= xprt_min_resvport || port > xprt_max_resvport)
1833                 return xprt_max_resvport;
1834         return --port;
1835 }
1836 static int xs_bind(struct sock_xprt *transport, struct socket *sock)
1837 {
1838         struct sockaddr_storage myaddr;
1839         int err, nloop = 0;
1840         unsigned short port = xs_get_srcport(transport);
1841         unsigned short last;
1842 
1843         /*
1844          * If we are asking for any ephemeral port (i.e. port == 0 &&
1845          * transport->xprt.resvport == 0), don't bind.  Let the local
1846          * port selection happen implicitly when the socket is used
1847          * (for example at connect time).
1848          *
1849          * This ensures that we can continue to establish TCP
1850          * connections even when all local ephemeral ports are already
1851          * a part of some TCP connection.  This makes no difference
1852          * for UDP sockets, but also doens't harm them.
1853          *
1854          * If we're asking for any reserved port (i.e. port == 0 &&
1855          * transport->xprt.resvport == 1) xs_get_srcport above will
1856          * ensure that port is non-zero and we will bind as needed.
1857          */
1858         if (port == 0)
1859                 return 0;
1860 
1861         memcpy(&myaddr, &transport->srcaddr, transport->xprt.addrlen);
1862         do {
1863                 rpc_set_port((struct sockaddr *)&myaddr, port);
1864                 err = kernel_bind(sock, (struct sockaddr *)&myaddr,
1865                                 transport->xprt.addrlen);
1866                 if (err == 0) {
1867                         transport->srcport = port;
1868                         break;
1869                 }
1870                 last = port;
1871                 port = xs_next_srcport(transport, port);
1872                 if (port > last)
1873                         nloop++;
1874         } while (err == -EADDRINUSE && nloop != 2);
1875 
1876         if (myaddr.ss_family == AF_INET)
1877                 dprintk("RPC:       %s %pI4:%u: %s (%d)\n", __func__,
1878                                 &((struct sockaddr_in *)&myaddr)->sin_addr,
1879                                 port, err ? "failed" : "ok", err);
1880         else
1881                 dprintk("RPC:       %s %pI6:%u: %s (%d)\n", __func__,
1882                                 &((struct sockaddr_in6 *)&myaddr)->sin6_addr,
1883                                 port, err ? "failed" : "ok", err);
1884         return err;
1885 }
1886 
1887 /*
1888  * We don't support autobind on AF_LOCAL sockets
1889  */
1890 static void xs_local_rpcbind(struct rpc_task *task)
1891 {
1892         xprt_set_bound(task->tk_xprt);
1893 }
1894 
1895 static void xs_local_set_port(struct rpc_xprt *xprt, unsigned short port)
1896 {
1897 }
1898 
1899 #ifdef CONFIG_DEBUG_LOCK_ALLOC
1900 static struct lock_class_key xs_key[2];
1901 static struct lock_class_key xs_slock_key[2];
1902 
1903 static inline void xs_reclassify_socketu(struct socket *sock)
1904 {
1905         struct sock *sk = sock->sk;
1906 
1907         sock_lock_init_class_and_name(sk, "slock-AF_LOCAL-RPC",
1908                 &xs_slock_key[1], "sk_lock-AF_LOCAL-RPC", &xs_key[1]);
1909 }
1910 
1911 static inline void xs_reclassify_socket4(struct socket *sock)
1912 {
1913         struct sock *sk = sock->sk;
1914 
1915         sock_lock_init_class_and_name(sk, "slock-AF_INET-RPC",
1916                 &xs_slock_key[0], "sk_lock-AF_INET-RPC", &xs_key[0]);
1917 }
1918 
1919 static inline void xs_reclassify_socket6(struct socket *sock)
1920 {
1921         struct sock *sk = sock->sk;
1922 
1923         sock_lock_init_class_and_name(sk, "slock-AF_INET6-RPC",
1924                 &xs_slock_key[1], "sk_lock-AF_INET6-RPC", &xs_key[1]);
1925 }
1926 
1927 static inline void xs_reclassify_socket(int family, struct socket *sock)
1928 {
1929         if (WARN_ON_ONCE(!sock_allow_reclassification(sock->sk)))
1930                 return;
1931 
1932         switch (family) {
1933         case AF_LOCAL:
1934                 xs_reclassify_socketu(sock);
1935                 break;
1936         case AF_INET:
1937                 xs_reclassify_socket4(sock);
1938                 break;
1939         case AF_INET6:
1940                 xs_reclassify_socket6(sock);
1941                 break;
1942         }
1943 }
1944 #else
1945 static inline void xs_reclassify_socket(int family, struct socket *sock)
1946 {
1947 }
1948 #endif
1949 
1950 static void xs_dummy_setup_socket(struct work_struct *work)
1951 {
1952 }
1953 
1954 static struct socket *xs_create_sock(struct rpc_xprt *xprt,
1955                 struct sock_xprt *transport, int family, int type,
1956                 int protocol, bool reuseport)
1957 {
1958         struct socket *sock;
1959         int err;
1960 
1961         err = __sock_create(xprt->xprt_net, family, type, protocol, &sock, 1);
1962         if (err < 0) {
1963                 dprintk("RPC:       can't create %d transport socket (%d).\n",
1964                                 protocol, -err);
1965                 goto out;
1966         }
1967         xs_reclassify_socket(family, sock);
1968 
1969         if (reuseport)
1970                 xs_sock_set_reuseport(sock);
1971 
1972         err = xs_bind(transport, sock);
1973         if (err) {
1974                 sock_release(sock);
1975                 goto out;
1976         }
1977 
1978         return sock;
1979 out:
1980         return ERR_PTR(err);
1981 }
1982 
1983 static int xs_local_finish_connecting(struct rpc_xprt *xprt,
1984                                       struct socket *sock)
1985 {
1986         struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
1987                                                                         xprt);
1988 
1989         if (!transport->inet) {
1990                 struct sock *sk = sock->sk;
1991 
1992                 write_lock_bh(&sk->sk_callback_lock);
1993 
1994                 xs_save_old_callbacks(transport, sk);
1995 
1996                 sk->sk_user_data = xprt;
1997                 sk->sk_data_ready = xs_data_ready;
1998                 sk->sk_write_space = xs_udp_write_space;
1999                 sock_set_flag(sk, SOCK_FASYNC);
2000                 sk->sk_error_report = xs_error_report;
2001                 sk->sk_allocation = GFP_NOIO;
2002 
2003                 xprt_clear_connected(xprt);
2004 
2005                 /* Reset to new socket */
2006                 transport->sock = sock;
2007                 transport->inet = sk;
2008 
2009                 write_unlock_bh(&sk->sk_callback_lock);
2010         }
2011 
2012         /* Tell the socket layer to start connecting... */
2013         xprt->stat.connect_count++;
2014         xprt->stat.connect_start = jiffies;
2015         return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, 0);
2016 }
2017 
2018 /**
2019  * xs_local_setup_socket - create AF_LOCAL socket, connect to a local endpoint
2020  * @transport: socket transport to connect
2021  */
2022 static int xs_local_setup_socket(struct sock_xprt *transport)
2023 {
2024         struct rpc_xprt *xprt = &transport->xprt;
2025         struct socket *sock;
2026         int status = -EIO;
2027 
2028         status = __sock_create(xprt->xprt_net, AF_LOCAL,
2029                                         SOCK_STREAM, 0, &sock, 1);
2030         if (status < 0) {
2031                 dprintk("RPC:       can't create AF_LOCAL "
2032                         "transport socket (%d).\n", -status);
2033                 goto out;
2034         }
2035         xs_reclassify_socket(AF_LOCAL, sock);
2036 
2037         dprintk("RPC:       worker connecting xprt %p via AF_LOCAL to %s\n",
2038                         xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
2039 
2040         status = xs_local_finish_connecting(xprt, sock);
2041         trace_rpc_socket_connect(xprt, sock, status);
2042         switch (status) {
2043         case 0:
2044                 dprintk("RPC:       xprt %p connected to %s\n",
2045                                 xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
2046                 xprt_set_connected(xprt);
2047         case -ENOBUFS:
2048                 break;
2049         case -ENOENT:
2050                 dprintk("RPC:       xprt %p: socket %s does not exist\n",
2051                                 xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
2052                 break;
2053         case -ECONNREFUSED:
2054                 dprintk("RPC:       xprt %p: connection refused for %s\n",
2055                                 xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
2056                 break;
2057         default:
2058                 printk(KERN_ERR "%s: unhandled error (%d) connecting to %s\n",
2059                                 __func__, -status,
2060                                 xprt->address_strings[RPC_DISPLAY_ADDR]);
2061         }
2062 
2063 out:
2064         xprt_clear_connecting(xprt);
2065         xprt_wake_pending_tasks(xprt, status);
2066         return status;
2067 }
2068 
2069 static void xs_local_connect(struct rpc_xprt *xprt, struct rpc_task *task)
2070 {
2071         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2072         int ret;
2073 
2074          if (RPC_IS_ASYNC(task)) {
2075                 /*
2076                  * We want the AF_LOCAL connect to be resolved in the
2077                  * filesystem namespace of the process making the rpc
2078                  * call.  Thus we connect synchronously.
2079                  *
2080                  * If we want to support asynchronous AF_LOCAL calls,
2081                  * we'll need to figure out how to pass a namespace to
2082                  * connect.
2083                  */
2084                 rpc_exit(task, -ENOTCONN);
2085                 return;
2086         }
2087         ret = xs_local_setup_socket(transport);
2088         if (ret && !RPC_IS_SOFTCONN(task))
2089                 msleep_interruptible(15000);
2090 }
2091 
2092 #if IS_ENABLED(CONFIG_SUNRPC_SWAP)
2093 /*
2094  * Note that this should be called with XPRT_LOCKED held (or when we otherwise
2095  * know that we have exclusive access to the socket), to guard against
2096  * races with xs_reset_transport.
2097  */
2098 static void xs_set_memalloc(struct rpc_xprt *xprt)
2099 {
2100         struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
2101                         xprt);
2102 
2103         /*
2104          * If there's no sock, then we have nothing to set. The
2105          * reconnecting process will get it for us.
2106          */
2107         if (!transport->inet)
2108                 return;
2109         if (atomic_read(&xprt->swapper))
2110                 sk_set_memalloc(transport->inet);
2111 }
2112 
2113 /**
2114  * xs_enable_swap - Tag this transport as being used for swap.
2115  * @xprt: transport to tag
2116  *
2117  * Take a reference to this transport on behalf of the rpc_clnt, and
2118  * optionally mark it for swapping if it wasn't already.
2119  */
2120 static int
2121 xs_enable_swap(struct rpc_xprt *xprt)
2122 {
2123         struct sock_xprt *xs = container_of(xprt, struct sock_xprt, xprt);
2124 
2125         if (atomic_inc_return(&xprt->swapper) != 1)
2126                 return 0;
2127         if (wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_KILLABLE))
2128                 return -ERESTARTSYS;
2129         if (xs->inet)
2130                 sk_set_memalloc(xs->inet);
2131         xprt_release_xprt(xprt, NULL);
2132         return 0;
2133 }
2134 
2135 /**
2136  * xs_disable_swap - Untag this transport as being used for swap.
2137  * @xprt: transport to tag
2138  *
2139  * Drop a "swapper" reference to this xprt on behalf of the rpc_clnt. If the
2140  * swapper refcount goes to 0, untag the socket as a memalloc socket.
2141  */
2142 static void
2143 xs_disable_swap(struct rpc_xprt *xprt)
2144 {
2145         struct sock_xprt *xs = container_of(xprt, struct sock_xprt, xprt);
2146 
2147         if (!atomic_dec_and_test(&xprt->swapper))
2148                 return;
2149         if (wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_KILLABLE))
2150                 return;
2151         if (xs->inet)
2152                 sk_clear_memalloc(xs->inet);
2153         xprt_release_xprt(xprt, NULL);
2154 }
2155 #else
2156 static void xs_set_memalloc(struct rpc_xprt *xprt)
2157 {
2158 }
2159 
2160 static int
2161 xs_enable_swap(struct rpc_xprt *xprt)
2162 {
2163         return -EINVAL;
2164 }
2165 
2166 static void
2167 xs_disable_swap(struct rpc_xprt *xprt)
2168 {
2169 }
2170 #endif
2171 
2172 static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
2173 {
2174         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2175 
2176         if (!transport->inet) {
2177                 struct sock *sk = sock->sk;
2178 
2179                 write_lock_bh(&sk->sk_callback_lock);
2180 
2181                 xs_save_old_callbacks(transport, sk);
2182 
2183                 sk->sk_user_data = xprt;
2184                 sk->sk_data_ready = xs_data_ready;
2185                 sk->sk_write_space = xs_udp_write_space;
2186                 sock_set_flag(sk, SOCK_FASYNC);
2187                 sk->sk_allocation = GFP_NOIO;
2188 
2189                 xprt_set_connected(xprt);
2190 
2191                 /* Reset to new socket */
2192                 transport->sock = sock;
2193                 transport->inet = sk;
2194 
2195                 xs_set_memalloc(xprt);
2196 
2197                 write_unlock_bh(&sk->sk_callback_lock);
2198         }
2199         xs_udp_do_set_buffer_size(xprt);
2200 
2201         xprt->stat.connect_start = jiffies;
2202 }
2203 
2204 static void xs_udp_setup_socket(struct work_struct *work)
2205 {
2206         struct sock_xprt *transport =
2207                 container_of(work, struct sock_xprt, connect_worker.work);
2208         struct rpc_xprt *xprt = &transport->xprt;
2209         struct socket *sock;
2210         int status = -EIO;
2211 
2212         sock = xs_create_sock(xprt, transport,
2213                         xs_addr(xprt)->sa_family, SOCK_DGRAM,
2214                         IPPROTO_UDP, false);
2215         if (IS_ERR(sock))
2216                 goto out;
2217 
2218         dprintk("RPC:       worker connecting xprt %p via %s to "
2219                                 "%s (port %s)\n", xprt,
2220                         xprt->address_strings[RPC_DISPLAY_PROTO],
2221                         xprt->address_strings[RPC_DISPLAY_ADDR],
2222                         xprt->address_strings[RPC_DISPLAY_PORT]);
2223 
2224         xs_udp_finish_connecting(xprt, sock);
2225         trace_rpc_socket_connect(xprt, sock, 0);
2226         status = 0;
2227 out:
2228         xprt_unlock_connect(xprt, transport);
2229         xprt_clear_connecting(xprt);
2230         xprt_wake_pending_tasks(xprt, status);
2231 }
2232 
2233 /**
2234  * xs_tcp_shutdown - gracefully shut down a TCP socket
2235  * @xprt: transport
2236  *
2237  * Initiates a graceful shutdown of the TCP socket by calling the
2238  * equivalent of shutdown(SHUT_RDWR);
2239  */
2240 static void xs_tcp_shutdown(struct rpc_xprt *xprt)
2241 {
2242         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2243         struct socket *sock = transport->sock;
2244 
2245         if (sock == NULL)
2246                 return;
2247         if (xprt_connected(xprt)) {
2248                 kernel_sock_shutdown(sock, SHUT_RDWR);
2249                 trace_rpc_socket_shutdown(xprt, sock);
2250         } else
2251                 xs_reset_transport(transport);
2252 }
2253 
2254 static void xs_tcp_set_socket_timeouts(struct rpc_xprt *xprt,
2255                 struct socket *sock)
2256 {
2257         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2258         unsigned int keepidle;
2259         unsigned int keepcnt;
2260         unsigned int opt_on = 1;
2261         unsigned int timeo;
2262 
2263         spin_lock_bh(&xprt->transport_lock);
2264         keepidle = DIV_ROUND_UP(xprt->timeout->to_initval, HZ);
2265         keepcnt = xprt->timeout->to_retries + 1;
2266         timeo = jiffies_to_msecs(xprt->timeout->to_initval) *
2267                 (xprt->timeout->to_retries + 1);
2268         clear_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state);
2269         spin_unlock_bh(&xprt->transport_lock);
2270 
2271         /* TCP Keepalive options */
2272         kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
2273                         (char *)&opt_on, sizeof(opt_on));
2274         kernel_setsockopt(sock, SOL_TCP, TCP_KEEPIDLE,
2275                         (char *)&keepidle, sizeof(keepidle));
2276         kernel_setsockopt(sock, SOL_TCP, TCP_KEEPINTVL,
2277                         (char *)&keepidle, sizeof(keepidle));
2278         kernel_setsockopt(sock, SOL_TCP, TCP_KEEPCNT,
2279                         (char *)&keepcnt, sizeof(keepcnt));
2280 
2281         /* TCP user timeout (see RFC5482) */
2282         kernel_setsockopt(sock, SOL_TCP, TCP_USER_TIMEOUT,
2283                         (char *)&timeo, sizeof(timeo));
2284 }
2285 
2286 static void xs_tcp_set_connect_timeout(struct rpc_xprt *xprt,
2287                 unsigned long connect_timeout,
2288                 unsigned long reconnect_timeout)
2289 {
2290         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2291         struct rpc_timeout to;
2292         unsigned long initval;
2293 
2294         spin_lock_bh(&xprt->transport_lock);
2295         if (reconnect_timeout < xprt->max_reconnect_timeout)
2296                 xprt->max_reconnect_timeout = reconnect_timeout;
2297         if (connect_timeout < xprt->connect_timeout) {
2298                 memcpy(&to, xprt->timeout, sizeof(to));
2299                 initval = DIV_ROUND_UP(connect_timeout, to.to_retries + 1);
2300                 /* Arbitrary lower limit */
2301                 if (initval <  XS_TCP_INIT_REEST_TO << 1)
2302                         initval = XS_TCP_INIT_REEST_TO << 1;
2303                 to.to_initval = initval;
2304                 to.to_maxval = initval;
2305                 memcpy(&transport->tcp_timeout, &to,
2306                                 sizeof(transport->tcp_timeout));
2307                 xprt->timeout = &transport->tcp_timeout;
2308                 xprt->connect_timeout = connect_timeout;
2309         }
2310         set_bit(XPRT_SOCK_UPD_TIMEOUT, &transport->sock_state);
2311         spin_unlock_bh(&xprt->transport_lock);
2312 }
2313 
2314 static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
2315 {
2316         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2317         int ret = -ENOTCONN;
2318 
2319         if (!transport->inet) {
2320                 struct sock *sk = sock->sk;
2321                 unsigned int addr_pref = IPV6_PREFER_SRC_PUBLIC;
2322 
2323                 /* Avoid temporary address, they are bad for long-lived
2324                  * connections such as NFS mounts.
2325                  * RFC4941, section 3.6 suggests that:
2326                  *    Individual applications, which have specific
2327                  *    knowledge about the normal duration of connections,
2328                  *    MAY override this as appropriate.
2329                  */
2330                 kernel_setsockopt(sock, SOL_IPV6, IPV6_ADDR_PREFERENCES,
2331                                 (char *)&addr_pref, sizeof(addr_pref));
2332 
2333                 xs_tcp_set_socket_timeouts(xprt, sock);
2334 
2335                 write_lock_bh(&sk->sk_callback_lock);
2336 
2337                 xs_save_old_callbacks(transport, sk);
2338 
2339                 sk->sk_user_data = xprt;
2340                 sk->sk_data_ready = xs_data_ready;
2341                 sk->sk_state_change = xs_tcp_state_change;
2342                 sk->sk_write_space = xs_tcp_write_space;
2343                 sock_set_flag(sk, SOCK_FASYNC);
2344                 sk->sk_error_report = xs_error_report;
2345                 sk->sk_allocation = GFP_NOIO;
2346 
2347                 /* socket options */
2348                 sock_reset_flag(sk, SOCK_LINGER);
2349                 tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
2350 
2351                 xprt_clear_connected(xprt);
2352 
2353                 /* Reset to new socket */
2354                 transport->sock = sock;
2355                 transport->inet = sk;
2356 
2357                 write_unlock_bh(&sk->sk_callback_lock);
2358         }
2359 
2360         if (!xprt_bound(xprt))
2361                 goto out;
2362 
2363         xs_set_memalloc(xprt);
2364 
2365         /* Tell the socket layer to start connecting... */
2366         xprt->stat.connect_count++;
2367         xprt->stat.connect_start = jiffies;
2368         set_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
2369         ret = kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
2370         switch (ret) {
2371         case 0:
2372                 xs_set_srcport(transport, sock);
2373                 /* fall through */
2374         case -EINPROGRESS:
2375                 /* SYN_SENT! */
2376                 if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2377                         xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2378                 break;
2379         case -EADDRNOTAVAIL:
2380                 /* Source port number is unavailable. Try a new one! */
2381                 transport->srcport = 0;
2382         }
2383 out:
2384         return ret;
2385 }
2386 
2387 /**
2388  * xs_tcp_setup_socket - create a TCP socket and connect to a remote endpoint
2389  *
2390  * Invoked by a work queue tasklet.
2391  */
2392 static void xs_tcp_setup_socket(struct work_struct *work)
2393 {
2394         struct sock_xprt *transport =
2395                 container_of(work, struct sock_xprt, connect_worker.work);
2396         struct socket *sock = transport->sock;
2397         struct rpc_xprt *xprt = &transport->xprt;
2398         int status = -EIO;
2399 
2400         if (!sock) {
2401                 sock = xs_create_sock(xprt, transport,
2402                                 xs_addr(xprt)->sa_family, SOCK_STREAM,
2403                                 IPPROTO_TCP, true);
2404                 if (IS_ERR(sock)) {
2405                         status = PTR_ERR(sock);
2406                         goto out;
2407                 }
2408         }
2409 
2410         dprintk("RPC:       worker connecting xprt %p via %s to "
2411                                 "%s (port %s)\n", xprt,
2412                         xprt->address_strings[RPC_DISPLAY_PROTO],
2413                         xprt->address_strings[RPC_DISPLAY_ADDR],
2414                         xprt->address_strings[RPC_DISPLAY_PORT]);
2415 
2416         status = xs_tcp_finish_connecting(xprt, sock);
2417         trace_rpc_socket_connect(xprt, sock, status);
2418         dprintk("RPC:       %p connect status %d connected %d sock state %d\n",
2419                         xprt, -status, xprt_connected(xprt),
2420                         sock->sk->sk_state);
2421         switch (status) {
2422         default:
2423                 printk("%s: connect returned unhandled error %d\n",
2424                         __func__, status);
2425                 /* fall through */
2426         case -EADDRNOTAVAIL:
2427                 /* We're probably in TIME_WAIT. Get rid of existing socket,
2428                  * and retry
2429                  */
2430                 xs_tcp_force_close(xprt);
2431                 break;
2432         case 0:
2433         case -EINPROGRESS:
2434         case -EALREADY:
2435                 xprt_unlock_connect(xprt, transport);
2436                 return;
2437         case -EINVAL:
2438                 /* Happens, for instance, if the user specified a link
2439                  * local IPv6 address without a scope-id.
2440                  */
2441         case -ECONNREFUSED:
2442         case -ECONNRESET:
2443         case -ENETDOWN:
2444         case -ENETUNREACH:
2445         case -EHOSTUNREACH:
2446         case -EADDRINUSE:
2447         case -ENOBUFS:
2448                 /*
2449                  * xs_tcp_force_close() wakes tasks with -EIO.
2450                  * We need to wake them first to ensure the
2451                  * correct error code.
2452                  */
2453                 xprt_wake_pending_tasks(xprt, status);
2454                 xs_tcp_force_close(xprt);
2455                 goto out;
2456         }
2457         status = -EAGAIN;
2458 out:
2459         xprt_unlock_connect(xprt, transport);
2460         xprt_clear_connecting(xprt);
2461         xprt_wake_pending_tasks(xprt, status);
2462 }
2463 
2464 static unsigned long xs_reconnect_delay(const struct rpc_xprt *xprt)
2465 {
2466         unsigned long start, now = jiffies;
2467 
2468         start = xprt->stat.connect_start + xprt->reestablish_timeout;
2469         if (time_after(start, now))
2470                 return start - now;
2471         return 0;
2472 }
2473 
2474 static void xs_reconnect_backoff(struct rpc_xprt *xprt)
2475 {
2476         xprt->reestablish_timeout <<= 1;
2477         if (xprt->reestablish_timeout > xprt->max_reconnect_timeout)
2478                 xprt->reestablish_timeout = xprt->max_reconnect_timeout;
2479         if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2480                 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2481 }
2482 
2483 /**
2484  * xs_connect - connect a socket to a remote endpoint
2485  * @xprt: pointer to transport structure
2486  * @task: address of RPC task that manages state of connect request
2487  *
2488  * TCP: If the remote end dropped the connection, delay reconnecting.
2489  *
2490  * UDP socket connects are synchronous, but we use a work queue anyway
2491  * to guarantee that even unprivileged user processes can set up a
2492  * socket on a privileged port.
2493  *
2494  * If a UDP socket connect fails, the delay behavior here prevents
2495  * retry floods (hard mounts).
2496  */
2497 static void xs_connect(struct rpc_xprt *xprt, struct rpc_task *task)
2498 {
2499         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2500         unsigned long delay = 0;
2501 
2502         WARN_ON_ONCE(!xprt_lock_connect(xprt, task, transport));
2503 
2504         if (transport->sock != NULL) {
2505                 dprintk("RPC:       xs_connect delayed xprt %p for %lu "
2506                                 "seconds\n",
2507                                 xprt, xprt->reestablish_timeout / HZ);
2508 
2509                 /* Start by resetting any existing state */
2510                 xs_reset_transport(transport);
2511 
2512                 delay = xs_reconnect_delay(xprt);
2513                 xs_reconnect_backoff(xprt);
2514 
2515         } else
2516                 dprintk("RPC:       xs_connect scheduled xprt %p\n", xprt);
2517 
2518         queue_delayed_work(xprtiod_workqueue,
2519                         &transport->connect_worker,
2520                         delay);
2521 }
2522 
2523 /**
2524  * xs_local_print_stats - display AF_LOCAL socket-specifc stats
2525  * @xprt: rpc_xprt struct containing statistics
2526  * @seq: output file
2527  *
2528  */
2529 static void xs_local_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2530 {
2531         long idle_time = 0;
2532 
2533         if (xprt_connected(xprt))
2534                 idle_time = (long)(jiffies - xprt->last_used) / HZ;
2535 
2536         seq_printf(seq, "\txprt:\tlocal %lu %lu %lu %ld %lu %lu %lu "
2537                         "%llu %llu %lu %llu %llu\n",
2538                         xprt->stat.bind_count,
2539                         xprt->stat.connect_count,
2540                         xprt->stat.connect_time,
2541                         idle_time,
2542                         xprt->stat.sends,
2543                         xprt->stat.recvs,
2544                         xprt->stat.bad_xids,
2545                         xprt->stat.req_u,
2546                         xprt->stat.bklog_u,
2547                         xprt->stat.max_slots,
2548                         xprt->stat.sending_u,
2549                         xprt->stat.pending_u);
2550 }
2551 
2552 /**
2553  * xs_udp_print_stats - display UDP socket-specifc stats
2554  * @xprt: rpc_xprt struct containing statistics
2555  * @seq: output file
2556  *
2557  */
2558 static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2559 {
2560         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2561 
2562         seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %llu %llu "
2563                         "%lu %llu %llu\n",
2564                         transport->srcport,
2565                         xprt->stat.bind_count,
2566                         xprt->stat.sends,
2567                         xprt->stat.recvs,
2568                         xprt->stat.bad_xids,
2569                         xprt->stat.req_u,
2570                         xprt->stat.bklog_u,
2571                         xprt->stat.max_slots,
2572                         xprt->stat.sending_u,
2573                         xprt->stat.pending_u);
2574 }
2575 
2576 /**
2577  * xs_tcp_print_stats - display TCP socket-specifc stats
2578  * @xprt: rpc_xprt struct containing statistics
2579  * @seq: output file
2580  *
2581  */
2582 static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2583 {
2584         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2585         long idle_time = 0;
2586 
2587         if (xprt_connected(xprt))
2588                 idle_time = (long)(jiffies - xprt->last_used) / HZ;
2589 
2590         seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu "
2591                         "%llu %llu %lu %llu %llu\n",
2592                         transport->srcport,
2593                         xprt->stat.bind_count,
2594                         xprt->stat.connect_count,
2595                         xprt->stat.connect_time,
2596                         idle_time,
2597                         xprt->stat.sends,
2598                         xprt->stat.recvs,
2599                         xprt->stat.bad_xids,
2600                         xprt->stat.req_u,
2601                         xprt->stat.bklog_u,
2602                         xprt->stat.max_slots,
2603                         xprt->stat.sending_u,
2604                         xprt->stat.pending_u);
2605 }
2606 
2607 /*
2608  * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
2609  * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
2610  * to use the server side send routines.
2611  */
2612 static int bc_malloc(struct rpc_task *task)
2613 {
2614         struct rpc_rqst *rqst = task->tk_rqstp;
2615         size_t size = rqst->rq_callsize;
2616         struct page *page;
2617         struct rpc_buffer *buf;
2618 
2619         if (size > PAGE_SIZE - sizeof(struct rpc_buffer)) {
2620                 WARN_ONCE(1, "xprtsock: large bc buffer request (size %zu)\n",
2621                           size);
2622                 return -EINVAL;
2623         }
2624 
2625         page = alloc_page(GFP_KERNEL);
2626         if (!page)
2627                 return -ENOMEM;
2628 
2629         buf = page_address(page);
2630         buf->len = PAGE_SIZE;
2631 
2632         rqst->rq_buffer = buf->data;
2633         rqst->rq_rbuffer = (char *)rqst->rq_buffer + rqst->rq_callsize;
2634         return 0;
2635 }
2636 
2637 /*
2638  * Free the space allocated in the bc_alloc routine
2639  */
2640 static void bc_free(struct rpc_task *task)
2641 {
2642         void *buffer = task->tk_rqstp->rq_buffer;
2643         struct rpc_buffer *buf;
2644 
2645         buf = container_of(buffer, struct rpc_buffer, data);
2646         free_page((unsigned long)buf);
2647 }
2648 
2649 /*
2650  * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
2651  * held. Borrows heavily from svc_tcp_sendto and xs_tcp_send_request.
2652  */
2653 static int bc_sendto(struct rpc_rqst *req)
2654 {
2655         int len;
2656         struct xdr_buf *xbufp = &req->rq_snd_buf;
2657         struct rpc_xprt *xprt = req->rq_xprt;
2658         struct sock_xprt *transport =
2659                                 container_of(xprt, struct sock_xprt, xprt);
2660         struct socket *sock = transport->sock;
2661         unsigned long headoff;
2662         unsigned long tailoff;
2663 
2664         xs_encode_stream_record_marker(xbufp);
2665 
2666         tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
2667         headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
2668         len = svc_send_common(sock, xbufp,
2669                               virt_to_page(xbufp->head[0].iov_base), headoff,
2670                               xbufp->tail[0].iov_base, tailoff);
2671 
2672         if (len != xbufp->len) {
2673                 printk(KERN_NOTICE "Error sending entire callback!\n");
2674                 len = -EAGAIN;
2675         }
2676 
2677         return len;
2678 }
2679 
2680 /*
2681  * The send routine. Borrows from svc_send
2682  */
2683 static int bc_send_request(struct rpc_task *task)
2684 {
2685         struct rpc_rqst *req = task->tk_rqstp;
2686         struct svc_xprt *xprt;
2687         int len;
2688 
2689         dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
2690         /*
2691          * Get the server socket associated with this callback xprt
2692          */
2693         xprt = req->rq_xprt->bc_xprt;
2694 
2695         /*
2696          * Grab the mutex to serialize data as the connection is shared
2697          * with the fore channel
2698          */
2699         if (!mutex_trylock(&xprt->xpt_mutex)) {
2700                 rpc_sleep_on(&xprt->xpt_bc_pending, task, NULL);
2701                 if (!mutex_trylock(&xprt->xpt_mutex))
2702                         return -EAGAIN;
2703                 rpc_wake_up_queued_task(&xprt->xpt_bc_pending, task);
2704         }
2705         if (test_bit(XPT_DEAD, &xprt->xpt_flags))
2706                 len = -ENOTCONN;
2707         else
2708                 len = bc_sendto(req);
2709         mutex_unlock(&xprt->xpt_mutex);
2710 
2711         if (len > 0)
2712                 len = 0;
2713 
2714         return len;
2715 }
2716 
2717 /*
2718  * The close routine. Since this is client initiated, we do nothing
2719  */
2720 
2721 static void bc_close(struct rpc_xprt *xprt)
2722 {
2723 }
2724 
2725 /*
2726  * The xprt destroy routine. Again, because this connection is client
2727  * initiated, we do nothing
2728  */
2729 
2730 static void bc_destroy(struct rpc_xprt *xprt)
2731 {
2732         dprintk("RPC:       bc_destroy xprt %p\n", xprt);
2733 
2734         xs_xprt_free(xprt);
2735         module_put(THIS_MODULE);
2736 }
2737 
2738 static const struct rpc_xprt_ops xs_local_ops = {
2739         .reserve_xprt           = xprt_reserve_xprt,
2740         .release_xprt           = xs_tcp_release_xprt,
2741         .alloc_slot             = xprt_alloc_slot,
2742         .rpcbind                = xs_local_rpcbind,
2743         .set_port               = xs_local_set_port,
2744         .connect                = xs_local_connect,
2745         .buf_alloc              = rpc_malloc,
2746         .buf_free               = rpc_free,
2747         .send_request           = xs_local_send_request,
2748         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
2749         .close                  = xs_close,
2750         .destroy                = xs_destroy,
2751         .print_stats            = xs_local_print_stats,
2752         .enable_swap            = xs_enable_swap,
2753         .disable_swap           = xs_disable_swap,
2754 };
2755 
2756 static const struct rpc_xprt_ops xs_udp_ops = {
2757         .set_buffer_size        = xs_udp_set_buffer_size,
2758         .reserve_xprt           = xprt_reserve_xprt_cong,
2759         .release_xprt           = xprt_release_xprt_cong,
2760         .alloc_slot             = xprt_alloc_slot,
2761         .rpcbind                = rpcb_getport_async,
2762         .set_port               = xs_set_port,
2763         .connect                = xs_connect,
2764         .buf_alloc              = rpc_malloc,
2765         .buf_free               = rpc_free,
2766         .send_request           = xs_udp_send_request,
2767         .set_retrans_timeout    = xprt_set_retrans_timeout_rtt,
2768         .timer                  = xs_udp_timer,
2769         .release_request        = xprt_release_rqst_cong,
2770         .close                  = xs_close,
2771         .destroy                = xs_destroy,
2772         .print_stats            = xs_udp_print_stats,
2773         .enable_swap            = xs_enable_swap,
2774         .disable_swap           = xs_disable_swap,
2775         .inject_disconnect      = xs_inject_disconnect,
2776 };
2777 
2778 static const struct rpc_xprt_ops xs_tcp_ops = {
2779         .reserve_xprt           = xprt_reserve_xprt,
2780         .release_xprt           = xs_tcp_release_xprt,
2781         .alloc_slot             = xprt_lock_and_alloc_slot,
2782         .rpcbind                = rpcb_getport_async,
2783         .set_port               = xs_set_port,
2784         .connect                = xs_connect,
2785         .buf_alloc              = rpc_malloc,
2786         .buf_free               = rpc_free,
2787         .send_request           = xs_tcp_send_request,
2788         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
2789         .close                  = xs_tcp_shutdown,
2790         .destroy                = xs_destroy,
2791         .set_connect_timeout    = xs_tcp_set_connect_timeout,
2792         .print_stats            = xs_tcp_print_stats,
2793         .enable_swap            = xs_enable_swap,
2794         .disable_swap           = xs_disable_swap,
2795         .inject_disconnect      = xs_inject_disconnect,
2796 #ifdef CONFIG_SUNRPC_BACKCHANNEL
2797         .bc_setup               = xprt_setup_bc,
2798         .bc_up                  = xs_tcp_bc_up,
2799         .bc_maxpayload          = xs_tcp_bc_maxpayload,
2800         .bc_free_rqst           = xprt_free_bc_rqst,
2801         .bc_destroy             = xprt_destroy_bc,
2802 #endif
2803 };
2804 
2805 /*
2806  * The rpc_xprt_ops for the server backchannel
2807  */
2808 
2809 static const struct rpc_xprt_ops bc_tcp_ops = {
2810         .reserve_xprt           = xprt_reserve_xprt,
2811         .release_xprt           = xprt_release_xprt,
2812         .alloc_slot             = xprt_alloc_slot,
2813         .buf_alloc              = bc_malloc,
2814         .buf_free               = bc_free,
2815         .send_request           = bc_send_request,
2816         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
2817         .close                  = bc_close,
2818         .destroy                = bc_destroy,
2819         .print_stats            = xs_tcp_print_stats,
2820         .enable_swap            = xs_enable_swap,
2821         .disable_swap           = xs_disable_swap,
2822         .inject_disconnect      = xs_inject_disconnect,
2823 };
2824 
2825 static int xs_init_anyaddr(const int family, struct sockaddr *sap)
2826 {
2827         static const struct sockaddr_in sin = {
2828                 .sin_family             = AF_INET,
2829                 .sin_addr.s_addr        = htonl(INADDR_ANY),
2830         };
2831         static const struct sockaddr_in6 sin6 = {
2832                 .sin6_family            = AF_INET6,
2833                 .sin6_addr              = IN6ADDR_ANY_INIT,
2834         };
2835 
2836         switch (family) {
2837         case AF_LOCAL:
2838                 break;
2839         case AF_INET:
2840                 memcpy(sap, &sin, sizeof(sin));
2841                 break;
2842         case AF_INET6:
2843                 memcpy(sap, &sin6, sizeof(sin6));
2844                 break;
2845         default:
2846                 dprintk("RPC:       %s: Bad address family\n", __func__);
2847                 return -EAFNOSUPPORT;
2848         }
2849         return 0;
2850 }
2851 
2852 static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
2853                                       unsigned int slot_table_size,
2854                                       unsigned int max_slot_table_size)
2855 {
2856         struct rpc_xprt *xprt;
2857         struct sock_xprt *new;
2858 
2859         if (args->addrlen > sizeof(xprt->addr)) {
2860                 dprintk("RPC:       xs_setup_xprt: address too large\n");
2861                 return ERR_PTR(-EBADF);
2862         }
2863 
2864         xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
2865                         max_slot_table_size);
2866         if (xprt == NULL) {
2867                 dprintk("RPC:       xs_setup_xprt: couldn't allocate "
2868                                 "rpc_xprt\n");
2869                 return ERR_PTR(-ENOMEM);
2870         }
2871 
2872         new = container_of(xprt, struct sock_xprt, xprt);
2873         mutex_init(&new->recv_mutex);
2874         memcpy(&xprt->addr, args->dstaddr, args->addrlen);
2875         xprt->addrlen = args->addrlen;
2876         if (args->srcaddr)
2877                 memcpy(&new->srcaddr, args->srcaddr, args->addrlen);
2878         else {
2879                 int err;
2880                 err = xs_init_anyaddr(args->dstaddr->sa_family,
2881                                         (struct sockaddr *)&new->srcaddr);
2882                 if (err != 0) {
2883                         xprt_free(xprt);
2884                         return ERR_PTR(err);
2885                 }
2886         }
2887 
2888         return xprt;
2889 }
2890 
2891 static const struct rpc_timeout xs_local_default_timeout = {
2892         .to_initval = 10 * HZ,
2893         .to_maxval = 10 * HZ,
2894         .to_retries = 2,
2895 };
2896 
2897 /**
2898  * xs_setup_local - Set up transport to use an AF_LOCAL socket
2899  * @args: rpc transport creation arguments
2900  *
2901  * AF_LOCAL is a "tpi_cots_ord" transport, just like TCP
2902  */
2903 static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
2904 {
2905         struct sockaddr_un *sun = (struct sockaddr_un *)args->dstaddr;
2906         struct sock_xprt *transport;
2907         struct rpc_xprt *xprt;
2908         struct rpc_xprt *ret;
2909 
2910         xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
2911                         xprt_max_tcp_slot_table_entries);
2912         if (IS_ERR(xprt))
2913                 return xprt;
2914         transport = container_of(xprt, struct sock_xprt, xprt);
2915 
2916         xprt->prot = 0;
2917         xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2918         xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2919 
2920         xprt->bind_timeout = XS_BIND_TO;
2921         xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2922         xprt->idle_timeout = XS_IDLE_DISC_TO;
2923 
2924         xprt->ops = &xs_local_ops;
2925         xprt->timeout = &xs_local_default_timeout;
2926 
2927         INIT_WORK(&transport->recv_worker, xs_local_data_receive_workfn);
2928         INIT_DELAYED_WORK(&transport->connect_worker,
2929                         xs_dummy_setup_socket);
2930 
2931         switch (sun->sun_family) {
2932         case AF_LOCAL:
2933                 if (sun->sun_path[0] != '/') {
2934                         dprintk("RPC:       bad AF_LOCAL address: %s\n",
2935                                         sun->sun_path);
2936                         ret = ERR_PTR(-EINVAL);
2937                         goto out_err;
2938                 }
2939                 xprt_set_bound(xprt);
2940                 xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
2941                 ret = ERR_PTR(xs_local_setup_socket(transport));
2942                 if (ret)
2943                         goto out_err;
2944                 break;
2945         default:
2946                 ret = ERR_PTR(-EAFNOSUPPORT);
2947                 goto out_err;
2948         }
2949 
2950         dprintk("RPC:       set up xprt to %s via AF_LOCAL\n",
2951                         xprt->address_strings[RPC_DISPLAY_ADDR]);
2952 
2953         if (try_module_get(THIS_MODULE))
2954                 return xprt;
2955         ret = ERR_PTR(-EINVAL);
2956 out_err:
2957         xs_xprt_free(xprt);
2958         return ret;
2959 }
2960 
2961 static const struct rpc_timeout xs_udp_default_timeout = {
2962         .to_initval = 5 * HZ,
2963         .to_maxval = 30 * HZ,
2964         .to_increment = 5 * HZ,
2965         .to_retries = 5,
2966 };
2967 
2968 /**
2969  * xs_setup_udp - Set up transport to use a UDP socket
2970  * @args: rpc transport creation arguments
2971  *
2972  */
2973 static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
2974 {
2975         struct sockaddr *addr = args->dstaddr;
2976         struct rpc_xprt *xprt;
2977         struct sock_xprt *transport;
2978         struct rpc_xprt *ret;
2979 
2980         xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
2981                         xprt_udp_slot_table_entries);
2982         if (IS_ERR(xprt))
2983                 return xprt;
2984         transport = container_of(xprt, struct sock_xprt, xprt);
2985 
2986         xprt->prot = IPPROTO_UDP;
2987         xprt->tsh_size = 0;
2988         /* XXX: header size can vary due to auth type, IPv6, etc. */
2989         xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
2990 
2991         xprt->bind_timeout = XS_BIND_TO;
2992         xprt->reestablish_timeout = XS_UDP_REEST_TO;
2993         xprt->idle_timeout = XS_IDLE_DISC_TO;
2994 
2995         xprt->ops = &xs_udp_ops;
2996 
2997         xprt->timeout = &xs_udp_default_timeout;
2998 
2999         INIT_WORK(&transport->recv_worker, xs_udp_data_receive_workfn);
3000         INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_setup_socket);
3001 
3002         switch (addr->sa_family) {
3003         case AF_INET:
3004                 if (((struct sockaddr_in *)addr)->sin_port != htons(0))
3005                         xprt_set_bound(xprt);
3006 
3007                 xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
3008                 break;
3009         case AF_INET6:
3010                 if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
3011                         xprt_set_bound(xprt);
3012 
3013                 xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
3014                 break;
3015         default:
3016                 ret = ERR_PTR(-EAFNOSUPPORT);
3017                 goto out_err;
3018         }
3019 
3020         if (xprt_bound(xprt))
3021                 dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3022                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3023                                 xprt->address_strings[RPC_DISPLAY_PORT],
3024                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3025         else
3026                 dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
3027                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3028                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3029 
3030         if (try_module_get(THIS_MODULE))
3031                 return xprt;
3032         ret = ERR_PTR(-EINVAL);
3033 out_err:
3034         xs_xprt_free(xprt);
3035         return ret;
3036 }
3037 
3038 static const struct rpc_timeout xs_tcp_default_timeout = {
3039         .to_initval = 60 * HZ,
3040         .to_maxval = 60 * HZ,
3041         .to_retries = 2,
3042 };
3043 
3044 /**
3045  * xs_setup_tcp - Set up transport to use a TCP socket
3046  * @args: rpc transport creation arguments
3047  *
3048  */
3049 static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
3050 {
3051         struct sockaddr *addr = args->dstaddr;
3052         struct rpc_xprt *xprt;
3053         struct sock_xprt *transport;
3054         struct rpc_xprt *ret;
3055         unsigned int max_slot_table_size = xprt_max_tcp_slot_table_entries;
3056 
3057         if (args->flags & XPRT_CREATE_INFINITE_SLOTS)
3058                 max_slot_table_size = RPC_MAX_SLOT_TABLE_LIMIT;
3059 
3060         xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3061                         max_slot_table_size);
3062         if (IS_ERR(xprt))
3063                 return xprt;
3064         transport = container_of(xprt, struct sock_xprt, xprt);
3065 
3066         xprt->prot = IPPROTO_TCP;
3067         xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
3068         xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3069 
3070         xprt->bind_timeout = XS_BIND_TO;
3071         xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
3072         xprt->idle_timeout = XS_IDLE_DISC_TO;
3073 
3074         xprt->ops = &xs_tcp_ops;
3075         xprt->timeout = &xs_tcp_default_timeout;
3076 
3077         xprt->max_reconnect_timeout = xprt->timeout->to_maxval;
3078         xprt->connect_timeout = xprt->timeout->to_initval *
3079                 (xprt->timeout->to_retries + 1);
3080 
3081         INIT_WORK(&transport->recv_worker, xs_tcp_data_receive_workfn);
3082         INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_setup_socket);
3083 
3084         switch (addr->sa_family) {
3085         case AF_INET:
3086                 if (((struct sockaddr_in *)addr)->sin_port != htons(0))
3087                         xprt_set_bound(xprt);
3088 
3089                 xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
3090                 break;
3091         case AF_INET6:
3092                 if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
3093                         xprt_set_bound(xprt);
3094 
3095                 xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
3096                 break;
3097         default:
3098                 ret = ERR_PTR(-EAFNOSUPPORT);
3099                 goto out_err;
3100         }
3101 
3102         if (xprt_bound(xprt))
3103                 dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3104                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3105                                 xprt->address_strings[RPC_DISPLAY_PORT],
3106                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3107         else
3108                 dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
3109                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3110                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3111 
3112         if (try_module_get(THIS_MODULE))
3113                 return xprt;
3114         ret = ERR_PTR(-EINVAL);
3115 out_err:
3116         xs_xprt_free(xprt);
3117         return ret;
3118 }
3119 
3120 /**
3121  * xs_setup_bc_tcp - Set up transport to use a TCP backchannel socket
3122  * @args: rpc transport creation arguments
3123  *
3124  */
3125 static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
3126 {
3127         struct sockaddr *addr = args->dstaddr;
3128         struct rpc_xprt *xprt;
3129         struct sock_xprt *transport;
3130         struct svc_sock *bc_sock;
3131         struct rpc_xprt *ret;
3132 
3133         xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3134                         xprt_tcp_slot_table_entries);
3135         if (IS_ERR(xprt))
3136                 return xprt;
3137         transport = container_of(xprt, struct sock_xprt, xprt);
3138 
3139         xprt->prot = IPPROTO_TCP;
3140         xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
3141         xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3142         xprt->timeout = &xs_tcp_default_timeout;
3143 
3144         /* backchannel */
3145         xprt_set_bound(xprt);
3146         xprt->bind_timeout = 0;
3147         xprt->reestablish_timeout = 0;
3148         xprt->idle_timeout = 0;
3149 
3150         xprt->ops = &bc_tcp_ops;
3151 
3152         switch (addr->sa_family) {
3153         case AF_INET:
3154                 xs_format_peer_addresses(xprt, "tcp",
3155                                          RPCBIND_NETID_TCP);
3156                 break;
3157         case AF_INET6:
3158                 xs_format_peer_addresses(xprt, "tcp",
3159                                    RPCBIND_NETID_TCP6);
3160                 break;
3161         default:
3162                 ret = ERR_PTR(-EAFNOSUPPORT);
3163                 goto out_err;
3164         }
3165 
3166         dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3167                         xprt->address_strings[RPC_DISPLAY_ADDR],
3168                         xprt->address_strings[RPC_DISPLAY_PORT],
3169                         xprt->address_strings[RPC_DISPLAY_PROTO]);
3170 
3171         /*
3172          * Once we've associated a backchannel xprt with a connection,
3173          * we want to keep it around as long as the connection lasts,
3174          * in case we need to start using it for a backchannel again;
3175          * this reference won't be dropped until bc_xprt is destroyed.
3176          */
3177         xprt_get(xprt);
3178         args->bc_xprt->xpt_bc_xprt = xprt;
3179         xprt->bc_xprt = args->bc_xprt;
3180         bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
3181         transport->sock = bc_sock->sk_sock;
3182         transport->inet = bc_sock->sk_sk;
3183 
3184         /*
3185          * Since we don't want connections for the backchannel, we set
3186          * the xprt status to connected
3187          */
3188         xprt_set_connected(xprt);
3189 
3190         if (try_module_get(THIS_MODULE))
3191                 return xprt;
3192 
3193         args->bc_xprt->xpt_bc_xprt = NULL;
3194         args->bc_xprt->xpt_bc_xps = NULL;
3195         xprt_put(xprt);
3196         ret = ERR_PTR(-EINVAL);
3197 out_err:
3198         xs_xprt_free(xprt);
3199         return ret;
3200 }
3201 
3202 static struct xprt_class        xs_local_transport = {
3203         .list           = LIST_HEAD_INIT(xs_local_transport.list),
3204         .name           = "named UNIX socket",
3205         .owner          = THIS_MODULE,
3206         .ident          = XPRT_TRANSPORT_LOCAL,
3207         .setup          = xs_setup_local,
3208 };
3209 
3210 static struct xprt_class        xs_udp_transport = {
3211         .list           = LIST_HEAD_INIT(xs_udp_transport.list),
3212         .name           = "udp",
3213         .owner          = THIS_MODULE,
3214         .ident          = XPRT_TRANSPORT_UDP,
3215         .setup          = xs_setup_udp,
3216 };
3217 
3218 static struct xprt_class        xs_tcp_transport = {
3219         .list           = LIST_HEAD_INIT(xs_tcp_transport.list),
3220         .name           = "tcp",
3221         .owner          = THIS_MODULE,
3222         .ident          = XPRT_TRANSPORT_TCP,
3223         .setup          = xs_setup_tcp,
3224 };
3225 
3226 static struct xprt_class        xs_bc_tcp_transport = {
3227         .list           = LIST_HEAD_INIT(xs_bc_tcp_transport.list),
3228         .name           = "tcp NFSv4.1 backchannel",
3229         .owner          = THIS_MODULE,
3230         .ident          = XPRT_TRANSPORT_BC_TCP,
3231         .setup          = xs_setup_bc_tcp,
3232 };
3233 
3234 /**
3235  * init_socket_xprt - set up xprtsock's sysctls, register with RPC client
3236  *
3237  */
3238 int init_socket_xprt(void)
3239 {
3240 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3241         if (!sunrpc_table_header)
3242                 sunrpc_table_header = register_sysctl_table(sunrpc_table);
3243 #endif
3244 
3245         xprt_register_transport(&xs_local_transport);
3246         xprt_register_transport(&xs_udp_transport);
3247         xprt_register_transport(&xs_tcp_transport);
3248         xprt_register_transport(&xs_bc_tcp_transport);
3249 
3250         return 0;
3251 }
3252 
3253 /**
3254  * cleanup_socket_xprt - remove xprtsock's sysctls, unregister
3255  *
3256  */
3257 void cleanup_socket_xprt(void)
3258 {
3259 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3260         if (sunrpc_table_header) {
3261                 unregister_sysctl_table(sunrpc_table_header);
3262                 sunrpc_table_header = NULL;
3263         }
3264 #endif
3265 
3266         xprt_unregister_transport(&xs_local_transport);
3267         xprt_unregister_transport(&xs_udp_transport);
3268         xprt_unregister_transport(&xs_tcp_transport);
3269         xprt_unregister_transport(&xs_bc_tcp_transport);
3270 }
3271 
3272 static int param_set_uint_minmax(const char *val,
3273                 const struct kernel_param *kp,
3274                 unsigned int min, unsigned int max)
3275 {
3276         unsigned int num;
3277         int ret;
3278 
3279         if (!val)
3280                 return -EINVAL;
3281         ret = kstrtouint(val, 0, &num);
3282         if (ret)
3283                 return ret;
3284         if (num < min || num > max)
3285                 return -EINVAL;
3286         *((unsigned int *)kp->arg) = num;
3287         return 0;
3288 }
3289 
3290 static int param_set_portnr(const char *val, const struct kernel_param *kp)
3291 {
3292         if (kp->arg == &xprt_min_resvport)
3293                 return param_set_uint_minmax(val, kp,
3294                         RPC_MIN_RESVPORT,
3295                         xprt_max_resvport);
3296         return param_set_uint_minmax(val, kp,
3297                         xprt_min_resvport,
3298                         RPC_MAX_RESVPORT);
3299 }
3300 
3301 static const struct kernel_param_ops param_ops_portnr = {
3302         .set = param_set_portnr,
3303         .get = param_get_uint,
3304 };
3305 
3306 #define param_check_portnr(name, p) \
3307         __param_check(name, p, unsigned int);
3308 
3309 module_param_named(min_resvport, xprt_min_resvport, portnr, 0644);
3310 module_param_named(max_resvport, xprt_max_resvport, portnr, 0644);
3311 
3312 static int param_set_slot_table_size(const char *val,
3313                                      const struct kernel_param *kp)
3314 {
3315         return param_set_uint_minmax(val, kp,
3316                         RPC_MIN_SLOT_TABLE,
3317                         RPC_MAX_SLOT_TABLE);
3318 }
3319 
3320 static const struct kernel_param_ops param_ops_slot_table_size = {
3321         .set = param_set_slot_table_size,
3322         .get = param_get_uint,
3323 };
3324 
3325 #define param_check_slot_table_size(name, p) \
3326         __param_check(name, p, unsigned int);
3327 
3328 static int param_set_max_slot_table_size(const char *val,
3329                                      const struct kernel_param *kp)
3330 {
3331         return param_set_uint_minmax(val, kp,
3332                         RPC_MIN_SLOT_TABLE,
3333                         RPC_MAX_SLOT_TABLE_LIMIT);
3334 }
3335 
3336 static const struct kernel_param_ops param_ops_max_slot_table_size = {
3337         .set = param_set_max_slot_table_size,
3338         .get = param_get_uint,
3339 };
3340 
3341 #define param_check_max_slot_table_size(name, p) \
3342         __param_check(name, p, unsigned int);
3343 
3344 module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
3345                    slot_table_size, 0644);
3346 module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
3347                    max_slot_table_size, 0644);
3348 module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
3349                    slot_table_size, 0644);
3350 
3351 

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp