Merge git://git.linux-nfs.org/pub/linux/nfs-2.6

* git://git.linux-nfs.org/pub/linux/nfs-2.6: (103 commits)
  SUNRPC,RPCSEC_GSS: spkm3--fix config dependencies
  SUNRPC,RPCSEC_GSS: spkm3: import contexts using NID_cast5_cbc
  LOCKD: Make nlmsvc_traverse_shares return void
  LOCKD: nlmsvc_traverse_blocks return is unused
  SUNRPC,RPCSEC_GSS: fix krb5 sequence numbers.
  NFSv4: Dont list system.nfs4_acl for filesystems that don't support it.
  SUNRPC,RPCSEC_GSS: remove unnecessary kmalloc of a checksum
  SUNRPC: Ensure rpc_call_async() always calls tk_ops->rpc_release()
  SUNRPC: Fix memory barriers for req->rq_received
  NFS: Fix a race in nfs_sync_inode()
  NFS: Clean up nfs_flush_list()
  NFS: Fix a race with PG_private and nfs_release_page()
  NFSv4: Ensure the callback daemon flushes signals
  SUNRPC: Fix a 'Busy inodes' error in rpc_pipefs
  NFS, NLM: Allow blocking locks to respect signals
  NFS: Make nfs_fhget() return appropriate error values
  NFSv4: Fix an oops in nfs4_fill_super
  lockd: blocks should hold a reference to the nlm_file
  NFSv4: SETCLIENTID_CONFIRM should handle NFS4ERR_DELAY/NFS4ERR_RESOURCE
  NFSv4: Send the delegation stateid for SETATTR calls
  ...

15 files changed, 330 insertions, 69 deletions

diff --git a/net/sunrpc/auth.c b/net/sunrpc/auth.c
index 8d6f1a176b15..55163af3dcaf 100644
--- a/net/sunrpc/auth.c
+++ b/net/sunrpc/auth.c
@@ -64,14 +64,26 @@ rpcauth_create(rpc_authflavor_t pseudoflavor, struct rpc_clnt *clnt)
         struct rpc_authops      *ops;
         u32                     flavor = pseudoflavor_to_flavor(pseudoflavor);
 
-        if (flavor >= RPC_AUTH_MAXFLAVOR || !(ops = auth_flavors[flavor]))
-                return ERR_PTR(-EINVAL);
+        auth = ERR_PTR(-EINVAL);
+        if (flavor >= RPC_AUTH_MAXFLAVOR)
+                goto out;
+
+        /* FIXME - auth_flavors[] really needs an rw lock,
+         * and module refcounting. */
+#ifdef CONFIG_KMOD
+        if ((ops = auth_flavors[flavor]) == NULL)
+                request_module("rpc-auth-%u", flavor);
+#endif
+        if ((ops = auth_flavors[flavor]) == NULL)
+                goto out;
         auth = ops->create(clnt, pseudoflavor);
         if (IS_ERR(auth))
                 return auth;
         if (clnt->cl_auth)
                 rpcauth_destroy(clnt->cl_auth);
         clnt->cl_auth = auth;
+
+out:
         return auth;
 }
 
diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index bb46efd92e57..900ef31f5a0e 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -721,6 +721,8 @@ gss_destroy(struct rpc_auth *auth)
 
         gss_auth = container_of(auth, struct gss_auth, rpc_auth);
         rpc_unlink(gss_auth->path);
+        dput(gss_auth->dentry);
+        gss_auth->dentry = NULL;
         gss_mech_put(gss_auth->mech);
 
         rpcauth_free_credcache(auth);
diff --git a/net/sunrpc/auth_gss/gss_krb5_seal.c b/net/sunrpc/auth_gss/gss_krb5_seal.c
index d0dfdfd5e79e..f43311221a72 100644
--- a/net/sunrpc/auth_gss/gss_krb5_seal.c
+++ b/net/sunrpc/auth_gss/gss_krb5_seal.c
@@ -70,15 +70,19 @@
 # define RPCDBG_FACILITY        RPCDBG_AUTH
 #endif
 
+spinlock_t krb5_seq_lock = SPIN_LOCK_UNLOCKED;
+
 u32
 gss_get_mic_kerberos(struct gss_ctx *gss_ctx, struct xdr_buf *text,
                 struct xdr_netobj *token)
 {
         struct krb5_ctx         *ctx = gss_ctx->internal_ctx_id;
         s32                     checksum_type;
-        struct xdr_netobj       md5cksum = {.len = 0, .data = NULL};
+        char                    cksumdata[16];
+        struct xdr_netobj       md5cksum = {.len = 0, .data = cksumdata};
         unsigned char           *ptr, *krb5_hdr, *msg_start;
         s32                     now;
+        u32                     seq_send;
 
         dprintk("RPC:     gss_krb5_seal\n");
 
@@ -133,16 +137,15 @@ gss_get_mic_kerberos(struct gss_ctx *gss_ctx, struct xdr_buf *text,
                 BUG();
         }
 
-        kfree(md5cksum.data);
+        spin_lock(&krb5_seq_lock);
+        seq_send = ctx->seq_send++;
+        spin_unlock(&krb5_seq_lock);
 
         if ((krb5_make_seq_num(ctx->seq, ctx->initiate ? 0 : 0xff,
-                               ctx->seq_send, krb5_hdr + 16, krb5_hdr + 8)))
+                               seq_send, krb5_hdr + 16, krb5_hdr + 8)))
                 goto out_err;
 
