SUNRPC: Ensure we return EAGAIN in xs_nospace if congestion is cleared
[linux-2.6.git] / net / sunrpc / xprtsock.c
index 04732d0..b60b750 100644 (file)
@@ -19,6 +19,7 @@
  */
 
 #include <linux/types.h>
+#include <linux/string.h>
 #include <linux/slab.h>
 #include <linux/module.h>
 #include <linux/capability.h>
@@ -28,6 +29,7 @@
 #include <linux/in.h>
 #include <linux/net.h>
 #include <linux/mm.h>
+#include <linux/un.h>
 #include <linux/udp.h>
 #include <linux/tcp.h>
 #include <linux/sunrpc/clnt.h>
@@ -35,7 +37,7 @@
 #include <linux/sunrpc/svcsock.h>
 #include <linux/sunrpc/xprtsock.h>
 #include <linux/file.h>
-#ifdef CONFIG_NFS_V4_1
+#ifdef CONFIG_SUNRPC_BACKCHANNEL
 #include <linux/sunrpc/bc_xprt.h>
 #endif
 
 #include <net/tcp.h>
 
 #include "sunrpc.h"
+
+static void xs_close(struct rpc_xprt *xprt);
+
 /*
  * xprtsock tunables
  */
 unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
-unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;
+unsigned int xprt_tcp_slot_table_entries = RPC_MIN_SLOT_TABLE;
+unsigned int xprt_max_tcp_slot_table_entries = RPC_MAX_SLOT_TABLE;
 
 unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
 unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
@@ -70,6 +76,7 @@ static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
 
 static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
 static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
+static unsigned int max_tcp_slot_table_limit = RPC_MAX_SLOT_TABLE_LIMIT;
 static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
 static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
 
@@ -99,6 +106,15 @@ static ctl_table xs_tunables_table[] = {
                .extra2         = &max_slot_table_size
        },
        {
+               .procname       = "tcp_max_slot_table_entries",
+               .data           = &xprt_max_tcp_slot_table_entries,
+               .maxlen         = sizeof(unsigned int),
+               .mode           = 0644,
+               .proc_handler   = proc_dointvec_minmax,
+               .extra1         = &min_slot_table_size,
+               .extra2         = &max_tcp_slot_table_limit
+       },
+       {
                .procname       = "min_resvport",
                .data           = &xprt_min_resvport,
                .maxlen         = sizeof(unsigned int),
@@ -138,20 +154,6 @@ static ctl_table sunrpc_table[] = {
 #endif
 
 /*
- * Time out for an RPC UDP socket connect.  UDP socket connects are
- * synchronous, but we set a timeout anyway in case of resource
- * exhaustion on the local host.
- */
-#define XS_UDP_CONN_TO         (5U * HZ)
-
-/*
- * Wait duration for an RPC TCP connection to be established.  Solaris
- * NFS over TCP uses 60 seconds, for example, which is in line with how
- * long a server takes to reboot.
- */
-#define XS_TCP_CONN_TO         (60U * HZ)
-
-/*
  * Wait duration for a reply from the RPC portmapper.
  */
 #define XS_BIND_TO             (60U * HZ)
@@ -224,7 +226,8 @@ struct sock_xprt {
         * State of TCP reply receive
         */
        __be32                  tcp_fraghdr,
-                               tcp_xid;
+                               tcp_xid,
+                               tcp_calldir;
 
        u32                     tcp_offset,
                                tcp_reclen;
@@ -274,6 +277,11 @@ static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
        return (struct sockaddr *) &xprt->addr;
 }
 
+static inline struct sockaddr_un *xs_addr_un(struct rpc_xprt *xprt)
+{
+       return (struct sockaddr_un *) &xprt->addr;
+}
+
 static inline struct sockaddr_in *xs_addr_in(struct rpc_xprt *xprt)
 {
        return (struct sockaddr_in *) &xprt->addr;
@@ -289,24 +297,34 @@ static void xs_format_common_peer_addresses(struct rpc_xprt *xprt)
        struct sockaddr *sap = xs_addr(xprt);
        struct sockaddr_in6 *sin6;
        struct sockaddr_in *sin;
+       struct sockaddr_un *sun;
        char buf[128];
 
-       (void)rpc_ntop(sap, buf, sizeof(buf));
-       xprt->address_strings[RPC_DISPLAY_ADDR] = kstrdup(buf, GFP_KERNEL);
-
        switch (sap->sa_family) {
+       case AF_LOCAL:
+               sun = xs_addr_un(xprt);
+               strlcpy(buf, sun->sun_path, sizeof(buf));
+               xprt->address_strings[RPC_DISPLAY_ADDR] =
+                                               kstrdup(buf, GFP_KERNEL);
+               break;
        case AF_INET:
+               (void)rpc_ntop(sap, buf, sizeof(buf));
+               xprt->address_strings[RPC_DISPLAY_ADDR] =
+                                               kstrdup(buf, GFP_KERNEL);
                sin = xs_addr_in(xprt);
-               (void)snprintf(buf, sizeof(buf), "%02x%02x%02x%02x",
-                                       NIPQUAD(sin->sin_addr.s_addr));
+               snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
                break;
        case AF_INET6:
+               (void)rpc_ntop(sap, buf, sizeof(buf));
+               xprt->address_strings[RPC_DISPLAY_ADDR] =
+                                               kstrdup(buf, GFP_KERNEL);
                sin6 = xs_addr_in6(xprt);
-               (void)snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
+               snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
                break;
        default:
                BUG();
        }
+
        xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
 }
 
@@ -315,10 +333,10 @@ static void xs_format_common_peer_ports(struct rpc_xprt *xprt)
        struct sockaddr *sap = xs_addr(xprt);
        char buf[128];
 
-       (void)snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
+       snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
        xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
 
-       (void)snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
+       snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
        xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
 }
 
@@ -478,7 +496,7 @@ static int xs_nospace(struct rpc_task *task)
        struct rpc_rqst *req = task->tk_rqstp;
        struct rpc_xprt *xprt = req->rq_xprt;
        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
-       int ret = 0;
+       int ret = -EAGAIN;
 
        dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
                        task->tk_pid, req->rq_slen - req->rq_bytes_sent,
@@ -490,7 +508,6 @@ static int xs_nospace(struct rpc_task *task)
        /* Don't race with disconnect */
        if (xprt_connected(xprt)) {
                if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
-                       ret = -EAGAIN;
                        /*
                         * Notify TCP that we're limited by the application
                         * window size
@@ -509,6 +526,70 @@ static int xs_nospace(struct rpc_task *task)
        return ret;
 }
 
+/*
+ * Construct a stream transport record marker in @buf.
+ */
+static inline void xs_encode_stream_record_marker(struct xdr_buf *buf)
+{
+       u32 reclen = buf->len - sizeof(rpc_fraghdr);
+       rpc_fraghdr *base = buf->head[0].iov_base;
+       *base = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT | reclen);
+}
+
+/**
+ * xs_local_send_request - write an RPC request to an AF_LOCAL socket
+ * @task: RPC task that manages the state of an RPC request
+ *
+ * Return values:
+ *        0:   The request has been sent
+ *   EAGAIN:   The socket was blocked, please call again later to
+ *             complete the request
+ * ENOTCONN:   Caller needs to invoke connect logic then call again
+ *    other:   Some other error occured, the request was not sent
+ */
+static int xs_local_send_request(struct rpc_task *task)
+{
+       struct rpc_rqst *req = task->tk_rqstp;
+       struct rpc_xprt *xprt = req->rq_xprt;
+       struct sock_xprt *transport =
+                               container_of(xprt, struct sock_xprt, xprt);
+       struct xdr_buf *xdr = &req->rq_snd_buf;
+       int status;
+
+       xs_encode_stream_record_marker(&req->rq_snd_buf);
+
+       xs_pktdump("packet data:",
+                       req->rq_svec->iov_base, req->rq_svec->iov_len);
+
+       status = xs_sendpages(transport->sock, NULL, 0,
+                                               xdr, req->rq_bytes_sent);
+       dprintk("RPC:       %s(%u) = %d\n",
+                       __func__, xdr->len - req->rq_bytes_sent, status);
+       if (likely(status >= 0)) {
+               req->rq_bytes_sent += status;
+               req->rq_xmit_bytes_sent += status;
+               if (likely(req->rq_bytes_sent >= req->rq_slen)) {
+                       req->rq_bytes_sent = 0;
+                       return 0;
+               }
+               status = -EAGAIN;
+       }
+
+       switch (status) {
+       case -EAGAIN:
+               status = xs_nospace(task);
+               break;
+       default:
+               dprintk("RPC:       sendmsg returned unrecognized error %d\n",
+                       -status);
+       case -EPIPE:
+               xs_close(xprt);
+               status = -ENOTCONN;
+       }
+
+       return status;
+}
+
 /**
  * xs_udp_send_request - write an RPC request to a UDP socket
  * @task: address of RPC task that manages the state of an RPC request
@@ -518,7 +599,7 @@ static int xs_nospace(struct rpc_task *task)
  *   EAGAIN:   The socket was blocked, please call again later to
  *             complete the request
  * ENOTCONN:   Caller needs to invoke connect logic then call again
- *    other:   Some other error occured, the request was not sent
+ *    other:   Some other error occurred, the request was not sent
  */
 static int xs_udp_send_request(struct rpc_task *task)
 {
@@ -543,14 +624,12 @@ static int xs_udp_send_request(struct rpc_task *task)
                        xdr->len - req->rq_bytes_sent, status);
 
        if (status >= 0) {
-               task->tk_bytes_sent += status;
+               req->rq_xmit_bytes_sent += status;
                if (status >= req->rq_slen)
                        return 0;
                /* Still some bytes left; set up for a retry later. */
                status = -EAGAIN;
        }
-       if (!transport->sock)
-               goto out;
 
        switch (status) {
        case -ENOTSOCK:
@@ -570,7 +649,7 @@ static int xs_udp_send_request(struct rpc_task *task)
                 * prompts ECONNREFUSED. */
                clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
        }
-out:
+
        return status;
 }
 
