1 files changed, 551 insertions, 247 deletions

diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index fe9306bf10cc..72abb7358933 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -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>
@@ -45,6 +47,9 @@
 #include <net/tcp.h>
 
 #include "sunrpc.h"
+
+static void xs_close(struct rpc_xprt *xprt);
+
 /*
  * xprtsock tunables
  */
@@ -261,6 +266,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;
@@ -276,23 +286,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);
                 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);
                 snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
                 break;
         default:
                 BUG();
         }
+
         xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
 }
 
@@ -495,6 +516,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
@@ -504,7 +589,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)
 {
@@ -574,13 +659,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
@@ -590,7 +668,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?
@@ -603,7 +681,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,
@@ -710,6 +788,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;
@@ -770,12 +850,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);
 }
 
@@ -784,6 +863,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
@@ -1343,7 +1504,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:
                 /*
@@ -1516,7 +1676,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;
 
@@ -1525,7 +1685,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;
@@ -1535,23 +1695,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) {
@@ -1559,52 +1714,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;
@@ -1622,7 +1772,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)
 {
 }
@@ -1630,11 +1799,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;
@@ -1644,13 +1844,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;
@@ -1658,85 +1857,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);
-                goto out;
-        }
-        xs_reclassify_socket6(sock);
-
-        if (xs_bind6(transport, sock) < 0) {
-                sock_release(sock);
+        sock = xs_create_sock(xprt, transport,
+                        xs_addr(xprt)->sa_family, SOCK_DGRAM, IPPROTO_UDP);
+        if (IS_ERR(sock))
                 goto out;
-        }
 
         dprintk("RPC:       worker connecting xprt %p via %s to "
                                 "%s (port %s)\n", xprt,
@@ -1755,12 +1982,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
@@ -1770,13 +1997,13 @@ 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;
 
@@ -1799,12 +2026,13 @@ static void xs_tcp_reuse_connection(struct rpc_xprt *xprt, struct sock_xprt *tra
                                 "sk_shutdown set to %d\n",
                                 __func__, transport->inet->sk_shutdown);
         }
-        xs_abort_connection(xprt, transport);
+        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 +2064,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 +2090,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 +2103,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 +2115,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;
@@ -1925,84 +2164,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
@@ -2041,6 +2202,32 @@ static void xs_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)
+{
+        long idle_time = 0;
+
+        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);
+}
+
+/**
  * xs_udp_print_stats - display UDP socket-specifc stats
  * @xprt: rpc_xprt struct containing statistics
  * @seq: output file
@@ -2138,10 +2325,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;
@@ -2213,6 +2397,21 @@ static void bc_destroy(struct rpc_xprt *xprt)
 {
 }
 
+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,
@@ -2262,6 +2461,33 @@ 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)
 {
@@ -2273,31 +2499,93 @@ 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);
+        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);
+        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,
@@ -2341,7 +2629,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:
@@ -2349,7 +2637,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_worker6);
+                                        xs_udp_setup_socket);
                 xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
                 break;
         default:
@@ -2371,8 +2659,7 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
                 return xprt;
         ret = ERR_PTR(-EINVAL);
 out_err:
-        kfree(xprt->slot);
-        kfree(xprt);
+        xprt_free(xprt);
         return ret;
 }
 
@@ -2416,7 +2703,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:
@@ -2424,7 +2711,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_worker6);
+                                        xs_tcp_setup_socket);
                 xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
                 break;
         default:
@@ -2447,8 +2734,7 @@ static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
                 return xprt;
         ret = ERR_PTR(-EINVAL);
 out_err:
-        kfree(xprt->slot);
-        kfree(xprt);
+        xprt_free(xprt);
         return ret;
 }
 
@@ -2465,6 +2751,15 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
         struct svc_sock *bc_sock;
         struct rpc_xprt *ret;
 
+        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);
         if (IS_ERR(xprt))
                 return xprt;
@@ -2481,16 +2776,6 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
         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) {
@@ -2507,15 +2792,24 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
                 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
@@ -2526,13 +2820,21 @@ static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
 
         if (try_module_get(THIS_MODULE))
                 return xprt;
+        xprt_put(xprt);
         ret = ERR_PTR(-EINVAL);
 out_err:
-        kfree(xprt->slot);
-        kfree(xprt);
+        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",
@@ -2568,6 +2870,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);
@@ -2588,6 +2891,7 @@ 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);