net/sunrpc/xprtsock.c

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