@@ -590,13 +669,6 @@ static void xs_tcp_shutdown(struct rpc_xprt *xprt)
                kernel_sock_shutdown(sock, SHUT_WR);
 }
 
-static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf)
-{
-       u32 reclen = buf->len - sizeof(rpc_fraghdr);
-       rpc_fraghdr *base = buf->head[0].iov_base;
-       *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen);
-}
-
 /**
  * xs_tcp_send_request - write an RPC request to a TCP socket
  * @task: address of RPC task that manages the state of an RPC request
@@ -606,7 +678,7 @@ static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf)
  *   EAGAIN:   The socket was blocked, please call again later to
  *             complete the request
  * ENOTCONN:   Caller needs to invoke connect logic then call again
- *    other:   Some other error occured, the request was not sent
+ *    other:   Some other error occurred, the request was not sent
  *
  * XXX: In the case of soft timeouts, should we eventually give up
  *     if sendmsg is not able to make progress?
@@ -619,7 +691,7 @@ static int xs_tcp_send_request(struct rpc_task *task)
        struct xdr_buf *xdr = &req->rq_snd_buf;
        int status;
 
-       xs_encode_tcp_record_marker(&req->rq_snd_buf);
+       xs_encode_stream_record_marker(&req->rq_snd_buf);
 
        xs_pktdump("packet data:",
                                req->rq_svec->iov_base,
@@ -641,7 +713,7 @@ static int xs_tcp_send_request(struct rpc_task *task)
                /* If we've sent the entire packet, immediately
                 * reset the count of bytes sent. */
                req->rq_bytes_sent += status;
-               task->tk_bytes_sent += status;
+               req->rq_xmit_bytes_sent += status;
                if (likely(req->rq_bytes_sent >= req->rq_slen)) {
                        req->rq_bytes_sent = 0;
                        return 0;
@@ -652,8 +724,6 @@ static int xs_tcp_send_request(struct rpc_task *task)
                status = -EAGAIN;
                break;
        }
-       if (!transport->sock)
-               goto out;
 
        switch (status) {
        case -ENOTSOCK:
@@ -673,7 +743,7 @@ static int xs_tcp_send_request(struct rpc_task *task)
        case -ENOTCONN:
                clear_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags);
        }
-out:
+
        return status;
 }
 
@@ -695,6 +765,8 @@ static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
        if (task == NULL)
                goto out_release;
        req = task->tk_rqstp;
+       if (req == NULL)
+               goto out_release;
        if (req->rq_bytes_sent == 0)
                goto out_release;
        if (req->rq_bytes_sent == req->rq_snd_buf.len)
@@ -728,6 +800,8 @@ static void xs_reset_transport(struct sock_xprt *transport)
        if (sk == NULL)
                return;
 
+       transport->srcport = 0;
+
        write_lock_bh(&sk->sk_callback_lock);
        transport->inet = NULL;
        transport->sock = NULL;
@@ -788,12 +862,11 @@ static void xs_destroy(struct rpc_xprt *xprt)
 
        dprintk("RPC:       xs_destroy xprt %p\n", xprt);
 
-       cancel_rearming_delayed_work(&transport->connect_worker);
+       cancel_delayed_work_sync(&transport->connect_worker);
 
        xs_close(xprt);
        xs_free_peer_addresses(xprt);
-       kfree(xprt->slot);
-       kfree(xprt);
+       xprt_free(xprt);
        module_put(THIS_MODULE);
 }
 
@@ -802,6 +875,88 @@ static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
        return (struct rpc_xprt *) sk->sk_user_data;
 }
 
+static int xs_local_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb)
+{
+       struct xdr_skb_reader desc = {
+               .skb            = skb,
+               .offset         = sizeof(rpc_fraghdr),
+               .count          = skb->len - sizeof(rpc_fraghdr),
+       };
+
+       if (xdr_partial_copy_from_skb(xdr, 0, &desc, xdr_skb_read_bits) < 0)
+               return -1;
+       if (desc.count)
+               return -1;
+       return 0;
+}
+
+/**
+ * xs_local_data_ready - "data ready" callback for AF_LOCAL sockets
+ * @sk: socket with data to read
+ * @len: how much data to read
+ *
+ * Currently this assumes we can read the whole reply in a single gulp.
+ */
+static void xs_local_data_ready(struct sock *sk, int len)
+{
+       struct rpc_task *task;
+       struct rpc_xprt *xprt;
+       struct rpc_rqst *rovr;
+       struct sk_buff *skb;
+       int err, repsize, copied;
+       u32 _xid;
+       __be32 *xp;
+
+       read_lock_bh(&sk->sk_callback_lock);
+       dprintk("RPC:       %s...\n", __func__);
+       xprt = xprt_from_sock(sk);
+       if (xprt == NULL)
+               goto out;
+
+       skb = skb_recv_datagram(sk, 0, 1, &err);
+       if (skb == NULL)
+               goto out;
+
+       if (xprt->shutdown)
+               goto dropit;
+
+       repsize = skb->len - sizeof(rpc_fraghdr);
+       if (repsize < 4) {
+               dprintk("RPC:       impossible RPC reply size %d\n", repsize);
+               goto dropit;
+       }
+
+       /* Copy the XID from the skb... */
+       xp = skb_header_pointer(skb, sizeof(rpc_fraghdr), sizeof(_xid), &_xid);
+       if (xp == NULL)
+               goto dropit;
+
+       /* Look up and lock the request corresponding to the given XID */
+       spin_lock(&xprt->transport_lock);
+       rovr = xprt_lookup_rqst(xprt, *xp);
+       if (!rovr)
+               goto out_unlock;
+       task = rovr->rq_task;
+
+       copied = rovr->rq_private_buf.buflen;
+       if (copied > repsize)
+               copied = repsize;
+
+       if (xs_local_copy_to_xdr(&rovr->rq_private_buf, skb)) {
+               dprintk("RPC:       sk_buff copy failed\n");
+               goto out_unlock;
+       }
+
+       xprt_complete_rqst(task, copied);
+
+ out_unlock:
+       spin_unlock(&xprt->transport_lock);
+ dropit:
+       skb_free_datagram(sk, skb);
+ out:
+       read_unlock_bh(&sk->sk_callback_lock);
+}
+
 /**
  * xs_udp_data_ready - "data ready" callback for UDP sockets
  * @sk: socket with data to read
@@ -818,7 +973,7 @@ static void xs_udp_data_ready(struct sock *sk, int len)
        u32 _xid;
        __be32 *xp;
 
-       read_lock(&sk->sk_callback_lock);
+       read_lock_bh(&sk->sk_callback_lock);
        dprintk("RPC:       xs_udp_data_ready...\n");
        if (!(xprt = xprt_from_sock(sk)))
                goto out;
@@ -863,7 +1018,6 @@ static void xs_udp_data_ready(struct sock *sk, int len)
        dst_confirm(skb_dst(skb));
 
        xprt_adjust_cwnd(task, copied);
-       xprt_update_rtt(task);
        xprt_complete_rqst(task, copied);
 
  out_unlock:
@@ -871,7 +1025,7 @@ static void xs_udp_data_ready(struct sock *sk, int len)
  dropit:
        skb_free_datagram(sk, skb);
  out:
-       read_unlock(&sk->sk_callback_lock);
+       read_unlock_bh(&sk->sk_callback_lock);
 }
 
 static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
@@ -947,7 +1101,7 @@ static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
 {
        size_t len, used;
        u32 offset;
-       __be32  calldir;
+       char *p;
 
        /*
         * We want transport->tcp_offset to be 8 at the end of this routine
@@ -956,26 +1110,33 @@ static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
         * transport->tcp_offset is 4 (after having already read the xid).
         */
        offset = transport->tcp_offset - sizeof(transport->tcp_xid);
