linux/net/sunrpc/xprtsock.c
<<
>>
Prefs
   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/slab.h>
  23#include <linux/module.h>
  24#include <linux/capability.h>
  25#include <linux/pagemap.h>
  26#include <linux/errno.h>
  27#include <linux/socket.h>
  28#include <linux/in.h>
  29#include <linux/net.h>
  30#include <linux/mm.h>
  31#include <linux/udp.h>
  32#include <linux/tcp.h>
  33#include <linux/sunrpc/clnt.h>
  34#include <linux/sunrpc/sched.h>
  35#include <linux/sunrpc/xprtsock.h>
  36#include <linux/file.h>
  37
  38#include <net/sock.h>
  39#include <net/checksum.h>
  40#include <net/udp.h>
  41#include <net/tcp.h>
  42
  43/*
  44 * xprtsock tunables
  45 */
  46unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
  47unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;
  48
  49unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
  50unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
  51
  52/*
  53 * We can register our own files under /proc/sys/sunrpc by
  54 * calling register_sysctl_table() again.  The files in that
  55 * directory become the union of all files registered there.
  56 *
  57 * We simply need to make sure that we don't collide with
  58 * someone else's file names!
  59 */
  60
  61#ifdef RPC_DEBUG
  62
  63static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
  64static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
  65static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
  66static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
  67
  68static struct ctl_table_header *sunrpc_table_header;
  69
  70/*
  71 * FIXME: changing the UDP slot table size should also resize the UDP
  72 *        socket buffers for existing UDP transports
  73 */
  74static ctl_table xs_tunables_table[] = {
  75        {
  76                .ctl_name       = CTL_SLOTTABLE_UDP,
  77                .procname       = "udp_slot_table_entries",
  78                .data           = &xprt_udp_slot_table_entries,
  79                .maxlen         = sizeof(unsigned int),
  80                .mode           = 0644,
  81                .proc_handler   = &proc_dointvec_minmax,
  82                .strategy       = &sysctl_intvec,
  83                .extra1         = &min_slot_table_size,
  84                .extra2         = &max_slot_table_size
  85        },
  86        {
  87                .ctl_name       = CTL_SLOTTABLE_TCP,
  88                .procname       = "tcp_slot_table_entries",
  89                .data           = &xprt_tcp_slot_table_entries,
  90                .maxlen         = sizeof(unsigned int),
  91                .mode           = 0644,
  92                .proc_handler   = &proc_dointvec_minmax,
  93                .strategy       = &sysctl_intvec,
  94                .extra1         = &min_slot_table_size,
  95                .extra2         = &max_slot_table_size
  96        },
  97        {
  98                .ctl_name       = CTL_MIN_RESVPORT,
  99                .procname       = "min_resvport",
 100                .data           = &xprt_min_resvport,
 101                .maxlen         = sizeof(unsigned int),
 102                .mode           = 0644,
 103                .proc_handler   = &proc_dointvec_minmax,
 104                .strategy       = &sysctl_intvec,
 105                .extra1         = &xprt_min_resvport_limit,
 106                .extra2         = &xprt_max_resvport_limit
 107        },
 108        {
 109                .ctl_name       = CTL_MAX_RESVPORT,
 110                .procname       = "max_resvport",
 111                .data           = &xprt_max_resvport,
 112                .maxlen         = sizeof(unsigned int),
 113                .mode           = 0644,
 114                .proc_handler   = &proc_dointvec_minmax,
 115                .strategy       = &sysctl_intvec,
 116                .extra1         = &xprt_min_resvport_limit,
 117                .extra2         = &xprt_max_resvport_limit
 118        },
 119        {
 120                .ctl_name = 0,
 121        },
 122};
 123
 124static ctl_table sunrpc_table[] = {
 125        {
 126                .ctl_name       = CTL_SUNRPC,
 127                .procname       = "sunrpc",
 128                .mode           = 0555,
 129                .child          = xs_tunables_table
 130        },
 131        {
 132                .ctl_name = 0,
 133        },
 134};
 135
 136#endif
 137
 138/*
 139 * Time out for an RPC UDP socket connect.  UDP socket connects are
 140 * synchronous, but we set a timeout anyway in case of resource
 141 * exhaustion on the local host.
 142 */
 143#define XS_UDP_CONN_TO          (5U * HZ)
 144
 145/*
 146 * Wait duration for an RPC TCP connection to be established.  Solaris
 147 * NFS over TCP uses 60 seconds, for example, which is in line with how
 148 * long a server takes to reboot.
 149 */
 150#define XS_TCP_CONN_TO          (60U * HZ)
 151
 152/*
 153 * Wait duration for a reply from the RPC portmapper.
 154 */
 155#define XS_BIND_TO              (60U * HZ)
 156
 157/*
 158 * Delay if a UDP socket connect error occurs.  This is most likely some
 159 * kind of resource problem on the local host.
 160 */
 161#define XS_UDP_REEST_TO         (2U * HZ)
 162
 163/*
 164 * The reestablish timeout allows clients to delay for a bit before attempting
 165 * to reconnect to a server that just dropped our connection.
 166 *
 167 * We implement an exponential backoff when trying to reestablish a TCP
 168 * transport connection with the server.  Some servers like to drop a TCP
 169 * connection when they are overworked, so we start with a short timeout and
 170 * increase over time if the server is down or not responding.
 171 */
 172#define XS_TCP_INIT_REEST_TO    (3U * HZ)
 173#define XS_TCP_MAX_REEST_TO     (5U * 60 * HZ)
 174
 175/*
 176 * TCP idle timeout; client drops the transport socket if it is idle
 177 * for this long.  Note that we also timeout UDP sockets to prevent
 178 * holding port numbers when there is no RPC traffic.
 179 */
 180#define XS_IDLE_DISC_TO         (5U * 60 * HZ)
 181
 182#ifdef RPC_DEBUG
 183# undef  RPC_DEBUG_DATA
 184# define RPCDBG_FACILITY        RPCDBG_TRANS
 185#endif
 186
 187#ifdef RPC_DEBUG_DATA
 188static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
 189{
 190        u8 *buf = (u8 *) packet;
 191        int j;
 192
 193        dprintk("RPC:       %s\n", msg);
 194        for (j = 0; j < count && j < 128; j += 4) {
 195                if (!(j & 31)) {
 196                        if (j)
 197                                dprintk("\n");
 198                        dprintk("0x%04x ", j);
 199                }
 200                dprintk("%02x%02x%02x%02x ",
 201                        buf[j], buf[j+1], buf[j+2], buf[j+3]);
 202        }
 203        dprintk("\n");
 204}
 205#else
 206static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
 207{
 208        /* NOP */
 209}
 210#endif
 211
 212struct sock_xprt {
 213        struct rpc_xprt         xprt;
 214
 215        /*
 216         * Network layer
 217         */
 218        struct socket *         sock;
 219        struct sock *           inet;
 220
 221        /*
 222         * State of TCP reply receive
 223         */
 224        __be32                  tcp_fraghdr,
 225                                tcp_xid;
 226
 227        u32                     tcp_offset,
 228                                tcp_reclen;
 229
 230        unsigned long           tcp_copied,
 231                                tcp_flags;
 232
 233        /*
 234         * Connection of transports
 235         */
 236        struct delayed_work     connect_worker;
 237        struct sockaddr_storage addr;
 238        unsigned short          port;
 239
 240        /*
 241         * UDP socket buffer size parameters
 242         */
 243        size_t                  rcvsize,
 244                                sndsize;
 245
 246        /*
 247         * Saved socket callback addresses
 248         */
 249        void                    (*old_data_ready)(struct sock *, int);
 250        void                    (*old_state_change)(struct sock *);
 251        void                    (*old_write_space)(struct sock *);
 252        void                    (*old_error_report)(struct sock *);
 253};
 254
 255/*
 256 * TCP receive state flags
 257 */
 258#define TCP_RCV_LAST_FRAG       (1UL << 0)
 259#define TCP_RCV_COPY_FRAGHDR    (1UL << 1)
 260#define TCP_RCV_COPY_XID        (1UL << 2)
 261#define TCP_RCV_COPY_DATA       (1UL << 3)
 262
 263static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
 264{
 265        return (struct sockaddr *) &xprt->addr;
 266}
 267
 268static inline struct sockaddr_in *xs_addr_in(struct rpc_xprt *xprt)
 269{
 270        return (struct sockaddr_in *) &xprt->addr;
 271}
 272
 273static inline struct sockaddr_in6 *xs_addr_in6(struct rpc_xprt *xprt)
 274{
 275        return (struct sockaddr_in6 *) &xprt->addr;
 276}
 277
 278static void xs_format_ipv4_peer_addresses(struct rpc_xprt *xprt,
 279                                          const char *protocol,
 280                                          const char *netid)
 281{
 282        struct sockaddr_in *addr = xs_addr_in(xprt);
 283        char *buf;
 284
 285        buf = kzalloc(20, GFP_KERNEL);
 286        if (buf) {
 287                snprintf(buf, 20, "%pI4", &addr->sin_addr.s_addr);
 288        }
 289        xprt->address_strings[RPC_DISPLAY_ADDR] = buf;
 290
 291        buf = kzalloc(8, GFP_KERNEL);
 292        if (buf) {
 293                snprintf(buf, 8, "%u",
 294                                ntohs(addr->sin_port));
 295        }
 296        xprt->address_strings[RPC_DISPLAY_PORT] = buf;
 297
 298        xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
 299
 300        buf = kzalloc(48, GFP_KERNEL);
 301        if (buf) {
 302                snprintf(buf, 48, "addr=%pI4 port=%u proto=%s",
 303                        &addr->sin_addr.s_addr,
 304                        ntohs(addr->sin_port),
 305                        protocol);
 306        }
 307        xprt->address_strings[RPC_DISPLAY_ALL] = buf;
 308
 309        buf = kzalloc(10, GFP_KERNEL);
 310        if (buf) {
 311                snprintf(buf, 10, "%02x%02x%02x%02x",
 312                                NIPQUAD(addr->sin_addr.s_addr));
 313        }
 314        xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = buf;
 315
 316        buf = kzalloc(8, GFP_KERNEL);
 317        if (buf) {
 318                snprintf(buf, 8, "%4hx",
 319                                ntohs(addr->sin_port));
 320        }
 321        xprt->address_strings[RPC_DISPLAY_HEX_PORT] = buf;
 322
 323        buf = kzalloc(30, GFP_KERNEL);
 324        if (buf) {
 325                snprintf(buf, 30, "%pI4.%u.%u",
 326                                &addr->sin_addr.s_addr,
 327                                ntohs(addr->sin_port) >> 8,
 328                                ntohs(addr->sin_port) & 0xff);
 329        }
 330        xprt->address_strings[RPC_DISPLAY_UNIVERSAL_ADDR] = buf;
 331
 332        xprt->address_strings[RPC_DISPLAY_NETID] = netid;
 333}
 334
 335static void xs_format_ipv6_peer_addresses(struct rpc_xprt *xprt,
 336                                          const char *protocol,
 337                                          const char *netid)
 338{
 339        struct sockaddr_in6 *addr = xs_addr_in6(xprt);
 340        char *buf;
 341
 342        buf = kzalloc(40, GFP_KERNEL);
 343        if (buf) {
 344                snprintf(buf, 40, "%pI6",&addr->sin6_addr);
 345        }
 346        xprt->address_strings[RPC_DISPLAY_ADDR] = buf;
 347
 348        buf = kzalloc(8, GFP_KERNEL);
 349        if (buf) {
 350                snprintf(buf, 8, "%u",
 351                                ntohs(addr->sin6_port));
 352        }
 353        xprt->address_strings[RPC_DISPLAY_PORT] = buf;
 354
 355        xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
 356
 357        buf = kzalloc(64, GFP_KERNEL);
 358        if (buf) {
 359                snprintf(buf, 64, "addr=%pI6 port=%u proto=%s",
 360                                &addr->sin6_addr,
 361                                ntohs(addr->sin6_port),
 362                                protocol);
 363        }
 364        xprt->address_strings[RPC_DISPLAY_ALL] = buf;
 365
 366        buf = kzalloc(36, GFP_KERNEL);
 367        if (buf)
 368                snprintf(buf, 36, "%pi6", &addr->sin6_addr);
 369
 370        xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = buf;
 371
 372        buf = kzalloc(8, GFP_KERNEL);
 373        if (buf) {
 374                snprintf(buf, 8, "%4hx",
 375                                ntohs(addr->sin6_port));
 376        }
 377        xprt->address_strings[RPC_DISPLAY_HEX_PORT] = buf;
 378
 379        buf = kzalloc(50, GFP_KERNEL);
 380        if (buf) {
 381                snprintf(buf, 50, "%pI6.%u.%u",
 382                         &addr->sin6_addr,
 383                         ntohs(addr->sin6_port) >> 8,
 384                         ntohs(addr->sin6_port) & 0xff);
 385        }
 386        xprt->address_strings[RPC_DISPLAY_UNIVERSAL_ADDR] = buf;
 387
 388        xprt->address_strings[RPC_DISPLAY_NETID] = netid;
 389}
 390
 391static void xs_free_peer_addresses(struct rpc_xprt *xprt)
 392{
 393        unsigned int i;
 394
 395        for (i = 0; i < RPC_DISPLAY_MAX; i++)
 396                switch (i) {
 397                case RPC_DISPLAY_PROTO:
 398                case RPC_DISPLAY_NETID:
 399                        continue;
 400                default:
 401                        kfree(xprt->address_strings[i]);
 402                }
 403}
 404
 405#define XS_SENDMSG_FLAGS        (MSG_DONTWAIT | MSG_NOSIGNAL)
 406
 407static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
 408{
 409        struct msghdr msg = {
 410                .msg_name       = addr,
 411                .msg_namelen    = addrlen,
 412                .msg_flags      = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
 413        };
 414        struct kvec iov = {
 415                .iov_base       = vec->iov_base + base,
 416                .iov_len        = vec->iov_len - base,
 417        };
 418
 419        if (iov.iov_len != 0)
 420                return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
 421        return kernel_sendmsg(sock, &msg, NULL, 0, 0);
 422}
 423
 424static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
 425{
 426        struct page **ppage;
 427        unsigned int remainder;
 428        int err, sent = 0;
 429
 430        remainder = xdr->page_len - base;
 431        base += xdr->page_base;
 432        ppage = xdr->pages + (base >> PAGE_SHIFT);
 433        base &= ~PAGE_MASK;
 434        for(;;) {
 435                unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
 436                int flags = XS_SENDMSG_FLAGS;
 437
 438                remainder -= len;
 439                if (remainder != 0 || more)
 440                        flags |= MSG_MORE;
 441                err = sock->ops->sendpage(sock, *ppage, base, len, flags);
 442                if (remainder == 0 || err != len)
 443                        break;
 444                sent += err;
 445                ppage++;
 446                base = 0;
 447        }
 448        if (sent == 0)
 449                return err;
 450        if (err > 0)
 451                sent += err;
 452        return sent;
 453}
 454
 455/**
 456 * xs_sendpages - write pages directly to a socket
 457 * @sock: socket to send on
 458 * @addr: UDP only -- address of destination
 459 * @addrlen: UDP only -- length of destination address
 460 * @xdr: buffer containing this request
 461 * @base: starting position in the buffer
 462 *
 463 */
 464static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
 465{
 466        unsigned int remainder = xdr->len - base;
 467        int err, sent = 0;
 468
 469        if (unlikely(!sock))
 470                return -ENOTSOCK;
 471
 472        clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
 473        if (base != 0) {
 474                addr = NULL;
 475                addrlen = 0;
 476        }
 477
 478        if (base < xdr->head[0].iov_len || addr != NULL) {
 479                unsigned int len = xdr->head[0].iov_len - base;
 480                remainder -= len;
 481                err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
 482                if (remainder == 0 || err != len)
 483                        goto out;
 484                sent += err;
 485                base = 0;
 486        } else
 487                base -= xdr->head[0].iov_len;
 488
 489        if (base < xdr->page_len) {
 490                unsigned int len = xdr->page_len - base;
 491                remainder -= len;
 492                err = xs_send_pagedata(sock, xdr, base, remainder != 0);
 493                if (remainder == 0 || err != len)
 494                        goto out;
 495                sent += err;
 496                base = 0;
 497        } else
 498                base -= xdr->page_len;
 499
 500        if (base >= xdr->tail[0].iov_len)
 501                return sent;
 502        err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
 503out:
 504        if (sent == 0)
 505                return err;
 506        if (err > 0)
 507                sent += err;
 508        return sent;
 509}
 510
 511static void xs_nospace_callback(struct rpc_task *task)
 512{
 513        struct sock_xprt *transport = container_of(task->tk_rqstp->rq_xprt, struct sock_xprt, xprt);
 514
 515        transport->inet->sk_write_pending--;
 516        clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
 517}
 518
 519/**
 520 * xs_nospace - place task on wait queue if transmit was incomplete
 521 * @task: task to put to sleep
 522 *
 523 */
 524static void xs_nospace(struct rpc_task *task)
 525{
 526        struct rpc_rqst *req = task->tk_rqstp;
 527        struct rpc_xprt *xprt = req->rq_xprt;
 528        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 529
 530        dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
 531                        task->tk_pid, req->rq_slen - req->rq_bytes_sent,
 532                        req->rq_slen);
 533
 534        /* Protect against races with write_space */
 535        spin_lock_bh(&xprt->transport_lock);
 536
 537        /* Don't race with disconnect */
 538        if (xprt_connected(xprt)) {
 539                if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
 540                        /*
 541                         * Notify TCP that we're limited by the application
 542                         * window size
 543                         */
 544                        set_bit(SOCK_NOSPACE, &transport->sock->flags);
 545                        transport->inet->sk_write_pending++;
 546                        /* ...and wait for more buffer space */
 547                        xprt_wait_for_buffer_space(task, xs_nospace_callback);
 548                }
 549        } else {
 550                clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
 551                task->tk_status = -ENOTCONN;
 552        }
 553
 554        spin_unlock_bh(&xprt->transport_lock);
 555}
 556
 557/**
 558 * xs_udp_send_request - write an RPC request to a UDP socket
 559 * @task: address of RPC task that manages the state of an RPC request
 560 *
 561 * Return values:
 562 *        0:    The request has been sent
 563 *   EAGAIN:    The socket was blocked, please call again later to
 564 *              complete the request
 565 * ENOTCONN:    Caller needs to invoke connect logic then call again
 566 *    other:    Some other error occured, the request was not sent
 567 */
 568static int xs_udp_send_request(struct rpc_task *task)
 569{
 570        struct rpc_rqst *req = task->tk_rqstp;
 571        struct rpc_xprt *xprt = req->rq_xprt;
 572        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 573        struct xdr_buf *xdr = &req->rq_snd_buf;
 574        int status;
 575
 576        xs_pktdump("packet data:",
 577                                req->rq_svec->iov_base,
 578                                req->rq_svec->iov_len);
 579
 580        if (!xprt_bound(xprt))
 581                return -ENOTCONN;
 582        status = xs_sendpages(transport->sock,
 583                              xs_addr(xprt),
 584                              xprt->addrlen, xdr,
 585                              req->rq_bytes_sent);
 586
 587        dprintk("RPC:       xs_udp_send_request(%u) = %d\n",
 588                        xdr->len - req->rq_bytes_sent, status);
 589
 590        if (status >= 0) {
 591                task->tk_bytes_sent += status;
 592                if (status >= req->rq_slen)
 593                        return 0;
 594                /* Still some bytes left; set up for a retry later. */
 595                status = -EAGAIN;
 596        }
 597
 598        switch (status) {
 599        case -ENOTSOCK:
 600                status = -ENOTCONN;
 601                /* Should we call xs_close() here? */
 602                break;
 603        case -EAGAIN:
 604                xs_nospace(task);
 605                break;
 606        case -ENETUNREACH:
 607        case -EPIPE:
 608        case -ECONNREFUSED:
 609                /* When the server has died, an ICMP port unreachable message
 610                 * prompts ECONNREFUSED. */
 611                clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
 612                break;
 613        default:
 614                clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
 615                dprintk("RPC:       sendmsg returned unrecognized error %d\n",
 616                        -status);
 617        }
 618
 619        return status;
 620}
 621
 622/**
 623 * xs_tcp_shutdown - gracefully shut down a TCP socket
 624 * @xprt: transport
 625 *
 626 * Initiates a graceful shutdown of the TCP socket by calling the
 627 * equivalent of shutdown(SHUT_WR);
 628 */
 629static void xs_tcp_shutdown(struct rpc_xprt *xprt)
 630{
 631        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 632        struct socket *sock = transport->sock;
 633
 634        if (sock != NULL)
 635                kernel_sock_shutdown(sock, SHUT_WR);
 636}
 637
 638static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf)
 639{
 640        u32 reclen = buf->len - sizeof(rpc_fraghdr);
 641        rpc_fraghdr *base = buf->head[0].iov_base;
 642        *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen);
 643}
 644
 645/**
 646 * xs_tcp_send_request - write an RPC request to a TCP socket
 647 * @task: address of RPC task that manages the state of an RPC request
 648 *
 649 * Return values:
 650 *        0:    The request has been sent
 651 *   EAGAIN:    The socket was blocked, please call again later to
 652 *              complete the request
 653 * ENOTCONN:    Caller needs to invoke connect logic then call again
 654 *    other:    Some other error occured, the request was not sent
 655 *
 656 * XXX: In the case of soft timeouts, should we eventually give up
 657 *      if sendmsg is not able to make progress?
 658 */
 659static int xs_tcp_send_request(struct rpc_task *task)
 660{
 661        struct rpc_rqst *req = task->tk_rqstp;
 662        struct rpc_xprt *xprt = req->rq_xprt;
 663        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 664        struct xdr_buf *xdr = &req->rq_snd_buf;
 665        int status;
 666
 667        xs_encode_tcp_record_marker(&req->rq_snd_buf);
 668
 669        xs_pktdump("packet data:",
 670                                req->rq_svec->iov_base,
 671                                req->rq_svec->iov_len);
 672
 673        /* Continue transmitting the packet/record. We must be careful
 674         * to cope with writespace callbacks arriving _after_ we have
 675         * called sendmsg(). */
 676        while (1) {
 677                status = xs_sendpages(transport->sock,
 678                                        NULL, 0, xdr, req->rq_bytes_sent);
 679
 680                dprintk("RPC:       xs_tcp_send_request(%u) = %d\n",
 681                                xdr->len - req->rq_bytes_sent, status);
 682
 683                if (unlikely(status < 0))
 684                        break;
 685
 686                /* If we've sent the entire packet, immediately
 687                 * reset the count of bytes sent. */
 688                req->rq_bytes_sent += status;
 689                task->tk_bytes_sent += status;
 690                if (likely(req->rq_bytes_sent >= req->rq_slen)) {
 691                        req->rq_bytes_sent = 0;
 692                        return 0;
 693                }
 694
 695                if (status != 0)
 696                        continue;
 697                status = -EAGAIN;
 698                break;
 699        }
 700
 701        switch (status) {
 702        case -ENOTSOCK:
 703                status = -ENOTCONN;
 704                /* Should we call xs_close() here? */
 705                break;
 706        case -EAGAIN:
 707                xs_nospace(task);
 708                break;
 709        case -ECONNRESET:
 710                xs_tcp_shutdown(xprt);
 711        case -ECONNREFUSED:
 712        case -ENOTCONN:
 713        case -EPIPE:
 714                status = -ENOTCONN;
 715                clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
 716                break;
 717        default:
 718                dprintk("RPC:       sendmsg returned unrecognized error %d\n",
 719                        -status);
 720                clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
 721                xs_tcp_shutdown(xprt);
 722        }
 723
 724        return status;
 725}
 726
 727/**
 728 * xs_tcp_release_xprt - clean up after a tcp transmission
 729 * @xprt: transport
 730 * @task: rpc task
 731 *
 732 * This cleans up if an error causes us to abort the transmission of a request.
 733 * In this case, the socket may need to be reset in order to avoid confusing
 734 * the server.
 735 */
 736static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
 737{
 738        struct rpc_rqst *req;
 739
 740        if (task != xprt->snd_task)
 741                return;
 742        if (task == NULL)
 743                goto out_release;
 744        req = task->tk_rqstp;
 745        if (req->rq_bytes_sent == 0)
 746                goto out_release;
 747        if (req->rq_bytes_sent == req->rq_snd_buf.len)
 748                goto out_release;
 749        set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state);
 750out_release:
 751        xprt_release_xprt(xprt, task);
 752}
 753
 754static void xs_save_old_callbacks(struct sock_xprt *transport, struct sock *sk)
 755{
 756        transport->old_data_ready = sk->sk_data_ready;
 757        transport->old_state_change = sk->sk_state_change;
 758        transport->old_write_space = sk->sk_write_space;
 759        transport->old_error_report = sk->sk_error_report;
 760}
 761
 762static void xs_restore_old_callbacks(struct sock_xprt *transport, struct sock *sk)
 763{
 764        sk->sk_data_ready = transport->old_data_ready;
 765        sk->sk_state_change = transport->old_state_change;
 766        sk->sk_write_space = transport->old_write_space;
 767        sk->sk_error_report = transport->old_error_report;
 768}
 769
 770/**
 771 * xs_close - close a socket
 772 * @xprt: transport
 773 *
 774 * This is used when all requests are complete; ie, no DRC state remains
 775 * on the server we want to save.
 776 */
 777static void xs_close(struct rpc_xprt *xprt)
 778{
 779        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 780        struct socket *sock = transport->sock;
 781        struct sock *sk = transport->inet;
 782
 783        if (!sk)
 784                goto clear_close_wait;
 785
 786        dprintk("RPC:       xs_close xprt %p\n", xprt);
 787
 788        write_lock_bh(&sk->sk_callback_lock);
 789        transport->inet = NULL;
 790        transport->sock = NULL;
 791
 792        sk->sk_user_data = NULL;
 793
 794        xs_restore_old_callbacks(transport, sk);
 795        write_unlock_bh(&sk->sk_callback_lock);
 796
 797        sk->sk_no_check = 0;
 798
 799        sock_release(sock);
 800clear_close_wait:
 801        smp_mb__before_clear_bit();
 802        clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
 803        clear_bit(XPRT_CLOSING, &xprt->state);
 804        smp_mb__after_clear_bit();
 805        xprt_disconnect_done(xprt);
 806}
 807
 808/**
 809 * xs_destroy - prepare to shutdown a transport
 810 * @xprt: doomed transport
 811 *
 812 */
 813static void xs_destroy(struct rpc_xprt *xprt)
 814{
 815        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 816
 817        dprintk("RPC:       xs_destroy xprt %p\n", xprt);
 818
 819        cancel_rearming_delayed_work(&transport->connect_worker);
 820
 821        xs_close(xprt);
 822        xs_free_peer_addresses(xprt);
 823        kfree(xprt->slot);
 824        kfree(xprt);
 825        module_put(THIS_MODULE);
 826}
 827
 828static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
 829{
 830        return (struct rpc_xprt *) sk->sk_user_data;
 831}
 832
 833/**
 834 * xs_udp_data_ready - "data ready" callback for UDP sockets
 835 * @sk: socket with data to read
 836 * @len: how much data to read
 837 *
 838 */
 839static void xs_udp_data_ready(struct sock *sk, int len)
 840{
 841        struct rpc_task *task;
 842        struct rpc_xprt *xprt;
 843        struct rpc_rqst *rovr;
 844        struct sk_buff *skb;
 845        int err, repsize, copied;
 846        u32 _xid;
 847        __be32 *xp;
 848
 849        read_lock(&sk->sk_callback_lock);
 850        dprintk("RPC:       xs_udp_data_ready...\n");
 851        if (!(xprt = xprt_from_sock(sk)))
 852                goto out;
 853
 854        if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
 855                goto out;
 856
 857        if (xprt->shutdown)
 858                goto dropit;
 859
 860        repsize = skb->len - sizeof(struct udphdr);
 861        if (repsize < 4) {
 862                dprintk("RPC:       impossible RPC reply size %d!\n", repsize);
 863                goto dropit;
 864        }
 865
 866        /* Copy the XID from the skb... */
 867        xp = skb_header_pointer(skb, sizeof(struct udphdr),
 868                                sizeof(_xid), &_xid);
 869        if (xp == NULL)
 870                goto dropit;
 871
 872        /* Look up and lock the request corresponding to the given XID */
 873        spin_lock(&xprt->transport_lock);
 874        rovr = xprt_lookup_rqst(xprt, *xp);
 875        if (!rovr)
 876                goto out_unlock;
 877        task = rovr->rq_task;
 878
 879        if ((copied = rovr->rq_private_buf.buflen) > repsize)
 880                copied = repsize;
 881
 882        /* Suck it into the iovec, verify checksum if not done by hw. */
 883        if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
 884                UDPX_INC_STATS_BH(sk, UDP_MIB_INERRORS);
 885                goto out_unlock;
 886        }
 887
 888        UDPX_INC_STATS_BH(sk, UDP_MIB_INDATAGRAMS);
 889
 890        /* Something worked... */
 891        dst_confirm(skb->dst);
 892
 893        xprt_adjust_cwnd(task, copied);
 894        xprt_update_rtt(task);
 895        xprt_complete_rqst(task, copied);
 896
 897 out_unlock:
 898        spin_unlock(&xprt->transport_lock);
 899 dropit:
 900        skb_free_datagram(sk, skb);
 901 out:
 902        read_unlock(&sk->sk_callback_lock);
 903}
 904
 905static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
 906{
 907        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 908        size_t len, used;
 909        char *p;
 910
 911        p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
 912        len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
 913        used = xdr_skb_read_bits(desc, p, len);
 914        transport->tcp_offset += used;
 915        if (used != len)
 916                return;
 917
 918        transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
 919        if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
 920                transport->tcp_flags |= TCP_RCV_LAST_FRAG;
 921        else
 922                transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
 923        transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
 924
 925        transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
 926        transport->tcp_offset = 0;
 927
 928        /* Sanity check of the record length */
 929        if (unlikely(transport->tcp_reclen < 4)) {
 930                dprintk("RPC:       invalid TCP record fragment length\n");
 931                xprt_force_disconnect(xprt);
 932                return;
 933        }
 934        dprintk("RPC:       reading TCP record fragment of length %d\n",
 935                        transport->tcp_reclen);
 936}
 937
 938static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
 939{
 940        if (transport->tcp_offset == transport->tcp_reclen) {
 941                transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
 942                transport->tcp_offset = 0;
 943                if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
 944                        transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
 945                        transport->tcp_flags |= TCP_RCV_COPY_XID;
 946                        transport->tcp_copied = 0;
 947                }
 948        }
 949}
 950
 951static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
 952{
 953        size_t len, used;
 954        char *p;
 955
 956        len = sizeof(transport->tcp_xid) - transport->tcp_offset;
 957        dprintk("RPC:       reading XID (%Zu bytes)\n", len);
 958        p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
 959        used = xdr_skb_read_bits(desc, p, len);
 960        transport->tcp_offset += used;
 961        if (used != len)
 962                return;
 963        transport->tcp_flags &= ~TCP_RCV_COPY_XID;
 964        transport->tcp_flags |= TCP_RCV_COPY_DATA;
 965        transport->tcp_copied = 4;
 966        dprintk("RPC:       reading reply for XID %08x\n",
 967                        ntohl(transport->tcp_xid));
 968        xs_tcp_check_fraghdr(transport);
 969}
 970
 971static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
 972{
 973        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 974        struct rpc_rqst *req;
 975        struct xdr_buf *rcvbuf;
 976        size_t len;
 977        ssize_t r;
 978
 979        /* Find and lock the request corresponding to this xid */
 980        spin_lock(&xprt->transport_lock);
 981        req = xprt_lookup_rqst(xprt, transport->tcp_xid);
 982        if (!req) {
 983                transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
 984                dprintk("RPC:       XID %08x request not found!\n",
 985                                ntohl(transport->tcp_xid));
 986                spin_unlock(&xprt->transport_lock);
 987                return;
 988        }
 989
 990        rcvbuf = &req->rq_private_buf;
 991        len = desc->count;
 992        if (len > transport->tcp_reclen - transport->tcp_offset) {
 993                struct xdr_skb_reader my_desc;
 994
 995                len = transport->tcp_reclen - transport->tcp_offset;
 996                memcpy(&my_desc, desc, sizeof(my_desc));
 997                my_desc.count = len;
 998                r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
 999                                          &my_desc, xdr_skb_read_bits);
1000                desc->count -= r;
1001                desc->offset += r;
1002        } else
1003                r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
1004                                          desc, xdr_skb_read_bits);
1005
1006        if (r > 0) {
1007                transport->tcp_copied += r;
1008                transport->tcp_offset += r;
1009        }
1010        if (r != len) {
1011                /* Error when copying to the receive buffer,
1012                 * usually because we weren't able to allocate
1013                 * additional buffer pages. All we can do now
1014                 * is turn off TCP_RCV_COPY_DATA, so the request
1015                 * will not receive any additional updates,
1016                 * and time out.
1017                 * Any remaining data from this record will
1018                 * be discarded.
1019                 */
1020                transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1021                dprintk("RPC:       XID %08x truncated request\n",
1022                                ntohl(transport->tcp_xid));
1023                dprintk("RPC:       xprt = %p, tcp_copied = %lu, "
1024                                "tcp_offset = %u, tcp_reclen = %u\n",
1025                                xprt, transport->tcp_copied,
1026                                transport->tcp_offset, transport->tcp_reclen);
1027                goto out;
1028        }
1029
1030        dprintk("RPC:       XID %08x read %Zd bytes\n",
1031                        ntohl(transport->tcp_xid), r);
1032        dprintk("RPC:       xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
1033                        "tcp_reclen = %u\n", xprt, transport->tcp_copied,
1034                        transport->tcp_offset, transport->tcp_reclen);
1035
1036        if (transport->tcp_copied == req->rq_private_buf.buflen)
1037                transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1038        else if (transport->tcp_offset == transport->tcp_reclen) {
1039                if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
1040                        transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
1041        }
1042
1043out:
1044        if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
1045                xprt_complete_rqst(req->rq_task, transport->tcp_copied);
1046        spin_unlock(&xprt->transport_lock);
1047        xs_tcp_check_fraghdr(transport);
1048}
1049
1050static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
1051{
1052        size_t len;
1053
1054        len = transport->tcp_reclen - transport->tcp_offset;
1055        if (len > desc->count)
1056                len = desc->count;
1057        desc->count -= len;
1058        desc->offset += len;
1059        transport->tcp_offset += len;
1060        dprintk("RPC:       discarded %Zu bytes\n", len);
1061        xs_tcp_check_fraghdr(transport);
1062}
1063
1064static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
1065{
1066        struct rpc_xprt *xprt = rd_desc->arg.data;
1067        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1068        struct xdr_skb_reader desc = {
1069                .skb    = skb,
1070                .offset = offset,
1071                .count  = len,
1072        };
1073
1074        dprintk("RPC:       xs_tcp_data_recv started\n");
1075        do {
1076                /* Read in a new fragment marker if necessary */
1077                /* Can we ever really expect to get completely empty fragments? */
1078                if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
1079                        xs_tcp_read_fraghdr(xprt, &desc);
1080                        continue;
1081                }
1082                /* Read in the xid if necessary */
1083                if (transport->tcp_flags & TCP_RCV_COPY_XID) {
1084                        xs_tcp_read_xid(transport, &desc);
1085                        continue;
1086                }
1087                /* Read in the request data */
1088                if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
1089                        xs_tcp_read_request(xprt, &desc);
1090                        continue;
1091                }
1092                /* Skip over any trailing bytes on short reads */
1093                xs_tcp_read_discard(transport, &desc);
1094        } while (desc.count);
1095        dprintk("RPC:       xs_tcp_data_recv done\n");
1096        return len - desc.count;
1097}
1098
1099/**
1100 * xs_tcp_data_ready - "data ready" callback for TCP sockets
1101 * @sk: socket with data to read
1102 * @bytes: how much data to read
1103 *
1104 */
1105static void xs_tcp_data_ready(struct sock *sk, int bytes)
1106{
1107        struct rpc_xprt *xprt;
1108        read_descriptor_t rd_desc;
1109        int read;
1110
1111        dprintk("RPC:       xs_tcp_data_ready...\n");
1112
1113        read_lock(&sk->sk_callback_lock);
1114        if (!(xprt = xprt_from_sock(sk)))
1115                goto out;
1116        if (xprt->shutdown)
1117                goto out;
1118
1119        /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
1120        rd_desc.arg.data = xprt;
1121        do {
1122                rd_desc.count = 65536;
1123                read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
1124        } while (read > 0);
1125out:
1126        read_unlock(&sk->sk_callback_lock);
1127}
1128
1129/**
1130 * xs_tcp_state_change - callback to handle TCP socket state changes
1131 * @sk: socket whose state has changed
1132 *
1133 */
1134static void xs_tcp_state_change(struct sock *sk)
1135{
1136        struct rpc_xprt *xprt;
1137
1138        read_lock(&sk->sk_callback_lock);
1139        if (!(xprt = xprt_from_sock(sk)))
1140                goto out;
1141        dprintk("RPC:       xs_tcp_state_change client %p...\n", xprt);
1142        dprintk("RPC:       state %x conn %d dead %d zapped %d\n",
1143                        sk->sk_state, xprt_connected(xprt),
1144                        sock_flag(sk, SOCK_DEAD),
1145                        sock_flag(sk, SOCK_ZAPPED));
1146
1147        switch (sk->sk_state) {
1148        case TCP_ESTABLISHED:
1149                spin_lock_bh(&xprt->transport_lock);
1150                if (!xprt_test_and_set_connected(xprt)) {
1151                        struct sock_xprt *transport = container_of(xprt,
1152                                        struct sock_xprt, xprt);
1153
1154                        /* Reset TCP record info */
1155                        transport->tcp_offset = 0;
1156                        transport->tcp_reclen = 0;
1157                        transport->tcp_copied = 0;
1158                        transport->tcp_flags =
1159                                TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
1160
1161                        xprt_wake_pending_tasks(xprt, 0);
1162                }
1163                spin_unlock_bh(&xprt->transport_lock);
1164                break;
1165        case TCP_FIN_WAIT1:
1166                /* The client initiated a shutdown of the socket */
1167                xprt->connect_cookie++;
1168                xprt->reestablish_timeout = 0;
1169                set_bit(XPRT_CLOSING, &xprt->state);
1170                smp_mb__before_clear_bit();
1171                clear_bit(XPRT_CONNECTED, &xprt->state);
1172                clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1173                smp_mb__after_clear_bit();
1174                break;
1175        case TCP_CLOSE_WAIT:
1176                /* The server initiated a shutdown of the socket */
1177                set_bit(XPRT_CLOSING, &xprt->state);
1178                xprt_force_disconnect(xprt);
1179        case TCP_SYN_SENT:
1180                xprt->connect_cookie++;
1181        case TCP_CLOSING:
1182                /*
1183                 * If the server closed down the connection, make sure that
1184                 * we back off before reconnecting
1185                 */
1186                if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
1187                        xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
1188                break;
1189        case TCP_LAST_ACK:
1190                smp_mb__before_clear_bit();
1191                clear_bit(XPRT_CONNECTED, &xprt->state);
1192                smp_mb__after_clear_bit();
1193                break;
1194        case TCP_CLOSE:
1195                smp_mb__before_clear_bit();
1196                clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
1197                clear_bit(XPRT_CLOSING, &xprt->state);
1198                smp_mb__after_clear_bit();
1199                /* Mark transport as closed and wake up all pending tasks */
1200                xprt_disconnect_done(xprt);
1201        }
1202 out:
1203        read_unlock(&sk->sk_callback_lock);
1204}
1205
1206/**
1207 * xs_tcp_error_report - callback mainly for catching RST events
1208 * @sk: socket
1209 */
1210static void xs_tcp_error_report(struct sock *sk)
1211{
1212        struct rpc_xprt *xprt;
1213
1214        read_lock(&sk->sk_callback_lock);
1215        if (sk->sk_err != ECONNRESET || sk->sk_state != TCP_ESTABLISHED)
1216                goto out;
1217        if (!(xprt = xprt_from_sock(sk)))
1218                goto out;
1219        dprintk("RPC:       %s client %p...\n"
1220                        "RPC:       error %d\n",
1221                        __func__, xprt, sk->sk_err);
1222
1223        xprt_force_disconnect(xprt);
1224out:
1225        read_unlock(&sk->sk_callback_lock);
1226}
1227
1228/**
1229 * xs_udp_write_space - callback invoked when socket buffer space
1230 *                             becomes available
1231 * @sk: socket whose state has changed
1232 *
1233 * Called when more output buffer space is available for this socket.
1234 * We try not to wake our writers until they can make "significant"
1235 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1236 * with a bunch of small requests.
1237 */
1238static void xs_udp_write_space(struct sock *sk)
1239{
1240        read_lock(&sk->sk_callback_lock);
1241
1242        /* from net/core/sock.c:sock_def_write_space */
1243        if (sock_writeable(sk)) {
1244                struct socket *sock;
1245                struct rpc_xprt *xprt;
1246
1247                if (unlikely(!(sock = sk->sk_socket)))
1248                        goto out;
1249                clear_bit(SOCK_NOSPACE, &sock->flags);
1250
1251                if (unlikely(!(xprt = xprt_from_sock(sk))))
1252                        goto out;
1253                if (test_and_clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags) == 0)
1254                        goto out;
1255
1256                xprt_write_space(xprt);
1257        }
1258
1259 out:
1260        read_unlock(&sk->sk_callback_lock);
1261}
1262
1263/**
1264 * xs_tcp_write_space - callback invoked when socket buffer space
1265 *                             becomes available
1266 * @sk: socket whose state has changed
1267 *
1268 * Called when more output buffer space is available for this socket.
1269 * We try not to wake our writers until they can make "significant"
1270 * progress, otherwise we'll waste resources thrashing kernel_sendmsg
1271 * with a bunch of small requests.
1272 */
1273static void xs_tcp_write_space(struct sock *sk)
1274{
1275        read_lock(&sk->sk_callback_lock);
1276
1277        /* from net/core/stream.c:sk_stream_write_space */
1278        if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
1279                struct socket *sock;
1280                struct rpc_xprt *xprt;
1281
1282                if (unlikely(!(sock = sk->sk_socket)))
1283                        goto out;
1284                clear_bit(SOCK_NOSPACE, &sock->flags);
1285
1286                if (unlikely(!(xprt = xprt_from_sock(sk))))
1287                        goto out;
1288                if (test_and_clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags) == 0)
1289                        goto out;
1290
1291                xprt_write_space(xprt);
1292        }
1293
1294 out:
1295        read_unlock(&sk->sk_callback_lock);
1296}
1297
1298static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
1299{
1300        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1301        struct sock *sk = transport->inet;
1302
1303        if (transport->rcvsize) {
1304                sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
1305                sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
1306        }
1307        if (transport->sndsize) {
1308                sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
1309                sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
1310                sk->sk_write_space(sk);
1311        }
1312}
1313
1314/**
1315 * xs_udp_set_buffer_size - set send and receive limits
1316 * @xprt: generic transport
1317 * @sndsize: requested size of send buffer, in bytes
1318 * @rcvsize: requested size of receive buffer, in bytes
1319 *
1320 * Set socket send and receive buffer size limits.
1321 */
1322static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
1323{
1324        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1325
1326        transport->sndsize = 0;
1327        if (sndsize)
1328                transport->sndsize = sndsize + 1024;
1329        transport->rcvsize = 0;
1330        if (rcvsize)
1331                transport->rcvsize = rcvsize + 1024;
1332
1333        xs_udp_do_set_buffer_size(xprt);
1334}
1335
1336/**
1337 * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
1338 * @task: task that timed out
1339 *
1340 * Adjust the congestion window after a retransmit timeout has occurred.
1341 */
1342static void xs_udp_timer(struct rpc_task *task)
1343{
1344        xprt_adjust_cwnd(task, -ETIMEDOUT);
1345}
1346
1347static unsigned short xs_get_random_port(void)
1348{
1349        unsigned short range = xprt_max_resvport - xprt_min_resvport;
1350        unsigned short rand = (unsigned short) net_random() % range;
1351        return rand + xprt_min_resvport;
1352}
1353
1354/**
1355 * xs_set_port - reset the port number in the remote endpoint address
1356 * @xprt: generic transport
1357 * @port: new port number
1358 *
1359 */
1360static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
1361{
1362        struct sockaddr *addr = xs_addr(xprt);
1363
1364        dprintk("RPC:       setting port for xprt %p to %u\n", xprt, port);
1365
1366        switch (addr->sa_family) {
1367        case AF_INET:
1368                ((struct sockaddr_in *)addr)->sin_port = htons(port);
1369                break;
1370        case AF_INET6:
1371                ((struct sockaddr_in6 *)addr)->sin6_port = htons(port);
1372                break;
1373        default:
1374                BUG();
1375        }
1376}
1377
1378static unsigned short xs_get_srcport(struct sock_xprt *transport, struct socket *sock)
1379{
1380        unsigned short port = transport->port;
1381
1382        if (port == 0 && transport->xprt.resvport)
1383                port = xs_get_random_port();
1384        return port;
1385}
1386
1387static unsigned short xs_next_srcport(struct sock_xprt *transport, struct socket *sock, unsigned short port)
1388{
1389        if (transport->port != 0)
1390                transport->port = 0;
1391        if (!transport->xprt.resvport)
1392                return 0;
1393        if (port <= xprt_min_resvport || port > xprt_max_resvport)
1394                return xprt_max_resvport;
1395        return --port;
1396}
1397
1398static int xs_bind4(struct sock_xprt *transport, struct socket *sock)
1399{
1400        struct sockaddr_in myaddr = {
1401                .sin_family = AF_INET,
1402        };
1403        struct sockaddr_in *sa;
1404        int err, nloop = 0;
1405        unsigned short port = xs_get_srcport(transport, sock);
1406        unsigned short last;
1407
1408        sa = (struct sockaddr_in *)&transport->addr;
1409        myaddr.sin_addr = sa->sin_addr;
1410        do {
1411                myaddr.sin_port = htons(port);
1412                err = kernel_bind(sock, (struct sockaddr *) &myaddr,
1413                                                sizeof(myaddr));
1414                if (port == 0)
1415                        break;
1416                if (err == 0) {
1417                        transport->port = port;
1418                        break;
1419                }
1420                last = port;
1421                port = xs_next_srcport(transport, sock, port);
1422                if (port > last)
1423                        nloop++;
1424        } while (err == -EADDRINUSE && nloop != 2);
1425        dprintk("RPC:       %s %pI4:%u: %s (%d)\n",
1426                        __func__, &myaddr.sin_addr,
1427                        port, err ? "failed" : "ok", err);
1428        return err;
1429}
1430
1431static int xs_bind6(struct sock_xprt *transport, struct socket *sock)
1432{
1433        struct sockaddr_in6 myaddr = {
1434                .sin6_family = AF_INET6,
1435        };
1436        struct sockaddr_in6 *sa;
1437        int err, nloop = 0;
1438        unsigned short port = xs_get_srcport(transport, sock);
1439        unsigned short last;
1440
1441        sa = (struct sockaddr_in6 *)&transport->addr;
1442        myaddr.sin6_addr = sa->sin6_addr;
1443        do {
1444                myaddr.sin6_port = htons(port);
1445                err = kernel_bind(sock, (struct sockaddr *) &myaddr,
1446                                                sizeof(myaddr));
1447                if (port == 0)
1448                        break;
1449                if (err == 0) {
1450                        transport->port = port;
1451                        break;
1452                }
1453                last = port;
1454                port = xs_next_srcport(transport, sock, port);
1455                if (port > last)
1456                        nloop++;
1457        } while (err == -EADDRINUSE && nloop != 2);
1458        dprintk("RPC:       xs_bind6 %pI6:%u: %s (%d)\n",
1459                &myaddr.sin6_addr, port, err ? "failed" : "ok", err);
1460        return err;
1461}
1462
1463#ifdef CONFIG_DEBUG_LOCK_ALLOC
1464static struct lock_class_key xs_key[2];
1465static struct lock_class_key xs_slock_key[2];
1466
1467static inline void xs_reclassify_socket4(struct socket *sock)
1468{
1469        struct sock *sk = sock->sk;
1470
1471        BUG_ON(sock_owned_by_user(sk));
1472        sock_lock_init_class_and_name(sk, "slock-AF_INET-RPC",
1473                &xs_slock_key[0], "sk_lock-AF_INET-RPC", &xs_key[0]);
1474}
1475
1476static inline void xs_reclassify_socket6(struct socket *sock)
1477{
1478        struct sock *sk = sock->sk;
1479
1480        BUG_ON(sock_owned_by_user(sk));
1481        sock_lock_init_class_and_name(sk, "slock-AF_INET6-RPC",
1482                &xs_slock_key[1], "sk_lock-AF_INET6-RPC", &xs_key[1]);
1483}
1484#else
1485static inline void xs_reclassify_socket4(struct socket *sock)
1486{
1487}
1488
1489static inline void xs_reclassify_socket6(struct socket *sock)
1490{
1491}
1492#endif
1493
1494static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
1495{
1496        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1497
1498        if (!transport->inet) {
1499                struct sock *sk = sock->sk;
1500
1501                write_lock_bh(&sk->sk_callback_lock);
1502
1503                xs_save_old_callbacks(transport, sk);
1504
1505                sk->sk_user_data = xprt;
1506                sk->sk_data_ready = xs_udp_data_ready;
1507                sk->sk_write_space = xs_udp_write_space;
1508                sk->sk_no_check = UDP_CSUM_NORCV;
1509                sk->sk_allocation = GFP_ATOMIC;
1510
1511                xprt_set_connected(xprt);
1512
1513                /* Reset to new socket */
1514                transport->sock = sock;
1515                transport->inet = sk;
1516
1517                write_unlock_bh(&sk->sk_callback_lock);
1518        }
1519        xs_udp_do_set_buffer_size(xprt);
1520}
1521
1522/**
1523 * xs_udp_connect_worker4 - set up a UDP socket
1524 * @work: RPC transport to connect
1525 *
1526 * Invoked by a work queue tasklet.
1527 */
1528static void xs_udp_connect_worker4(struct work_struct *work)
1529{
1530        struct sock_xprt *transport =
1531                container_of(work, struct sock_xprt, connect_worker.work);
1532        struct rpc_xprt *xprt = &transport->xprt;
1533        struct socket *sock = transport->sock;
1534        int err, status = -EIO;
1535
1536        if (xprt->shutdown)
1537                goto out;
1538
1539        /* Start by resetting any existing state */
1540        xs_close(xprt);
1541
1542        if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
1543                dprintk("RPC:       can't create UDP transport socket (%d).\n", -err);
1544                goto out;
1545        }
1546        xs_reclassify_socket4(sock);
1547
1548        if (xs_bind4(transport, sock)) {
1549                sock_release(sock);
1550                goto out;
1551        }
1552
1553        dprintk("RPC:       worker connecting xprt %p to address: %s\n",
1554                        xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
1555
1556        xs_udp_finish_connecting(xprt, sock);
1557        status = 0;
1558out:
1559        xprt_wake_pending_tasks(xprt, status);
1560        xprt_clear_connecting(xprt);
1561}
1562
1563/**
1564 * xs_udp_connect_worker6 - set up a UDP socket
1565 * @work: RPC transport to connect
1566 *
1567 * Invoked by a work queue tasklet.
1568 */
1569static void xs_udp_connect_worker6(struct work_struct *work)
1570{
1571        struct sock_xprt *transport =
1572                container_of(work, struct sock_xprt, connect_worker.work);
1573        struct rpc_xprt *xprt = &transport->xprt;
1574        struct socket *sock = transport->sock;
1575        int err, status = -EIO;
1576
1577        if (xprt->shutdown)
1578                goto out;
1579
1580        /* Start by resetting any existing state */
1581        xs_close(xprt);
1582
1583        if ((err = sock_create_kern(PF_INET6, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
1584                dprintk("RPC:       can't create UDP transport socket (%d).\n", -err);
1585                goto out;
1586        }
1587        xs_reclassify_socket6(sock);
1588
1589        if (xs_bind6(transport, sock) < 0) {
1590                sock_release(sock);
1591                goto out;
1592        }
1593
1594        dprintk("RPC:       worker connecting xprt %p to address: %s\n",
1595                        xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
1596
1597        xs_udp_finish_connecting(xprt, sock);
1598        status = 0;
1599out:
1600        xprt_wake_pending_tasks(xprt, status);
1601        xprt_clear_connecting(xprt);
1602}
1603
1604/*
1605 * We need to preserve the port number so the reply cache on the server can
1606 * find our cached RPC replies when we get around to reconnecting.
1607 */
1608static void xs_tcp_reuse_connection(struct rpc_xprt *xprt)
1609{
1610        int result;
1611        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1612        struct sockaddr any;
1613
1614        dprintk("RPC:       disconnecting xprt %p to reuse port\n", xprt);
1615
1616        /*
1617         * Disconnect the transport socket by doing a connect operation
1618         * with AF_UNSPEC.  This should return immediately...
1619         */
1620        memset(&any, 0, sizeof(any));
1621        any.sa_family = AF_UNSPEC;
1622        result = kernel_connect(transport->sock, &any, sizeof(any), 0);
1623        if (result)
1624                dprintk("RPC:       AF_UNSPEC connect return code %d\n",
1625                                result);
1626}
1627
1628static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
1629{
1630        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1631
1632        if (!transport->inet) {
1633                struct sock *sk = sock->sk;
1634
1635                write_lock_bh(&sk->sk_callback_lock);
1636
1637                xs_save_old_callbacks(transport, sk);
1638
1639                sk->sk_user_data = xprt;
1640                sk->sk_data_ready = xs_tcp_data_ready;
1641                sk->sk_state_change = xs_tcp_state_change;
1642                sk->sk_write_space = xs_tcp_write_space;
1643                sk->sk_error_report = xs_tcp_error_report;
1644                sk->sk_allocation = GFP_ATOMIC;
1645
1646                /* socket options */
1647                sk->sk_userlocks |= SOCK_BINDPORT_LOCK;
1648                sock_reset_flag(sk, SOCK_LINGER);
1649                tcp_sk(sk)->linger2 = 0;
1650                tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
1651
1652                xprt_clear_connected(xprt);
1653
1654                /* Reset to new socket */
1655                transport->sock = sock;
1656                transport->inet = sk;
1657
1658                write_unlock_bh(&sk->sk_callback_lock);
1659        }
1660
1661        if (!xprt_bound(xprt))
1662                return -ENOTCONN;
1663
1664        /* Tell the socket layer to start connecting... */
1665        xprt->stat.connect_count++;
1666        xprt->stat.connect_start = jiffies;
1667        return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
1668}
1669
1670/**
1671 * xs_tcp_connect_worker4 - connect a TCP socket to a remote endpoint
1672 * @work: RPC transport to connect
1673 *
1674 * Invoked by a work queue tasklet.
1675 */
1676static void xs_tcp_connect_worker4(struct work_struct *work)
1677{
1678        struct sock_xprt *transport =
1679                container_of(work, struct sock_xprt, connect_worker.work);
1680        struct rpc_xprt *xprt = &transport->xprt;
1681        struct socket *sock = transport->sock;
1682        int err, status = -EIO;
1683
1684        if (xprt->shutdown)
1685                goto out;
1686
1687        if (!sock) {
1688                /* start from scratch */
1689                if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
1690                        dprintk("RPC:       can't create TCP transport socket (%d).\n", -err);
1691                        goto out;
1692                }
1693                xs_reclassify_socket4(sock);
1694
1695                if (xs_bind4(transport, sock) < 0) {
1696                        sock_release(sock);
1697                        goto out;
1698                }
1699        } else
1700                /* "close" the socket, preserving the local port */
1701                xs_tcp_reuse_connection(xprt);
1702
1703        dprintk("RPC:       worker connecting xprt %p to address: %s\n",
1704                        xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
1705
1706        status = xs_tcp_finish_connecting(xprt, sock);
1707        dprintk("RPC:       %p connect status %d connected %d sock state %d\n",
1708                        xprt, -status, xprt_connected(xprt),
1709                        sock->sk->sk_state);
1710        if (status < 0) {
1711                switch (status) {
1712                        case -EINPROGRESS:
1713                        case -EALREADY:
1714                                goto out_clear;
1715                        case -ECONNREFUSED:
1716                        case -ECONNRESET:
1717                                /* retry with existing socket, after a delay */
1718                                break;
1719                        default:
1720                                /* get rid of existing socket, and retry */
1721                                xs_tcp_shutdown(xprt);
1722                }
1723        }
1724out:
1725        xprt_wake_pending_tasks(xprt, status);
1726out_clear:
1727        xprt_clear_connecting(xprt);
1728}
1729
1730/**
1731 * xs_tcp_connect_worker6 - connect a TCP socket to a remote endpoint
1732 * @work: RPC transport to connect
1733 *
1734 * Invoked by a work queue tasklet.
1735 */
1736static void xs_tcp_connect_worker6(struct work_struct *work)
1737{
1738        struct sock_xprt *transport =
1739                container_of(work, struct sock_xprt, connect_worker.work);
1740        struct rpc_xprt *xprt = &transport->xprt;
1741        struct socket *sock = transport->sock;
1742        int err, status = -EIO;
1743
1744        if (xprt->shutdown)
1745                goto out;
1746
1747        if (!sock) {
1748                /* start from scratch */
1749                if ((err = sock_create_kern(PF_INET6, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
1750                        dprintk("RPC:       can't create TCP transport socket (%d).\n", -err);
1751                        goto out;
1752                }
1753                xs_reclassify_socket6(sock);
1754
1755                if (xs_bind6(transport, sock) < 0) {
1756                        sock_release(sock);
1757                        goto out;
1758                }
1759        } else
1760                /* "close" the socket, preserving the local port */
1761                xs_tcp_reuse_connection(xprt);
1762
1763        dprintk("RPC:       worker connecting xprt %p to address: %s\n",
1764                        xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
1765
1766        status = xs_tcp_finish_connecting(xprt, sock);
1767        dprintk("RPC:       %p connect status %d connected %d sock state %d\n",
1768                        xprt, -status, xprt_connected(xprt), sock->sk->sk_state);
1769        if (status < 0) {
1770                switch (status) {
1771                        case -EINPROGRESS:
1772                        case -EALREADY:
1773                                goto out_clear;
1774                        case -ECONNREFUSED:
1775                        case -ECONNRESET:
1776                                /* retry with existing socket, after a delay */
1777                                break;
1778                        default:
1779                                /* get rid of existing socket, and retry */
1780                                xs_tcp_shutdown(xprt);
1781                }
1782        }
1783out:
1784        xprt_wake_pending_tasks(xprt, status);
1785out_clear:
1786        xprt_clear_connecting(xprt);
1787}
1788
1789/**
1790 * xs_connect - connect a socket to a remote endpoint
1791 * @task: address of RPC task that manages state of connect request
1792 *
1793 * TCP: If the remote end dropped the connection, delay reconnecting.
1794 *
1795 * UDP socket connects are synchronous, but we use a work queue anyway
1796 * to guarantee that even unprivileged user processes can set up a
1797 * socket on a privileged port.
1798 *
1799 * If a UDP socket connect fails, the delay behavior here prevents
1800 * retry floods (hard mounts).
1801 */
1802static void xs_connect(struct rpc_task *task)
1803{
1804        struct rpc_xprt *xprt = task->tk_xprt;
1805        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1806
1807        if (xprt_test_and_set_connecting(xprt))
1808                return;
1809
1810        if (transport->sock != NULL) {
1811                dprintk("RPC:       xs_connect delayed xprt %p for %lu "
1812                                "seconds\n",
1813                                xprt, xprt->reestablish_timeout / HZ);
1814                queue_delayed_work(rpciod_workqueue,
1815                                   &transport->connect_worker,
1816                                   xprt->reestablish_timeout);
1817                xprt->reestablish_timeout <<= 1;
1818                if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
1819                        xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
1820        } else {
1821                dprintk("RPC:       xs_connect scheduled xprt %p\n", xprt);
1822                queue_delayed_work(rpciod_workqueue,
1823                                   &transport->connect_worker, 0);
1824        }
1825}
1826
1827static void xs_tcp_connect(struct rpc_task *task)
1828{
1829        struct rpc_xprt *xprt = task->tk_xprt;
1830
1831        /* Initiate graceful shutdown of the socket if not already done */
1832        if (test_bit(XPRT_CONNECTED, &xprt->state))
1833                xs_tcp_shutdown(xprt);
1834        /* Exit if we need to wait for socket shutdown to complete */
1835        if (test_bit(XPRT_CLOSING, &xprt->state))
1836                return;
1837        xs_connect(task);
1838}
1839
1840/**
1841 * xs_udp_print_stats - display UDP socket-specifc stats
1842 * @xprt: rpc_xprt struct containing statistics
1843 * @seq: output file
1844 *
1845 */
1846static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
1847{
1848        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1849
1850        seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
1851                        transport->port,
1852                        xprt->stat.bind_count,
1853                        xprt->stat.sends,
1854                        xprt->stat.recvs,
1855                        xprt->stat.bad_xids,
1856                        xprt->stat.req_u,
1857                        xprt->stat.bklog_u);
1858}
1859
1860/**
1861 * xs_tcp_print_stats - display TCP socket-specifc stats
1862 * @xprt: rpc_xprt struct containing statistics
1863 * @seq: output file
1864 *
1865 */
1866static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
1867{
1868        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
1869        long idle_time = 0;
1870
1871        if (xprt_connected(xprt))
1872                idle_time = (long)(jiffies - xprt->last_used) / HZ;
1873
1874        seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n",
1875                        transport->port,
1876                        xprt->stat.bind_count,
1877                        xprt->stat.connect_count,
1878                        xprt->stat.connect_time,
1879                        idle_time,
1880                        xprt->stat.sends,
1881                        xprt->stat.recvs,
1882                        xprt->stat.bad_xids,
1883                        xprt->stat.req_u,
1884                        xprt->stat.bklog_u);
1885}
1886
1887static struct rpc_xprt_ops xs_udp_ops = {
1888        .set_buffer_size        = xs_udp_set_buffer_size,
1889        .reserve_xprt           = xprt_reserve_xprt_cong,
1890        .release_xprt           = xprt_release_xprt_cong,
1891        .rpcbind                = rpcb_getport_async,
1892        .set_port               = xs_set_port,
1893        .connect                = xs_connect,
1894        .buf_alloc              = rpc_malloc,
1895        .buf_free               = rpc_free,
1896        .send_request           = xs_udp_send_request,
1897        .set_retrans_timeout    = xprt_set_retrans_timeout_rtt,
1898        .timer                  = xs_udp_timer,
1899        .release_request        = xprt_release_rqst_cong,
1900        .close                  = xs_close,
1901        .destroy                = xs_destroy,
1902        .print_stats            = xs_udp_print_stats,
1903};
1904
1905static struct rpc_xprt_ops xs_tcp_ops = {
1906        .reserve_xprt           = xprt_reserve_xprt,
1907        .release_xprt           = xs_tcp_release_xprt,
1908        .rpcbind                = rpcb_getport_async,
1909        .set_port               = xs_set_port,
1910        .connect                = xs_tcp_connect,
1911        .buf_alloc              = rpc_malloc,
1912        .buf_free               = rpc_free,
1913        .send_request           = xs_tcp_send_request,
1914        .set_retrans_timeout    = xprt_set_retrans_timeout_def,
1915        .close                  = xs_tcp_shutdown,
1916        .destroy                = xs_destroy,
1917        .print_stats            = xs_tcp_print_stats,
1918};
1919
1920static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
1921                                      unsigned int slot_table_size)
1922{
1923        struct rpc_xprt *xprt;
1924        struct sock_xprt *new;
1925
1926        if (args->addrlen > sizeof(xprt->addr)) {
1927                dprintk("RPC:       xs_setup_xprt: address too large\n");
1928                return ERR_PTR(-EBADF);
1929        }
1930
1931        new = kzalloc(sizeof(*new), GFP_KERNEL);
1932        if (new == NULL) {
1933                dprintk("RPC:       xs_setup_xprt: couldn't allocate "
1934                                "rpc_xprt\n");
1935                return ERR_PTR(-ENOMEM);
1936        }
1937        xprt = &new->xprt;
1938
1939        xprt->max_reqs = slot_table_size;
1940        xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL);
1941        if (xprt->slot == NULL) {
1942                kfree(xprt);
1943                dprintk("RPC:       xs_setup_xprt: couldn't allocate slot "
1944                                "table\n");
1945                return ERR_PTR(-ENOMEM);
1946        }
1947
1948        memcpy(&xprt->addr, args->dstaddr, args->addrlen);
1949        xprt->addrlen = args->addrlen;
1950        if (args->srcaddr)
1951                memcpy(&new->addr, args->srcaddr, args->addrlen);
1952
1953        return xprt;
1954}
1955
1956static const struct rpc_timeout xs_udp_default_timeout = {
1957        .to_initval = 5 * HZ,
1958        .to_maxval = 30 * HZ,
1959        .to_increment = 5 * HZ,
1960        .to_retries = 5,
1961};
1962
1963/**
1964 * xs_setup_udp - Set up transport to use a UDP socket
1965 * @args: rpc transport creation arguments
1966 *
1967 */
1968static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
1969{
1970        struct sockaddr *addr = args->dstaddr;
1971        struct rpc_xprt *xprt;
1972        struct sock_xprt *transport;
1973
1974        xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries);
1975        if (IS_ERR(xprt))
1976                return xprt;
1977        transport = container_of(xprt, struct sock_xprt, xprt);
1978
1979        xprt->prot = IPPROTO_UDP;
1980        xprt->tsh_size = 0;
1981        /* XXX: header size can vary due to auth type, IPv6, etc. */
1982        xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
1983
1984        xprt->bind_timeout = XS_BIND_TO;
1985        xprt->connect_timeout = XS_UDP_CONN_TO;
1986        xprt->reestablish_timeout = XS_UDP_REEST_TO;
1987        xprt->idle_timeout = XS_IDLE_DISC_TO;
1988
1989        xprt->ops = &xs_udp_ops;
1990
1991        xprt->timeout = &xs_udp_default_timeout;
1992
1993        switch (addr->sa_family) {
1994        case AF_INET:
1995                if (((struct sockaddr_in *)addr)->sin_port != htons(0))
1996                        xprt_set_bound(xprt);
1997
1998                INIT_DELAYED_WORK(&transport->connect_worker,
1999                                        xs_udp_connect_worker4);
2000                xs_format_ipv4_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
2001                break;
2002        case AF_INET6:
2003                if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
2004                        xprt_set_bound(xprt);
2005
2006                INIT_DELAYED_WORK(&transport->connect_worker,
2007                                        xs_udp_connect_worker6);
2008                xs_format_ipv6_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
2009                break;
2010        default:
2011                kfree(xprt);
2012                return ERR_PTR(-EAFNOSUPPORT);
2013        }
2014
2015        dprintk("RPC:       set up transport to address %s\n",
2016                        xprt->address_strings[RPC_DISPLAY_ALL]);
2017
2018        if (try_module_get(THIS_MODULE))
2019                return xprt;
2020
2021        kfree(xprt->slot);
2022        kfree(xprt);
2023        return ERR_PTR(-EINVAL);
2024}
2025
2026static const struct rpc_timeout xs_tcp_default_timeout = {
2027        .to_initval = 60 * HZ,
2028        .to_maxval = 60 * HZ,
2029        .to_retries = 2,
2030};
2031
2032/**
2033 * xs_setup_tcp - Set up transport to use a TCP socket
2034 * @args: rpc transport creation arguments
2035 *
2036 */
2037static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
2038{
2039        struct sockaddr *addr = args->dstaddr;
2040        struct rpc_xprt *xprt;
2041        struct sock_xprt *transport;
2042
2043        xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
2044        if (IS_ERR(xprt))
2045                return xprt;
2046        transport = container_of(xprt, struct sock_xprt, xprt);
2047
2048        xprt->prot = IPPROTO_TCP;
2049        xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
2050        xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
2051
2052        xprt->bind_timeout = XS_BIND_TO;
2053        xprt->connect_timeout = XS_TCP_CONN_TO;
2054        xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
2055        xprt->idle_timeout = XS_IDLE_DISC_TO;
2056
2057        xprt->ops = &xs_tcp_ops;
2058        xprt->timeout = &xs_tcp_default_timeout;
2059
2060        switch (addr->sa_family) {
2061        case AF_INET:
2062                if (((struct sockaddr_in *)addr)->sin_port != htons(0))
2063                        xprt_set_bound(xprt);
2064
2065                INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker4);
2066                xs_format_ipv4_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
2067                break;
2068        case AF_INET6:
2069                if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
2070                        xprt_set_bound(xprt);
2071
2072                INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker6);
2073                xs_format_ipv6_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
2074                break;
2075        default:
2076                kfree(xprt);
2077                return ERR_PTR(-EAFNOSUPPORT);
2078        }
2079
2080        dprintk("RPC:       set up transport to address %s\n",
2081                        xprt->address_strings[RPC_DISPLAY_ALL]);
2082
2083        if (try_module_get(THIS_MODULE))
2084                return xprt;
2085
2086        kfree(xprt->slot);
2087        kfree(xprt);
2088        return ERR_PTR(-EINVAL);
2089}
2090
2091static struct xprt_class        xs_udp_transport = {
2092        .list           = LIST_HEAD_INIT(xs_udp_transport.list),
2093        .name           = "udp",
2094        .owner          = THIS_MODULE,
2095        .ident          = IPPROTO_UDP,
2096        .setup          = xs_setup_udp,
2097};
2098
2099static struct xprt_class        xs_tcp_transport = {
2100        .list           = LIST_HEAD_INIT(xs_tcp_transport.list),
2101        .name           = "tcp",
2102        .owner          = THIS_MODULE,
2103        .ident          = IPPROTO_TCP,
2104        .setup          = xs_setup_tcp,
2105};
2106
2107/**
2108 * init_socket_xprt - set up xprtsock's sysctls, register with RPC client
2109 *
2110 */
2111int init_socket_xprt(void)
2112{
2113#ifdef RPC_DEBUG
2114        if (!sunrpc_table_header)
2115                sunrpc_table_header = register_sysctl_table(sunrpc_table);
2116#endif
2117
2118        xprt_register_transport(&xs_udp_transport);
2119        xprt_register_transport(&xs_tcp_transport);
2120
2121        return 0;
2122}
2123
2124/**
2125 * cleanup_socket_xprt - remove xprtsock's sysctls, unregister
2126 *
2127 */
2128void cleanup_socket_xprt(void)
2129{
2130#ifdef RPC_DEBUG
2131        if (sunrpc_table_header) {
2132                unregister_sysctl_table(sunrpc_table_header);
2133                sunrpc_table_header = NULL;
2134        }
2135#endif
2136
2137        xprt_unregister_transport(&xs_udp_transport);
2138        xprt_unregister_transport(&xs_tcp_transport);
2139}
2140