-        ctx->seq_send++;
-
         return ((ctx->endtime < now) ? GSS_S_CONTEXT_EXPIRED : GSS_S_COMPLETE);
 out_err:
-        kfree(md5cksum.data);
         return GSS_S_FAILURE;
 }
diff --git a/net/sunrpc/auth_gss/gss_krb5_unseal.c b/net/sunrpc/auth_gss/gss_krb5_unseal.c
index db055fd7d778..0828cf64100f 100644
--- a/net/sunrpc/auth_gss/gss_krb5_unseal.c
+++ b/net/sunrpc/auth_gss/gss_krb5_unseal.c
@@ -79,7 +79,8 @@ gss_verify_mic_kerberos(struct gss_ctx *gss_ctx,
         int                     signalg;
         int                     sealalg;
         s32                     checksum_type;
-        struct xdr_netobj       md5cksum = {.len = 0, .data = NULL};
+        char                    cksumdata[16];
+        struct xdr_netobj       md5cksum = {.len = 0, .data = cksumdata};
         s32                     now;
         int                     direction;
         s32                     seqnum;
@@ -176,6 +177,5 @@ gss_verify_mic_kerberos(struct gss_ctx *gss_ctx,
 
         ret = GSS_S_COMPLETE;
 out:
-        kfree(md5cksum.data);
         return ret;
 }
diff --git a/net/sunrpc/auth_gss/gss_krb5_wrap.c b/net/sunrpc/auth_gss/gss_krb5_wrap.c
index af777cf9f251..89d1f3e14128 100644
--- a/net/sunrpc/auth_gss/gss_krb5_wrap.c
+++ b/net/sunrpc/auth_gss/gss_krb5_wrap.c
@@ -121,12 +121,14 @@ gss_wrap_kerberos(struct gss_ctx *ctx, int offset,
 {
         struct krb5_ctx         *kctx = ctx->internal_ctx_id;
         s32                     checksum_type;
-        struct xdr_netobj       md5cksum = {.len = 0, .data = NULL};
+        char                    cksumdata[16];
+        struct xdr_netobj       md5cksum = {.len = 0, .data = cksumdata};
         int                     blocksize = 0, plainlen;
         unsigned char           *ptr, *krb5_hdr, *msg_start;
         s32                     now;
         int                     headlen;
         struct page             **tmp_pages;
+        u32                     seq_send;
 
         dprintk("RPC:     gss_wrap_kerberos\n");
 
@@ -205,23 +207,22 @@ gss_wrap_kerberos(struct gss_ctx *ctx, int offset,
                 BUG();
         }
 
-        kfree(md5cksum.data);
+        spin_lock(&krb5_seq_lock);
+        seq_send = kctx->seq_send++;
+        spin_unlock(&krb5_seq_lock);
 
         /* XXX would probably be more efficient to compute checksum
          * and encrypt at the same time: */
         if ((krb5_make_seq_num(kctx->seq, kctx->initiate ? 0 : 0xff,
-                               kctx->seq_send, krb5_hdr + 16, krb5_hdr + 8)))
+                               seq_send, krb5_hdr + 16, krb5_hdr + 8)))
                 goto out_err;
 
         if (gss_encrypt_xdr_buf(kctx->enc, buf, offset + headlen - blocksize,
                                                                         pages))
                 goto out_err;
 
-        kctx->seq_send++;
-
         return ((kctx->endtime < now) ? GSS_S_CONTEXT_EXPIRED : GSS_S_COMPLETE);
 out_err:
-        if (md5cksum.data) kfree(md5cksum.data);
         return GSS_S_FAILURE;
 }
 
@@ -232,7 +233,8 @@ gss_unwrap_kerberos(struct gss_ctx *ctx, int offset, struct xdr_buf *buf)
         int                     signalg;
         int                     sealalg;
         s32                     checksum_type;
-        struct xdr_netobj       md5cksum = {.len = 0, .data = NULL};
+        char                    cksumdata[16];
+        struct xdr_netobj       md5cksum = {.len = 0, .data = cksumdata};
         s32                     now;
         int                     direction;
         s32                     seqnum;
@@ -358,6 +360,5 @@ gss_unwrap_kerberos(struct gss_ctx *ctx, int offset, struct xdr_buf *buf)
 
         ret = GSS_S_COMPLETE;
 out:
-        if (md5cksum.data) kfree(md5cksum.data);
         return ret;
 }
diff --git a/net/sunrpc/auth_gss/gss_spkm3_mech.c b/net/sunrpc/auth_gss/gss_spkm3_mech.c
index 58400807d4df..5bf11ccba7cd 100644
--- a/net/sunrpc/auth_gss/gss_spkm3_mech.c
+++ b/net/sunrpc/auth_gss/gss_spkm3_mech.c
@@ -102,6 +102,12 @@ get_key(const void *p, const void *end, struct crypto_tfm **res, int *resalg)
                         alg_mode = CRYPTO_TFM_MODE_CBC;
                         setkey = 1;
                         break;