-       len = sizeof(calldir) - offset;
+       len = sizeof(transport->tcp_calldir) - offset;
        dprintk("RPC:       reading CALL/REPLY flag (%Zu bytes)\n", len);
-       used = xdr_skb_read_bits(desc, &calldir, len);
+       p = ((char *) &transport->tcp_calldir) + offset;
+       used = xdr_skb_read_bits(desc, p, len);
        transport->tcp_offset += used;
        if (used != len)
                return;
        transport->tcp_flags &= ~TCP_RCV_READ_CALLDIR;
-       transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
-       transport->tcp_flags |= TCP_RCV_COPY_DATA;
        /*
         * We don't yet have the XDR buffer, so we will write the calldir
         * out after we get the buffer from the 'struct rpc_rqst'
         */
-       if (ntohl(calldir) == RPC_REPLY)
+       switch (ntohl(transport->tcp_calldir)) {
+       case RPC_REPLY:
+               transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
+               transport->tcp_flags |= TCP_RCV_COPY_DATA;
                transport->tcp_flags |= TCP_RPC_REPLY;
-       else
+               break;
+       case RPC_CALL:
+               transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
+               transport->tcp_flags |= TCP_RCV_COPY_DATA;
                transport->tcp_flags &= ~TCP_RPC_REPLY;
-       dprintk("RPC:       reading %s CALL/REPLY flag %08x\n",
-                       (transport->tcp_flags & TCP_RPC_REPLY) ?
-                               "reply for" : "request with", calldir);
+               break;
+       default:
+               dprintk("RPC:       invalid request message type\n");
+               xprt_force_disconnect(&transport->xprt);
+       }
        xs_tcp_check_fraghdr(transport);
 }
 
@@ -995,12 +1156,10 @@ static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
                /*
                 * Save the RPC direction in the XDR buffer
                 */
-               __be32  calldir = transport->tcp_flags & TCP_RPC_REPLY ?
-                                       htonl(RPC_REPLY) : 0;
-
                memcpy(rcvbuf->head[0].iov_base + transport->tcp_copied,
-                       &calldir, sizeof(calldir));
-               transport->tcp_copied += sizeof(calldir);
+                       &transport->tcp_calldir,
+                       sizeof(transport->tcp_calldir));
+               transport->tcp_copied += sizeof(transport->tcp_calldir);
                transport->tcp_flags &= ~TCP_RCV_COPY_CALLDIR;
        }
 
@@ -1055,8 +1214,6 @@ static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
                if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
                        transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
        }
-
-       return;
 }
 
 /*
@@ -1091,7 +1248,7 @@ static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
        return 0;
 }
 
-#if defined(CONFIG_NFS_V4_1)
+#if defined(CONFIG_SUNRPC_BACKCHANNEL)
 /*
  * Obtains an rpc_rqst previously allocated and invokes the common
  * tcp read code to read the data.  The result is placed in the callback
@@ -1154,7 +1311,7 @@ static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
 {
        return xs_tcp_read_reply(xprt, desc);
 }
-#endif /* CONFIG_NFS_V4_1 */
+#endif /* CONFIG_SUNRPC_BACKCHANNEL */
 
 /*
  * Read data off the transport.  This can be either an RPC_CALL or an
@@ -1245,7 +1402,7 @@ static void xs_tcp_data_ready(struct sock *sk, int bytes)
 
        dprintk("RPC:       xs_tcp_data_ready...\n");
 
-       read_lock(&sk->sk_callback_lock);
+       read_lock_bh(&sk->sk_callback_lock);
        if (!(xprt = xprt_from_sock(sk)))
                goto out;
        if (xprt->shutdown)
@@ -1264,7 +1421,7 @@ static void xs_tcp_data_ready(struct sock *sk, int bytes)
                read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
        } while (read > 0);
 out:
-       read_unlock(&sk->sk_callback_lock);
+       read_unlock_bh(&sk->sk_callback_lock);
 }
 
 /*
@@ -1317,18 +1474,19 @@ static void xs_tcp_state_change(struct sock *sk)
 {
        struct rpc_xprt *xprt;
 
-       read_lock(&sk->sk_callback_lock);
+       read_lock_bh(&sk->sk_callback_lock);
        if (!(xprt = xprt_from_sock(sk)))
                goto out;
        dprintk("RPC:       xs_tcp_state_change client %p...\n", xprt);
-       dprintk("RPC:       state %x conn %d dead %d zapped %d\n",
+       dprintk("RPC:       state %x conn %d dead %d zapped %d sk_shutdown %d\n",
                        sk->sk_state, xprt_connected(xprt),
                        sock_flag(sk, SOCK_DEAD),
-                       sock_flag(sk, SOCK_ZAPPED));
+                       sock_flag(sk, SOCK_ZAPPED),
+                       sk->sk_shutdown);
 
        switch (sk->sk_state) {
        case TCP_ESTABLISHED:
-               spin_lock_bh(&xprt->transport_lock);
+               spin_lock(&xprt->transport_lock);
                if (!xprt_test_and_set_connected(xprt)) {
                        struct sock_xprt *transport = container_of(xprt,
                                        struct sock_xprt, xprt);
@@ -1342,7 +1500,7 @@ static void xs_tcp_state_change(struct sock *sk)
 
                        xprt_wake_pending_tasks(xprt, -EAGAIN);
                }
-               spin_unlock_bh(&xprt->transport_lock);
+               spin_unlock(&xprt->transport_lock);
                break;
        case TCP_FIN_WAIT1:
                /* The client initiated a shutdown of the socket */
@@ -1358,7 +1516,6 @@ static void xs_tcp_state_change(struct sock *sk)
        case TCP_CLOSE_WAIT:
                /* The server initiated a shutdown of the socket */
                xprt_force_disconnect(xprt);
-       case TCP_SYN_SENT:
                xprt->connect_cookie++;
        case TCP_CLOSING:
                /*
@@ -1380,7 +1537,7 @@ static void xs_tcp_state_change(struct sock *sk)
                xs_sock_mark_closed(xprt);
        }
  out:
-       read_unlock(&sk->sk_callback_lock);
+       read_unlock_bh(&sk->sk_callback_lock);
 }
 
 /**
@@ -1391,7 +1548,7 @@ static void xs_error_report(struct sock *sk)
 {
        struct rpc_xprt *xprt;
 
-       read_lock(&sk->sk_callback_lock);
+       read_lock_bh(&sk->sk_callback_lock);
        if (!(xprt = xprt_from_sock(sk)))
                goto out;
        dprintk("RPC:       %s client %p...\n"
@@ -1399,7 +1556,7 @@ static void xs_error_report(struct sock *sk)
                        __func__, xprt, sk->sk_err);
        xprt_wake_pending_tasks(xprt, -EAGAIN);
 out:
-       read_unlock(&sk->sk_callback_lock);
+       read_unlock_bh(&sk->sk_callback_lock);
 }
 
 static void xs_write_space(struct sock *sk)
@@ -1431,13 +1588,13 @@ static void xs_write_space(struct sock *sk)
  */
 static void xs_udp_write_space(struct sock *sk)
 {
-       read_lock(&sk->sk_callback_lock);
+       read_lock_bh(&sk->sk_callback_lock);
 
        /* from net/core/sock.c:sock_def_write_space */
        if (sock_writeable(sk))
                xs_write_space(sk);
 
-       read_unlock(&sk->sk_callback_lock);
+       read_unlock_bh(&sk->sk_callback_lock);
 }
 
 /**
@@ -1452,13 +1609,13 @@ static void xs_udp_write_space(struct sock *sk)
  */
 static void xs_tcp_write_space(struct sock *sk)
 {
-       read_lock(&sk->sk_callback_lock);
+       read_lock_bh(&sk->sk_callback_lock);
 
        /* from net/core/stream.c:sk_stream_write_space */
        if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk))
                xs_write_space(sk);
 
