net/sunrpc/xprtsock.c

   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 -ENETUNREACH:
2444         case -EADDRINUSE:
2445         case -ENOBUFS:
2446                 /*
2447                  * xs_tcp_force_close() wakes tasks with -EIO.
2448                  * We need to wake them first to ensure the
2449                  * correct error code.
2450                  */
2451                 xprt_wake_pending_tasks(xprt, status);
2452                 xs_tcp_force_close(xprt);
2453                 goto out;
2454         }
2455         status = -EAGAIN;
2456 out:
2457         xprt_unlock_connect(xprt, transport);
2458         xprt_clear_connecting(xprt);
2459         xprt_wake_pending_tasks(xprt, status);
2460 }
2461
2462 static unsigned long xs_reconnect_delay(const struct rpc_xprt *xprt)
2463 {
2464         unsigned long start, now = jiffies;
2465
2466         start = xprt->stat.connect_start + xprt->reestablish_timeout;
2467         if (time_after(start, now))
2468                 return start - now;
2469         return 0;
2470 }
2471
2472 static void xs_reconnect_backoff(struct rpc_xprt *xprt)
2473 {
2474         xprt->reestablish_timeout <<= 1;
2475         if (xprt->reestablish_timeout > xprt->max_reconnect_timeout)
2476                 xprt->reestablish_timeout = xprt->max_reconnect_timeout;
2477         if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
2478                 xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2479 }
2480
2481 /**
2482  * xs_connect - connect a socket to a remote endpoint
2483  * @xprt: pointer to transport structure
2484  * @task: address of RPC task that manages state of connect request
2485  *
2486  * TCP: If the remote end dropped the connection, delay reconnecting.
2487  *
2488  * UDP socket connects are synchronous, but we use a work queue anyway
2489  * to guarantee that even unprivileged user processes can set up a
2490  * socket on a privileged port.
2491  *
2492  * If a UDP socket connect fails, the delay behavior here prevents
2493  * retry floods (hard mounts).
2494  */
2495 static void xs_connect(struct rpc_xprt *xprt, struct rpc_task *task)
2496 {
2497         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2498         unsigned long delay = 0;
2499
2500         WARN_ON_ONCE(!xprt_lock_connect(xprt, task, transport));
2501
2502         if (transport->sock != NULL) {
2503                 dprintk("RPC:       xs_connect delayed xprt %p for %lu "
2504                                 "seconds\n",
2505                                 xprt, xprt->reestablish_timeout / HZ);
2506
2507                 /* Start by resetting any existing state */
2508                 xs_reset_transport(transport);
2509
2510                 delay = xs_reconnect_delay(xprt);
2511                 xs_reconnect_backoff(xprt);
2512
2513         } else
2514                 dprintk("RPC:       xs_connect scheduled xprt %p\n", xprt);
2515
2516         queue_delayed_work(xprtiod_workqueue,
2517                         &transport->connect_worker,
2518                         delay);
2519 }
2520
2521 /**
2522  * xs_local_print_stats - display AF_LOCAL socket-specifc stats
2523  * @xprt: rpc_xprt struct containing statistics
2524  * @seq: output file
2525  *
2526  */
2527 static void xs_local_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2528 {
2529         long idle_time = 0;
2530
2531         if (xprt_connected(xprt))
2532                 idle_time = (long)(jiffies - xprt->last_used) / HZ;
2533
2534         seq_printf(seq, "\txprt:\tlocal %lu %lu %lu %ld %lu %lu %lu "
2535                         "%llu %llu %lu %llu %llu\n",
2536                         xprt->stat.bind_count,
2537                         xprt->stat.connect_count,
2538                         xprt->stat.connect_time,
2539                         idle_time,
2540                         xprt->stat.sends,
2541                         xprt->stat.recvs,
2542                         xprt->stat.bad_xids,
2543                         xprt->stat.req_u,
2544                         xprt->stat.bklog_u,
2545                         xprt->stat.max_slots,
2546                         xprt->stat.sending_u,
2547                         xprt->stat.pending_u);
2548 }
2549
2550 /**
2551  * xs_udp_print_stats - display UDP socket-specifc stats
2552  * @xprt: rpc_xprt struct containing statistics
2553  * @seq: output file
2554  *
2555  */
2556 static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2557 {
2558         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2559
2560         seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %llu %llu "
2561                         "%lu %llu %llu\n",
2562                         transport->srcport,
2563                         xprt->stat.bind_count,
2564                         xprt->stat.sends,
2565                         xprt->stat.recvs,
2566                         xprt->stat.bad_xids,
2567                         xprt->stat.req_u,
2568                         xprt->stat.bklog_u,
2569                         xprt->stat.max_slots,
2570                         xprt->stat.sending_u,
2571                         xprt->stat.pending_u);
2572 }
2573
2574 /**
2575  * xs_tcp_print_stats - display TCP socket-specifc stats
2576  * @xprt: rpc_xprt struct containing statistics
2577  * @seq: output file
2578  *
2579  */
2580 static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
2581 {
2582         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
2583         long idle_time = 0;
2584
2585         if (xprt_connected(xprt))
2586                 idle_time = (long)(jiffies - xprt->last_used) / HZ;
2587
2588         seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu "
2589                         "%llu %llu %lu %llu %llu\n",
2590                         transport->srcport,
2591                         xprt->stat.bind_count,
2592                         xprt->stat.connect_count,
2593                         xprt->stat.connect_time,
2594                         idle_time,
2595                         xprt->stat.sends,
2596                         xprt->stat.recvs,
2597                         xprt->stat.bad_xids,
2598                         xprt->stat.req_u,
2599                         xprt->stat.bklog_u,
2600                         xprt->stat.max_slots,
2601                         xprt->stat.sending_u,
2602                         xprt->stat.pending_u);
2603 }
2604
2605 /*
2606  * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
2607  * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
2608  * to use the server side send routines.
2609  */
2610 static int bc_malloc(struct rpc_task *task)
2611 {
2612         struct rpc_rqst *rqst = task->tk_rqstp;
2613         size_t size = rqst->rq_callsize;
2614         struct page *page;
2615         struct rpc_buffer *buf;
2616
2617         if (size > PAGE_SIZE - sizeof(struct rpc_buffer)) {
2618                 WARN_ONCE(1, "xprtsock: large bc buffer request (size %zu)\n",
2619                           size);
2620                 return -EINVAL;
2621         }
2622
2623         page = alloc_page(GFP_KERNEL);
2624         if (!page)
2625                 return -ENOMEM;
2626
2627         buf = page_address(page);
2628         buf->len = PAGE_SIZE;
2629
2630         rqst->rq_buffer = buf->data;
2631         rqst->rq_rbuffer = (char *)rqst->rq_buffer + rqst->rq_callsize;
2632         return 0;
2633 }
2634
2635 /*
2636  * Free the space allocated in the bc_alloc routine
2637  */
2638 static void bc_free(struct rpc_task *task)
2639 {
2640         void *buffer = task->tk_rqstp->rq_buffer;
2641         struct rpc_buffer *buf;
2642
2643         buf = container_of(buffer, struct rpc_buffer, data);
2644         free_page((unsigned long)buf);
2645 }
2646
2647 /*
2648  * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
2649  * held. Borrows heavily from svc_tcp_sendto and xs_tcp_send_request.
2650  */
2651 static int bc_sendto(struct rpc_rqst *req)
2652 {
2653         int len;
2654         struct xdr_buf *xbufp = &req->rq_snd_buf;
2655         struct rpc_xprt *xprt = req->rq_xprt;
2656         struct sock_xprt *transport =
2657                                 container_of(xprt, struct sock_xprt, xprt);
2658         struct socket *sock = transport->sock;
2659         unsigned long headoff;
2660         unsigned long tailoff;
2661
2662         xs_encode_stream_record_marker(xbufp);
2663
2664         tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
2665         headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
2666         len = svc_send_common(sock, xbufp,
2667                               virt_to_page(xbufp->head[0].iov_base), headoff,
2668                               xbufp->tail[0].iov_base, tailoff);
2669
2670         if (len != xbufp->len) {
2671                 printk(KERN_NOTICE "Error sending entire callback!\n");
2672                 len = -EAGAIN;
2673         }
2674
2675         return len;
2676 }
2677
2678 /*
2679  * The send routine. Borrows from svc_send
2680  */
2681 static int bc_send_request(struct rpc_task *task)
2682 {
2683         struct rpc_rqst *req = task->tk_rqstp;
2684         struct svc_xprt *xprt;
2685         int len;
2686
2687         dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
2688         /*
2689          * Get the server socket associated with this callback xprt
2690          */
2691         xprt = req->rq_xprt->bc_xprt;
2692
2693         /*
2694          * Grab the mutex to serialize data as the connection is shared
2695          * with the fore channel
2696          */
2697         if (!mutex_trylock(&xprt->xpt_mutex)) {
2698                 rpc_sleep_on(&xprt->xpt_bc_pending, task, NULL);
2699                 if (!mutex_trylock(&xprt->xpt_mutex))
2700                         return -EAGAIN;
2701                 rpc_wake_up_queued_task(&xprt->xpt_bc_pending, task);
2702         }
2703         if (test_bit(XPT_DEAD, &xprt->xpt_flags))
2704                 len = -ENOTCONN;
2705         else
2706                 len = bc_sendto(req);
2707         mutex_unlock(&xprt->xpt_mutex);
2708
2709         if (len > 0)
2710                 len = 0;
2711
2712         return len;
2713 }
2714
2715 /*
2716  * The close routine. Since this is client initiated, we do nothing
2717  */
2718
2719 static void bc_close(struct rpc_xprt *xprt)
2720 {
2721 }
2722
2723 /*
2724  * The xprt destroy routine. Again, because this connection is client
2725  * initiated, we do nothing
2726  */
2727
2728 static void bc_destroy(struct rpc_xprt *xprt)
2729 {
2730         dprintk("RPC:       bc_destroy xprt %p\n", xprt);
2731
2732         xs_xprt_free(xprt);
2733         module_put(THIS_MODULE);
2734 }
2735
2736 static const struct rpc_xprt_ops xs_local_ops = {
2737         .reserve_xprt           = xprt_reserve_xprt,
2738         .release_xprt           = xs_tcp_release_xprt,
2739         .alloc_slot             = xprt_alloc_slot,
2740         .rpcbind                = xs_local_rpcbind,
2741         .set_port               = xs_local_set_port,
2742         .connect                = xs_local_connect,
2743         .buf_alloc              = rpc_malloc,
2744         .buf_free               = rpc_free,
2745         .send_request           = xs_local_send_request,
2746         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
2747         .close                  = xs_close,
2748         .destroy                = xs_destroy,
2749         .print_stats            = xs_local_print_stats,
2750         .enable_swap            = xs_enable_swap,
2751         .disable_swap           = xs_disable_swap,
2752 };
2753
2754 static const struct rpc_xprt_ops xs_udp_ops = {
2755         .set_buffer_size        = xs_udp_set_buffer_size,
2756         .reserve_xprt           = xprt_reserve_xprt_cong,
2757         .release_xprt           = xprt_release_xprt_cong,
2758         .alloc_slot             = xprt_alloc_slot,
2759         .rpcbind                = rpcb_getport_async,
2760         .set_port               = xs_set_port,
2761         .connect                = xs_connect,
2762         .buf_alloc              = rpc_malloc,
2763         .buf_free               = rpc_free,
2764         .send_request           = xs_udp_send_request,
2765         .set_retrans_timeout    = xprt_set_retrans_timeout_rtt,
2766         .timer                  = xs_udp_timer,
2767         .release_request        = xprt_release_rqst_cong,
2768         .close                  = xs_close,
2769         .destroy                = xs_destroy,
2770         .print_stats            = xs_udp_print_stats,
2771         .enable_swap            = xs_enable_swap,
2772         .disable_swap           = xs_disable_swap,
2773         .inject_disconnect      = xs_inject_disconnect,
2774 };
2775
2776 static const struct rpc_xprt_ops xs_tcp_ops = {
2777         .reserve_xprt           = xprt_reserve_xprt,
2778         .release_xprt           = xs_tcp_release_xprt,
2779         .alloc_slot             = xprt_lock_and_alloc_slot,
2780         .rpcbind                = rpcb_getport_async,
2781         .set_port               = xs_set_port,
2782         .connect                = xs_connect,
2783         .buf_alloc              = rpc_malloc,
2784         .buf_free               = rpc_free,
2785         .send_request           = xs_tcp_send_request,
2786         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
2787         .close                  = xs_tcp_shutdown,
2788         .destroy                = xs_destroy,
2789         .set_connect_timeout    = xs_tcp_set_connect_timeout,
2790         .print_stats            = xs_tcp_print_stats,
2791         .enable_swap            = xs_enable_swap,
2792         .disable_swap           = xs_disable_swap,
2793         .inject_disconnect      = xs_inject_disconnect,
2794 #ifdef CONFIG_SUNRPC_BACKCHANNEL
2795         .bc_setup               = xprt_setup_bc,
2796         .bc_up                  = xs_tcp_bc_up,
2797         .bc_maxpayload          = xs_tcp_bc_maxpayload,
2798         .bc_free_rqst           = xprt_free_bc_rqst,
2799         .bc_destroy             = xprt_destroy_bc,
2800 #endif
2801 };
2802
2803 /*
2804  * The rpc_xprt_ops for the server backchannel
2805  */
2806
2807 static const struct rpc_xprt_ops bc_tcp_ops = {
2808         .reserve_xprt           = xprt_reserve_xprt,
2809         .release_xprt           = xprt_release_xprt,
2810         .alloc_slot             = xprt_alloc_slot,
2811         .buf_alloc              = bc_malloc,
2812         .buf_free               = bc_free,
2813         .send_request           = bc_send_request,
2814         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
2815         .close                  = bc_close,
2816         .destroy                = bc_destroy,
2817         .print_stats            = xs_tcp_print_stats,
2818         .enable_swap            = xs_enable_swap,
2819         .disable_swap           = xs_disable_swap,
2820         .inject_disconnect      = xs_inject_disconnect,
2821 };
2822
2823 static int xs_init_anyaddr(const int family, struct sockaddr *sap)
2824 {
2825         static const struct sockaddr_in sin = {
2826                 .sin_family             = AF_INET,
2827                 .sin_addr.s_addr        = htonl(INADDR_ANY),
2828         };
2829         static const struct sockaddr_in6 sin6 = {
2830                 .sin6_family            = AF_INET6,
2831                 .sin6_addr              = IN6ADDR_ANY_INIT,
2832         };
2833
2834         switch (family) {
2835         case AF_LOCAL:
2836                 break;
2837         case AF_INET:
2838                 memcpy(sap, &sin, sizeof(sin));
2839                 break;
2840         case AF_INET6:
2841                 memcpy(sap, &sin6, sizeof(sin6));
2842                 break;
2843         default:
2844                 dprintk("RPC:       %s: Bad address family\n", __func__);
2845                 return -EAFNOSUPPORT;
2846         }
2847         return 0;
2848 }
2849
2850 static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
2851                                       unsigned int slot_table_size,
2852                                       unsigned int max_slot_table_size)
2853 {
2854         struct rpc_xprt *xprt;
2855         struct sock_xprt *new;
2856
2857         if (args->addrlen > sizeof(xprt->addr)) {
2858                 dprintk("RPC:       xs_setup_xprt: address too large\n");
2859                 return ERR_PTR(-EBADF);
2860         }
2861
2862         xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
2863                         max_slot_table_size);
2864         if (xprt == NULL) {
2865                 dprintk("RPC:       xs_setup_xprt: couldn't allocate "
2866                                 "rpc_xprt\n");
2867                 return ERR_PTR(-ENOMEM);
2868         }
2869
2870         new = container_of(xprt, struct sock_xprt, xprt);
2871         mutex_init(&new->recv_mutex);
2872         memcpy(&xprt->addr, args->dstaddr, args->addrlen);
2873         xprt->addrlen = args->addrlen;
2874         if (args->srcaddr)
2875                 memcpy(&new->srcaddr, args->srcaddr, args->addrlen);
2876         else {
2877                 int err;
2878                 err = xs_init_anyaddr(args->dstaddr->sa_family,
2879                                         (struct sockaddr *)&new->srcaddr);
2880                 if (err != 0) {
2881                         xprt_free(xprt);
2882                         return ERR_PTR(err);
2883                 }
2884         }
2885
2886         return xprt;
2887 }
2888
2889 static const struct rpc_timeout xs_local_default_timeout = {
2890         .to_initval = 10 * HZ,
2891         .to_maxval = 10 * HZ,
2892         .to_retries = 2,
2893 };
2894
2895 /**
2896  * xs_setup_local - Set up transport to use an AF_LOCAL socket
2897  * @args: rpc transport creation arguments
2898  *
2899  * AF_LOCAL is a "tpi_cots_ord" transport, just like TCP
2900  */
2901 static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
2902 {
2903         struct sockaddr_un *sun = (struct sockaddr_un *)args->dstaddr;
2904         struct sock_xprt *transport;
2905         struct rpc_xprt *xprt;
2906         struct rpc_xprt *ret;
2907
2908         xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
2909                         xprt_max_tcp_slot_table_entries);
2910         if (IS_ERR(xprt))
2911                 return xprt;
2912         transport = container_of(xprt, struct sock_xprt, xprt);
2913
2914         xprt->prot = 0;
2915         xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2916         xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2917
2918         xprt->bind_timeout = XS_BIND_TO;
2919         xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2920         xprt->idle_timeout = XS_IDLE_DISC_TO;
2921
2922         xprt->ops = &xs_local_ops;
2923         xprt->timeout = &xs_local_default_timeout;
2924
2925         INIT_WORK(&transport->recv_worker, xs_local_data_receive_workfn);
2926         INIT_DELAYED_WORK(&transport->connect_worker,
2927                         xs_dummy_setup_socket);
2928
2929         switch (sun->sun_family) {
2930         case AF_LOCAL:
2931                 if (sun->sun_path[0] != '/') {
2932                         dprintk("RPC:       bad AF_LOCAL address: %s\n",
2933                                         sun->sun_path);
2934                         ret = ERR_PTR(-EINVAL);
2935                         goto out_err;
2936                 }
2937                 xprt_set_bound(xprt);
2938                 xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
2939                 ret = ERR_PTR(xs_local_setup_socket(transport));
2940                 if (ret)
2941                         goto out_err;
2942                 break;
2943         default:
2944                 ret = ERR_PTR(-EAFNOSUPPORT);
2945                 goto out_err;
2946         }
2947
2948         dprintk("RPC:       set up xprt to %s via AF_LOCAL\n",
2949                         xprt->address_strings[RPC_DISPLAY_ADDR]);
2950
2951         if (try_module_get(THIS_MODULE))
2952                 return xprt;
2953         ret = ERR_PTR(-EINVAL);
2954 out_err:
2955         xs_xprt_free(xprt);
2956         return ret;
2957 }
2958
2959 static const struct rpc_timeout xs_udp_default_timeout = {
2960         .to_initval = 5 * HZ,
2961         .to_maxval = 30 * HZ,
2962         .to_increment = 5 * HZ,
2963         .to_retries = 5,
2964 };
2965
2966 /**
2967  * xs_setup_udp - Set up transport to use a UDP socket
2968  * @args: rpc transport creation arguments
2969  *
2970  */
2971 static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
2972 {
2973         struct sockaddr *addr = args->dstaddr;
2974         struct rpc_xprt *xprt;
2975         struct sock_xprt *transport;
2976         struct rpc_xprt *ret;
2977
2978         xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
2979                         xprt_udp_slot_table_entries);
2980         if (IS_ERR(xprt))
2981                 return xprt;
2982         transport = container_of(xprt, struct sock_xprt, xprt);
2983
2984         xprt->prot = IPPROTO_UDP;
2985         xprt->tsh_size = 0;
2986         /* XXX: header size can vary due to auth type, IPv6, etc. */
2987         xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
2988
2989         xprt->bind_timeout = XS_BIND_TO;
2990         xprt->reestablish_timeout = XS_UDP_REEST_TO;
2991         xprt->idle_timeout = XS_IDLE_DISC_TO;
2992
2993         xprt->ops = &xs_udp_ops;
2994
2995         xprt->timeout = &xs_udp_default_timeout;
2996
2997         INIT_WORK(&transport->recv_worker, xs_udp_data_receive_workfn);
2998         INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_setup_socket);
2999
3000         switch (addr->sa_family) {
3001         case AF_INET:
3002                 if (((struct sockaddr_in *)addr)->sin_port != htons(0))
3003                         xprt_set_bound(xprt);
3004
3005                 xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
3006                 break;
3007         case AF_INET6:
3008                 if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
3009                         xprt_set_bound(xprt);
3010
3011                 xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
3012                 break;
3013         default:
3014                 ret = ERR_PTR(-EAFNOSUPPORT);
3015                 goto out_err;
3016         }
3017
3018         if (xprt_bound(xprt))
3019                 dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3020                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3021                                 xprt->address_strings[RPC_DISPLAY_PORT],
3022                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3023         else
3024                 dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
3025                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3026                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3027
3028         if (try_module_get(THIS_MODULE))
3029                 return xprt;
3030         ret = ERR_PTR(-EINVAL);
3031 out_err:
3032         xs_xprt_free(xprt);
3033         return ret;
3034 }
3035
3036 static const struct rpc_timeout xs_tcp_default_timeout = {
3037         .to_initval = 60 * HZ,
3038         .to_maxval = 60 * HZ,
3039         .to_retries = 2,
3040 };
3041
3042 /**
3043  * xs_setup_tcp - Set up transport to use a TCP socket
3044  * @args: rpc transport creation arguments
3045  *
3046  */
3047 static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
3048 {
3049         struct sockaddr *addr = args->dstaddr;
3050         struct rpc_xprt *xprt;
3051         struct sock_xprt *transport;
3052         struct rpc_xprt *ret;
3053         unsigned int max_slot_table_size = xprt_max_tcp_slot_table_entries;
3054
3055         if (args->flags & XPRT_CREATE_INFINITE_SLOTS)
3056                 max_slot_table_size = RPC_MAX_SLOT_TABLE_LIMIT;
3057
3058         xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3059                         max_slot_table_size);
3060         if (IS_ERR(xprt))
3061                 return xprt;
3062         transport = container_of(xprt, struct sock_xprt, xprt);
3063
3064         xprt->prot = IPPROTO_TCP;
3065         xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
3066         xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3067
3068         xprt->bind_timeout = XS_BIND_TO;
3069         xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
3070         xprt->idle_timeout = XS_IDLE_DISC_TO;
3071
3072         xprt->ops = &xs_tcp_ops;
3073         xprt->timeout = &xs_tcp_default_timeout;
3074
3075         xprt->max_reconnect_timeout = xprt->timeout->to_maxval;
3076         xprt->connect_timeout = xprt->timeout->to_initval *
3077                 (xprt->timeout->to_retries + 1);
3078
3079         INIT_WORK(&transport->recv_worker, xs_tcp_data_receive_workfn);
3080         INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_setup_socket);
3081
3082         switch (addr->sa_family) {
3083         case AF_INET:
3084                 if (((struct sockaddr_in *)addr)->sin_port != htons(0))
3085                         xprt_set_bound(xprt);
3086
3087                 xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
3088                 break;
3089         case AF_INET6:
3090                 if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
3091                         xprt_set_bound(xprt);
3092
3093                 xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
3094                 break;
3095         default:
3096                 ret = ERR_PTR(-EAFNOSUPPORT);
3097                 goto out_err;
3098         }
3099
3100         if (xprt_bound(xprt))
3101                 dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3102                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3103                                 xprt->address_strings[RPC_DISPLAY_PORT],
3104                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3105         else
3106                 dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
3107                                 xprt->address_strings[RPC_DISPLAY_ADDR],
3108                                 xprt->address_strings[RPC_DISPLAY_PROTO]);
3109
3110         if (try_module_get(THIS_MODULE))
3111                 return xprt;
3112         ret = ERR_PTR(-EINVAL);
3113 out_err:
3114         xs_xprt_free(xprt);
3115         return ret;
3116 }
3117
3118 /**
3119  * xs_setup_bc_tcp - Set up transport to use a TCP backchannel socket
3120  * @args: rpc transport creation arguments
3121  *
3122  */
3123 static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
3124 {
3125         struct sockaddr *addr = args->dstaddr;
3126         struct rpc_xprt *xprt;
3127         struct sock_xprt *transport;
3128         struct svc_sock *bc_sock;
3129         struct rpc_xprt *ret;
3130
3131         xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
3132                         xprt_tcp_slot_table_entries);
3133         if (IS_ERR(xprt))
3134                 return xprt;
3135         transport = container_of(xprt, struct sock_xprt, xprt);
3136
3137         xprt->prot = IPPROTO_TCP;
3138         xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
3139         xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
3140         xprt->timeout = &xs_tcp_default_timeout;
3141
3142         /* backchannel */
3143         xprt_set_bound(xprt);
3144         xprt->bind_timeout = 0;
3145         xprt->reestablish_timeout = 0;
3146         xprt->idle_timeout = 0;
3147
3148         xprt->ops = &bc_tcp_ops;
3149
3150         switch (addr->sa_family) {
3151         case AF_INET:
3152                 xs_format_peer_addresses(xprt, "tcp",
3153                                          RPCBIND_NETID_TCP);
3154                 break;
3155         case AF_INET6:
3156                 xs_format_peer_addresses(xprt, "tcp",
3157                                    RPCBIND_NETID_TCP6);
3158                 break;
3159         default:
3160                 ret = ERR_PTR(-EAFNOSUPPORT);
3161                 goto out_err;
3162         }
3163
3164         dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
3165                         xprt->address_strings[RPC_DISPLAY_ADDR],
3166                         xprt->address_strings[RPC_DISPLAY_PORT],
3167                         xprt->address_strings[RPC_DISPLAY_PROTO]);
3168
3169         /*
3170          * Once we've associated a backchannel xprt with a connection,
3171          * we want to keep it around as long as the connection lasts,
3172          * in case we need to start using it for a backchannel again;
3173          * this reference won't be dropped until bc_xprt is destroyed.
3174          */
3175         xprt_get(xprt);
3176         args->bc_xprt->xpt_bc_xprt = xprt;
3177         xprt->bc_xprt = args->bc_xprt;
3178         bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
3179         transport->sock = bc_sock->sk_sock;
3180         transport->inet = bc_sock->sk_sk;
3181
3182         /*
3183          * Since we don't want connections for the backchannel, we set
3184          * the xprt status to connected
3185          */
3186         xprt_set_connected(xprt);
3187
3188         if (try_module_get(THIS_MODULE))
3189                 return xprt;
3190
3191         args->bc_xprt->xpt_bc_xprt = NULL;
3192         args->bc_xprt->xpt_bc_xps = NULL;
3193         xprt_put(xprt);
3194         ret = ERR_PTR(-EINVAL);
3195 out_err:
3196         xs_xprt_free(xprt);
3197         return ret;
3198 }
3199
3200 static struct xprt_class        xs_local_transport = {
3201         .list           = LIST_HEAD_INIT(xs_local_transport.list),
3202         .name           = "named UNIX socket",
3203         .owner          = THIS_MODULE,
3204         .ident          = XPRT_TRANSPORT_LOCAL,
3205         .setup          = xs_setup_local,
3206 };
3207
3208 static struct xprt_class        xs_udp_transport = {
3209         .list           = LIST_HEAD_INIT(xs_udp_transport.list),
3210         .name           = "udp",
3211         .owner          = THIS_MODULE,
3212         .ident          = XPRT_TRANSPORT_UDP,
3213         .setup          = xs_setup_udp,
3214 };
3215
3216 static struct xprt_class        xs_tcp_transport = {
3217         .list           = LIST_HEAD_INIT(xs_tcp_transport.list),
3218         .name           = "tcp",
3219         .owner          = THIS_MODULE,
3220         .ident          = XPRT_TRANSPORT_TCP,
3221         .setup          = xs_setup_tcp,
3222 };
3223
3224 static struct xprt_class        xs_bc_tcp_transport = {
3225         .list           = LIST_HEAD_INIT(xs_bc_tcp_transport.list),
3226         .name           = "tcp NFSv4.1 backchannel",
3227         .owner          = THIS_MODULE,
3228         .ident          = XPRT_TRANSPORT_BC_TCP,
3229         .setup          = xs_setup_bc_tcp,
3230 };
3231
3232 /**
3233  * init_socket_xprt - set up xprtsock's sysctls, register with RPC client
3234  *
3235  */
3236 int init_socket_xprt(void)
3237 {
3238 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3239         if (!sunrpc_table_header)
3240                 sunrpc_table_header = register_sysctl_table(sunrpc_table);
3241 #endif
3242
3243         xprt_register_transport(&xs_local_transport);
3244         xprt_register_transport(&xs_udp_transport);
3245         xprt_register_transport(&xs_tcp_transport);
3246         xprt_register_transport(&xs_bc_tcp_transport);
3247
3248         return 0;
3249 }
3250
3251 /**
3252  * cleanup_socket_xprt - remove xprtsock's sysctls, unregister
3253  *
3254  */
3255 void cleanup_socket_xprt(void)
3256 {
3257 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
3258         if (sunrpc_table_header) {
3259                 unregister_sysctl_table(sunrpc_table_header);
3260                 sunrpc_table_header = NULL;
3261         }
3262 #endif
3263
3264         xprt_unregister_transport(&xs_local_transport);
3265         xprt_unregister_transport(&xs_udp_transport);
3266         xprt_unregister_transport(&xs_tcp_transport);
3267         xprt_unregister_transport(&xs_bc_tcp_transport);
3268 }
3269
3270 static int param_set_uint_minmax(const char *val,
3271                 const struct kernel_param *kp,
3272                 unsigned int min, unsigned int max)
3273 {
3274         unsigned int num;
3275         int ret;
3276
3277         if (!val)
3278                 return -EINVAL;
3279         ret = kstrtouint(val, 0, &num);
3280         if (ret)
3281                 return ret;
3282         if (num < min || num > max)
3283                 return -EINVAL;
3284         *((unsigned int *)kp->arg) = num;
3285         return 0;
3286 }
3287
3288 static int param_set_portnr(const char *val, const struct kernel_param *kp)
3289 {
3290         if (kp->arg == &xprt_min_resvport)
3291                 return param_set_uint_minmax(val, kp,
3292                         RPC_MIN_RESVPORT,
3293                         xprt_max_resvport);
3294         return param_set_uint_minmax(val, kp,
3295                         xprt_min_resvport,
3296                         RPC_MAX_RESVPORT);
3297 }
3298
3299 static const struct kernel_param_ops param_ops_portnr = {
3300         .set = param_set_portnr,
3301         .get = param_get_uint,
3302 };
3303
3304 #define param_check_portnr(name, p) \
3305         __param_check(name, p, unsigned int);
3306
3307 module_param_named(min_resvport, xprt_min_resvport, portnr, 0644);
3308 module_param_named(max_resvport, xprt_max_resvport, portnr, 0644);
3309
3310 static int param_set_slot_table_size(const char *val,
3311                                      const struct kernel_param *kp)
3312 {
3313         return param_set_uint_minmax(val, kp,
3314                         RPC_MIN_SLOT_TABLE,
3315                         RPC_MAX_SLOT_TABLE);
3316 }
3317
3318 static const struct kernel_param_ops param_ops_slot_table_size = {
3319         .set = param_set_slot_table_size,
3320         .get = param_get_uint,
3321 };
3322
3323 #define param_check_slot_table_size(name, p) \
3324         __param_check(name, p, unsigned int);
3325
3326 static int param_set_max_slot_table_size(const char *val,
3327                                      const struct kernel_param *kp)
3328 {
3329         return param_set_uint_minmax(val, kp,
3330                         RPC_MIN_SLOT_TABLE,
3331                         RPC_MAX_SLOT_TABLE_LIMIT);
3332 }
3333
3334 static const struct kernel_param_ops param_ops_max_slot_table_size = {
3335         .set = param_set_max_slot_table_size,
3336         .get = param_get_uint,
3337 };
3338
3339 #define param_check_max_slot_table_size(name, p) \
3340         __param_check(name, p, unsigned int);
3341
3342 module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
3343                    slot_table_size, 0644);
3344 module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
3345                    max_slot_table_size, 0644);
3346 module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
3347                    slot_table_size, 0644);
3348