Merge branch 'nfs-for-2.6.38' of git://git.linux-nfs.org/projects/trondmy/nfs-2.6

* 'nfs-for-2.6.38' of git://git.linux-nfs.org/projects/trondmy/nfs-2.6: (89 commits)
  NFS fix the setting of exchange id flag
  NFS: Don't use vm_map_ram() in readdir
  NFSv4: Ensure continued open and lockowner name uniqueness
  NFS: Move cl_delegations to the nfs_server struct
  NFS: Introduce nfs_detach_delegations()
  NFS: Move cl_state_owners and related fields to the nfs_server struct
  NFS: Allow walking nfs_client.cl_superblocks list outside client.c
  pnfs: layout roc code
  pnfs: update nfs4_callback_recallany to handle layouts
  pnfs: add CB_LAYOUTRECALL handling
  pnfs: CB_LAYOUTRECALL xdr code
  pnfs: change lo refcounting to atomic_t
  pnfs: check that partial LAYOUTGET return is ignored
  pnfs: add layout to client list before sending rpc
  pnfs: serialize LAYOUTGET(openstateid)
  pnfs: layoutget rpc code cleanup
  pnfs: change how lsegs are removed from layout list
  pnfs: change layout state seqlock to a spinlock
  pnfs: add prefix to struct pnfs_layout_hdr fields
  pnfs: add prefix to struct pnfs_layout_segment fields
  ...

9 files changed, 345 insertions, 196 deletions

diff --git a/net/sunrpc/auth.c b/net/sunrpc/auth.c
index afe67849269f..67e31276682a 100644
--- a/net/sunrpc/auth.c
+++ b/net/sunrpc/auth.c
@@ -563,8 +563,17 @@ rpcauth_checkverf(struct rpc_task *task, __be32 *p)
         return cred->cr_ops->crvalidate(task, p);
 }
 
+static void rpcauth_wrap_req_encode(kxdreproc_t encode, struct rpc_rqst *rqstp,
+                                   __be32 *data, void *obj)
+{
+        struct xdr_stream xdr;
+
+        xdr_init_encode(&xdr, &rqstp->rq_snd_buf, data);
+        encode(rqstp, &xdr, obj);
+}
+
 int