-       read_unlock(&sk->sk_callback_lock);
+       read_unlock_bh(&sk->sk_callback_lock);
 }
 
 static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
@@ -1531,7 +1688,7 @@ static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
        xs_update_peer_port(xprt);
 }
 
-static unsigned short xs_get_srcport(struct sock_xprt *transport, struct socket *sock)
+static unsigned short xs_get_srcport(struct sock_xprt *transport)
 {
        unsigned short port = transport->srcport;
 
@@ -1540,7 +1697,7 @@ static unsigned short xs_get_srcport(struct sock_xprt *transport, struct socket
        return port;
 }
 
-static unsigned short xs_next_srcport(struct sock_xprt *transport, struct socket *sock, unsigned short port)
+static unsigned short xs_next_srcport(struct sock_xprt *transport, unsigned short port)
 {
        if (transport->srcport != 0)
                transport->srcport = 0;
@@ -1550,23 +1707,18 @@ static unsigned short xs_next_srcport(struct sock_xprt *transport, struct socket
                return xprt_max_resvport;
        return --port;
 }
-
-static int xs_bind4(struct sock_xprt *transport, struct socket *sock)
+static int xs_bind(struct sock_xprt *transport, struct socket *sock)
 {
-       struct sockaddr_in myaddr = {
-               .sin_family = AF_INET,
-       };
-       struct sockaddr_in *sa;
+       struct sockaddr_storage myaddr;
        int err, nloop = 0;
-       unsigned short port = xs_get_srcport(transport, sock);
+       unsigned short port = xs_get_srcport(transport);
        unsigned short last;
 
-       sa = (struct sockaddr_in *)&transport->srcaddr;
-       myaddr.sin_addr = sa->sin_addr;
+       memcpy(&myaddr, &transport->srcaddr, transport->xprt.addrlen);
        do {
-               myaddr.sin_port = htons(port);
-               err = kernel_bind(sock, (struct sockaddr *) &myaddr,
-                                               sizeof(myaddr));
+               rpc_set_port((struct sockaddr *)&myaddr, port);
+               err = kernel_bind(sock, (struct sockaddr *)&myaddr,
+                               transport->xprt.addrlen);
                if (port == 0)
                        break;
                if (err == 0) {
@@ -1574,52 +1726,47 @@ static int xs_bind4(struct sock_xprt *transport, struct socket *sock)
                        break;
                }
                last = port;
-               port = xs_next_srcport(transport, sock, port);
+               port = xs_next_srcport(transport, port);
                if (port > last)
                        nloop++;
        } while (err == -EADDRINUSE && nloop != 2);
-       dprintk("RPC:       %s %pI4:%u: %s (%d)\n",
-                       __func__, &myaddr.sin_addr,
-                       port, err ? "failed" : "ok", err);
+
+       if (myaddr.ss_family == AF_INET)
+               dprintk("RPC:       %s %pI4:%u: %s (%d)\n", __func__,
+                               &((struct sockaddr_in *)&myaddr)->sin_addr,
+                               port, err ? "failed" : "ok", err);
+       else
+               dprintk("RPC:       %s %pI6:%u: %s (%d)\n", __func__,
+                               &((struct sockaddr_in6 *)&myaddr)->sin6_addr,
+                               port, err ? "failed" : "ok", err);
        return err;
 }
 
-static int xs_bind6(struct sock_xprt *transport, struct socket *sock)
+/*
+ * We don't support autobind on AF_LOCAL sockets
+ */
+static void xs_local_rpcbind(struct rpc_task *task)
 {
-       struct sockaddr_in6 myaddr = {
-               .sin6_family = AF_INET6,
-       };
-       struct sockaddr_in6 *sa;
-       int err, nloop = 0;
-       unsigned short port = xs_get_srcport(transport, sock);
-       unsigned short last;
+       xprt_set_bound(task->tk_xprt);
+}
 
-       sa = (struct sockaddr_in6 *)&transport->srcaddr;
-       myaddr.sin6_addr = sa->sin6_addr;
-       do {
-               myaddr.sin6_port = htons(port);
-               err = kernel_bind(sock, (struct sockaddr *) &myaddr,
-                                               sizeof(myaddr));
-               if (port == 0)
-                       break;
-               if (err == 0) {
-                       transport->srcport = port;
-                       break;
-               }
-               last = port;
-               port = xs_next_srcport(transport, sock, port);
-               if (port > last)
-                       nloop++;
-       } while (err == -EADDRINUSE && nloop != 2);
-       dprintk("RPC:       xs_bind6 %pI6:%u: %s (%d)\n",
-               &myaddr.sin6_addr, port, err ? "failed" : "ok", err);
-       return err;
+static void xs_local_set_port(struct rpc_xprt *xprt, unsigned short port)
+{
 }
 
 #ifdef CONFIG_DEBUG_LOCK_ALLOC
 static struct lock_class_key xs_key[2];
 static struct lock_class_key xs_slock_key[2];
 
+static inline void xs_reclassify_socketu(struct socket *sock)
+{
+       struct sock *sk = sock->sk;
+
+       BUG_ON(sock_owned_by_user(sk));
+       sock_lock_init_class_and_name(sk, "slock-AF_LOCAL-RPC",
+               &xs_slock_key[1], "sk_lock-AF_LOCAL-RPC", &xs_key[1]);
+}
+
 static inline void xs_reclassify_socket4(struct socket *sock)
 {
        struct sock *sk = sock->sk;
@@ -1637,7 +1784,26 @@ static inline void xs_reclassify_socket6(struct socket *sock)
        sock_lock_init_class_and_name(sk, "slock-AF_INET6-RPC",
                &xs_slock_key[1], "sk_lock-AF_INET6-RPC", &xs_key[1]);
 }
+
+static inline void xs_reclassify_socket(int family, struct socket *sock)
+{
+       switch (family) {
+       case AF_LOCAL:
+               xs_reclassify_socketu(sock);
+               break;
+       case AF_INET:
+               xs_reclassify_socket4(sock);
+               break;
+       case AF_INET6:
+               xs_reclassify_socket6(sock);
+               break;
+       }
+}
 #else
+static inline void xs_reclassify_socketu(struct socket *sock)
+{
+}
+
 static inline void xs_reclassify_socket4(struct socket *sock)
 {
 }
@@ -1645,11 +1811,42 @@ static inline void xs_reclassify_socket4(struct socket *sock)
 static inline void xs_reclassify_socket6(struct socket *sock)
 {
 }
+
+static inline void xs_reclassify_socket(int family, struct socket *sock)
+{
+}
 #endif
 
-static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
+static struct socket *xs_create_sock(struct rpc_xprt *xprt,
+               struct sock_xprt *transport, int family, int type, int protocol)
 {
-       struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
+       struct socket *sock;
+       int err;
+
+       err = __sock_create(xprt->xprt_net, family, type, protocol, &sock, 1);
+       if (err < 0) {
+               dprintk("RPC:       can't create %d transport socket (%d).\n",
+                               protocol, -err);
+               goto out;
+       }
+       xs_reclassify_socket(family, sock);
+
+       err = xs_bind(transport, sock);
+       if (err) {
+               sock_release(sock);
+               goto out;
+       }
+
+       return sock;
+out:
+       return ERR_PTR(err);
+}
+
+static int xs_local_finish_connecting(struct rpc_xprt *xprt,
+                                     struct socket *sock)
+{
+       struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
+                                                                       xprt);
 
        if (!transport->inet) {
                struct sock *sk = sock->sk;
@@ -1659,13 +1856,12 @@ static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
                xs_save_old_callbacks(transport, sk);
 
                sk->sk_user_data = xprt;
-               sk->sk_data_ready = xs_udp_data_ready;
+               sk->sk_data_ready = xs_local_data_ready;
                sk->sk_write_space = xs_udp_write_space;
                sk->sk_error_report = xs_error_report;
-               sk->sk_no_check = UDP_CSUM_NORCV;
                sk->sk_allocation = GFP_ATOMIC;
 
-               xprt_set_connected(xprt);
+               xprt_clear_connected(xprt);
 
                /* Reset to new socket */
                transport->sock = sock;
@@ -1673,85 +1869,113 @@ static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
 
                write_unlock_bh(&sk->sk_callback_lock);
        }
-       xs_udp_do_set_buffer_size(xprt);
+
+       /* Tell the socket layer to start connecting... */
+       xprt->stat.connect_count++;
+       xprt->stat.connect_start = jiffies;
+       return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, 0);
 }
 
 /**
- * xs_udp_connect_worker4 - set up a UDP socket
- * @work: RPC transport to connect
+ * xs_local_setup_socket - create AF_LOCAL socket, connect to a local endpoint
+ * @xprt: RPC transport to connect
+ * @transport: socket transport to connect
+ * @create_sock: function to create a socket of the correct type
  *
  * Invoked by a work queue tasklet.
  */
