Merge tag 'nfs-for-3.7-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs

Pull NFS client updates from Trond Myklebust:
 "Features include:

   - Remove CONFIG_EXPERIMENTAL dependency from NFSv4.1
     Aside from the issues discussed at the LKS, distros are shipping
     NFSv4.1 with all the trimmings.
   - Fix fdatasync()/fsync() for the corner case of a server reboot.
   - NFSv4 OPEN access fix: finally distinguish correctly between
     open-for-read and open-for-execute permissions in all situations.
   - Ensure that the TCP socket is closed when we're in CLOSE_WAIT
   - More idmapper bugfixes
   - Lots of pNFS bugfixes and cleanups to remove unnecessary state and
     make the code easier to read.
   - In cases where a pNFS read or write fails, allow the client to
     resume trying layoutgets after two minutes of read/write-
     through-mds.
   - More net namespace fixes to the NFSv4 callback code.
   - More net namespace fixes to the NFSv3 locking code.
   - More NFSv4 migration preparatory patches.
     Including patches to detect network trunking in both NFSv4 and
     NFSv4.1
   - pNFS block updates to optimise LAYOUTGET calls."

* tag 'nfs-for-3.7-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs: (113 commits)
  pnfsblock: cleanup nfs4_blkdev_get
  NFS41: send real read size in layoutget
  NFS41: send real write size in layoutget
  NFS: track direct IO left bytes
  NFSv4.1: Cleanup ugliness in pnfs_layoutgets_blocked()
  NFSv4.1: Ensure that the layout sequence id stays 'close' to the current
  NFSv4.1: Deal with seqid wraparound in the pNFS return-on-close code
  NFSv4 set open access operation call flag in nfs4_init_opendata_res
  NFSv4.1: Remove the dependency on CONFIG_EXPERIMENTAL
  NFSv4 reduce attribute requests for open reclaim
  NFSv4: nfs4_open_done first must check that GETATTR decoded a file type
  NFSv4.1: Deal with wraparound when updating the layout "barrier" seqid
  NFSv4.1: Deal with wraparound issues when updating the layout stateid
  NFSv4.1: Always set the layout stateid if this is the first layoutget
  NFSv4.1: Fix another refcount issue in pnfs_find_alloc_layout
  NFSv4: don't put ACCESS in OPEN compound if O_EXCL
  NFSv4: don't check MAY_WRITE access bit in OPEN
  NFS: Set key construction data for the legacy upcall
  NFSv4.1: don't do two EXCHANGE_IDs on mount
  NFS: nfs41_walk_client_list(): re-lock before iterating
  ...

8 files changed, 139 insertions, 125 deletions

diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index 34c522021004..909dc0c31aab 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -239,7 +239,7 @@ gss_fill_context(const void *p, const void *end, struct gss_cl_ctx *ctx, struct
         }
         return q;
 err:
-        dprintk("RPC:       gss_fill_context returning %ld\n", -PTR_ERR(p));
+        dprintk("RPC:       %s returning %ld\n", __func__, -PTR_ERR(p));
         return p;
 }
 
@@ -301,10 +301,10 @@ __gss_find_upcall(struct rpc_pipe *pipe, uid_t uid)
                 if (pos->uid != uid)
                         continue;
                 atomic_inc(&pos->count);
-                dprintk("RPC:       gss_find_upcall found msg %p\n", pos);
+                dprintk("RPC:       %s found msg %p\n", __func__, pos);
                 return pos;
         }
-        dprintk("RPC:       gss_find_upcall found nothing\n");
+        dprintk("RPC:       %s found nothing\n", __func__);
         return NULL;
 }
 
@@ -507,8 +507,8 @@ gss_refresh_upcall(struct rpc_task *task)
         struct rpc_pipe *pipe;
         int err = 0;
 