-rpcauth_wrap_req(struct rpc_task *task, kxdrproc_t encode, void *rqstp,
+rpcauth_wrap_req(struct rpc_task *task, kxdreproc_t encode, void *rqstp,
                 __be32 *data, void *obj)
 {
         struct rpc_cred *cred = task->tk_rqstp->rq_cred;
@@ -574,11 +583,22 @@ rpcauth_wrap_req(struct rpc_task *task, kxdrproc_t encode, void *rqstp,
         if (cred->cr_ops->crwrap_req)
                 return cred->cr_ops->crwrap_req(task, encode, rqstp, data, obj);
         /* By default, we encode the arguments normally. */
-        return encode(rqstp, data, obj);
+        rpcauth_wrap_req_encode(encode, rqstp, data, obj);
+        return 0;
+}
+
+static int
+rpcauth_unwrap_req_decode(kxdrdproc_t decode, struct rpc_rqst *rqstp,
+                          __be32 *data, void *obj)
+{
+        struct xdr_stream xdr;
+
+        xdr_init_decode(&xdr, &rqstp->rq_rcv_buf, data);
+        return decode(rqstp, &xdr, obj);
 }
 
 int
-rpcauth_unwrap_resp(struct rpc_task *task, kxdrproc_t decode, void *rqstp,
+rpcauth_unwrap_resp(struct rpc_task *task, kxdrdproc_t decode, void *rqstp,
                 __be32 *data, void *obj)
 {
         struct rpc_cred *cred = task->tk_rqstp->rq_cred;
@@ -589,7 +609,7 @@ rpcauth_unwrap_resp(struct rpc_task *task, kxdrproc_t decode, void *rqstp,
                 return cred->cr_ops->crunwrap_resp(task, decode, rqstp,
                                                    data, obj);
         /* By default, we decode the arguments normally. */
-        return decode(rqstp, data, obj);
+        return rpcauth_unwrap_req_decode(decode, rqstp, data, obj);
 }
 
 int
diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index 3835ce35e224..45dbf1521b9a 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -1231,9 +1231,19 @@ out_bad:
         return NULL;
 }
 
+static void gss_wrap_req_encode(kxdreproc_t encode, struct rpc_rqst *rqstp,
+                                __be32 *p, void *obj)
+{
+        struct xdr_stream xdr;
+
+        xdr_init_encode(&xdr, &rqstp->rq_snd_buf, p);
+        encode(rqstp, &xdr, obj);
+}
+
 static inline int
 gss_wrap_req_integ(struct rpc_cred *cred, struct gss_cl_ctx *ctx,
-                kxdrproc_t encode, struct rpc_rqst *rqstp, __be32 *p, void *obj)
+                   kxdreproc_t encode, struct rpc_rqst *rqstp,
+                   __be32 *p, void *obj)
 {
         struct xdr_buf  *snd_buf = &rqstp->rq_snd_buf;
         struct xdr_buf  integ_buf;
@@ -1249,9 +1259,7 @@ gss_wrap_req_integ(struct rpc_cred *cred, struct gss_cl_ctx *ctx,
         offset = (u8 *)p - (u8 *)snd_buf->head[0].iov_base;
         *p++ = htonl(rqstp->rq_seqno);
 
-        status = encode(rqstp, p, obj);
-        if (status)
-                return status;
+        gss_wrap_req_encode(encode, rqstp, p, obj);
 
         if (xdr_buf_subsegment(snd_buf, &integ_buf,
                                 offset, snd_buf->len - offset))
@@ -1325,7 +1333,8 @@ out:
 
 static inline int
 gss_wrap_req_priv(struct rpc_cred *cred, struct gss_cl_ctx *ctx,
-                kxdrproc_t encode, struct rpc_rqst *rqstp, __be32 *p, void *obj)
+                  kxdreproc_t encode, struct rpc_rqst *rqstp,
+                  __be32 *p, void *obj)
 {
         struct xdr_buf  *snd_buf = &rqstp->rq_snd_buf;
         u32             offset;
@@ -1342,9 +1351,7 @@ gss_wrap_req_priv(struct rpc_cred *cred, struct gss_cl_ctx *ctx,
         offset = (u8 *)p - (u8 *)snd_buf->head[0].iov_base;
         *p++ = htonl(rqstp->rq_seqno);
 
-        status = encode(rqstp, p, obj);
-        if (status)
-                return status;
+        gss_wrap_req_encode(encode, rqstp, p, obj);
 
         status = alloc_enc_pages(rqstp);
         if (status)
@@ -1394,7 +1401,7 @@ gss_wrap_req_priv(struct rpc_cred *cred, struct gss_cl_ctx *ctx,
 
 static int
 gss_wrap_req(struct rpc_task *task,
-             kxdrproc_t encode, void *rqstp, __be32 *p, void *obj)
+             kxdreproc_t encode, void *rqstp, __be32 *p, void *obj)
 {
         struct rpc_cred *cred = task->tk_rqstp->rq_cred;
         struct gss_cred *gss_cred = container_of(cred, struct gss_cred,
@@ -1407,12 +1414,14 @@ gss_wrap_req(struct rpc_task *task,
                 /* The spec seems a little ambiguous here, but I think that not
                  * wrapping context destruction requests makes the most sense.
                  */
-                status = encode(rqstp, p, obj);
+                gss_wrap_req_encode(encode, rqstp, p, obj);
+                status = 0;
                 goto out;
         }
         switch (gss_cred->gc_service) {
                 case RPC_GSS_SVC_NONE:
-                        status = encode(rqstp, p, obj);
+                        gss_wrap_req_encode(encode, rqstp, p, obj);
+                        status = 0;
                         break;
                 case RPC_GSS_SVC_INTEGRITY:
                         status = gss_wrap_req_integ(cred, ctx, encode,
@@ -1494,10 +1503,19 @@ gss_unwrap_resp_priv(struct rpc_cred *cred, struct gss_cl_ctx *ctx,
         return 0;
 }
 
+static int
+gss_unwrap_req_decode(kxdrdproc_t decode, struct rpc_rqst *rqstp,
+                      __be32 *p, void *obj)
+{
+        struct xdr_stream xdr;
+
+        xdr_init_decode(&xdr, &rqstp->rq_rcv_buf, p);
+        return decode(rqstp, &xdr, obj);
+}
 
 static int
 gss_unwrap_resp(struct rpc_task *task,
-                kxdrproc_t decode, void *rqstp, __be32 *p, void *obj)
+                kxdrdproc_t decode, void *rqstp, __be32 *p, void *obj)
 {
         struct rpc_cred *cred = task->tk_rqstp->rq_cred;
         struct gss_cred *gss_cred = container_of(cred, struct gss_cred,
@@ -1528,7 +1546,7 @@ gss_unwrap_resp(struct rpc_task *task,
         cred->cr_auth->au_rslack = cred->cr_auth->au_verfsize + (p - savedp)
                                                 + (savedlen - head->iov_len);
 out_decode:
-        status = decode(rqstp, p, obj);
+        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,
diff --git a/net/sunrpc/bc_svc.c b/net/sunrpc/bc_svc.c
index 7dcfe0cc3500..1dd1a6890007 100644
--- a/net/sunrpc/bc_svc.c
+++ b/net/sunrpc/bc_svc.c
@@ -59,8 +59,8 @@ int bc_send(struct rpc_rqst *req)
                 ret = task->tk_status;
                 rpc_put_task(task);
         }
-        return ret;
         dprintk("RPC:       bc_send ret= %d\n", ret);
+        return ret;
 }
 
 #endif /* CONFIG_NFS_V4_1 */
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index 92ce94f5146b..57d344cf2256 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -1095,7 +1095,7 @@ static void
 rpc_xdr_encode(struct rpc_task *task)
 {
         struct rpc_rqst *req = task->tk_rqstp;
-        kxdrproc_t      encode;
+        kxdreproc_t     encode;
         __be32          *p;
 
         dprint_status(task);
@@ -1535,7 +1535,7 @@ call_decode(struct rpc_task *task)
 {
         struct rpc_clnt *clnt = task->tk_client;
         struct rpc_rqst *req = task->tk_rqstp;
-        kxdrproc_t      decode = task->tk_msg.rpc_proc->p_decode;
+        kxdrdproc_t     decode = task->tk_msg.rpc_proc->p_decode;
         __be32          *p;
 
         dprintk("RPC: %5u call_decode (status %d)\n",
@@ -1776,12 +1776,11 @@ out_overflow:
         goto out_garbage;
 }
 
-static int rpcproc_encode_null(void *rqstp, __be32 *data, void *obj)
+static void rpcproc_encode_null(void *rqstp, struct xdr_stream *xdr, void *obj)
 {
-        return 0;
 }
 
-static int rpcproc_decode_null(void *rqstp, __be32 *data, void *obj)
+static int rpcproc_decode_null(void *rqstp, struct xdr_stream *xdr, void *obj)
 {
         return 0;
 }
@@ -1830,23 +1829,15 @@ static void rpc_show_task(const struct rpc_clnt *clnt,
                           const struct rpc_task *task)
 {
         const char *rpc_waitq = "none";
-        char *p, action[KSYM_SYMBOL_LEN];
 
         if (RPC_IS_QUEUED(task))
                 rpc_waitq = rpc_qname(task->tk_waitqueue);
 
-        /* map tk_action pointer to a function name; then trim off
-         * the "+0x0 [sunrpc]" */
-        sprint_symbol(action, (unsigned long)task->tk_action);
-        p = strchr(action, '+');
-        if (p)
-                *p = '\0';
-
-        printk(KERN_INFO "%5u %04x %6d %8p %8p %8ld %8p %sv%u %s a:%s q:%s\n",
+        printk(KERN_INFO "%5u %04x %6d %8p %8p %8ld %8p %sv%u %s a:%ps q:%s\n",
                 task->tk_pid, task->tk_flags, task->tk_status,
                 clnt, task->tk_rqstp, task->tk_timeout, task->tk_ops,
                 clnt->cl_protname, clnt->cl_vers, rpc_proc_name(task),
-                action, rpc_waitq);
+                task->tk_action, rpc_waitq);
 }
 
 void rpc_show_tasks(void)
diff --git a/net/sunrpc/rpc_pipe.c b/net/sunrpc/rpc_pipe.c
index 09f01f41e55a..72bc53683965 100644
--- a/net/sunrpc/rpc_pipe.c
+++ b/net/sunrpc/rpc_pipe.c
@@ -474,7 +474,7 @@ static int __rpc_create_common(struct inode *dir, struct dentry *dentry,
 {
         struct inode *inode;
 
-        BUG_ON(!d_unhashed(dentry));
+        d_drop(dentry);
         inode = rpc_get_inode(dir->i_sb, mode);
         if (!inode)
                 goto out_err;
diff --git a/net/sunrpc/rpcb_clnt.c b/net/sunrpc/rpcb_clnt.c
index fa6d7ca2c851..c652e4cc9fe9 100644
--- a/net/sunrpc/rpcb_clnt.c
+++ b/net/sunrpc/rpcb_clnt.c
@@ -57,10 +57,6 @@ enum {
         RPCBPROC_GETSTAT,
 };
 
-#define RPCB_HIGHPROC_2         RPCBPROC_CALLIT
-#define RPCB_HIGHPROC_3         RPCBPROC_TADDR2UADDR
-#define RPCB_HIGHPROC_4         RPCBPROC_GETSTAT
-
 /*
  * r_owner
  *
@@ -693,46 +689,37 @@ static void rpcb_getport_done(struct rpc_task *child, void *data)
  * XDR functions for rpcbind
  */
 
-static int rpcb_enc_mapping(struct rpc_rqst *req, __be32 *p,
-                            const struct rpcbind_args *rpcb)
+static void rpcb_enc_mapping(struct rpc_rqst *req, struct xdr_stream *xdr,
+                             const struct rpcbind_args *rpcb)
 {
         struct rpc_task *task = req->rq_task;
-        struct xdr_stream xdr;
+        __be32 *p;
 
         dprintk("RPC: %5u encoding PMAP_%s call (%u, %u, %d, %u)\n",
                         task->tk_pid, task->tk_msg.rpc_proc->p_name,
                         rpcb->r_prog, rpcb->r_vers, rpcb->r_prot, rpcb->r_port);
 
-        xdr_init_encode(&xdr, &req->rq_snd_buf, p);
-
-        p = xdr_reserve_space(&xdr, sizeof(__be32) * RPCB_mappingargs_sz);
-        if (unlikely(p == NULL))
-                return -EIO;
-
-        *p++ = htonl(rpcb->r_prog);
-        *p++ = htonl(rpcb->r_vers);
-        *p++ = htonl(rpcb->r_prot);
-        *p   = htonl(rpcb->r_port);
-
-        return 0;
+        p = xdr_reserve_space(xdr, RPCB_mappingargs_sz << 2);
+        *p++ = cpu_to_be32(rpcb->r_prog);
+        *p++ = cpu_to_be32(rpcb->r_vers);
+        *p++ = cpu_to_be32(rpcb->r_prot);
+        *p   = cpu_to_be32(rpcb->r_port);
 }
 
-static int rpcb_dec_getport(struct rpc_rqst *req, __be32 *p,
+static int rpcb_dec_getport(struct rpc_rqst *req, struct xdr_stream *xdr,
                             struct rpcbind_args *rpcb)
 {
         struct rpc_task *task = req->rq_task;
-        struct xdr_stream xdr;
         unsigned long port;
-
-        xdr_init_decode(&xdr, &req->rq_rcv_buf, p);
+        __be32 *p;
 
         rpcb->r_port = 0;
 
-        p = xdr_inline_decode(&xdr, sizeof(__be32));
+        p = xdr_inline_decode(xdr, 4);
         if (unlikely(p == NULL))
                 return -EIO;
 
-        port = ntohl(*p);
+        port = be32_to_cpup(p);
         dprintk("RPC: %5u PMAP_%s result: %lu\n", task->tk_pid,
                         task->tk_msg.rpc_proc->p_name, port);
         if (unlikely(port > USHRT_MAX))
@@ -742,20 +729,18 @@ static int rpcb_dec_getport(struct rpc_rqst *req, __be32 *p,
         return 0;
 }
 
-static int rpcb_dec_set(struct rpc_rqst *req, __be32 *p,
+static int rpcb_dec_set(struct rpc_rqst *req, struct xdr_stream *xdr,
                         unsigned int *boolp)
 {
         struct rpc_task *task = req->rq_task;
-        struct xdr_stream xdr;
-
-        xdr_init_decode(&xdr, &req->rq_rcv_buf, p);
+        __be32 *p;
 
-        p = xdr_inline_decode(&xdr, sizeof(__be32));
+        p = xdr_inline_decode(xdr, 4);
         if (unlikely(p == NULL))
                 return -EIO;
 
         *boolp = 0;
-        if (*p)
+        if (*p != xdr_zero)
                 *boolp = 1;
 
         dprintk("RPC: %5u RPCB_%s call %s\n",
@@ -764,73 +749,53 @@ static int rpcb_dec_set(struct rpc_rqst *req, __be32 *p,
         return 0;
 }
 
-static int encode_rpcb_string(struct xdr_stream *xdr, const char *string,
-                                const u32 maxstrlen)
+static void encode_rpcb_string(struct xdr_stream *xdr, const char *string,
+                               const u32 maxstrlen)
 {
-        u32 len;
         __be32 *p;
+        u32 len;
 
-        if (unlikely(string == NULL))
-                return -EIO;
         len = strlen(string);
-        if (unlikely(len > maxstrlen))
-                return -EIO;
-
-        p = xdr_reserve_space(xdr, sizeof(__be32) + len);
-        if (unlikely(p == NULL))
-                return -EIO;
+        BUG_ON(len > maxstrlen);
+        p = xdr_reserve_space(xdr, 4 + len);
         xdr_encode_opaque(p, string, len);
-
-        return 0;
 }
 
-static int rpcb_enc_getaddr(struct rpc_rqst *req, __be32 *p,
-                            const struct rpcbind_args *rpcb)
+static void rpcb_enc_getaddr(struct rpc_rqst *req, struct xdr_stream *xdr,
+                             const struct rpcbind_args *rpcb)
 {
         struct rpc_task *task = req->rq_task;
-        struct xdr_stream xdr;
+        __be32 *p;
 
         dprintk("RPC: %5u encoding RPCB_%s call (%u, %u, '%s', '%s')\n",
                         task->tk_pid, task->tk_msg.rpc_proc->p_name,
                         rpcb->r_prog, rpcb->r_vers,
                         rpcb->r_netid, rpcb->r_addr);
 
-        xdr_init_encode(&xdr, &req->rq_snd_buf, p);
-
-        p = xdr_reserve_space(&xdr,
-                        sizeof(__be32) * (RPCB_program_sz + RPCB_version_sz));
-        if (unlikely(p == NULL))
-                return -EIO;
-        *p++ = htonl(rpcb->r_prog);
-        *p = htonl(rpcb->r_vers);
-
-        if (encode_rpcb_string(&xdr, rpcb->r_netid, RPCBIND_MAXNETIDLEN))
-                return -EIO;
-        if (encode_rpcb_string(&xdr, rpcb->r_addr, RPCBIND_MAXUADDRLEN))
-                return -EIO;
-        if (encode_rpcb_string(&xdr, rpcb->r_owner, RPCB_MAXOWNERLEN))
-                return -EIO;
+        p = xdr_reserve_space(xdr, (RPCB_program_sz + RPCB_version_sz) << 2);
+        *p++ = cpu_to_be32(rpcb->r_prog);
+        *p = cpu_to_be32(rpcb->r_vers);
 
-        return 0;
+        encode_rpcb_string(xdr, rpcb->r_netid, RPCBIND_MAXNETIDLEN);
+        encode_rpcb_string(xdr, rpcb->r_addr, RPCBIND_MAXUADDRLEN);
+        encode_rpcb_string(xdr, rpcb->r_owner, RPCB_MAXOWNERLEN);
 }
 
-static int rpcb_dec_getaddr(struct rpc_rqst *req, __be32 *p,
+static int rpcb_dec_getaddr(struct rpc_rqst *req, struct xdr_stream *xdr,
                             struct rpcbind_args *rpcb)
 {
         struct sockaddr_storage address;
         struct sockaddr *sap = (struct sockaddr *)&address;
         struct rpc_task *task = req->rq_task;
-        struct xdr_stream xdr;
+        __be32 *p;
         u32 len;
 
         rpcb->r_port = 0;
 
-        xdr_init_decode(&xdr, &req->rq_rcv_buf, p);
-
-        p = xdr_inline_decode(&xdr, sizeof(__be32));
+        p = xdr_inline_decode(xdr, 4);
         if (unlikely(p == NULL))
                 goto out_fail;
-        len = ntohl(*p);
+        len = be32_to_cpup(p);
 
         /*
          * If the returned universal address is a null string,
@@ -845,7 +810,7 @@ static int rpcb_dec_getaddr(struct rpc_rqst *req, __be32 *p,
         if (unlikely(len > RPCBIND_MAXUADDRLEN))
                 goto out_fail;
 
-        p = xdr_inline_decode(&xdr, len);
+        p = xdr_inline_decode(xdr, len);
         if (unlikely(p == NULL))
                 goto out_fail;
         dprintk("RPC: %5u RPCB_%s reply: %s\n", task->tk_pid,
@@ -871,8 +836,8 @@ out_fail:
 static struct rpc_procinfo rpcb_procedures2[] = {
         [RPCBPROC_SET] = {
                 .p_proc         = RPCBPROC_SET,
-                .p_encode       = (kxdrproc_t)rpcb_enc_mapping,
-                .p_decode       = (kxdrproc_t)rpcb_dec_set,
+                .p_encode       = (kxdreproc_t)rpcb_enc_mapping,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_set,
                 .p_arglen       = RPCB_mappingargs_sz,
                 .p_replen       = RPCB_setres_sz,
                 .p_statidx      = RPCBPROC_SET,
@@ -881,8 +846,8 @@ static struct rpc_procinfo rpcb_procedures2[] = {
         },
         [RPCBPROC_UNSET] = {
                 .p_proc         = RPCBPROC_UNSET,
-                .p_encode       = (kxdrproc_t)rpcb_enc_mapping,
-                .p_decode       = (kxdrproc_t)rpcb_dec_set,
+                .p_encode       = (kxdreproc_t)rpcb_enc_mapping,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_set,
                 .p_arglen       = RPCB_mappingargs_sz,
                 .p_replen       = RPCB_setres_sz,
                 .p_statidx      = RPCBPROC_UNSET,
@@ -891,8 +856,8 @@ static struct rpc_procinfo rpcb_procedures2[] = {
         },
         [RPCBPROC_GETPORT] = {
                 .p_proc         = RPCBPROC_GETPORT,
-                .p_encode       = (kxdrproc_t)rpcb_enc_mapping,
-                .p_decode       = (kxdrproc_t)rpcb_dec_getport,
+                .p_encode       = (kxdreproc_t)rpcb_enc_mapping,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_getport,
                 .p_arglen       = RPCB_mappingargs_sz,
                 .p_replen       = RPCB_getportres_sz,
                 .p_statidx      = RPCBPROC_GETPORT,
@@ -904,8 +869,8 @@ static struct rpc_procinfo rpcb_procedures2[] = {
 static struct rpc_procinfo rpcb_procedures3[] = {
         [RPCBPROC_SET] = {
                 .p_proc         = RPCBPROC_SET,
-                .p_encode       = (kxdrproc_t)rpcb_enc_getaddr,
-                .p_decode       = (kxdrproc_t)rpcb_dec_set,
+                .p_encode       = (kxdreproc_t)rpcb_enc_getaddr,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_set,
                 .p_arglen       = RPCB_getaddrargs_sz,
                 .p_replen       = RPCB_setres_sz,
                 .p_statidx      = RPCBPROC_SET,
@@ -914,8 +879,8 @@ static struct rpc_procinfo rpcb_procedures3[] = {
         },
         [RPCBPROC_UNSET] = {
                 .p_proc         = RPCBPROC_UNSET,
-                .p_encode       = (kxdrproc_t)rpcb_enc_getaddr,
-                .p_decode       = (kxdrproc_t)rpcb_dec_set,
+                .p_encode       = (kxdreproc_t)rpcb_enc_getaddr,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_set,
                 .p_arglen       = RPCB_getaddrargs_sz,
                 .p_replen       = RPCB_setres_sz,
                 .p_statidx      = RPCBPROC_UNSET,
@@ -924,8 +889,8 @@ static struct rpc_procinfo rpcb_procedures3[] = {
         },
         [RPCBPROC_GETADDR] = {
                 .p_proc         = RPCBPROC_GETADDR,
-                .p_encode       = (kxdrproc_t)rpcb_enc_getaddr,
-                .p_decode       = (kxdrproc_t)rpcb_dec_getaddr,
+                .p_encode       = (kxdreproc_t)rpcb_enc_getaddr,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_getaddr,
                 .p_arglen       = RPCB_getaddrargs_sz,
                 .p_replen       = RPCB_getaddrres_sz,
                 .p_statidx      = RPCBPROC_GETADDR,
@@ -937,8 +902,8 @@ static struct rpc_procinfo rpcb_procedures3[] = {
 static struct rpc_procinfo rpcb_procedures4[] = {
         [RPCBPROC_SET] = {
                 .p_proc         = RPCBPROC_SET,
-                .p_encode       = (kxdrproc_t)rpcb_enc_getaddr,
-                .p_decode       = (kxdrproc_t)rpcb_dec_set,
+                .p_encode       = (kxdreproc_t)rpcb_enc_getaddr,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_set,
                 .p_arglen       = RPCB_getaddrargs_sz,
                 .p_replen       = RPCB_setres_sz,
                 .p_statidx      = RPCBPROC_SET,
@@ -947,8 +912,8 @@ static struct rpc_procinfo rpcb_procedures4[] = {
         },
         [RPCBPROC_UNSET] = {
                 .p_proc         = RPCBPROC_UNSET,
-                .p_encode       = (kxdrproc_t)rpcb_enc_getaddr,
-                .p_decode       = (kxdrproc_t)rpcb_dec_set,
+                .p_encode       = (kxdreproc_t)rpcb_enc_getaddr,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_set,
                 .p_arglen       = RPCB_getaddrargs_sz,
                 .p_replen       = RPCB_setres_sz,
                 .p_statidx      = RPCBPROC_UNSET,
@@ -957,8 +922,8 @@ static struct rpc_procinfo rpcb_procedures4[] = {
         },
         [RPCBPROC_GETADDR] = {
                 .p_proc         = RPCBPROC_GETADDR,
-                .p_encode       = (kxdrproc_t)rpcb_enc_getaddr,
-                .p_decode       = (kxdrproc_t)rpcb_dec_getaddr,
+                .p_encode       = (kxdreproc_t)rpcb_enc_getaddr,
+                .p_decode       = (kxdrdproc_t)rpcb_dec_getaddr,
                 .p_arglen       = RPCB_getaddrargs_sz,
                 .p_replen       = RPCB_getaddrres_sz,
                 .p_statidx      = RPCBPROC_GETADDR,
@@ -993,19 +958,19 @@ static struct rpcb_info rpcb_next_version6[] = {
 
 static struct rpc_version rpcb_version2 = {
         .number         = RPCBVERS_2,
-        .nrprocs        = RPCB_HIGHPROC_2,
+        .nrprocs        = ARRAY_SIZE(rpcb_procedures2),
         .procs          = rpcb_procedures2
 };
 
 static struct rpc_version rpcb_version3 = {
         .number         = RPCBVERS_3,
-        .nrprocs        = RPCB_HIGHPROC_3,
+        .nrprocs        = ARRAY_SIZE(rpcb_procedures3),
         .procs          = rpcb_procedures3
 };
 
 static struct rpc_version rpcb_version4 = {
         .number         = RPCBVERS_4,
-        .nrprocs        = RPCB_HIGHPROC_4,
+        .nrprocs        = ARRAY_SIZE(rpcb_procedures4),
         .procs          = rpcb_procedures4
 };
 
diff --git a/net/sunrpc/svc.c b/net/sunrpc/svc.c
index 6359c42c4941..0e659c665a8d 100644
--- a/net/sunrpc/svc.c
+++ b/net/sunrpc/svc.c
@@ -488,10 +488,6 @@ svc_destroy(struct svc_serv *serv)
         if (svc_serv_is_pooled(serv))
                 svc_pool_map_put();
 
-#if defined(CONFIG_NFS_V4_1)
-        svc_sock_destroy(serv->bc_xprt);
-#endif /* CONFIG_NFS_V4_1 */
-
         svc_unregister(serv);
         kfree(serv->sv_pools);
         kfree(serv);
@@ -1147,7 +1143,6 @@ svc_process_common(struct svc_rqst *rqstp, struct kvec *argv, struct kvec *resv)
  dropit:
         svc_authorise(rqstp);   /* doesn't hurt to call this twice */
         dprintk("svc: svc_process dropit\n");
-        svc_drop(rqstp);
         return 0;
 
 err_short_len:
@@ -1218,7 +1213,6 @@ svc_process(struct svc_rqst *rqstp)
         struct kvec             *resv = &rqstp->rq_res.head[0];
         struct svc_serv         *serv = rqstp->rq_server;
         u32                     dir;
-        int                     error;
 
         /*
          * Setup response xdr_buf.
@@ -1246,11 +1240,13 @@ svc_process(struct svc_rqst *rqstp)
                 return 0;
         }
 
-        error = svc_process_common(rqstp, argv, resv);
-        if (error <= 0)
-                return error;
-
-        return svc_send(rqstp);
+        /* Returns 1 for send, 0 for drop */
+        if (svc_process_common(rqstp, argv, resv))
+                return svc_send(rqstp);
+        else {
+                svc_drop(rqstp);
+                return 0;
+        }
 }
 
 #if defined(CONFIG_NFS_V4_1)
@@ -1264,10 +1260,9 @@ bc_svc_process(struct svc_serv *serv, struct rpc_rqst *req,
 {
         struct kvec     *argv = &rqstp->rq_arg.head[0];
         struct kvec     *resv = &rqstp->rq_res.head[0];
-        int             error;
 
         /* Build the svc_rqst used by the common processing routine */
-        rqstp->rq_xprt = serv->bc_xprt;
+        rqstp->rq_xprt = serv->sv_bc_xprt;
         rqstp->rq_xid = req->rq_xid;
         rqstp->rq_prot = req->rq_xprt->prot;
         rqstp->rq_server = serv;
@@ -1292,12 +1287,15 @@ bc_svc_process(struct svc_serv *serv, struct rpc_rqst *req,
         svc_getu32(argv);       /* XID */
         svc_getnl(argv);        /* CALLDIR */
 
-        error = svc_process_common(rqstp, argv, resv);
-        if (error <= 0)
-                return error;
-
-        memcpy(&req->rq_snd_buf, &rqstp->rq_res, sizeof(req->rq_snd_buf));
-        return bc_send(req);
+        /* Returns 1 for send, 0 for drop */
+        if (svc_process_common(rqstp, argv, resv)) {
+                memcpy(&req->rq_snd_buf, &rqstp->rq_res,
+                                                sizeof(req->rq_snd_buf));
+                return bc_send(req);
+        } else {
+                /* Nothing to do to drop request */
+                return 0;
+        }
 }
 EXPORT_SYMBOL(bc_svc_process);
 #endif /* CONFIG_NFS_V4_1 */
diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 07919e16be3e..d265aa700bb3 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -66,6 +66,13 @@ static void		svc_sock_free(struct svc_xprt *);
 static struct svc_xprt *svc_create_socket(struct svc_serv *, int,
                                           struct net *, struct sockaddr *,
                                           int, int);
+#if defined(CONFIG_NFS_V4_1)
+static struct svc_xprt *svc_bc_create_socket(struct svc_serv *, int,
+                                             struct net *, struct sockaddr *,
+                                             int, int);
+static void svc_bc_sock_free(struct svc_xprt *xprt);
+#endif /* CONFIG_NFS_V4_1 */
+
 #ifdef CONFIG_DEBUG_LOCK_ALLOC
 static struct lock_class_key svc_key[2];
 static struct lock_class_key svc_slock_key[2];
@@ -1184,6 +1191,57 @@ static struct svc_xprt *svc_tcp_create(struct svc_serv *serv,
         return svc_create_socket(serv, IPPROTO_TCP, net, sa, salen, flags);
 }
 
+#if defined(CONFIG_NFS_V4_1)
+static struct svc_xprt *svc_bc_create_socket(struct svc_serv *, int,
+                                             struct net *, struct sockaddr *,
+                                             int, int);
+static void svc_bc_sock_free(struct svc_xprt *xprt);
+
+static struct svc_xprt *svc_bc_tcp_create(struct svc_serv *serv,
+                                       struct net *net,
+                                       struct sockaddr *sa, int salen,
+                                       int flags)
+{
+        return svc_bc_create_socket(serv, IPPROTO_TCP, net, sa, salen, flags);
+}
+
+static void svc_bc_tcp_sock_detach(struct svc_xprt *xprt)
+{
+}
+
+static struct svc_xprt_ops svc_tcp_bc_ops = {
+        .xpo_create = svc_bc_tcp_create,
+        .xpo_detach = svc_bc_tcp_sock_detach,
+        .xpo_free = svc_bc_sock_free,
+        .xpo_prep_reply_hdr = svc_tcp_prep_reply_hdr,
+};
+
+static struct svc_xprt_class svc_tcp_bc_class = {
+        .xcl_name = "tcp-bc",
+        .xcl_owner = THIS_MODULE,
+        .xcl_ops = &svc_tcp_bc_ops,
+        .xcl_max_payload = RPCSVC_MAXPAYLOAD_TCP,
+};
+
+static void svc_init_bc_xprt_sock(void)
+{
+        svc_reg_xprt_class(&svc_tcp_bc_class);
+}
+
+static void svc_cleanup_bc_xprt_sock(void)
+{
+        svc_unreg_xprt_class(&svc_tcp_bc_class);
+}
+#else /* CONFIG_NFS_V4_1 */
+static void svc_init_bc_xprt_sock(void)
+{
+}
+
+static void svc_cleanup_bc_xprt_sock(void)
+{
+}
+#endif /* CONFIG_NFS_V4_1 */
+
 static struct svc_xprt_ops svc_tcp_ops = {
         .xpo_create = svc_tcp_create,
         .xpo_recvfrom = svc_tcp_recvfrom,
@@ -1207,12 +1265,14 @@ void svc_init_xprt_sock(void)
 {
         svc_reg_xprt_class(&svc_tcp_class);
         svc_reg_xprt_class(&svc_udp_class);
+        svc_init_bc_xprt_sock();
 }
 
 void svc_cleanup_xprt_sock(void)
 {
         svc_unreg_xprt_class(&svc_tcp_class);
         svc_unreg_xprt_class(&svc_udp_class);
+        svc_cleanup_bc_xprt_sock();
 }
 
 static void svc_tcp_init(struct svc_sock *svsk, struct svc_serv *serv)
@@ -1509,41 +1569,45 @@ static void svc_sock_free(struct svc_xprt *xprt)
         kfree(svsk);
 }
 
+#if defined(CONFIG_NFS_V4_1)
 /*
- * Create a svc_xprt.
- *
- * For internal use only (e.g. nfsv4.1 backchannel).
- * Callers should typically use the xpo_create() method.
+ * Create a back channel svc_xprt which shares the fore channel socket.
  */
-struct svc_xprt *svc_sock_create(struct svc_serv *serv, int prot)
+static struct svc_xprt *svc_bc_create_socket(struct svc_serv *serv,
+                                             int protocol,
+                                             struct net *net,
+                                             struct sockaddr *sin, int len,
+                                             int flags)
 {
         struct svc_sock *svsk;
-        struct svc_xprt *xprt = NULL;
+        struct svc_xprt *xprt;
+
+        if (protocol != IPPROTO_TCP) {
+                printk(KERN_WARNING "svc: only TCP sockets"
+                        " supported on shared back channel\n");
+                return ERR_PTR(-EINVAL);
+        }
 
-        dprintk("svc: %s\n", __func__);
         svsk = kzalloc(sizeof(*svsk), GFP_KERNEL);
         if (!svsk)
-                goto out;
+                return ERR_PTR(-ENOMEM);
 
         xprt = &svsk->sk_xprt;
-        if (prot == IPPROTO_TCP)
-                svc_xprt_init(&svc_tcp_class, xprt, serv);
-        else if (prot == IPPROTO_UDP)
-                svc_xprt_init(&svc_udp_class, xprt, serv);
-        else
-                BUG();
-out:
-        dprintk("svc: %s return %p\n", __func__, xprt);
+        svc_xprt_init(&svc_tcp_bc_class, xprt, serv);
+
+        serv->sv_bc_xprt = xprt;
+
         return xprt;
 }
-EXPORT_SYMBOL_GPL(svc_sock_create);
 
 /*
- * Destroy a svc_sock.
+ * Free a back channel svc_sock.
  */
-void svc_sock_destroy(struct svc_xprt *xprt)
+static void svc_bc_sock_free(struct svc_xprt *xprt)
 {
-        if (xprt)
+        if (xprt) {
+                kfree(xprt->xpt_bc_sid);
                 kfree(container_of(xprt, struct svc_sock, sk_xprt));
+        }
 }
-EXPORT_SYMBOL_GPL(svc_sock_destroy);
+#endif /* CONFIG_NFS_V4_1 */
diff --git a/net/sunrpc/xdr.c b/net/sunrpc/xdr.c
index cd9e841e7492..679cd674b81d 100644
--- a/net/sunrpc/xdr.c
+++ b/net/sunrpc/xdr.c
@@ -552,6 +552,74 @@ void xdr_write_pages(struct xdr_stream *xdr, struct page **pages, unsigned int b
 }
 EXPORT_SYMBOL_GPL(xdr_write_pages);
 
+static void xdr_set_iov(struct xdr_stream *xdr, struct kvec *iov,
+                __be32 *p, unsigned int len)
+{
+        if (len > iov->iov_len)
+                len = iov->iov_len;
+        if (p == NULL)
+                p = (__be32*)iov->iov_base;
+        xdr->p = p;
+        xdr->end = (__be32*)(iov->iov_base + len);
+        xdr->iov = iov;
+        xdr->page_ptr = NULL;
+}
+
+static int xdr_set_page_base(struct xdr_stream *xdr,
+                unsigned int base, unsigned int len)
+{
+        unsigned int pgnr;
+        unsigned int maxlen;
+        unsigned int pgoff;
+        unsigned int pgend;
+        void *kaddr;
+
+        maxlen = xdr->buf->page_len;
+        if (base >= maxlen)
+                return -EINVAL;
+        maxlen -= base;
+        if (len > maxlen)
+                len = maxlen;
+
+        base += xdr->buf->page_base;
+
+        pgnr = base >> PAGE_SHIFT;
+        xdr->page_ptr = &xdr->buf->pages[pgnr];
+        kaddr = page_address(*xdr->page_ptr);
+
+        pgoff = base & ~PAGE_MASK;
+        xdr->p = (__be32*)(kaddr + pgoff);
+
+        pgend = pgoff + len;
+        if (pgend > PAGE_SIZE)
+                pgend = PAGE_SIZE;
+        xdr->end = (__be32*)(kaddr + pgend);
+        xdr->iov = NULL;
+        return 0;
+}
+
+static void xdr_set_next_page(struct xdr_stream *xdr)
+{
+        unsigned int newbase;
+
+        newbase = (1 + xdr->page_ptr - xdr->buf->pages) << PAGE_SHIFT;
+        newbase -= xdr->buf->page_base;
+
+        if (xdr_set_page_base(xdr, newbase, PAGE_SIZE) < 0)
+                xdr_set_iov(xdr, xdr->buf->tail, NULL, xdr->buf->len);
+}
+
+static bool xdr_set_next_buffer(struct xdr_stream *xdr)
+{
+        if (xdr->page_ptr != NULL)
+                xdr_set_next_page(xdr);
+        else if (xdr->iov == xdr->buf->head) {
+                if (xdr_set_page_base(xdr, 0, PAGE_SIZE) < 0)
+                        xdr_set_iov(xdr, xdr->buf->tail, NULL, xdr->buf->len);
+        }
+        return xdr->p != xdr->end;
+}
+
 /**
  * xdr_init_decode - Initialize an xdr_stream for decoding data.
  * @xdr: pointer to xdr_stream struct
@@ -560,41 +628,67 @@ EXPORT_SYMBOL_GPL(xdr_write_pages);
  */
 void xdr_init_decode(struct xdr_stream *xdr, struct xdr_buf *buf, __be32 *p)
 {
-        struct kvec *iov = buf->head;
-        unsigned int len = iov->iov_len;
-
-        if (len > buf->len)
-                len = buf->len;
         xdr->buf = buf;
-        xdr->iov = iov;
-        xdr->p = p;
-        xdr->end = (__be32 *)((char *)iov->iov_base + len);
+        xdr->scratch.iov_base = NULL;
+        xdr->scratch.iov_len = 0;
+        if (buf->head[0].iov_len != 0)
+                xdr_set_iov(xdr, buf->head, p, buf->len);
+        else if (buf->page_len != 0)
+                xdr_set_page_base(xdr, 0, buf->len);
 }
 EXPORT_SYMBOL_GPL(xdr_init_decode);
 
-/**
- * xdr_inline_peek - Allow read-ahead in the XDR data stream
- * @xdr: pointer to xdr_stream struct
- * @nbytes: number of bytes of data to decode
- *
- * Check if the input buffer is long enough to enable us to decode
- * 'nbytes' more bytes of data starting at the current position.
- * If so return the current pointer without updating the current
- * pointer position.
- */
-__be32 * xdr_inline_peek(struct xdr_stream *xdr, size_t nbytes)
+static __be32 * __xdr_inline_decode(struct xdr_stream *xdr, size_t nbytes)
 {
         __be32 *p = xdr->p;
         __be32 *q = p + XDR_QUADLEN(nbytes);
 
         if (unlikely(q > xdr->end || q < p))
                 return NULL;
+        xdr->p = q;
         return p;
 }
-EXPORT_SYMBOL_GPL(xdr_inline_peek);
 
 /**
- * xdr_inline_decode - Retrieve non-page XDR data to decode
+ * xdr_set_scratch_buffer - Attach a scratch buffer for decoding data.
+ * @xdr: pointer to xdr_stream struct
+ * @buf: pointer to an empty buffer
+ * @buflen: size of 'buf'
+ *
+ * The scratch buffer is used when decoding from an array of pages.
+ * If an xdr_inline_decode() call spans across page boundaries, then
+ * we copy the data into the scratch buffer in order to allow linear
+ * access.
+ */
+void xdr_set_scratch_buffer(struct xdr_stream *xdr, void *buf, size_t buflen)
+{
+        xdr->scratch.iov_base = buf;
+        xdr->scratch.iov_len = buflen;
+}
+EXPORT_SYMBOL_GPL(xdr_set_scratch_buffer);
+
+static __be32 *xdr_copy_to_scratch(struct xdr_stream *xdr, size_t nbytes)
+{
+        __be32 *p;
+        void *cpdest = xdr->scratch.iov_base;
+        size_t cplen = (char *)xdr->end - (char *)xdr->p;
+
+        if (nbytes > xdr->scratch.iov_len)
+                return NULL;
+        memcpy(cpdest, xdr->p, cplen);
+        cpdest += cplen;
+        nbytes -= cplen;
+        if (!xdr_set_next_buffer(xdr))
+                return NULL;
+        p = __xdr_inline_decode(xdr, nbytes);
+        if (p == NULL)
+                return NULL;
+        memcpy(cpdest, p, nbytes);
+        return xdr->scratch.iov_base;
+}
+
+/**
+ * xdr_inline_decode - Retrieve XDR data to decode
  * @xdr: pointer to xdr_stream struct
  * @nbytes: number of bytes of data to decode
  *
@@ -605,13 +699,16 @@ EXPORT_SYMBOL_GPL(xdr_inline_peek);
  */
 __be32 * xdr_inline_decode(struct xdr_stream *xdr, size_t nbytes)
 {
-        __be32 *p = xdr->p;
-        __be32 *q = p + XDR_QUADLEN(nbytes);
+        __be32 *p;
 
-        if (unlikely(q > xdr->end || q < p))
+        if (nbytes == 0)
+                return xdr->p;
+        if (xdr->p == xdr->end && !xdr_set_next_buffer(xdr))
                 return NULL;
-        xdr->p = q;
-        return p;
+        p = __xdr_inline_decode(xdr, nbytes);
+        if (p != NULL)
+                return p;
+        return xdr_copy_to_scratch(xdr, nbytes);
 }
 EXPORT_SYMBOL_GPL(xdr_inline_decode);
 
@@ -671,16 +768,12 @@ EXPORT_SYMBOL_GPL(xdr_read_pages);
  */
 void xdr_enter_page(struct xdr_stream *xdr, unsigned int len)
 {
-        char * kaddr = page_address(xdr->buf->pages[0]);
         xdr_read_pages(xdr, len);
         /*
          * Position current pointer at beginning of tail, and
          * set remaining message length.
          */
-        if (len > PAGE_CACHE_SIZE - xdr->buf->page_base)
-                len = PAGE_CACHE_SIZE - xdr->buf->page_base;
-        xdr->p = (__be32 *)(kaddr + xdr->buf->page_base);
-        xdr->end = (__be32 *)((char *)xdr->p + len);
+        xdr_set_page_base(xdr, 0, len);
 }
 EXPORT_SYMBOL_GPL(xdr_enter_page);