-static void xs_udp_connect_worker4(struct work_struct *work)
+static void xs_local_setup_socket(struct work_struct *work)
 {
        struct sock_xprt *transport =
                container_of(work, struct sock_xprt, connect_worker.work);
        struct rpc_xprt *xprt = &transport->xprt;
-       struct socket *sock = transport->sock;
-       int err, status = -EIO;
+       struct socket *sock;
+       int status = -EIO;
 
        if (xprt->shutdown)
                goto out;
 
-       /* Start by resetting any existing state */
-       xs_reset_transport(transport);
-
-       err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock);
-       if (err < 0) {
-               dprintk("RPC:       can't create UDP transport socket (%d).\n", -err);
+       clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
+       status = __sock_create(xprt->xprt_net, AF_LOCAL,
+                                       SOCK_STREAM, 0, &sock, 1);
+       if (status < 0) {
+               dprintk("RPC:       can't create AF_LOCAL "
+                       "transport socket (%d).\n", -status);
                goto out;
        }
-       xs_reclassify_socket4(sock);
+       xs_reclassify_socketu(sock);
 
-       if (xs_bind4(transport, sock)) {
-               sock_release(sock);
-               goto out;
-       }
+       dprintk("RPC:       worker connecting xprt %p via AF_LOCAL to %s\n",
+                       xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
 
-       dprintk("RPC:       worker connecting xprt %p via %s to "
-                               "%s (port %s)\n", xprt,
-                       xprt->address_strings[RPC_DISPLAY_PROTO],
-                       xprt->address_strings[RPC_DISPLAY_ADDR],
-                       xprt->address_strings[RPC_DISPLAY_PORT]);
+       status = xs_local_finish_connecting(xprt, sock);
+       switch (status) {
+       case 0:
+               dprintk("RPC:       xprt %p connected to %s\n",
+                               xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
+               xprt_set_connected(xprt);
+               break;
+       case -ENOENT:
+               dprintk("RPC:       xprt %p: socket %s does not exist\n",
+                               xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
+               break;
+       default:
+               printk(KERN_ERR "%s: unhandled error (%d) connecting to %s\n",
+                               __func__, -status,
+                               xprt->address_strings[RPC_DISPLAY_ADDR]);
+       }
 
-       xs_udp_finish_connecting(xprt, sock);
-       status = 0;
 out:
        xprt_clear_connecting(xprt);
        xprt_wake_pending_tasks(xprt, status);
 }
 
-/**
- * xs_udp_connect_worker6 - set up a UDP socket
- * @work: RPC transport to connect
- *
- * Invoked by a work queue tasklet.
- */
-static void xs_udp_connect_worker6(struct work_struct *work)
+static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
+{
+       struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
+
+       if (!transport->inet) {
+               struct sock *sk = sock->sk;
+
+               write_lock_bh(&sk->sk_callback_lock);
+
+               xs_save_old_callbacks(transport, sk);
+
+               sk->sk_user_data = xprt;
+               sk->sk_data_ready = xs_udp_data_ready;
+               sk->sk_write_space = xs_udp_write_space;
+               sk->sk_error_report = xs_error_report;
+               sk->sk_no_check = UDP_CSUM_NORCV;
+               sk->sk_allocation = GFP_ATOMIC;
+
+               xprt_set_connected(xprt);
+
+               /* Reset to new socket */
+               transport->sock = sock;
+               transport->inet = sk;
+
+               write_unlock_bh(&sk->sk_callback_lock);
+       }
+       xs_udp_do_set_buffer_size(xprt);
+}
+
+static void xs_udp_setup_socket(struct work_struct *work)
 {
        struct sock_xprt *transport =
                container_of(work, struct sock_xprt, connect_worker.work);
        struct rpc_xprt *xprt = &transport->xprt;
        struct socket *sock = transport->sock;
-       int err, status = -EIO;
+       int status = -EIO;
 
        if (xprt->shutdown)
                goto out;
 
        /* Start by resetting any existing state */
        xs_reset_transport(transport);
-
-       err = sock_create_kern(PF_INET6, SOCK_DGRAM, IPPROTO_UDP, &sock);
-       if (err < 0) {
-               dprintk("RPC:       can't create UDP transport socket (%d).\n", -err);
+       sock = xs_create_sock(xprt, transport,
+                       xs_addr(xprt)->sa_family, SOCK_DGRAM, IPPROTO_UDP);
+       if (IS_ERR(sock))
                goto out;
-       }
-       xs_reclassify_socket6(sock);
-
-       if (xs_bind6(transport, sock) < 0) {
-               sock_release(sock);
-               goto out;
-       }
 
        dprintk("RPC:       worker connecting xprt %p via %s to "
                                "%s (port %s)\n", xprt,
@@ -1770,12 +1994,12 @@ out:
  * We need to preserve the port number so the reply cache on the server can
  * find our cached RPC replies when we get around to reconnecting.
  */
-static void xs_abort_connection(struct rpc_xprt *xprt, struct sock_xprt *transport)
+static void xs_abort_connection(struct sock_xprt *transport)
 {
        int result;
        struct sockaddr any;
 
-       dprintk("RPC:       disconnecting xprt %p to reuse port\n", xprt);
+       dprintk("RPC:       disconnecting xprt %p to reuse port\n", transport);
 
        /*
         * Disconnect the transport socket by doing a connect operation
@@ -1785,26 +2009,42 @@ static void xs_abort_connection(struct rpc_xprt *xprt, struct sock_xprt *transpo
        any.sa_family = AF_UNSPEC;
        result = kernel_connect(transport->sock, &any, sizeof(any), 0);
        if (!result)
-               xs_sock_mark_closed(xprt);
+               xs_sock_mark_closed(&transport->xprt);
        else
                dprintk("RPC:       AF_UNSPEC connect return code %d\n",
                                result);
 }
 
-static void xs_tcp_reuse_connection(struct rpc_xprt *xprt, struct sock_xprt *transport)
+static void xs_tcp_reuse_connection(struct sock_xprt *transport)
 {
        unsigned int state = transport->inet->sk_state;
 
-       if (state == TCP_CLOSE && transport->sock->state == SS_UNCONNECTED)
-               return;
-       if ((1 << state) & (TCPF_ESTABLISHED|TCPF_SYN_SENT))
-               return;
-       xs_abort_connection(xprt, transport);
+       if (state == TCP_CLOSE && transport->sock->state == SS_UNCONNECTED) {
+               /* we don't need to abort the connection if the socket
+                * hasn't undergone a shutdown
+                */
+               if (transport->inet->sk_shutdown == 0)
+                       return;
+               dprintk("RPC:       %s: TCP_CLOSEd and sk_shutdown set to %d\n",
+                               __func__, transport->inet->sk_shutdown);
+       }
+       if ((1 << state) & (TCPF_ESTABLISHED|TCPF_SYN_SENT)) {
+               /* we don't need to abort the connection if the socket
+                * hasn't undergone a shutdown
+                */
+               if (transport->inet->sk_shutdown == 0)
+                       return;
+               dprintk("RPC:       %s: ESTABLISHED/SYN_SENT "
+                               "sk_shutdown set to %d\n",
+                               __func__, transport->inet->sk_shutdown);
+       }
+       xs_abort_connection(transport);
 }
 
 static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
 {
        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
+       int ret = -ENOTCONN;
 
        if (!transport->inet) {
                struct sock *sk = sock->sk;
@@ -1836,12 +2076,22 @@ static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
        }
 
        if (!xprt_bound(xprt))
-               return -ENOTCONN;
+               goto out;
 
        /* Tell the socket layer to start connecting... */
        xprt->stat.connect_count++;
        xprt->stat.connect_start = jiffies;
-       return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
+       ret = kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
+       switch (ret) {
+       case 0:
+       case -EINPROGRESS:
+               /* SYN_SENT! */
+               xprt->connect_cookie++;
+               if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
+                       xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
+       }
+out:
+       return ret;
 }
 
 /**
@@ -1852,12 +2102,12 @@ static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
  *
  * Invoked by a work queue tasklet.
  */
-static void xs_tcp_setup_socket(struct rpc_xprt *xprt,
-               struct sock_xprt *transport,
-               struct socket *(*create_sock)(struct rpc_xprt *,
-                       struct sock_xprt *))
+static void xs_tcp_setup_socket(struct work_struct *work)
 {
+       struct sock_xprt *transport =
+               container_of(work, struct sock_xprt, connect_worker.work);
        struct socket *sock = transport->sock;
+       struct rpc_xprt *xprt = &transport->xprt;
        int status = -EIO;
 
        if (xprt->shutdown)
@@ -1865,7 +2115,8 @@ static void xs_tcp_setup_socket(struct rpc_xprt *xprt,
 
        if (!sock) {
                clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
-               sock = create_sock(xprt, transport);
+               sock = xs_create_sock(xprt, transport,
+                               xs_addr(xprt)->sa_family, SOCK_STREAM, IPPROTO_TCP);
                if (IS_ERR(sock)) {
                        status = PTR_ERR(sock);
                        goto out;
@@ -1876,7 +2127,7 @@ static void xs_tcp_setup_socket(struct rpc_xprt *xprt,
                abort_and_exit = test_and_clear_bit(XPRT_CONNECTION_ABORT,
                                &xprt->state);
                /* "close" the socket, preserving the local port */
-               xs_tcp_reuse_connection(xprt, transport);
+               xs_tcp_reuse_connection(transport);
 
                if (abort_and_exit)
                        goto out_eagain;
@@ -1912,6 +2163,11 @@ static void xs_tcp_setup_socket(struct rpc_xprt *xprt,
        case -EALREADY:
                xprt_clear_connecting(xprt);
                return;
+       case -EINVAL:
+               /* Happens, for instance, if the user specified a link
+                * local IPv6 address without a scope-id.
+                */
+               goto out;
        }
 out_eagain:
        status = -EAGAIN;
@@ -1920,84 +2176,6 @@ out:
        xprt_wake_pending_tasks(xprt, status);
 }
 
-static struct socket *xs_create_tcp_sock4(struct rpc_xprt *xprt,
-               struct sock_xprt *transport)
-{
-       struct socket *sock;
-       int err;
-
-       /* start from scratch */
-       err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
-       if (err < 0) {
-               dprintk("RPC:       can't create TCP transport socket (%d).\n",
-                               -err);
-               goto out_err;
-       }
-       xs_reclassify_socket4(sock);
-
-       if (xs_bind4(transport, sock) < 0) {
-               sock_release(sock);
-               goto out_err;
-       }
-       return sock;
-out_err:
-       return ERR_PTR(-EIO);
-}
-
-/**
- * xs_tcp_connect_worker4 - connect a TCP socket to a remote endpoint
- * @work: RPC transport to connect
- *
- * Invoked by a work queue tasklet.
- */
-static void xs_tcp_connect_worker4(struct work_struct *work)
-{
-       struct sock_xprt *transport =
-               container_of(work, struct sock_xprt, connect_worker.work);
-       struct rpc_xprt *xprt = &transport->xprt;
-
-       xs_tcp_setup_socket(xprt, transport, xs_create_tcp_sock4);
-}
-
-static struct socket *xs_create_tcp_sock6(struct rpc_xprt *xprt,
-               struct sock_xprt *transport)
-{
-       struct socket *sock;
-       int err;
-
-       /* start from scratch */
-       err = sock_create_kern(PF_INET6, SOCK_STREAM, IPPROTO_TCP, &sock);
-       if (err < 0) {
-               dprintk("RPC:       can't create TCP transport socket (%d).\n",
-                               -err);
-               goto out_err;
-       }
-       xs_reclassify_socket6(sock);
-
-       if (xs_bind6(transport, sock) < 0) {
-               sock_release(sock);
-               goto out_err;
-       }
-       return sock;
-out_err:
-       return ERR_PTR(-EIO);
-}
-
-/**
- * xs_tcp_connect_worker6 - connect a TCP socket to a remote endpoint
- * @work: RPC transport to connect
- *
- * Invoked by a work queue tasklet.
- */
-static void xs_tcp_connect_worker6(struct work_struct *work)
-{
-       struct sock_xprt *transport =
-               container_of(work, struct sock_xprt, connect_worker.work);
-       struct rpc_xprt *xprt = &transport->xprt;
-
-       xs_tcp_setup_socket(xprt, transport, xs_create_tcp_sock6);
-}
-
 /**
  * xs_connect - connect a socket to a remote endpoint
  * @task: address of RPC task that manages state of connect request
@@ -2016,10 +2194,7 @@ static void xs_connect(struct rpc_task *task)
        struct rpc_xprt *xprt = task->tk_xprt;
        struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
 
-       if (xprt_test_and_set_connecting(xprt))
-               return;
-
-       if (transport->sock != NULL) {
+       if (transport->sock != NULL && !RPC_IS_SOFTCONN(task)) {
                dprintk("RPC:       xs_connect delayed xprt %p for %lu "
                                "seconds\n",
                                xprt, xprt->reestablish_timeout / HZ);
@@ -2038,14 +2213,30 @@ static void xs_connect(struct rpc_task *task)
        }
 }
 
-static void xs_tcp_connect(struct rpc_task *task)
+/**
+ * xs_local_print_stats - display AF_LOCAL socket-specifc stats
+ * @xprt: rpc_xprt struct containing statistics
+ * @seq: output file
+ *
+ */
+static void xs_local_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
 {
-       struct rpc_xprt *xprt = task->tk_xprt;
+       long idle_time = 0;
 
-       /* Exit if we need to wait for socket shutdown to complete */
-       if (test_bit(XPRT_CLOSING, &xprt->state))
-               return;
-       xs_connect(task);
+       if (xprt_connected(xprt))
+               idle_time = (long)(jiffies - xprt->last_used) / HZ;
+
+       seq_printf(seq, "\txprt:\tlocal %lu %lu %lu %ld %lu %lu %lu "
+                       "%llu %llu\n",
+                       xprt->stat.bind_count,
+                       xprt->stat.connect_count,
+                       xprt->stat.connect_time,
+                       idle_time,
+                       xprt->stat.sends,
+                       xprt->stat.recvs,
+                       xprt->stat.bad_xids,
+                       xprt->stat.req_u,
+                       xprt->stat.bklog_u);
 }
 
 /**
@@ -2100,7 +2291,7 @@ static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
  * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
  * to use the server side send routines.
  */
-void *bc_malloc(struct rpc_task *task, size_t size)
+static void *bc_malloc(struct rpc_task *task, size_t size)
 {
        struct page *page;
        struct rpc_buffer *buf;
@@ -2120,7 +2311,7 @@ void *bc_malloc(struct rpc_task *task, size_t size)
 /*
  * Free the space allocated in the bc_alloc routine
  */
-void bc_free(void *buffer)
+static void bc_free(void *buffer)
 {
        struct rpc_buffer *buf;
 
@@ -2146,10 +2337,7 @@ static int bc_sendto(struct rpc_rqst *req)
        unsigned long headoff;
        unsigned long tailoff;
 
-       /*
-        * Set up the rpc header and record marker stuff
-        */
-       xs_encode_tcp_record_marker(xbufp);
+       xs_encode_stream_record_marker(xbufp);
 
        tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
        headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
@@ -2210,7 +2398,6 @@ static int bc_send_request(struct rpc_task *task)
 
 static void bc_close(struct rpc_xprt *xprt)
 {
-       return;
 }
 
 /*
@@ -2220,9 +2407,23 @@ static void bc_close(struct rpc_xprt *xprt)
 
 static void bc_destroy(struct rpc_xprt *xprt)
 {
-       return;
 }
 
+static struct rpc_xprt_ops xs_local_ops = {
+       .reserve_xprt           = xprt_reserve_xprt,
+       .release_xprt           = xs_tcp_release_xprt,
+       .rpcbind                = xs_local_rpcbind,
+       .set_port               = xs_local_set_port,
+       .connect                = xs_connect,
+       .buf_alloc              = rpc_malloc,
+       .buf_free               = rpc_free,
+       .send_request           = xs_local_send_request,
+       .set_retrans_timeout    = xprt_set_retrans_timeout_def,
+       .close                  = xs_close,
+       .destroy                = xs_destroy,
+       .print_stats            = xs_local_print_stats,
+};
+
 static struct rpc_xprt_ops xs_udp_ops = {
        .set_buffer_size        = xs_udp_set_buffer_size,
        .reserve_xprt           = xprt_reserve_xprt_cong,
@@ -2246,14 +2447,11 @@ static struct rpc_xprt_ops xs_tcp_ops = {
        .release_xprt           = xs_tcp_release_xprt,
        .rpcbind                = rpcb_getport_async,
        .set_port               = xs_set_port,
-       .connect                = xs_tcp_connect,
+       .connect                = xs_connect,
        .buf_alloc              = rpc_malloc,
        .buf_free               = rpc_free,
        .send_request           = xs_tcp_send_request,
        .set_retrans_timeout    = xprt_set_retrans_timeout_def,
-#if defined(CONFIG_NFS_V4_1)
-       .release_request        = bc_release_request,
-#endif /* CONFIG_NFS_V4_1 */
        .close                  = xs_tcp_close,
        .destroy                = xs_destroy,
        .print_stats            = xs_tcp_print_stats,
@@ -2275,8 +2473,36 @@ static struct rpc_xprt_ops bc_tcp_ops = {
        .print_stats            = xs_tcp_print_stats,
 };
 
+static int xs_init_anyaddr(const int family, struct sockaddr *sap)
+{
+       static const struct sockaddr_in sin = {
+               .sin_family             = AF_INET,
+               .sin_addr.s_addr        = htonl(INADDR_ANY),
+       };
+       static const struct sockaddr_in6 sin6 = {
+               .sin6_family            = AF_INET6,
+               .sin6_addr              = IN6ADDR_ANY_INIT,
+       };
+
+       switch (family) {
+       case AF_LOCAL:
+               break;
+       case AF_INET:
+               memcpy(sap, &sin, sizeof(sin));
+               break;
+       case AF_INET6:
+               memcpy(sap, &sin6, sizeof(sin6));
+               break;
+       default:
+               dprintk("RPC:       %s: Bad address family\n", __func__);
+               return -EAFNOSUPPORT;
+       }
+       return 0;
+}
+
 static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
-                                     unsigned int slot_table_size)
+                                     unsigned int slot_table_size,
+                                     unsigned int max_slot_table_size)
 {
        struct rpc_xprt *xprt;
        struct sock_xprt *new;
@@ -2286,31 +2512,95 @@ static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
                return ERR_PTR(-EBADF);
        }
 
-       new = kzalloc(sizeof(*new), GFP_KERNEL);
-       if (new == NULL) {
+       xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
+                       max_slot_table_size);
+       if (xprt == NULL) {
                dprintk("RPC:       xs_setup_xprt: couldn't allocate "
                                "rpc_xprt\n");
                return ERR_PTR(-ENOMEM);
        }
-       xprt = &new->xprt;
-
-       xprt->max_reqs = slot_table_size;
-       xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL);
-       if (xprt->slot == NULL) {
-               kfree(xprt);
-               dprintk("RPC:       xs_setup_xprt: couldn't allocate slot "
-                               "table\n");
-               return ERR_PTR(-ENOMEM);
-       }
 
+       new = container_of(xprt, struct sock_xprt, xprt);
        memcpy(&xprt->addr, args->dstaddr, args->addrlen);
        xprt->addrlen = args->addrlen;
        if (args->srcaddr)
                memcpy(&new->srcaddr, args->srcaddr, args->addrlen);
+       else {
+               int err;
+               err = xs_init_anyaddr(args->dstaddr->sa_family,
+                                       (struct sockaddr *)&new->srcaddr);
+               if (err != 0)
+                       return ERR_PTR(err);
+       }
 
        return xprt;
 }
 
+static const struct rpc_timeout xs_local_default_timeout = {
+       .to_initval = 10 * HZ,
+       .to_maxval = 10 * HZ,
+       .to_retries = 2,
+};
+
+/**
+ * xs_setup_local - Set up transport to use an AF_LOCAL socket
+ * @args: rpc transport creation arguments
+ *
+ * AF_LOCAL is a "tpi_cots_ord" transport, just like TCP
+ */
+static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
+{
+       struct sockaddr_un *sun = (struct sockaddr_un *)args->dstaddr;
+       struct sock_xprt *transport;
+       struct rpc_xprt *xprt;
+       struct rpc_xprt *ret;
+
+       xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
+                       xprt_max_tcp_slot_table_entries);
+       if (IS_ERR(xprt))
+               return xprt;
+       transport = container_of(xprt, struct sock_xprt, xprt);
+
+       xprt->prot = 0;
+       xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
+       xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
+
+       xprt->bind_timeout = XS_BIND_TO;
+       xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
+       xprt->idle_timeout = XS_IDLE_DISC_TO;
+
+       xprt->ops = &xs_local_ops;
+       xprt->timeout = &xs_local_default_timeout;
+
+       switch (sun->sun_family) {
+       case AF_LOCAL:
+               if (sun->sun_path[0] != '/') {
+                       dprintk("RPC:       bad AF_LOCAL address: %s\n",
+                                       sun->sun_path);
+                       ret = ERR_PTR(-EINVAL);
+                       goto out_err;
+               }
+               xprt_set_bound(xprt);
+               INIT_DELAYED_WORK(&transport->connect_worker,
+                                       xs_local_setup_socket);
+               xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
+               break;
+       default:
+               ret = ERR_PTR(-EAFNOSUPPORT);
+               goto out_err;
+       }
+
+       dprintk("RPC:       set up xprt to %s via AF_LOCAL\n",
+                       xprt->address_strings[RPC_DISPLAY_ADDR]);
+
+       if (try_module_get(THIS_MODULE))
+               return xprt;
+       ret = ERR_PTR(-EINVAL);
+out_err:
+       xprt_free(xprt);
+       return ret;
+}
+
 static const struct rpc_timeout xs_udp_default_timeout = {
        .to_initval = 5 * HZ,
        .to_maxval = 30 * HZ,
@@ -2328,8 +2618,10 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
        struct sockaddr *addr = args->dstaddr;
        struct rpc_xprt *xprt;
        struct sock_xprt *transport;
+       struct rpc_xprt *ret;
 
-       xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries);
+       xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
+                       xprt_udp_slot_table_entries);
        if (IS_ERR(xprt))
                return xprt;
        transport = container_of(xprt, struct sock_xprt, xprt);
@@ -2340,7 +2632,6 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
        xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
 
        xprt->bind_timeout = XS_BIND_TO;
-       xprt->connect_timeout = XS_UDP_CONN_TO;
        xprt->reestablish_timeout = XS_UDP_REEST_TO;
        xprt->idle_timeout = XS_IDLE_DISC_TO;
 
@@ -2354,7 +2645,7 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
                        xprt_set_bound(xprt);
 
                INIT_DELAYED_WORK(&transport->connect_worker,
-                                       xs_udp_connect_worker4);
+                                       xs_udp_setup_socket);
                xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
                break;
        case AF_INET6:
@@ -2362,12 +2653,12 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
                        xprt_set_bound(xprt);
 
                INIT_DELAYED_WORK(&transport->connect_worker,
-                                       xs_udp_connect_worker6);
+                                       xs_udp_setup_socket);
                xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
                break;
        default:
-               kfree(xprt);
-               return ERR_PTR(-EAFNOSUPPORT);
+               ret = ERR_PTR(-EAFNOSUPPORT);
+               goto out_err;
        }
 
        if (xprt_bound(xprt))
@@ -2382,10 +2673,10 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
 
        if (try_module_get(THIS_MODULE))
                return xprt;
-
-       kfree(xprt->slot);
-       kfree(xprt);
-       return ERR_PTR(-EINVAL);
+       ret = ERR_PTR(-EINVAL);
+out_err:
+       xprt_free(xprt);
+       return ret;
 }
 
 static const struct rpc_timeout xs_tcp_default_timeout = {
@@ -2404,8 +2695,10 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
        struct sockaddr *addr = args->dstaddr;
        struct rpc_xprt *xprt;
        struct sock_xprt *transport;
+       struct rpc_xprt *ret;
 
-       xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
+       xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
+                       xprt_max_tcp_slot_table_entries);
        if (IS_ERR(xprt))
                return xprt;
        transport = container_of(xprt, struct sock_xprt, xprt);