-        dprintk("RPC: %5u gss_refresh_upcall for uid %u\n", task->tk_pid,
-                                                                cred->cr_uid);
+        dprintk("RPC: %5u %s for uid %u\n",
+                task->tk_pid, __func__, cred->cr_uid);
         gss_msg = gss_setup_upcall(task->tk_client, gss_auth, cred);
         if (PTR_ERR(gss_msg) == -EAGAIN) {
                 /* XXX: warning on the first, under the assumption we
@@ -539,8 +539,8 @@ gss_refresh_upcall(struct rpc_task *task)
         spin_unlock(&pipe->lock);
         gss_release_msg(gss_msg);
 out:
-        dprintk("RPC: %5u gss_refresh_upcall for uid %u result %d\n",
-                        task->tk_pid, cred->cr_uid, err);
+        dprintk("RPC: %5u %s for uid %u result %d\n",
+                task->tk_pid, __func__, cred->cr_uid, err);
         return err;
 }
 
@@ -553,7 +553,7 @@ gss_create_upcall(struct gss_auth *gss_auth, struct gss_cred *gss_cred)
         DEFINE_WAIT(wait);
         int err = 0;
 
-        dprintk("RPC:       gss_upcall for uid %u\n", cred->cr_uid);
+        dprintk("RPC:       %s for uid %u\n", __func__, cred->cr_uid);
 retry:
         gss_msg = gss_setup_upcall(gss_auth->client, gss_auth, cred);
         if (PTR_ERR(gss_msg) == -EAGAIN) {
@@ -594,8 +594,8 @@ out_intr:
         finish_wait(&gss_msg->waitqueue, &wait);
         gss_release_msg(gss_msg);
 out:
-        dprintk("RPC:       gss_create_upcall for uid %u result %d\n",
-                        cred->cr_uid, err);
+        dprintk("RPC:       %s for uid %u result %d\n",
+                __func__, cred->cr_uid, err);
         return err;
 }
 
@@ -681,7 +681,7 @@ err_put_ctx:
 err:
         kfree(buf);
 out:
-        dprintk("RPC:       gss_pipe_downcall returning %Zd\n", err);
+        dprintk("RPC:       %s returning %Zd\n", __func__, err);
         return err;
 }
 
@@ -747,8 +747,8 @@ gss_pipe_destroy_msg(struct rpc_pipe_msg *msg)
         struct gss_upcall_msg *gss_msg = container_of(msg, struct gss_upcall_msg, msg);
 
         if (msg->errno < 0) {
-                dprintk("RPC:       gss_pipe_destroy_msg releasing msg %p\n",
-                                gss_msg);
+                dprintk("RPC:       %s releasing msg %p\n",
+                        __func__, gss_msg);
                 atomic_inc(&gss_msg->count);
                 gss_unhash_msg(gss_msg);
                 if (msg->errno == -ETIMEDOUT)
@@ -976,7 +976,7 @@ gss_destroying_context(struct rpc_cred *cred)
 static void
 gss_do_free_ctx(struct gss_cl_ctx *ctx)
 {
-        dprintk("RPC:       gss_free_ctx\n");
+        dprintk("RPC:       %s\n", __func__);
 
         gss_delete_sec_context(&ctx->gc_gss_ctx);
         kfree(ctx->gc_wire_ctx.data);
@@ -999,7 +999,7 @@ gss_free_ctx(struct gss_cl_ctx *ctx)
 static void
 gss_free_cred(struct gss_cred *gss_cred)
 {
-        dprintk("RPC:       gss_free_cred %p\n", gss_cred);
+        dprintk("RPC:       %s cred=%p\n", __func__, gss_cred);
         kfree(gss_cred);
 }
 
@@ -1049,8 +1049,8 @@ gss_create_cred(struct rpc_auth *auth, struct auth_cred *acred, int flags)
         struct gss_cred *cred = NULL;
         int err = -ENOMEM;
 
-        dprintk("RPC:       gss_create_cred for uid %d, flavor %d\n",
-                acred->uid, auth->au_flavor);
+        dprintk("RPC:       %s for uid %d, flavor %d\n",
+                __func__, acred->uid, auth->au_flavor);
 
         if (!(cred = kzalloc(sizeof(*cred), GFP_NOFS)))
                 goto out_err;
@@ -1069,7 +1069,7 @@ gss_create_cred(struct rpc_auth *auth, struct auth_cred *acred, int flags)
         return &cred->gc_base;
 
 out_err:
-        dprintk("RPC:       gss_create_cred failed with error %d\n", err);
+        dprintk("RPC:       %s failed with error %d\n", __func__, err);
         return ERR_PTR(err);
 }
 
@@ -1127,7 +1127,7 @@ gss_marshal(struct rpc_task *task, __be32 *p)
         struct kvec     iov;
         struct xdr_buf  verf_buf;
 
-        dprintk("RPC: %5u gss_marshal\n", task->tk_pid);
+        dprintk("RPC: %5u %s\n", task->tk_pid, __func__);
 
         *p++ = htonl(RPC_AUTH_GSS);
         cred_len = p++;
@@ -1253,7 +1253,7 @@ gss_validate(struct rpc_task *task, __be32 *p)
         u32             flav,len;
         u32             maj_stat;
 
-        dprintk("RPC: %5u gss_validate\n", task->tk_pid);
+        dprintk("RPC: %5u %s\n", task->tk_pid, __func__);
 
         flav = ntohl(*p++);
         if ((len = ntohl(*p++)) > RPC_MAX_AUTH_SIZE)
@@ -1271,20 +1271,20 @@ gss_validate(struct rpc_task *task, __be32 *p)
         if (maj_stat == GSS_S_CONTEXT_EXPIRED)
                 clear_bit(RPCAUTH_CRED_UPTODATE, &cred->cr_flags);
         if (maj_stat) {
-                dprintk("RPC: %5u gss_validate: gss_verify_mic returned "
-                                "error 0x%08x\n", task->tk_pid, maj_stat);
+                dprintk("RPC: %5u %s: gss_verify_mic returned error 0x%08x\n",
+                        task->tk_pid, __func__, maj_stat);
                 goto out_bad;
         }
         /* We leave it to unwrap to calculate au_rslack. For now we just
          * calculate the length of the verifier: */
         cred->cr_auth->au_verfsize = XDR_QUADLEN(len) + 2;
         gss_put_ctx(ctx);
-        dprintk("RPC: %5u gss_validate: gss_verify_mic succeeded.\n",
-                        task->tk_pid);
+        dprintk("RPC: %5u %s: gss_verify_mic succeeded.\n",
+                        task->tk_pid, __func__);
         return p + XDR_QUADLEN(len);
 out_bad:
         gss_put_ctx(ctx);
-        dprintk("RPC: %5u gss_validate failed.\n", task->tk_pid);
+        dprintk("RPC: %5u %s failed.\n", task->tk_pid, __func__);
         return NULL;
 }
 