+                case NID_cast5_cbc:
+                        /* XXXX here in name only, not used */
+                        alg_name = "cast5";
+                        alg_mode = CRYPTO_TFM_MODE_CBC;
+                        setkey = 0; /* XXX will need to set to 1 */
+                        break;
                 case NID_md5:
                         if (key.len == 0) {
                                 dprintk("RPC: SPKM3 get_key: NID_md5 zero Key length\n");
diff --git a/net/sunrpc/auth_gss/gss_spkm3_seal.c b/net/sunrpc/auth_gss/gss_spkm3_seal.c
index 86fbf7c3e39c..18c7862bc234 100644
--- a/net/sunrpc/auth_gss/gss_spkm3_seal.c
+++ b/net/sunrpc/auth_gss/gss_spkm3_seal.c
@@ -57,7 +57,8 @@ spkm3_make_token(struct spkm3_ctx *ctx,
 {
         s32                     checksum_type;
         char                    tokhdrbuf[25];
-        struct xdr_netobj       md5cksum = {.len = 0, .data = NULL};
+        char                    cksumdata[16];
+        struct xdr_netobj       md5cksum = {.len = 0, .data = cksumdata};
         struct xdr_netobj       mic_hdr = {.len = 0, .data = tokhdrbuf};
         int                     tokenlen = 0;
         unsigned char           *ptr;
@@ -115,13 +116,11 @@ spkm3_make_token(struct spkm3_ctx *ctx,
                 dprintk("RPC: gss_spkm3_seal: SPKM_WRAP_TOK not supported\n");
                 goto out_err;
         }
-        kfree(md5cksum.data);
 
         /* XXX need to implement sequence numbers, and ctx->expired */
 
         return  GSS_S_COMPLETE;
 out_err:
-        kfree(md5cksum.data);
         token->data = NULL;
         token->len = 0;
         return GSS_S_FAILURE;
diff --git a/net/sunrpc/auth_gss/gss_spkm3_unseal.c b/net/sunrpc/auth_gss/gss_spkm3_unseal.c
index 96851b0ba1ba..8537f581ef9b 100644
--- a/net/sunrpc/auth_gss/gss_spkm3_unseal.c
+++ b/net/sunrpc/auth_gss/gss_spkm3_unseal.c
@@ -56,7 +56,8 @@ spkm3_read_token(struct spkm3_ctx *ctx,
 {
         s32                     code;
         struct xdr_netobj       wire_cksum = {.len =0, .data = NULL};
-        struct xdr_netobj       md5cksum = {.len = 0, .data = NULL};
+        char                    cksumdata[16];
+        struct xdr_netobj       md5cksum = {.len = 0, .data = cksumdata};
         unsigned char           *ptr = (unsigned char *)read_token->data;
         unsigned char           *cksum;
         int                     bodysize, md5elen;
@@ -120,7 +121,6 @@ spkm3_read_token(struct spkm3_ctx *ctx,
         /* XXX: need to add expiration and sequencing */
         ret = GSS_S_COMPLETE;
 out:
-        kfree(md5cksum.data);
         kfree(wire_cksum.data);
         return ret;
 }
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index d78479782045..aa8965e9d307 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -28,12 +28,11 @@
 #include <linux/mm.h>
 #include <linux/slab.h>
 #include <linux/utsname.h>
+#include <linux/workqueue.h>
 
 #include <linux/sunrpc/clnt.h>
-#include <linux/workqueue.h>
 #include <linux/sunrpc/rpc_pipe_fs.h>
-
-#include <linux/nfs.h>
+#include <linux/sunrpc/metrics.h>
 
 
 #define RPC_SLACK_SPACE         (1024)  /* total overkill */
@@ -71,8 +70,15 @@ rpc_setup_pipedir(struct rpc_clnt *clnt, char *dir_name)
         static uint32_t clntid;
         int error;
 
+        clnt->cl_vfsmnt = ERR_PTR(-ENOENT);
+        clnt->cl_dentry = ERR_PTR(-ENOENT);
         if (dir_name == NULL)
                 return 0;
+
+        clnt->cl_vfsmnt = rpc_get_mount();
+        if (IS_ERR(clnt->cl_vfsmnt))
+                return PTR_ERR(clnt->cl_vfsmnt);
+
         for (;;) {
                 snprintf(clnt->cl_pathname, sizeof(clnt->cl_pathname),
                                 "%s/clnt%x", dir_name,
@@ -85,6 +91,7 @@ rpc_setup_pipedir(struct rpc_clnt *clnt, char *dir_name)
                 if (error != -EEXIST) {
                         printk(KERN_INFO "RPC: Couldn't create pipefs entry %s, error %d\n",
                                         clnt->cl_pathname, error);
+                        rpc_put_mount();
                         return error;
                 }
         }
@@ -147,6 +154,7 @@ rpc_new_client(struct rpc_xprt *xprt, char *servname,
         clnt->cl_vers     = version->number;
         clnt->cl_prot     = xprt->prot;
         clnt->cl_stats    = program->stats;
+        clnt->cl_metrics  = rpc_alloc_iostats(clnt);
         rpc_init_wait_queue(&clnt->cl_pmap_default.pm_bindwait, "bindwait");
 
         if (!clnt->cl_port)
@@ -175,7 +183,11 @@ rpc_new_client(struct rpc_xprt *xprt, char *servname,
         return clnt;
 
 out_no_auth:
-        rpc_rmdir(clnt->cl_pathname);
+        if (!IS_ERR(clnt->cl_dentry)) {
+                rpc_rmdir(clnt->cl_pathname);
+                dput(clnt->cl_dentry);
+                rpc_put_mount();
+        }
 out_no_path:
         if (clnt->cl_server != clnt->cl_inline_name)
                 kfree(clnt->cl_server);
@@ -240,11 +252,15 @@ rpc_clone_client(struct rpc_clnt *clnt)
         new->cl_autobind = 0;
         new->cl_oneshot = 0;
         new->cl_dead = 0;
+        if (!IS_ERR(new->cl_dentry)) {
+                dget(new->cl_dentry);
+                rpc_get_mount();
+        }
         rpc_init_rtt(&new->cl_rtt_default, clnt->cl_xprt->timeout.to_initval);
         if (new->cl_auth)
                 atomic_inc(&new->cl_auth->au_count);
         new->cl_pmap            = &new->cl_pmap_default;
-        rpc_init_wait_queue(&new->cl_pmap_default.pm_bindwait, "bindwait");
+        new->cl_metrics         = rpc_alloc_iostats(clnt);
         return new;
 out_no_clnt:
         printk(KERN_INFO "RPC: out of memory in %s\n", __FUNCTION__);
@@ -314,6 +330,12 @@ rpc_destroy_client(struct rpc_clnt *clnt)
         if (clnt->cl_server != clnt->cl_inline_name)
                 kfree(clnt->cl_server);
 out_free:
+        rpc_free_iostats(clnt->cl_metrics);
+        clnt->cl_metrics = NULL;
+        if (!IS_ERR(clnt->cl_dentry)) {
+                dput(clnt->cl_dentry);
+                rpc_put_mount();
+        }
         kfree(clnt);
         return 0;
 }
@@ -473,15 +495,16 @@ rpc_call_async(struct rpc_clnt *clnt, struct rpc_message *msg, int flags,
         int             status;
 
         /* If this client is slain all further I/O fails */
+        status = -EIO;
         if (clnt->cl_dead) 
-                return -EIO;
+                goto out_release;
 
         flags |= RPC_TASK_ASYNC;
 
         /* Create/initialize a new RPC task */
         status = -ENOMEM;
         if (!(task = rpc_new_task(clnt, flags, tk_ops, data)))
-                goto out;
+                goto out_release;
 
         /* Mask signals on GSS_AUTH upcalls */
         rpc_task_sigmask(task, &oldset);                
@@ -496,7 +519,10 @@ rpc_call_async(struct rpc_clnt *clnt, struct rpc_message *msg, int flags,
                 rpc_release_task(task);
 
         rpc_restore_sigmask(&oldset);           
-out:
+        return status;
+out_release:
+        if (tk_ops->rpc_release != NULL)
+                tk_ops->rpc_release(data);
         return status;
 }
 
@@ -993,6 +1019,8 @@ call_timeout(struct rpc_task *task)
         }
 
         dprintk("RPC: %4d call_timeout (major)\n", task->tk_pid);
+        task->tk_timeouts++;
+
         if (RPC_IS_SOFT(task)) {
                 printk(KERN_NOTICE "%s: server %s not responding, timed out\n",
                                 clnt->cl_protname, clnt->cl_server);
@@ -1045,6 +1073,11 @@ call_decode(struct rpc_task *task)
                 return;
         }
 
+        /*
+         * Ensure that we see all writes made by xprt_complete_rqst()
+         * before it changed req->rq_received.
+         */
+        smp_rmb();
         req->rq_rcv_buf.len = req->rq_private_buf.len;
 
         /* Check that the softirq receive buffer is valid */
@@ -1194,8 +1227,8 @@ call_verify(struct rpc_task *task)
                         task->tk_action = call_bind;
                         goto out_retry;
                 case RPC_AUTH_TOOWEAK:
-                        printk(KERN_NOTICE "call_verify: server requires stronger "
-                               "authentication.\n");
+                        printk(KERN_NOTICE "call_verify: server %s requires stronger "
+                               "authentication.\n", task->tk_client->cl_server);
                         break;
                 default:
                         printk(KERN_WARNING "call_verify: unknown auth error: %x\n", n);
diff --git a/net/sunrpc/pmap_clnt.c b/net/sunrpc/pmap_clnt.c
index 8139ce68e915..d25b054ec921 100644
--- a/net/sunrpc/pmap_clnt.c
+++ b/net/sunrpc/pmap_clnt.c
@@ -82,6 +82,7 @@ rpc_getport(struct rpc_task *task, struct rpc_clnt *clnt)
         rpc_call_setup(child, &msg, 0);
 
         /* ... and run the child task */
+        task->tk_xprt->stat.bind_count++;
         rpc_run_child(task, child, pmap_getport_done);
         return;
 
@@ -103,6 +104,11 @@ rpc_getport_external(struct sockaddr_in *sin, __u32 prog, __u32 vers, int prot)
                 .pm_prot        = prot,
                 .pm_port        = 0
         };
+        struct rpc_message msg = {
+                .rpc_proc       = &pmap_procedures[PMAP_GETPORT],
+                .rpc_argp       = &map,
+                .rpc_resp       = &map.pm_port,
+        };
         struct rpc_clnt *pmap_clnt;
         char            hostname[32];
         int             status;
@@ -116,7 +122,7 @@ rpc_getport_external(struct sockaddr_in *sin, __u32 prog, __u32 vers, int prot)
                 return PTR_ERR(pmap_clnt);
 
         /* Setup the call info struct */
-        status = rpc_call(pmap_clnt, PMAP_GETPORT, &map, &map.pm_port, 0);
+        status = rpc_call_sync(pmap_clnt, &msg, 0);
 
         if (status >= 0) {
                 if (map.pm_port != 0)
@@ -161,16 +167,27 @@ pmap_getport_done(struct rpc_task *task)
 int
 rpc_register(u32 prog, u32 vers, int prot, unsigned short port, int *okay)
 {
-        struct sockaddr_in      sin;
-        struct rpc_portmap      map;
+        struct sockaddr_in      sin = {
+                .sin_family     = AF_INET,
+                .sin_addr.s_addr = htonl(INADDR_LOOPBACK),
+        };
+        struct rpc_portmap      map = {
+                .pm_prog        = prog,
+                .pm_vers        = vers,
+                .pm_prot        = prot,
+                .pm_port        = port,
+        };
+        struct rpc_message msg = {
+                .rpc_proc       = &pmap_procedures[port ? PMAP_SET : PMAP_UNSET],
+                .rpc_argp       = &map,
+                .rpc_resp       = okay,
+        };
         struct rpc_clnt         *pmap_clnt;
         int error = 0;
 
         dprintk("RPC: registering (%d, %d, %d, %d) with portmapper.\n",
                         prog, vers, prot, port);
 
-        sin.sin_family = AF_INET;
-        sin.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
         pmap_clnt = pmap_create("localhost", &sin, IPPROTO_UDP, 1);
         if (IS_ERR(pmap_clnt)) {
                 error = PTR_ERR(pmap_clnt);
@@ -178,13 +195,7 @@ rpc_register(u32 prog, u32 vers, int prot, unsigned short port, int *okay)
                 return error;
         }
 
-        map.pm_prog = prog;
-        map.pm_vers = vers;
-        map.pm_prot = prot;
-        map.pm_port = port;
-
-        error = rpc_call(pmap_clnt, port? PMAP_SET : PMAP_UNSET,
-                                        &map, okay, 0);
+        error = rpc_call_sync(pmap_clnt, &msg, 0);
 
         if (error < 0) {
                 printk(KERN_WARNING
@@ -260,6 +271,8 @@ static struct rpc_procinfo	pmap_procedures[] = {
           .p_decode             = (kxdrproc_t) xdr_decode_bool,
           .p_bufsiz             = 4,
           .p_count              = 1,
+          .p_statidx            = PMAP_SET,
+          .p_name               = "SET",
         },
 [PMAP_UNSET] = {
           .p_proc               = PMAP_UNSET,
@@ -267,6 +280,8 @@ static struct rpc_procinfo	pmap_procedures[] = {
           .p_decode             = (kxdrproc_t) xdr_decode_bool,
           .p_bufsiz             = 4,
           .p_count              = 1,
+          .p_statidx            = PMAP_UNSET,
+          .p_name               = "UNSET",
         },
 [PMAP_GETPORT] = {
           .p_proc               = PMAP_GETPORT,
@@ -274,6 +289,8 @@ static struct rpc_procinfo	pmap_procedures[] = {
           .p_decode             = (kxdrproc_t) xdr_decode_port,
           .p_bufsiz             = 4,
           .p_count              = 1,
+          .p_statidx            = PMAP_GETPORT,
+          .p_name               = "GETPORT",
         },
 };
 
diff --git a/net/sunrpc/rpc_pipe.c b/net/sunrpc/rpc_pipe.c
index ad9d9fc4e734..aa4158be9900 100644
--- a/net/sunrpc/rpc_pipe.c
+++ b/net/sunrpc/rpc_pipe.c
@@ -91,7 +91,8 @@ rpc_queue_upcall(struct inode *inode, struct rpc_pipe_msg *msg)
                 res = 0;
         } else if (rpci->flags & RPC_PIPE_WAIT_FOR_OPEN) {
                 if (list_empty(&rpci->pipe))
-                        schedule_delayed_work(&rpci->queue_timeout,
+                        queue_delayed_work(rpciod_workqueue,
+                                        &rpci->queue_timeout,
                                         RPC_UPCALL_TIMEOUT);
                 list_add_tail(&msg->list, &rpci->pipe);
                 rpci->pipelen += msg->len;
@@ -132,7 +133,7 @@ rpc_close_pipes(struct inode *inode)
                 if (ops->release_pipe)
                         ops->release_pipe(inode);
                 cancel_delayed_work(&rpci->queue_timeout);
-                flush_scheduled_work();
+                flush_workqueue(rpciod_workqueue);
         }
         rpc_inode_setowner(inode, NULL);
         mutex_unlock(&inode->i_mutex);
@@ -434,14 +435,17 @@ static struct rpc_filelist authfiles[] = {
         },
 };
 
-static int
-rpc_get_mount(void)
+struct vfsmount *rpc_get_mount(void)
 {
-        return simple_pin_fs("rpc_pipefs", &rpc_mount, &rpc_mount_count);
+        int err;
+
+        err = simple_pin_fs("rpc_pipefs", &rpc_mount, &rpc_mount_count);
+        if (err != 0)
+                return ERR_PTR(err);
+        return rpc_mount;
 }
 
-static void
-rpc_put_mount(void)
+void rpc_put_mount(void)
 {
         simple_release_fs(&rpc_mount, &rpc_mount_count);
 }
@@ -451,12 +455,13 @@ rpc_lookup_parent(char *path, struct nameidata *nd)
 {
         if (path[0] == '\0')
                 return -ENOENT;
-        if (rpc_get_mount()) {
+        nd->mnt = rpc_get_mount();
+        if (IS_ERR(nd->mnt)) {
                 printk(KERN_WARNING "%s: %s failed to mount "
                                "pseudofilesystem \n", __FILE__, __FUNCTION__);
-                return -ENODEV;
+                return PTR_ERR(nd->mnt);
         }
-        nd->mnt = mntget(rpc_mount);
+        mntget(nd->mnt);
         nd->dentry = dget(rpc_mount->mnt_root);
         nd->last_type = LAST_ROOT;
         nd->flags = LOOKUP_PARENT;
@@ -593,7 +598,6 @@ __rpc_mkdir(struct inode *dir, struct dentry *dentry)
         d_instantiate(dentry, inode);
         dir->i_nlink++;
         inode_dir_notify(dir, DN_CREATE);
-        rpc_get_mount();
         return 0;
 out_err:
         printk(KERN_WARNING "%s: %s failed to allocate inode for dentry %s\n",
@@ -614,7 +618,6 @@ __rpc_rmdir(struct inode *dir, struct dentry *dentry)
         if (!error) {
                 inode_dir_notify(dir, DN_DELETE);
                 d_drop(dentry);
-                rpc_put_mount();
         }
         return 0;
 }
@@ -668,7 +671,7 @@ rpc_mkdir(char *path, struct rpc_clnt *rpc_client)
 out:
         mutex_unlock(&dir->i_mutex);
         rpc_release_path(&nd);
-        return dentry;
+        return dget(dentry);
 err_depopulate:
         rpc_depopulate(dentry);
         __rpc_rmdir(dir, dentry);
@@ -732,7 +735,7 @@ rpc_mkpipe(char *path, void *private, struct rpc_pipe_ops *ops, int flags)
 out:
         mutex_unlock(&dir->i_mutex);
         rpc_release_path(&nd);
-        return dentry;
+        return dget(dentry);
 err_dput:
         dput(dentry);
         dentry = ERR_PTR(-ENOMEM);
diff --git a/net/sunrpc/sched.c b/net/sunrpc/sched.c
index dff07795bd16..b9969b91a9f7 100644
--- a/net/sunrpc/sched.c
+++ b/net/sunrpc/sched.c
@@ -65,7 +65,7 @@ static LIST_HEAD(all_tasks);
  */
 static DEFINE_MUTEX(rpciod_mutex);
 static unsigned int             rpciod_users;
-static struct workqueue_struct *rpciod_workqueue;
+struct workqueue_struct *rpciod_workqueue;
 
 /*
  * Spinlock for other critical sections of code.
@@ -182,6 +182,7 @@ static void __rpc_add_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *
         else
                 list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
         task->u.tk_wait.rpc_waitq = queue;
+        queue->qlen++;
         rpc_set_queued(task);
 
         dprintk("RPC: %4d added to queue %p \"%s\"\n",
@@ -216,6 +217,7 @@ static void __rpc_remove_wait_queue(struct rpc_task *task)
                 __rpc_remove_wait_queue_priority(task);
         else
                 list_del(&task->u.tk_wait.list);
+        queue->qlen--;
         dprintk("RPC: %4d removed from queue %p \"%s\"\n",
                                 task->tk_pid, queue, rpc_qname(queue));
 }
@@ -816,6 +818,9 @@ void rpc_init_task(struct rpc_task *task, struct rpc_clnt *clnt, int flags, cons
 
         BUG_ON(task->tk_ops == NULL);
 
+        /* starting timestamp */
+        task->tk_start = jiffies;
+
         dprintk("RPC: %4d new task procpid %d\n", task->tk_pid,
                                 current->pid);
 }
@@ -917,8 +922,11 @@ struct rpc_task *rpc_run_task(struct rpc_clnt *clnt, int flags,
 {
         struct rpc_task *task;
         task = rpc_new_task(clnt, flags, ops, data);
-        if (task == NULL)
+        if (task == NULL) {
+                if (ops->rpc_release != NULL)
+                        ops->rpc_release(data);
                 return ERR_PTR(-ENOMEM);
+        }
         atomic_inc(&task->tk_count);
         rpc_execute(task);
         return task;
diff --git a/net/sunrpc/stats.c b/net/sunrpc/stats.c
index 4979f226e285..790941e8af4d 100644
--- a/net/sunrpc/stats.c
+++ b/net/sunrpc/stats.c
@@ -21,6 +21,7 @@
 #include <linux/seq_file.h>
 #include <linux/sunrpc/clnt.h>
 #include <linux/sunrpc/svcsock.h>
+#include <linux/sunrpc/metrics.h>
 
 #define RPCDBG_FACILITY RPCDBG_MISC
 
@@ -106,6 +107,120 @@ void svc_seq_show(struct seq_file *seq, const struct svc_stat *statp) {
         }
 }
 
+/**
+ * rpc_alloc_iostats - allocate an rpc_iostats structure
+ * @clnt: RPC program, version, and xprt
+ *
+ */
+struct rpc_iostats *rpc_alloc_iostats(struct rpc_clnt *clnt)
+{
+        unsigned int ops = clnt->cl_maxproc;
+        size_t size = ops * sizeof(struct rpc_iostats);
+        struct rpc_iostats *new;
+
+        new = kmalloc(size, GFP_KERNEL);
+        if (new)
+                memset(new, 0 , size);
+        return new;
+}
+EXPORT_SYMBOL(rpc_alloc_iostats);
+
+/**
+ * rpc_free_iostats - release an rpc_iostats structure
+ * @stats: doomed rpc_iostats structure
+ *
+ */
+void rpc_free_iostats(struct rpc_iostats *stats)
+{
+        kfree(stats);
+}
+EXPORT_SYMBOL(rpc_free_iostats);
+
+/**
+ * rpc_count_iostats - tally up per-task stats
+ * @task: completed rpc_task
+ *
+ * Relies on the caller for serialization.
+ */
+void rpc_count_iostats(struct rpc_task *task)
+{
+        struct rpc_rqst *req = task->tk_rqstp;
+        struct rpc_iostats *stats = task->tk_client->cl_metrics;
+        struct rpc_iostats *op_metrics;
+        long rtt, execute, queue;
+
+        if (!stats || !req)
+                return;
+        op_metrics = &stats[task->tk_msg.rpc_proc->p_statidx];
+
+        op_metrics->om_ops++;
+        op_metrics->om_ntrans += req->rq_ntrans;
+        op_metrics->om_timeouts += task->tk_timeouts;
+
+        op_metrics->om_bytes_sent += task->tk_bytes_sent;
+        op_metrics->om_bytes_recv += req->rq_received;
+
+        queue = (long)req->rq_xtime - task->tk_start;
+        if (queue < 0)
+                queue = -queue;
+        op_metrics->om_queue += queue;
+
+        rtt = task->tk_rtt;
+        if (rtt < 0)
+                rtt = -rtt;
+        op_metrics->om_rtt += rtt;
+
+        execute = (long)jiffies - task->tk_start;
+        if (execute < 0)
+                execute = -execute;
+        op_metrics->om_execute += execute;
+}
+
+void _print_name(struct seq_file *seq, unsigned int op, struct rpc_procinfo *procs)
+{
+        if (procs[op].p_name)
+                seq_printf(seq, "\t%12s: ", procs[op].p_name);
+        else if (op == 0)
+                seq_printf(seq, "\t        NULL: ");
+        else
+                seq_printf(seq, "\t%12u: ", op);
+}
+
+#define MILLISECS_PER_JIFFY     (1000 / HZ)
+
+void rpc_print_iostats(struct seq_file *seq, struct rpc_clnt *clnt)
+{
+        struct rpc_iostats *stats = clnt->cl_metrics;
+        struct rpc_xprt *xprt = clnt->cl_xprt;
+        unsigned int op, maxproc = clnt->cl_maxproc;
+
+        if (!stats)
+                return;
+
+        seq_printf(seq, "\tRPC iostats version: %s  ", RPC_IOSTATS_VERS);
+        seq_printf(seq, "p/v: %u/%u (%s)\n",
+                        clnt->cl_prog, clnt->cl_vers, clnt->cl_protname);
+
+        if (xprt)
+                xprt->ops->print_stats(xprt, seq);
+
+        seq_printf(seq, "\tper-op statistics\n");
+        for (op = 0; op < maxproc; op++) {
+                struct rpc_iostats *metrics = &stats[op];
+                _print_name(seq, op, clnt->cl_procinfo);
+                seq_printf(seq, "%lu %lu %lu %Lu %Lu %Lu %Lu %Lu\n",
+                                metrics->om_ops,
+                                metrics->om_ntrans,
+                                metrics->om_timeouts,
+                                metrics->om_bytes_sent,
+                                metrics->om_bytes_recv,
+                                metrics->om_queue * MILLISECS_PER_JIFFY,
+                                metrics->om_rtt * MILLISECS_PER_JIFFY,
+                                metrics->om_execute * MILLISECS_PER_JIFFY);
+        }
+}
+EXPORT_SYMBOL(rpc_print_iostats);
+
 /*
  * Register/unregister RPC proc files
  */
diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index 8ff2c8acb223..4dd5b3cfe754 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -44,13 +44,13 @@
 #include <linux/random.h>
 
 #include <linux/sunrpc/clnt.h>
+#include <linux/sunrpc/metrics.h>
 
 /*
  * Local variables
  */
 
 #ifdef RPC_DEBUG
-# undef  RPC_DEBUG_DATA
 # define RPCDBG_FACILITY        RPCDBG_XPRT
 #endif
 
@@ -548,6 +548,7 @@ void xprt_connect(struct rpc_task *task)
 
                 task->tk_timeout = xprt->connect_timeout;
                 rpc_sleep_on(&xprt->pending, task, xprt_connect_status, NULL);
+                xprt->stat.connect_start = jiffies;
                 xprt->ops->connect(task);
         }
         return;
@@ -558,6 +559,8 @@ static void xprt_connect_status(struct rpc_task *task)
         struct rpc_xprt *xprt = task->tk_xprt;
 
         if (task->tk_status >= 0) {
+                xprt->stat.connect_count++;
+                xprt->stat.connect_time += (long)jiffies - xprt->stat.connect_start;
                 dprintk("RPC: %4d xprt_connect_status: connection established\n",
                                 task->tk_pid);
                 return;
@@ -601,16 +604,14 @@ static void xprt_connect_status(struct rpc_task *task)
 struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, u32 xid)
 {
         struct list_head *pos;
-        struct rpc_rqst *req = NULL;
 
         list_for_each(pos, &xprt->recv) {
                 struct rpc_rqst *entry = list_entry(pos, struct rpc_rqst, rq_list);
-                if (entry->rq_xid == xid) {
-                        req = entry;
-                        break;
-                }
+                if (entry->rq_xid == xid)
+                        return entry;
         }
-        return req;
+        xprt->stat.bad_xids++;
+        return NULL;
 }
 
 /**
@@ -646,7 +647,12 @@ void xprt_complete_rqst(struct rpc_task *task, int copied)
         dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
                         task->tk_pid, ntohl(req->rq_xid), copied);
 
+        task->tk_xprt->stat.recvs++;
+        task->tk_rtt = (long)jiffies - req->rq_xtime;
+
         list_del_init(&req->rq_list);
+        /* Ensure all writes are done before we update req->rq_received */
+        smp_wmb();
         req->rq_received = req->rq_private_buf.len = copied;
         rpc_wake_up_task(task);
 }
@@ -723,7 +729,6 @@ void xprt_transmit(struct rpc_task *task)
 
         dprintk("RPC: %4d xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
 
-        smp_rmb();
         if (!req->rq_received) {
                 if (list_empty(&req->rq_list)) {
                         spin_lock_bh(&xprt->transport_lock);
@@ -744,12 +749,19 @@ void xprt_transmit(struct rpc_task *task)
         if (status == 0) {
                 dprintk("RPC: %4d xmit complete\n", task->tk_pid);
                 spin_lock_bh(&xprt->transport_lock);
+
                 xprt->ops->set_retrans_timeout(task);
+
+                xprt->stat.sends++;
+                xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs;
+                xprt->stat.bklog_u += xprt->backlog.qlen;
+
                 /* Don't race with disconnect */
                 if (!xprt_connected(xprt))
                         task->tk_status = -ENOTCONN;
                 else if (!req->rq_received)
                         rpc_sleep_on(&xprt->pending, task, NULL, xprt_timer);
+
                 xprt->ops->release_xprt(xprt, task);
                 spin_unlock_bh(&xprt->transport_lock);
                 return;
@@ -848,6 +860,7 @@ void xprt_release(struct rpc_task *task)
 
         if (!(req = task->tk_rqstp))
                 return;
+        rpc_count_iostats(task);
         spin_lock_bh(&xprt->transport_lock);
         xprt->ops->release_xprt(xprt, task);
         if (xprt->ops->release_request)
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index c458f8d1d6d1..4b4e7dfdff14 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -382,6 +382,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;
                 if (likely(req->rq_bytes_sent >= req->rq_slen)) {
                         req->rq_bytes_sent = 0;
                         return 0;
@@ -1114,6 +1115,8 @@ static void xs_tcp_connect_worker(void *args)
         }
 
         /* Tell the socket layer to start connecting... */
+        xprt->stat.connect_count++;
+        xprt->stat.connect_start = jiffies;
         status = sock->ops->connect(sock, (struct sockaddr *) &xprt->addr,
                         sizeof(xprt->addr), O_NONBLOCK);
         dprintk("RPC: %p  connect status %d connected %d sock state %d\n",
@@ -1177,6 +1180,50 @@ static void xs_connect(struct rpc_task *task)
         }
 }
 
+/**
+ * xs_udp_print_stats - display UDP socket-specifc stats
+ * @xprt: rpc_xprt struct containing statistics
+ * @seq: output file
+ *
+ */
+static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
+{
+        seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
+                        xprt->port,
+                        xprt->stat.bind_count,
+                        xprt->stat.sends,
+                        xprt->stat.recvs,
+                        xprt->stat.bad_xids,
+                        xprt->stat.req_u,
+                        xprt->stat.bklog_u);
+}
+
+/**
+ * xs_tcp_print_stats - display TCP socket-specifc stats
+ * @xprt: rpc_xprt struct containing statistics
+ * @seq: output file
+ *
+ */
+static void xs_tcp_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:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n",
+                        xprt->port,
+                        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);
+}
+
 static struct rpc_xprt_ops xs_udp_ops = {
         .set_buffer_size        = xs_udp_set_buffer_size,
         .reserve_xprt           = xprt_reserve_xprt_cong,
@@ -1191,6 +1238,7 @@ static struct rpc_xprt_ops xs_udp_ops = {
         .release_request        = xprt_release_rqst_cong,
         .close                  = xs_close,
         .destroy                = xs_destroy,
+        .print_stats            = xs_udp_print_stats,
 };
 
 static struct rpc_xprt_ops xs_tcp_ops = {
@@ -1204,6 +1252,7 @@ static struct rpc_xprt_ops xs_tcp_ops = {
         .set_retrans_timeout    = xprt_set_retrans_timeout_def,
         .close                  = xs_close,
         .destroy                = xs_destroy,
+        .print_stats            = xs_tcp_print_stats,
 };
 
 /**