@@ -2415,7 +2708,6 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
        xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
 
        xprt->bind_timeout = XS_BIND_TO;
-       xprt->connect_timeout = XS_TCP_CONN_TO;
        xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
        xprt->idle_timeout = XS_IDLE_DISC_TO;
 
@@ -2428,7 +2720,7 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
                        xprt_set_bound(xprt);
 
                INIT_DELAYED_WORK(&transport->connect_worker,
-                                       xs_tcp_connect_worker4);
+                                       xs_tcp_setup_socket);
                xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
                break;
        case AF_INET6:
@@ -2436,12 +2728,12 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
                        xprt_set_bound(xprt);
 
                INIT_DELAYED_WORK(&transport->connect_worker,
-                                       xs_tcp_connect_worker6);
+                                       xs_tcp_setup_socket);
                xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
                break;
        default:
-               kfree(xprt);
-               return ERR_PTR(-EAFNOSUPPORT);
+               ret = ERR_PTR(-EAFNOSUPPORT);
+               goto out_err;
        }
 
        if (xprt_bound(xprt))
@@ -2457,10 +2749,10 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
 
        if (try_module_get(THIS_MODULE))
                return xprt;
-
-       kfree(xprt->slot);
-       kfree(xprt);
-       return ERR_PTR(-EINVAL);
+       ret = ERR_PTR(-EINVAL);
+out_err:
+       xprt_free(xprt);
+       return ret;
 }
 
 /**
@@ -2474,11 +2766,19 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
        struct rpc_xprt *xprt;
        struct sock_xprt *transport;
        struct svc_sock *bc_sock;
+       struct rpc_xprt *ret;
 
-       if (!args->bc_xprt)
-               ERR_PTR(-EINVAL);
-
-       xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries);
+       if (args->bc_xprt->xpt_bc_xprt) {
+               /*
+                * This server connection already has a backchannel
+                * export; we can't create a new one, as we wouldn't be
+                * able to match replies based on xid any more.  So,
+                * reuse the already-existing one:
+                */
+                return args->bc_xprt->xpt_bc_xprt;
+       }
+       xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
+                       xprt_tcp_slot_table_entries);
        if (IS_ERR(xprt))
                return xprt;
        transport = container_of(xprt, struct sock_xprt, xprt);