@@ -1466,7 +1466,7 @@ gss_wrap_req(struct rpc_task *task,
         struct gss_cl_ctx *ctx = gss_cred_get_ctx(cred);
         int             status = -EIO;
 
-        dprintk("RPC: %5u gss_wrap_req\n", task->tk_pid);
+        dprintk("RPC: %5u %s\n", task->tk_pid, __func__);
         if (ctx->gc_proc != RPC_GSS_PROC_DATA) {
                 /* The spec seems a little ambiguous here, but I think that not
                  * wrapping context destruction requests makes the most sense.
@@ -1489,7 +1489,7 @@ gss_wrap_req(struct rpc_task *task,
         }
 out:
         gss_put_ctx(ctx);
-        dprintk("RPC: %5u gss_wrap_req returning %d\n", task->tk_pid, status);
+        dprintk("RPC: %5u %s returning %d\n", task->tk_pid, __func__, status);
         return status;
 }
 
@@ -1604,8 +1604,8 @@ out_decode:
         status = gss_unwrap_req_decode(decode, rqstp, p, obj);
 out:
         gss_put_ctx(ctx);
-        dprintk("RPC: %5u gss_unwrap_resp returning %d\n", task->tk_pid,
-                        status);
+        dprintk("RPC: %5u %s returning %d\n",
+                task->tk_pid, __func__, status);
         return status;
 }
 
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index fa48c60aef23..cdc7564b4512 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -490,61 +490,86 @@ EXPORT_SYMBOL_GPL(rpc_create);
  * same transport while varying parameters such as the authentication
  * flavour.
  */
-struct rpc_clnt *
-rpc_clone_client(struct rpc_clnt *clnt)
+static struct rpc_clnt *__rpc_clone_client(struct rpc_create_args *args,
+                                           struct rpc_clnt *clnt)
 {
-        struct rpc_clnt *new;
         struct rpc_xprt *xprt;
-        int err = -ENOMEM;
+        struct rpc_clnt *new;
+        int err;
 
-        new = kmemdup(clnt, sizeof(*new), GFP_KERNEL);
-        if (!new)
-                goto out_no_clnt;
-        new->cl_parent = clnt;
-        /* Turn off autobind on clones */
-        new->cl_autobind = 0;
-        INIT_LIST_HEAD(&new->cl_tasks);
-        spin_lock_init(&new->cl_lock);
-        rpc_init_rtt(&new->cl_rtt_default, clnt->cl_timeout->to_initval);
-        new->cl_metrics = rpc_alloc_iostats(clnt);
-        if (new->cl_metrics == NULL)
-                goto out_no_stats;
-        if (clnt->cl_principal) {
-                new->cl_principal = kstrdup(clnt->cl_principal, GFP_KERNEL);
-                if (new->cl_principal == NULL)
-                        goto out_no_principal;
-        }
+        err = -ENOMEM;
         rcu_read_lock();
         xprt = xprt_get(rcu_dereference(clnt->cl_xprt));
         rcu_read_unlock();
         if (xprt == NULL)
-                goto out_no_transport;
-        rcu_assign_pointer(new->cl_xprt, xprt);
-        atomic_set(&new->cl_count, 1);
-        err = rpc_setup_pipedir(new, clnt->cl_program->pipe_dir_name);
-        if (err != 0)
-                goto out_no_path;
-        rpc_clnt_set_nodename(new, utsname()->nodename);
-        if (new->cl_auth)
-                atomic_inc(&new->cl_auth->au_count);
+                goto out_err;
+        args->servername = xprt->servername;
+
+        new = rpc_new_client(args, xprt);
+        if (IS_ERR(new)) {
+                err = PTR_ERR(new);
+                goto out_put;
+        }
+
         atomic_inc(&clnt->cl_count);
-        rpc_register_client(new);
-        rpciod_up();
+        new->cl_parent = clnt;
+
+        /* Turn off autobind on clones */
+        new->cl_autobind = 0;
+        new->cl_softrtry = clnt->cl_softrtry;
+        new->cl_discrtry = clnt->cl_discrtry;
+        new->cl_chatty = clnt->cl_chatty;
         return new;
-out_no_path:
+
+out_put:
         xprt_put(xprt);
-out_no_transport:
-        kfree(new->cl_principal);
-out_no_principal:
-        rpc_free_iostats(new->cl_metrics);
-out_no_stats:
-        kfree(new);
-out_no_clnt:
+out_err:
         dprintk("RPC:       %s: returned error %d\n", __func__, err);
         return ERR_PTR(err);
 }
+
+/**
+ * rpc_clone_client - Clone an RPC client structure
+ *
+ * @clnt: RPC client whose parameters are copied
+ *
+ * Returns a fresh RPC client or an ERR_PTR.
+ */
+struct rpc_clnt *rpc_clone_client(struct rpc_clnt *clnt)
+{
+        struct rpc_create_args args = {
+                .program        = clnt->cl_program,
+                .prognumber     = clnt->cl_prog,
+                .version        = clnt->cl_vers,
+                .authflavor     = clnt->cl_auth->au_flavor,
+                .client_name    = clnt->cl_principal,
+        };
+        return __rpc_clone_client(&args, clnt);
+}
 EXPORT_SYMBOL_GPL(rpc_clone_client);
 
+/**
+ * rpc_clone_client_set_auth - Clone an RPC client structure and set its auth
+ *
+ * @clnt: RPC client whose parameters are copied
+ * @auth: security flavor for new client
+ *
+ * Returns a fresh RPC client or an ERR_PTR.
+ */
+struct rpc_clnt *
+rpc_clone_client_set_auth(struct rpc_clnt *clnt, rpc_authflavor_t flavor)
+{
+        struct rpc_create_args args = {
+                .program        = clnt->cl_program,
+                .prognumber     = clnt->cl_prog,
+                .version        = clnt->cl_vers,
+                .authflavor     = flavor,
+                .client_name    = clnt->cl_principal,
+        };
+        return __rpc_clone_client(&args, clnt);
+}
+EXPORT_SYMBOL_GPL(rpc_clone_client_set_auth);
+
 /*
  * Kill all tasks for the given client.
  * XXX: kill their descendants as well?
diff --git a/net/sunrpc/rpc_pipe.c b/net/sunrpc/rpc_pipe.c
index 21fde99e5c56..80f5dd23417d 100644
--- a/net/sunrpc/rpc_pipe.c
+++ b/net/sunrpc/rpc_pipe.c
@@ -1119,8 +1119,8 @@ rpc_fill_super(struct super_block *sb, void *data, int silent)
                 return -ENOMEM;
         if (rpc_populate(root, files, RPCAUTH_lockd, RPCAUTH_RootEOF, NULL))
                 return -ENOMEM;
-        dprintk("RPC:   sending pipefs MOUNT notification for net %p%s\n", net,
-                                                                NET_NAME(net));
+        dprintk("RPC:       sending pipefs MOUNT notification for net %p%s\n",
+                net, NET_NAME(net));
         sn->pipefs_sb = sb;
         err = blocking_notifier_call_chain(&rpc_pipefs_notifier_list,
                                            RPC_PIPEFS_MOUNT,
@@ -1155,8 +1155,8 @@ static void rpc_kill_sb(struct super_block *sb)
         sn->pipefs_sb = NULL;
         mutex_unlock(&sn->pipefs_sb_lock);
         put_net(net);
-        dprintk("RPC:   sending pipefs UMOUNT notification for net %p%s\n", net,
-                                                                NET_NAME(net));
+        dprintk("RPC:       sending pipefs UMOUNT notification for net %p%s\n",
+                net, NET_NAME(net));
         blocking_notifier_call_chain(&rpc_pipefs_notifier_list,
                                            RPC_PIPEFS_UMOUNT,
                                            sb);
diff --git a/net/sunrpc/sched.c b/net/sunrpc/sched.c
index 128494ec9a64..6357fcb00c7e 100644
--- a/net/sunrpc/sched.c
+++ b/net/sunrpc/sched.c
@@ -1022,7 +1022,7 @@ static int rpciod_start(void)
          * Create the rpciod thread and wait for it to start.
          */
         dprintk("RPC:       creating workqueue rpciod\n");
-        wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM, 0);
+        wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM, 1);
         rpciod_workqueue = wq;
         return rpciod_workqueue != NULL;
 }