@@ -2491,20 +2791,9 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
        /* backchannel */
        xprt_set_bound(xprt);
        xprt->bind_timeout = 0;
-       xprt->connect_timeout = 0;
        xprt->reestablish_timeout = 0;
        xprt->idle_timeout = 0;
 
-       /*
-        * The backchannel uses the same socket connection as the
-        * forechannel
-        */
-       xprt->bc_xprt = args->bc_xprt;
-       bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
-       bc_sock->sk_bc_xprt = xprt;
-       transport->sock = bc_sock->sk_sock;
-       transport->inet = bc_sock->sk_sk;
-
        xprt->ops = &bc_tcp_ops;
 
        switch (addr->sa_family) {
@@ -2517,19 +2806,28 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
                                   RPCBIND_NETID_TCP6);
                break;
        default:
-               kfree(xprt);
-               return ERR_PTR(-EAFNOSUPPORT);
+               ret = ERR_PTR(-EAFNOSUPPORT);
+               goto out_err;
        }
 
-       if (xprt_bound(xprt))
-               dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
-                               xprt->address_strings[RPC_DISPLAY_ADDR],
-                               xprt->address_strings[RPC_DISPLAY_PORT],
-                               xprt->address_strings[RPC_DISPLAY_PROTO]);
-       else
-               dprintk("RPC:       set up xprt to %s (autobind) via %s\n",
-                               xprt->address_strings[RPC_DISPLAY_ADDR],
-                               xprt->address_strings[RPC_DISPLAY_PROTO]);
+       dprintk("RPC:       set up xprt to %s (port %s) via %s\n",
+                       xprt->address_strings[RPC_DISPLAY_ADDR],
+                       xprt->address_strings[RPC_DISPLAY_PORT],
+                       xprt->address_strings[RPC_DISPLAY_PROTO]);
+
+       /*
+        * Once we've associated a backchannel xprt with a connection,
+        * we want to keep it around as long as long as the connection
+        * lasts, in case we need to start using it for a backchannel
+        * again; this reference won't be dropped until bc_xprt is
+        * destroyed.
+        */
+       xprt_get(xprt);
+       args->bc_xprt->xpt_bc_xprt = xprt;
+       xprt->bc_xprt = args->bc_xprt;
+       bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
+       transport->sock = bc_sock->sk_sock;
+       transport->inet = bc_sock->sk_sk;
 
        /*
         * Since we don't want connections for the backchannel, we set
@@ -2540,11 +2838,21 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
 
        if (try_module_get(THIS_MODULE))
                return xprt;
-       kfree(xprt->slot);
-       kfree(xprt);
-       return ERR_PTR(-EINVAL);
+       xprt_put(xprt);
+       ret = ERR_PTR(-EINVAL);
+out_err:
+       xprt_free(xprt);
+       return ret;
 }
 
+static struct xprt_class       xs_local_transport = {
+       .list           = LIST_HEAD_INIT(xs_local_transport.list),
+       .name           = "named UNIX socket",
+       .owner          = THIS_MODULE,
+       .ident          = XPRT_TRANSPORT_LOCAL,
+       .setup          = xs_setup_local,
+};
+
 static struct xprt_class       xs_udp_transport = {
        .list           = LIST_HEAD_INIT(xs_udp_transport.list),
        .name           = "udp",
@@ -2580,6 +2888,7 @@ int init_socket_xprt(void)
                sunrpc_table_header = register_sysctl_table(sunrpc_table);
 #endif
 
+       xprt_register_transport(&xs_local_transport);
        xprt_register_transport(&xs_udp_transport);
        xprt_register_transport(&xs_tcp_transport);
        xprt_register_transport(&xs_bc_tcp_transport);
@@ -2600,12 +2909,14 @@ void cleanup_socket_xprt(void)
        }
 #endif
 
+       xprt_unregister_transport(&xs_local_transport);
        xprt_unregister_transport(&xs_udp_transport);
        xprt_unregister_transport(&xs_tcp_transport);
        xprt_unregister_transport(&xs_bc_tcp_transport);
 }
 
-static int param_set_uint_minmax(const char *val, struct kernel_param *kp,
+static int param_set_uint_minmax(const char *val,
+               const struct kernel_param *kp,
                unsigned int min, unsigned int max)
 {
        unsigned long num;
@@ -2620,39 +2931,60 @@ static int param_set_uint_minmax(const char *val, struct kernel_param *kp,
        return 0;
 }
 
-static int param_set_portnr(const char *val, struct kernel_param *kp)
+static int param_set_portnr(const char *val, const struct kernel_param *kp)
 {
        return param_set_uint_minmax(val, kp,
                        RPC_MIN_RESVPORT,
                        RPC_MAX_RESVPORT);
 }
 
-static int param_get_portnr(char *buffer, struct kernel_param *kp)
-{
-       return param_get_uint(buffer, kp);
-}
+static struct kernel_param_ops param_ops_portnr = {
+       .set = param_set_portnr,
+       .get = param_get_uint,
+};
+
 #define param_check_portnr(name, p) \
        __param_check(name, p, unsigned int);
 
 module_param_named(min_resvport, xprt_min_resvport, portnr, 0644);
 module_param_named(max_resvport, xprt_max_resvport, portnr, 0644);
 
-static int param_set_slot_table_size(const char *val, struct kernel_param *kp)
+static int param_set_slot_table_size(const char *val,
+                                    const struct kernel_param *kp)
 {
        return param_set_uint_minmax(val, kp,
                        RPC_MIN_SLOT_TABLE,
                        RPC_MAX_SLOT_TABLE);
 }
 
-static int param_get_slot_table_size(char *buffer, struct kernel_param *kp)
+static struct kernel_param_ops param_ops_slot_table_size = {
+       .set = param_set_slot_table_size,
+       .get = param_get_uint,
+};
+
+#define param_check_slot_table_size(name, p) \
+       __param_check(name, p, unsigned int);
+
+static int param_set_max_slot_table_size(const char *val,
+                                    const struct kernel_param *kp)
 {
-       return param_get_uint(buffer, kp);
+       return param_set_uint_minmax(val, kp,
+                       RPC_MIN_SLOT_TABLE,
+                       RPC_MAX_SLOT_TABLE_LIMIT);
 }
-#define param_check_slot_table_size(name, p) \
+
+static struct kernel_param_ops param_ops_max_slot_table_size = {
+       .set = param_set_max_slot_table_size,
+       .get = param_get_uint,
+};
+
+#define param_check_max_slot_table_size(name, p) \
        __param_check(name, p, unsigned int);
 
 module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
                   slot_table_size, 0644);
+module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
+                  max_slot_table_size, 0644);
 module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
                   slot_table_size, 0644);