diff --git a/net/sunrpc/xdr.c b/net/sunrpc/xdr.c
index 0afba1b4b656..08f50afd5f2a 100644
--- a/net/sunrpc/xdr.c
+++ b/net/sunrpc/xdr.c
@@ -730,19 +730,24 @@ static unsigned int xdr_align_pages(struct xdr_stream *xdr, unsigned int len)
 
         if (xdr->nwords == 0)
                 return 0;
-        if (nwords > xdr->nwords) {
-                nwords = xdr->nwords;
-                len = nwords << 2;
-        }
         /* Realign pages to current pointer position */
         iov  = buf->head;
-        if (iov->iov_len > cur)
+        if (iov->iov_len > cur) {
                 xdr_shrink_bufhead(buf, iov->iov_len - cur);
+                xdr->nwords = XDR_QUADLEN(buf->len - cur);
+        }
 
-        /* Truncate page data and move it into the tail */
-        if (buf->page_len > len)
+        if (nwords > xdr->nwords) {
+                nwords = xdr->nwords;
+                len = nwords << 2;
+        }
+        if (buf->page_len <= len)
+                len = buf->page_len;
+        else if (nwords < xdr->nwords) {
+                /* Truncate page data and move it into the tail */
                 xdr_shrink_pagelen(buf, buf->page_len - len);
-        xdr->nwords = XDR_QUADLEN(buf->len - cur);
+                xdr->nwords = XDR_QUADLEN(buf->len - cur);
+        }
         return len;
 }
 
diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index 5d7f61d7559c..bd462a532acf 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -231,7 +231,7 @@ EXPORT_SYMBOL_GPL(xprt_reserve_xprt);
 static void xprt_clear_locked(struct rpc_xprt *xprt)
 {
         xprt->snd_task = NULL;
-        if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state) || xprt->shutdown) {
+        if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state)) {
                 smp_mb__before_clear_bit();
                 clear_bit(XPRT_LOCKED, &xprt->state);
                 smp_mb__after_clear_bit();
@@ -504,9 +504,6 @@ EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space);
  */
 void xprt_write_space(struct rpc_xprt *xprt)
 {
-        if (unlikely(xprt->shutdown))
-                return;
-
         spin_lock_bh(&xprt->transport_lock);
         if (xprt->snd_task) {
                 dprintk("RPC:       write space: waking waiting task on "
@@ -679,7 +676,7 @@ xprt_init_autodisconnect(unsigned long data)
         struct rpc_xprt *xprt = (struct rpc_xprt *)data;
 
         spin_lock(&xprt->transport_lock);
-        if (!list_empty(&xprt->recv) || xprt->shutdown)
+        if (!list_empty(&xprt->recv))
                 goto out_abort;
         if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
                 goto out_abort;
@@ -1262,7 +1259,6 @@ out:
 static void xprt_destroy(struct rpc_xprt *xprt)
 {
         dprintk("RPC:       destroying transport %p\n", xprt);
-        xprt->shutdown = 1;
         del_timer_sync(&xprt->timer);
 
         rpc_destroy_wait_queue(&xprt->binding);
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index 5d9202dc7cb1..c9aa7a35f3bf 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -199,21 +199,15 @@ xprt_rdma_connect_worker(struct work_struct *work)
         struct rpc_xprt *xprt = &r_xprt->xprt;
         int rc = 0;
 
-        if (!xprt->shutdown) {
-                current->flags |= PF_FSTRANS;
-                xprt_clear_connected(xprt);
-
-                dprintk("RPC:       %s: %sconnect\n", __func__,
-                                r_xprt->rx_ep.rep_connected != 0 ? "re" : "");
-                rc = rpcrdma_ep_connect(&r_xprt->rx_ep, &r_xprt->rx_ia);
-                if (rc)
-                        goto out;
-        }
-        goto out_clear;
+        current->flags |= PF_FSTRANS;
+        xprt_clear_connected(xprt);
+
+        dprintk("RPC:       %s: %sconnect\n", __func__,
+                        r_xprt->rx_ep.rep_connected != 0 ? "re" : "");
+        rc = rpcrdma_ep_connect(&r_xprt->rx_ep, &r_xprt->rx_ia);
+        if (rc)
+                xprt_wake_pending_tasks(xprt, rc);
 
-out:
-        xprt_wake_pending_tasks(xprt, rc);
-out_clear:
         dprintk("RPC:       %s: exit\n", __func__);
         xprt_clear_connecting(xprt);
         current->flags &= ~PF_FSTRANS;
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index a35b8e52e551..aaaadfbe36e9 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -917,9 +917,6 @@ static void xs_local_data_ready(struct sock *sk, int len)
         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);
@@ -981,9 +978,6 @@ static void xs_udp_data_ready(struct sock *sk, int len)
         if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
                 goto out;
 
-        if (xprt->shutdown)
-                goto dropit;
-
         repsize = skb->len - sizeof(struct udphdr);
         if (repsize < 4) {
                 dprintk("RPC:       impossible RPC reply size %d!\n", repsize);
@@ -1025,6 +1019,16 @@ static void xs_udp_data_ready(struct sock *sk, int len)
         read_unlock_bh(&sk->sk_callback_lock);
 }
 
+/*
+ * Helper function to force a TCP close if the server is sending
+ * junk and/or it has put us in CLOSE_WAIT
+ */
+static void xs_tcp_force_close(struct rpc_xprt *xprt)
+{
+        set_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
+        xprt_force_disconnect(xprt);
+}
+
 static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
 {
         struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
@@ -1051,7 +1055,7 @@ static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_rea
         /* Sanity check of the record length */
         if (unlikely(transport->tcp_reclen < 8)) {
                 dprintk("RPC:       invalid TCP record fragment length\n");
-                xprt_force_disconnect(xprt);
+                xs_tcp_force_close(xprt);
                 return;
         }
         dprintk("RPC:       reading TCP record fragment of length %d\n",
@@ -1132,7 +1136,7 @@ static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
                 break;
         default:
                 dprintk("RPC:       invalid request message type\n");
-                xprt_force_disconnect(&transport->xprt);
+                xs_tcp_force_close(&transport->xprt);
         }
         xs_tcp_check_fraghdr(transport);
 }
@@ -1402,9 +1406,6 @@ static void xs_tcp_data_ready(struct sock *sk, int bytes)
         read_lock_bh(&sk->sk_callback_lock);
         if (!(xprt = xprt_from_sock(sk)))
                 goto out;
-        if (xprt->shutdown)
-                goto out;
-
         /* Any data means we had a useful conversation, so
          * the we don't need to delay the next reconnect
          */
@@ -1455,6 +1456,8 @@ static void xs_tcp_cancel_linger_timeout(struct rpc_xprt *xprt)
 static void xs_sock_mark_closed(struct rpc_xprt *xprt)
 {
         smp_mb__before_clear_bit();
+        clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
+        clear_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
         clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
         clear_bit(XPRT_CLOSING, &xprt->state);
         smp_mb__after_clear_bit();
@@ -1512,8 +1515,8 @@ static void xs_tcp_state_change(struct sock *sk)
                 break;
         case TCP_CLOSE_WAIT:
                 /* The server initiated a shutdown of the socket */
-                xprt_force_disconnect(xprt);
                 xprt->connect_cookie++;
+                xs_tcp_force_close(xprt);
         case TCP_CLOSING:
                 /*
                  * If the server closed down the connection, make sure that
@@ -1889,9 +1892,6 @@ static void xs_local_setup_socket(struct work_struct *work)
         struct socket *sock;
         int status = -EIO;
 
-        if (xprt->shutdown)
-                goto out;
-
         current->flags |= PF_FSTRANS;
 
         clear_bit(XPRT_CONNECTION_ABORT, &xprt->state);
@@ -2008,9 +2008,6 @@ static void xs_udp_setup_socket(struct work_struct *work)
         struct socket *sock = transport->sock;
         int status = -EIO;
 
-        if (xprt->shutdown)
-                goto out;
-
         current->flags |= PF_FSTRANS;
 
         /* Start by resetting any existing state */
@@ -2156,9 +2153,6 @@ static void xs_tcp_setup_socket(struct work_struct *work)
         struct rpc_xprt *xprt = &transport->xprt;
         int status = -EIO;
 
-        if (xprt->shutdown)
-                goto out;
-
         current->flags |= PF_FSTRANS;
 
         if (!sock) {
@@ -2199,8 +2193,7 @@ static void xs_tcp_setup_socket(struct work_struct *work)
                 /* We're probably in TIME_WAIT. Get rid of existing socket,
                  * and retry
                  */
-                set_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
-                xprt_force_disconnect(xprt);
+                xs_tcp_force_close(xprt);
                 break;
         case -ECONNREFUSED:
         case -ECONNRESET:
@@ -2528,6 +2521,7 @@ static struct rpc_xprt_ops xs_tcp_ops = {
 static struct rpc_xprt_ops bc_tcp_ops = {
         .reserve_xprt           = xprt_reserve_xprt,
         .release_xprt           = xprt_release_xprt,
+        .alloc_slot             = xprt_alloc_slot,
         .rpcbind                = xs_local_rpcbind,
         .buf_alloc              = bc_malloc,
         .buf_free               = bc_free,