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

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

   - stable fix for a bug in nfs3_list_one_acl()
   - speed up NFS path walks by supporting LOOKUP_RCU
   - more read/write code cleanups
   - pNFS fixes for layout return on close
   - fixes for the RCU handling in the rpcsec_gss code
   - more NFS/RDMA fixes"

* tag 'nfs-for-3.17-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs: (79 commits)
  nfs: reject changes to resvport and sharecache during remount
  NFS: Avoid infinite loop when RELEASE_LOCKOWNER getting expired error
  SUNRPC: remove all refcounting of groupinfo from rpcauth_lookupcred
  NFS: fix two problems in lookup_revalidate in RCU-walk
  NFS: allow lockless access to access_cache
  NFS: teach nfs_lookup_verify_inode to handle LOOKUP_RCU
  NFS: teach nfs_neg_need_reval to understand LOOKUP_RCU
  NFS: support RCU_WALK in nfs_permission()
  sunrpc/auth: allow lockless (rcu) lookup of credential cache.
  NFS: prepare for RCU-walk support but pushing tests later in code.
  NFS: nfs4_lookup_revalidate: only evaluate parent if it will be used.
  NFS: add checks for returned value of try_module_get()
  nfs: clear_request_commit while holding i_lock
  pnfs: add pnfs_put_lseg_async
  pnfs: find swapped pages on pnfs commit lists too
  nfs: fix comment and add warn_on for PG_INODE_REF
  nfs: check wait_on_bit_lock err in page_group_lock
  sunrpc: remove "ec" argument from encrypt_v2 operation
  sunrpc: clean up sparse endianness warnings in gss_krb5_wrap.c
  sunrpc: clean up sparse endianness warnings in gss_krb5_seal.c
  ...

30 files changed, 1124 insertions, 870 deletions

diff --git a/fs/nfs/blocklayout/blocklayout.c b/fs/nfs/blocklayout/blocklayout.c
index 9b431f44fad9..cbb1797149d5 100644
--- a/fs/nfs/blocklayout/blocklayout.c
+++ b/fs/nfs/blocklayout/blocklayout.c
@@ -210,8 +210,7 @@ static void bl_end_io_read(struct bio *bio, int err)
                         SetPageUptodate(bvec->bv_page);
 
         if (err) {
-                struct nfs_pgio_data *rdata = par->data;
-                struct nfs_pgio_header *header = rdata->header;
+                struct nfs_pgio_header *header = par->data;
 
                 if (!header->pnfs_error)
                         header->pnfs_error = -EIO;
@@ -224,43 +223,44 @@ static void bl_end_io_read(struct bio *bio, int err)
 static void bl_read_cleanup(struct work_struct *work)
 {
         struct rpc_task *task;
-        struct nfs_pgio_data *rdata;
+        struct nfs_pgio_header *hdr;
         dprintk("%s enter\n", __func__);
         task = container_of(work, struct rpc_task, u.tk_work);
-        rdata = container_of(task, struct nfs_pgio_data, task);
-        pnfs_ld_read_done(rdata);
+        hdr = container_of(task, struct nfs_pgio_header, task);
+        pnfs_ld_read_done(hdr);
 }
 
 static void
 bl_end_par_io_read(void *data, int unused)
 {
-        struct nfs_pgio_data *rdata = data;
+        struct nfs_pgio_header *hdr = data;
 
-        rdata->task.tk_status = rdata->header->pnfs_error;
-        INIT_WORK(&rdata->task.u.tk_work, bl_read_cleanup);
-        schedule_work(&rdata->task.u.tk_work);
+        hdr->task.tk_status = hdr->pnfs_error;
+        INIT_WORK(&hdr->task.u.tk_work, bl_read_cleanup);
+        schedule_work(&hdr->task.u.tk_work);
 }
 
 static enum pnfs_try_status
-bl_read_pagelist(struct nfs_pgio_data *rdata)
+bl_read_pagelist(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *header = rdata->header;
+        struct nfs_pgio_header *header = hdr;
         int i, hole;
         struct bio *bio = NULL;
         struct pnfs_block_extent *be = NULL, *cow_read = NULL;
         sector_t isect, extent_length = 0;
         struct parallel_io *par;
-        loff_t f_offset = rdata->args.offset;
-        size_t bytes_left = rdata->args.count;
+        loff_t f_offset = hdr->args.offset;
+        size_t bytes_left = hdr->args.count;
         unsigned int pg_offset, pg_len;
-        struct page **pages = rdata->args.pages;
-        int pg_index = rdata->args.pgbase >> PAGE_CACHE_SHIFT;
+        struct page **pages = hdr->args.pages;
+        int pg_index = hdr->args.pgbase >> PAGE_CACHE_SHIFT;
         const bool is_dio = (header->dreq != NULL);
 
         dprintk("%s enter nr_pages %u offset %lld count %u\n", __func__,
-               rdata->pages.npages, f_offset, (unsigned int)rdata->args.count);
+                hdr->page_array.npages, f_offset,
+                (unsigned int)hdr->args.count);
 
-        par = alloc_parallel(rdata);
+        par = alloc_parallel(hdr);
         if (!par)
                 goto use_mds;
         par->pnfs_callback = bl_end_par_io_read;
@@ -268,7 +268,7 @@ bl_read_pagelist(struct nfs_pgio_data *rdata)
 
         isect = (sector_t) (f_offset >> SECTOR_SHIFT);
         /* Code assumes extents are page-aligned */
-        for (i = pg_index; i < rdata->pages.npages; i++) {
+        for (i = pg_index; i < hdr->page_array.npages; i++) {
                 if (!extent_length) {
                         /* We've used up the previous extent */
                         bl_put_extent(be);
@@ -317,7 +317,8 @@ bl_read_pagelist(struct nfs_pgio_data *rdata)
                         struct pnfs_block_extent *be_read;
 
                         be_read = (hole && cow_read) ? cow_read : be;
-                        bio = do_add_page_to_bio(bio, rdata->pages.npages - i,
+                        bio = do_add_page_to_bio(bio,
+                                                 hdr->page_array.npages - i,
                                                  READ,
                                                  isect, pages[i], be_read,
                                                  bl_end_io_read, par,
@@ -332,10 +333,10 @@ bl_read_pagelist(struct nfs_pgio_data *rdata)
                 extent_length -= PAGE_CACHE_SECTORS;
         }
         if ((isect << SECTOR_SHIFT) >= header->inode->i_size) {
-                rdata->res.eof = 1;
-                rdata->res.count = header->inode->i_size - rdata->args.offset;
+                hdr->res.eof = 1;
+                hdr->res.count = header->inode->i_size - hdr->args.offset;
         } else {
-                rdata->res.count = (isect << SECTOR_SHIFT) - rdata->args.offset;
+                hdr->res.count = (isect << SECTOR_SHIFT) - hdr->args.offset;
         }
 out:
         bl_put_extent(be);
@@ -390,8 +391,7 @@ static void bl_end_io_write_zero(struct bio *bio, int err)
         }
 
         if (unlikely(err)) {
-                struct nfs_pgio_data *data = par->data;
-                struct nfs_pgio_header *header = data->header;
+                struct nfs_pgio_header *header = par->data;
 
                 if (!header->pnfs_error)
                         header->pnfs_error = -EIO;
@@ -405,8 +405,7 @@ static void bl_end_io_write(struct bio *bio, int err)
 {
         struct parallel_io *par = bio->bi_private;
         const int uptodate = test_bit(BIO_UPTODATE, &bio->bi_flags);
-        struct nfs_pgio_data *data = par->data;
-        struct nfs_pgio_header *header = data->header;
+        struct nfs_pgio_header *header = par->data;
 
         if (!uptodate) {
                 if (!header->pnfs_error)
@@ -423,32 +422,32 @@ static void bl_end_io_write(struct bio *bio, int err)
 static void bl_write_cleanup(struct work_struct *work)
 {
         struct rpc_task *task;
-        struct nfs_pgio_data *wdata;
+        struct nfs_pgio_header *hdr;
         dprintk("%s enter\n", __func__);
         task = container_of(work, struct rpc_task, u.tk_work);
-        wdata = container_of(task, struct nfs_pgio_data, task);
-        if (likely(!wdata->header->pnfs_error)) {
+        hdr = container_of(task, struct nfs_pgio_header, task);
+        if (likely(!hdr->pnfs_error)) {
                 /* Marks for LAYOUTCOMMIT */
-                mark_extents_written(BLK_LSEG2EXT(wdata->header->lseg),
-                                     wdata->args.offset, wdata->args.count);
+                mark_extents_written(BLK_LSEG2EXT(hdr->lseg),
+                                     hdr->args.offset, hdr->args.count);
         }
-        pnfs_ld_write_done(wdata);
+        pnfs_ld_write_done(hdr);
 }
 
 /* Called when last of bios associated with a bl_write_pagelist call finishes */
 static void bl_end_par_io_write(void *data, int num_se)
 {
-        struct nfs_pgio_data *wdata = data;
+        struct nfs_pgio_header *hdr = data;
 
-        if (unlikely(wdata->header->pnfs_error)) {
-                bl_free_short_extents(&BLK_LSEG2EXT(wdata->header->lseg)->bl_inval,
+        if (unlikely(hdr->pnfs_error)) {
+                bl_free_short_extents(&BLK_LSEG2EXT(hdr->lseg)->bl_inval,
                                         num_se);
         }
 
-        wdata->task.tk_status = wdata->header->pnfs_error;
-        wdata->verf.committed = NFS_FILE_SYNC;
-        INIT_WORK(&wdata->task.u.tk_work, bl_write_cleanup);
-        schedule_work(&wdata->task.u.tk_work);
+        hdr->task.tk_status = hdr->pnfs_error;
+        hdr->verf.committed = NFS_FILE_SYNC;
+        INIT_WORK(&hdr->task.u.tk_work, bl_write_cleanup);
+        schedule_work(&hdr->task.u.tk_work);
 }
 
 /* FIXME STUB - mark intersection of layout and page as bad, so is not
@@ -673,18 +672,17 @@ check_page:
 }
 
 static enum pnfs_try_status
-bl_write_pagelist(struct nfs_pgio_data *wdata, int sync)
+bl_write_pagelist(struct nfs_pgio_header *header, int sync)
 {
-        struct nfs_pgio_header *header = wdata->header;
         int i, ret, npg_zero, pg_index, last = 0;
         struct bio *bio = NULL;
         struct pnfs_block_extent *be = NULL, *cow_read = NULL;
         sector_t isect, last_isect = 0, extent_length = 0;
         struct parallel_io *par = NULL;
-        loff_t offset = wdata->args.offset;
-        size_t count = wdata->args.count;
+        loff_t offset = header->args.offset;
+        size_t count = header->args.count;
         unsigned int pg_offset, pg_len, saved_len;
-        struct page **pages = wdata->args.pages;
+        struct page **pages = header->args.pages;
         struct page *page;
         pgoff_t index;
         u64 temp;
@@ -699,11 +697,11 @@ bl_write_pagelist(struct nfs_pgio_data *wdata, int sync)
                 dprintk("pnfsblock nonblock aligned DIO writes. Resend MDS\n");
                 goto out_mds;
         }
-        /* At this point, wdata->pages is a (sequential) list of nfs_pages.
+        /* At this point, header->page_aray is a (sequential) list of nfs_pages.
          * We want to write each, and if there is an error set pnfs_error
          * to have it redone using nfs.
          */
-        par = alloc_parallel(wdata);
+        par = alloc_parallel(header);
         if (!par)
                 goto out_mds;
         par->pnfs_callback = bl_end_par_io_write;
@@ -790,8 +788,8 @@ next_page:
         bio = bl_submit_bio(WRITE, bio);
 
         /* Middle pages */
-        pg_index = wdata->args.pgbase >> PAGE_CACHE_SHIFT;
-        for (i = pg_index; i < wdata->pages.npages; i++) {
+        pg_index = header->args.pgbase >> PAGE_CACHE_SHIFT;
+        for (i = pg_index; i < header->page_array.npages; i++) {
                 if (!extent_length) {
                         /* We've used up the previous extent */
                         bl_put_extent(be);
@@ -862,7 +860,8 @@ next_page:
                 }
 
 
-                bio = do_add_page_to_bio(bio, wdata->pages.npages - i, WRITE,
+                bio = do_add_page_to_bio(bio, header->page_array.npages - i,
+                                         WRITE,
                                          isect, pages[i], be,
                                          bl_end_io_write, par,
                                          pg_offset, pg_len);
@@ -890,7 +889,7 @@ next_page:
         }
 
 write_done:
-        wdata->res.count = wdata->args.count;
+        header->res.count = header->args.count;
 out:
         bl_put_extent(be);
         bl_put_extent(cow_read);
@@ -1063,7 +1062,7 @@ nfs4_blk_get_deviceinfo(struct nfs_server *server, const struct nfs_fh *fh,
                 return ERR_PTR(-ENOMEM);
         }
 
-        pages = kzalloc(max_pages * sizeof(struct page *), GFP_NOFS);
+        pages = kcalloc(max_pages, sizeof(struct page *), GFP_NOFS);
         if (pages == NULL) {
                 kfree(dev);
                 return ERR_PTR(-ENOMEM);
diff --git a/fs/nfs/callback.c b/fs/nfs/callback.c
index 073b4cf67ed9..54de482143cc 100644
--- a/fs/nfs/callback.c
+++ b/fs/nfs/callback.c
@@ -428,6 +428,18 @@ check_gss_callback_principal(struct nfs_client *clp, struct svc_rqst *rqstp)
         if (p == NULL)
                 return 0;
 
+        /*
+         * Did we get the acceptor from userland during the SETCLIENID
+         * negotiation?
+         */
+        if (clp->cl_acceptor)
+                return !strcmp(p, clp->cl_acceptor);
+
+        /*
+         * Otherwise try to verify it using the cl_hostname. Note that this
+         * doesn't work if a non-canonical hostname was used in the devname.
+         */
+
         /* Expect a GSS_C_NT_HOSTBASED_NAME like "nfs@serverhostname" */
 
         if (memcmp(p, "nfs@", 4) != 0)
diff --git a/fs/nfs/client.c b/fs/nfs/client.c
index 180d1ec9c32e..1c5ff6d58385 100644
--- a/fs/nfs/client.c
+++ b/fs/nfs/client.c
@@ -110,8 +110,8 @@ struct nfs_subversion *get_nfs_version(unsigned int version)
                 mutex_unlock(&nfs_version_mutex);
         }
 
-        if (!IS_ERR(nfs))
-                try_module_get(nfs->owner);
+        if (!IS_ERR(nfs) && !try_module_get(nfs->owner))
+                return ERR_PTR(-EAGAIN);
         return nfs;
 }
 
@@ -158,7 +158,8 @@ struct nfs_client *nfs_alloc_client(const struct nfs_client_initdata *cl_init)
                 goto error_0;
 
         clp->cl_nfs_mod = cl_init->nfs_mod;
-        try_module_get(clp->cl_nfs_mod->owner);
+        if (!try_module_get(clp->cl_nfs_mod->owner))
+                goto error_dealloc;
 
         clp->rpc_ops = clp->cl_nfs_mod->rpc_ops;
 
@@ -190,6 +191,7 @@ struct nfs_client *nfs_alloc_client(const struct nfs_client_initdata *cl_init)
 
 error_cleanup:
         put_nfs_version(clp->cl_nfs_mod);
+error_dealloc:
         kfree(clp);
 error_0:
         return ERR_PTR(err);
@@ -252,6 +254,7 @@ void nfs_free_client(struct nfs_client *clp)
         put_net(clp->cl_net);
         put_nfs_version(clp->cl_nfs_mod);
         kfree(clp->cl_hostname);
+        kfree(clp->cl_acceptor);
         kfree(clp);
 
         dprintk("<-- nfs_free_client()\n");
@@ -482,8 +485,13 @@ nfs_get_client(const struct nfs_client_initdata *cl_init,
         struct nfs_net *nn = net_generic(cl_init->net, nfs_net_id);
         const struct nfs_rpc_ops *rpc_ops = cl_init->nfs_mod->rpc_ops;
 
+        if (cl_init->hostname == NULL) {
+                WARN_ON(1);
+                return NULL;
+        }
+
         dprintk("--> nfs_get_client(%s,v%u)\n",
-                cl_init->hostname ?: "", rpc_ops->version);
+                cl_init->hostname, rpc_ops->version);
 
         /* see if the client already exists */
         do {
@@ -510,7 +518,7 @@ nfs_get_client(const struct nfs_client_initdata *cl_init,
         } while (!IS_ERR(new));
 
         dprintk("<-- nfs_get_client() Failed to find %s (%ld)\n",
-                cl_init->hostname ?: "", PTR_ERR(new));
+                cl_init->hostname, PTR_ERR(new));
         return new;
 }
 EXPORT_SYMBOL_GPL(nfs_get_client);
diff --git a/fs/nfs/delegation.c b/fs/nfs/delegation.c
index 5d8ccecf5f5c..5853f53db732 100644
--- a/fs/nfs/delegation.c
+++ b/fs/nfs/delegation.c
@@ -41,14 +41,8 @@ void nfs_mark_delegation_referenced(struct nfs_delegation *delegation)
         set_bit(NFS_DELEGATION_REFERENCED, &delegation->flags);
 }
 
-/**
- * nfs_have_delegation - check if inode has a delegation
- * @inode: inode to check
- * @flags: delegation types to check for
- *
- * Returns one if inode has the indicated delegation, otherwise zero.
- */
-int nfs4_have_delegation(struct inode *inode, fmode_t flags)
+static int
+nfs4_do_check_delegation(struct inode *inode, fmode_t flags, bool mark)
 {
         struct nfs_delegation *delegation;
         int ret = 0;
@@ -58,12 +52,34 @@ int nfs4_have_delegation(struct inode *inode, fmode_t flags)
         delegation = rcu_dereference(NFS_I(inode)->delegation);
         if (delegation != NULL && (delegation->type & flags) == flags &&
             !test_bit(NFS_DELEGATION_RETURNING, &delegation->flags)) {
-                nfs_mark_delegation_referenced(delegation);
+                if (mark)
+                        nfs_mark_delegation_referenced(delegation);
                 ret = 1;
         }
         rcu_read_unlock();
         return ret;
 }
+/**
+ * nfs_have_delegation - check if inode has a delegation, mark it
+ * NFS_DELEGATION_REFERENCED if there is one.
+ * @inode: inode to check
+ * @flags: delegation types to check for
+ *
+ * Returns one if inode has the indicated delegation, otherwise zero.
+ */
+int nfs4_have_delegation(struct inode *inode, fmode_t flags)
+{
+        return nfs4_do_check_delegation(inode, flags, true);
+}
+
+/*
+ * nfs4_check_delegation - check if inode has a delegation, do not mark
+ * NFS_DELEGATION_REFERENCED if it has one.
+ */
+int nfs4_check_delegation(struct inode *inode, fmode_t flags)
+{
+        return nfs4_do_check_delegation(inode, flags, false);
+}
 
 static int nfs_delegation_claim_locks(struct nfs_open_context *ctx, struct nfs4_state *state, const nfs4_stateid *stateid)
 {
diff --git a/fs/nfs/delegation.h b/fs/nfs/delegation.h
index 9a79c7a99d6d..5c1cce39297f 100644
--- a/fs/nfs/delegation.h
+++ b/fs/nfs/delegation.h
@@ -59,6 +59,7 @@ bool nfs4_copy_delegation_stateid(nfs4_stateid *dst, struct inode *inode, fmode_
 
 void nfs_mark_delegation_referenced(struct nfs_delegation *delegation);
 int nfs4_have_delegation(struct inode *inode, fmode_t flags);
+int nfs4_check_delegation(struct inode *inode, fmode_t flags);
 
 #endif
 
diff --git a/fs/nfs/dir.c b/fs/nfs/dir.c
index 4a3d4ef76127..36d921f0c602 100644
--- a/fs/nfs/dir.c
+++ b/fs/nfs/dir.c
@@ -988,9 +988,13 @@ EXPORT_SYMBOL_GPL(nfs_force_lookup_revalidate);
  * A check for whether or not the parent directory has changed.
  * In the case it has, we assume that the dentries are untrustworthy
  * and may need to be looked up again.
+ * If rcu_walk prevents us from performing a full check, return 0.
  */
-static int nfs_check_verifier(struct inode *dir, struct dentry *dentry)
+static int nfs_check_verifier(struct inode *dir, struct dentry *dentry,
+                              int rcu_walk)
 {
+        int ret;
+
         if (IS_ROOT(dentry))
                 return 1;
         if (NFS_SERVER(dir)->flags & NFS_MOUNT_LOOKUP_CACHE_NONE)
@@ -998,7 +1002,11 @@ static int nfs_check_verifier(struct inode *dir, struct dentry *dentry)
         if (!nfs_verify_change_attribute(dir, dentry->d_time))
                 return 0;
         /* Revalidate nfsi->cache_change_attribute before we declare a match */
-        if (nfs_revalidate_inode(NFS_SERVER(dir), dir) < 0)
+        if (rcu_walk)
+                ret = nfs_revalidate_inode_rcu(NFS_SERVER(dir), dir);
+        else
+                ret = nfs_revalidate_inode(NFS_SERVER(dir), dir);
+        if (ret < 0)
                 return 0;
         if (!nfs_verify_change_attribute(dir, dentry->d_time))
                 return 0;
@@ -1042,6 +1050,8 @@ int nfs_lookup_verify_inode(struct inode *inode, unsigned int flags)
 out:
         return (inode->i_nlink == 0) ? -ENOENT : 0;
 out_force:
+        if (flags & LOOKUP_RCU)
+                return -ECHILD;
         ret = __nfs_revalidate_inode(server, inode);
         if (ret != 0)
                 return ret;
@@ -1054,6 +1064,9 @@ out_force:
  *
  * If parent mtime has changed, we revalidate, else we wait for a
  * period corresponding to the parent's attribute cache timeout value.
+ *
+ * If LOOKUP_RCU prevents us from performing a full check, return 1
+ * suggesting a reval is needed.
  */
 static inline
 int nfs_neg_need_reval(struct inode *dir, struct dentry *dentry,
@@ -1064,7 +1077,7 @@ int nfs_neg_need_reval(struct inode *dir, struct dentry *dentry,
                 return 0;
         if (NFS_SERVER(dir)->flags & NFS_MOUNT_LOOKUP_CACHE_NONEG)
                 return 1;
-        return !nfs_check_verifier(dir, dentry);
+        return !nfs_check_verifier(dir, dentry, flags & LOOKUP_RCU);
 }
 
 /*
@@ -1088,21 +1101,30 @@ static int nfs_lookup_revalidate(struct dentry *dentry, unsigned int flags)
         struct nfs4_label *label = NULL;
         int error;
 
-        if (flags & LOOKUP_RCU)
-                return -ECHILD;
-
-        parent = dget_parent(dentry);
-        dir = parent->d_inode;
+        if (flags & LOOKUP_RCU) {
+                parent = ACCESS_ONCE(dentry->d_parent);
+                dir = ACCESS_ONCE(parent->d_inode);
+                if (!dir)
+                        return -ECHILD;
+        } else {
+                parent = dget_parent(dentry);
+                dir = parent->d_inode;
+        }
         nfs_inc_stats(dir, NFSIOS_DENTRYREVALIDATE);
         inode = dentry->d_inode;
 
         if (!inode) {
-                if (nfs_neg_need_reval(dir, dentry, flags))
+                if (nfs_neg_need_reval(dir, dentry, flags)) {
+                        if (flags & LOOKUP_RCU)
+                                return -ECHILD;
                         goto out_bad;
+                }
                 goto out_valid_noent;
         }
 
         if (is_bad_inode(inode)) {
+                if (flags & LOOKUP_RCU)
+                        return -ECHILD;
                 dfprintk(LOOKUPCACHE, "%s: %pd2 has dud inode\n",
                                 __func__, dentry);
                 goto out_bad;
@@ -1112,12 +1134,20 @@ static int nfs_lookup_revalidate(struct dentry *dentry, unsigned int flags)
                 goto out_set_verifier;
 
         /* Force a full look up iff the parent directory has changed */
-        if (!nfs_is_exclusive_create(dir, flags) && nfs_check_verifier(dir, dentry)) {
-                if (nfs_lookup_verify_inode(inode, flags))
+        if (!nfs_is_exclusive_create(dir, flags) &&
+            nfs_check_verifier(dir, dentry, flags & LOOKUP_RCU)) {
+
+                if (nfs_lookup_verify_inode(inode, flags)) {
+                        if (flags & LOOKUP_RCU)
+                                return -ECHILD;
                         goto out_zap_parent;
+                }
                 goto out_valid;
         }
 
+        if (flags & LOOKUP_RCU)
+                return -ECHILD;
+
         if (NFS_STALE(inode))
                 goto out_bad;
 
@@ -1153,13 +1183,18 @@ out_set_verifier:
         /* Success: notify readdir to use READDIRPLUS */
         nfs_advise_use_readdirplus(dir);
  out_valid_noent:
-        dput(parent);
+        if (flags & LOOKUP_RCU) {
+                if (parent != ACCESS_ONCE(dentry->d_parent))
+                        return -ECHILD;
+        } else
+                dput(parent);
         dfprintk(LOOKUPCACHE, "NFS: %s(%pd2) is valid\n",
                         __func__, dentry);
         return 1;
 out_zap_parent:
         nfs_zap_caches(dir);
  out_bad:
+        WARN_ON(flags & LOOKUP_RCU);
         nfs_free_fattr(fattr);
         nfs_free_fhandle(fhandle);
         nfs4_label_free(label);
@@ -1185,6 +1220,7 @@ out_zap_parent:
                         __func__, dentry);
         return 0;
 out_error:
+        WARN_ON(flags & LOOKUP_RCU);
         nfs_free_fattr(fattr);
         nfs_free_fhandle(fhandle);
         nfs4_label_free(label);
@@ -1529,14 +1565,9 @@ EXPORT_SYMBOL_GPL(nfs_atomic_open);
 
 static int nfs4_lookup_revalidate(struct dentry *dentry, unsigned int flags)
 {
-        struct dentry *parent = NULL;
         struct inode *inode;
-        struct inode *dir;
         int ret = 0;
 
-        if (flags & LOOKUP_RCU)
-                return -ECHILD;
-
         if (!(flags & LOOKUP_OPEN) || (flags & LOOKUP_DIRECTORY))
                 goto no_open;
         if (d_mountpoint(dentry))
@@ -1545,34 +1576,47 @@ static int nfs4_lookup_revalidate(struct dentry *dentry, unsigned int flags)
                 goto no_open;
 
         inode = dentry->d_inode;
-        parent = dget_parent(dentry);
-        dir = parent->d_inode;
 
         /* We can't create new files in nfs_open_revalidate(), so we
          * optimize away revalidation of negative dentries.
          */
         if (inode == NULL) {
+                struct dentry *parent;
+                struct inode *dir;
+
+                if (flags & LOOKUP_RCU) {
+                        parent = ACCESS_ONCE(dentry->d_parent);
+                        dir = ACCESS_ONCE(parent->d_inode);
+                        if (!dir)
+                                return -ECHILD;
+                } else {
+                        parent = dget_parent(dentry);
+                        dir = parent->d_inode;
+                }
                 if (!nfs_neg_need_reval(dir, dentry, flags))
                         ret = 1;
+                else if (flags & LOOKUP_RCU)
+                        ret = -ECHILD;
+                if (!(flags & LOOKUP_RCU))
+                        dput(parent);
+                else if (parent != ACCESS_ONCE(dentry->d_parent))
+                        return -ECHILD;
                 goto out;
         }
 
         /* NFS only supports OPEN on regular files */
         if (!S_ISREG(inode->i_mode))
-                goto no_open_dput;
+                goto no_open;
         /* We cannot do exclusive creation on a positive dentry */
         if (flags & LOOKUP_EXCL)
-                goto no_open_dput;
+                goto no_open;
 
         /* Let f_op->open() actually open (and revalidate) the file */
         ret = 1;
 
 out:
-        dput(parent);
         return ret;
 
-no_open_dput:
-        dput(parent);
 no_open:
         return nfs_lookup_revalidate(dentry, flags);
 }
@@ -2028,10 +2072,14 @@ static DEFINE_SPINLOCK(nfs_access_lru_lock);
 static LIST_HEAD(nfs_access_lru_list);
 static atomic_long_t nfs_access_nr_entries;
 
+static unsigned long nfs_access_max_cachesize = ULONG_MAX;
+module_param(nfs_access_max_cachesize, ulong, 0644);
+MODULE_PARM_DESC(nfs_access_max_cachesize, "NFS access maximum total cache length");
+
 static void nfs_access_free_entry(struct nfs_access_entry *entry)
 {
         put_rpccred(entry->cred);
-        kfree(entry);
+        kfree_rcu(entry, rcu_head);
         smp_mb__before_atomic();
         atomic_long_dec(&nfs_access_nr_entries);
         smp_mb__after_atomic();
@@ -2048,19 +2096,14 @@ static void nfs_access_free_list(struct list_head *head)
         }
 }
 
-unsigned long
-nfs_access_cache_scan(struct shrinker *shrink, struct shrink_control *sc)
+static unsigned long
+nfs_do_access_cache_scan(unsigned int nr_to_scan)
 {
         LIST_HEAD(head);
         struct nfs_inode *nfsi, *next;
         struct nfs_access_entry *cache;
-        int nr_to_scan = sc->nr_to_scan;
-        gfp_t gfp_mask = sc->gfp_mask;
         long freed = 0;
 
-        if ((gfp_mask & GFP_KERNEL) != GFP_KERNEL)
-                return SHRINK_STOP;
-
         spin_lock(&nfs_access_lru_lock);
         list_for_each_entry_safe(nfsi, next, &nfs_access_lru_list, access_cache_inode_lru) {
                 struct inode *inode;
@@ -2094,11 +2137,39 @@ remove_lru_entry:
 }
 
 unsigned long
+nfs_access_cache_scan(struct shrinker *shrink, struct shrink_control *sc)
+{
+        int nr_to_scan = sc->nr_to_scan;
+        gfp_t gfp_mask = sc->gfp_mask;
+
+        if ((gfp_mask & GFP_KERNEL) != GFP_KERNEL)
+                return SHRINK_STOP;
+        return nfs_do_access_cache_scan(nr_to_scan);
+}
+
+
+unsigned long
 nfs_access_cache_count(struct shrinker *shrink, struct shrink_control *sc)
 {
         return vfs_pressure_ratio(atomic_long_read(&nfs_access_nr_entries));
 }
 
+static void
+nfs_access_cache_enforce_limit(void)
+{
+        long nr_entries = atomic_long_read(&nfs_access_nr_entries);
+        unsigned long diff;
+        unsigned int nr_to_scan;
+
+        if (nr_entries < 0 || nr_entries <= nfs_access_max_cachesize)
+                return;
+        nr_to_scan = 100;
+        diff = nr_entries - nfs_access_max_cachesize;
+        if (diff < nr_to_scan)
+                nr_to_scan = diff;
+        nfs_do_access_cache_scan(nr_to_scan);
+}
+
 static void __nfs_access_zap_cache(struct nfs_inode *nfsi, struct list_head *head)
 {
         struct rb_root *root_node = &nfsi->access_cache;
@@ -2186,6 +2257,38 @@ out_zap:
         return -ENOENT;
 }
 
+static int nfs_access_get_cached_rcu(struct inode *inode, struct rpc_cred *cred, struct nfs_access_entry *res)
+{
+        /* Only check the most recently returned cache entry,
+         * but do it without locking.
+         */
+        struct nfs_inode *nfsi = NFS_I(inode);
+        struct nfs_access_entry *cache;
+        int err = -ECHILD;
+        struct list_head *lh;
+
+        rcu_read_lock();
+        if (nfsi->cache_validity & NFS_INO_INVALID_ACCESS)
+                goto out;
+        lh = rcu_dereference(nfsi->access_cache_entry_lru.prev);
+        cache = list_entry(lh, struct nfs_access_entry, lru);
+        if (lh == &nfsi->access_cache_entry_lru ||
+            cred != cache->cred)
+                cache = NULL;
+        if (cache == NULL)
+                goto out;
+        if (!nfs_have_delegated_attributes(inode) &&
+            !time_in_range_open(jiffies, cache->jiffies, cache->jiffies + nfsi->attrtimeo))
+                goto out;
+        res->jiffies = cache->jiffies;
+        res->cred = cache->cred;
+        res->mask = cache->mask;
+        err = 0;
+out:
+        rcu_read_unlock();
+        return err;
+}
+
 static void nfs_access_add_rbtree(struct inode *inode, struct nfs_access_entry *set)
 {
         struct nfs_inode *nfsi = NFS_I(inode);
@@ -2229,6 +2332,11 @@ void nfs_access_add_cache(struct inode *inode, struct nfs_access_entry *set)
         cache->cred = get_rpccred(set->cred);
         cache->mask = set->mask;
 
+        /* The above field assignments must be visible
+         * before this item appears on the lru.  We cannot easily
+         * use rcu_assign_pointer, so just force the memory barrier.
+         */
+        smp_wmb();
         nfs_access_add_rbtree(inode, cache);
 
         /* Update accounting */
@@ -2244,6 +2352,7 @@ void nfs_access_add_cache(struct inode *inode, struct nfs_access_entry *set)
                                         &nfs_access_lru_list);
                 spin_unlock(&nfs_access_lru_lock);
         }
+        nfs_access_cache_enforce_limit();
 }
 EXPORT_SYMBOL_GPL(nfs_access_add_cache);
 
@@ -2267,10 +2376,16 @@ static int nfs_do_access(struct inode *inode, struct rpc_cred *cred, int mask)
 
         trace_nfs_access_enter(inode);
 
-        status = nfs_access_get_cached(inode, cred, &cache);
+        status = nfs_access_get_cached_rcu(inode, cred, &cache);
+        if (status != 0)
+                status = nfs_access_get_cached(inode, cred, &cache);
         if (status == 0)
                 goto out_cached;
 
+        status = -ECHILD;
+        if (mask & MAY_NOT_BLOCK)
+                goto out;
+
         /* Be clever: ask server to check for all possible rights */
         cache.mask = MAY_EXEC | MAY_WRITE | MAY_READ;
         cache.cred = cred;
@@ -2321,9 +2436,6 @@ int nfs_permission(struct inode *inode, int mask)
         struct rpc_cred *cred;
         int res = 0;
 
-        if (mask & MAY_NOT_BLOCK)
-                return -ECHILD;
-
         nfs_inc_stats(inode, NFSIOS_VFSACCESS);
 
         if ((mask & (MAY_READ | MAY_WRITE | MAY_EXEC)) == 0)
@@ -2350,12 +2462,23 @@ force_lookup:
         if (!NFS_PROTO(inode)->access)
                 goto out_notsup;
 
-        cred = rpc_lookup_cred();
-        if (!IS_ERR(cred)) {
-                res = nfs_do_access(inode, cred, mask);
-                put_rpccred(cred);
-        } else
+        /* Always try fast lookups first */
+        rcu_read_lock();
+        cred = rpc_lookup_cred_nonblock();
+        if (!IS_ERR(cred))
+                res = nfs_do_access(inode, cred, mask|MAY_NOT_BLOCK);
+        else
                 res = PTR_ERR(cred);
+        rcu_read_unlock();
+        if (res == -ECHILD && !(mask & MAY_NOT_BLOCK)) {
+                /* Fast lookup failed, try the slow way */
+                cred = rpc_lookup_cred();
+                if (!IS_ERR(cred)) {
+                        res = nfs_do_access(inode, cred, mask);
+                        put_rpccred(cred);
+                } else
+                        res = PTR_ERR(cred);
+        }
 out:
         if (!res && (mask & MAY_EXEC) && !execute_ok(inode))
                 res = -EACCES;
@@ -2364,6 +2487,9 @@ out:
                 inode->i_sb->s_id, inode->i_ino, mask, res);
         return res;
 out_notsup:
+        if (mask & MAY_NOT_BLOCK)
+                return -ECHILD;
+
         res = nfs_revalidate_inode(NFS_SERVER(inode), inode);
         if (res == 0)
                 res = generic_permission(inode, mask);
diff --git a/fs/nfs/direct.c b/fs/nfs/direct.c
index f11b9eed0de1..65ef6e00deee 100644
--- a/fs/nfs/direct.c
+++ b/fs/nfs/direct.c
@@ -148,8 +148,8 @@ static void nfs_direct_set_hdr_verf(struct nfs_direct_req *dreq,
 {
         struct nfs_writeverf *verfp;
 
-        verfp = nfs_direct_select_verf(dreq, hdr->data->ds_clp,
-                                      hdr->data->ds_idx);
+        verfp = nfs_direct_select_verf(dreq, hdr->ds_clp,
+                                      hdr->ds_idx);
         WARN_ON_ONCE(verfp->committed >= 0);
         memcpy(verfp, &hdr->verf, sizeof(struct nfs_writeverf));
         WARN_ON_ONCE(verfp->committed < 0);
@@ -169,8 +169,8 @@ static int nfs_direct_set_or_cmp_hdr_verf(struct nfs_direct_req *dreq,
 {
         struct nfs_writeverf *verfp;
 
-        verfp = nfs_direct_select_verf(dreq, hdr->data->ds_clp,
-                                         hdr->data->ds_idx);
+        verfp = nfs_direct_select_verf(dreq, hdr->ds_clp,
+                                         hdr->ds_idx);
         if (verfp->committed < 0) {
                 nfs_direct_set_hdr_verf(dreq, hdr);
                 return 0;
@@ -715,7 +715,7 @@ static void nfs_direct_write_completion(struct nfs_pgio_header *hdr)
 {
         struct nfs_direct_req *dreq = hdr->dreq;
         struct nfs_commit_info cinfo;
-        int bit = -1;
+        bool request_commit = false;
         struct nfs_page *req = nfs_list_entry(hdr->pages.next);
 
         if (test_bit(NFS_IOHDR_REDO, &hdr->flags))
@@ -729,27 +729,20 @@ static void nfs_direct_write_completion(struct nfs_pgio_header *hdr)
                 dreq->flags = 0;
                 dreq->error = hdr->error;
         }
-        if (dreq->error != 0)
-                bit = NFS_IOHDR_ERROR;
-        else {
+        if (dreq->error == 0) {
                 dreq->count += hdr->good_bytes;
-                if (test_bit(NFS_IOHDR_NEED_RESCHED, &hdr->flags)) {
-                        dreq->flags = NFS_ODIRECT_RESCHED_WRITES;
-                        bit = NFS_IOHDR_NEED_RESCHED;
-                } else if (test_bit(NFS_IOHDR_NEED_COMMIT, &hdr->flags)) {
+                if (nfs_write_need_commit(hdr)) {
                         if (dreq->flags == NFS_ODIRECT_RESCHED_WRITES)
-                                bit = NFS_IOHDR_NEED_RESCHED;
+                                request_commit = true;
                         else if (dreq->flags == 0) {
                                 nfs_direct_set_hdr_verf(dreq, hdr);
-                                bit = NFS_IOHDR_NEED_COMMIT;
+                                request_commit = true;
                                 dreq->flags = NFS_ODIRECT_DO_COMMIT;
                         } else if (dreq->flags == NFS_ODIRECT_DO_COMMIT) {
-                                if (nfs_direct_set_or_cmp_hdr_verf(dreq, hdr)) {
+                                request_commit = true;
+                                if (nfs_direct_set_or_cmp_hdr_verf(dreq, hdr))
                                         dreq->flags =
                                                 NFS_ODIRECT_RESCHED_WRITES;
-                                        bit = NFS_IOHDR_NEED_RESCHED;
-                                } else
-                                        bit = NFS_IOHDR_NEED_COMMIT;
                         }
                 }
         }
@@ -759,9 +752,7 @@ static void nfs_direct_write_completion(struct nfs_pgio_header *hdr)
 
                 req = nfs_list_entry(hdr->pages.next);
                 nfs_list_remove_request(req);
-                switch (bit) {
-                case NFS_IOHDR_NEED_RESCHED:
-                case NFS_IOHDR_NEED_COMMIT:
+                if (request_commit) {
                         kref_get(&req->wb_kref);
                         nfs_mark_request_commit(req, hdr->lseg, &cinfo);
                 }
diff --git a/fs/nfs/filelayout/filelayout.c b/fs/nfs/filelayout/filelayout.c
index d2eba1c13b7e..1359c4a27393 100644
--- a/fs/nfs/filelayout/filelayout.c
+++ b/fs/nfs/filelayout/filelayout.c
@@ -84,45 +84,37 @@ filelayout_get_dserver_offset(struct pnfs_layout_segment *lseg, loff_t offset)
         BUG();
 }
 
-static void filelayout_reset_write(struct nfs_pgio_data *data)
+static void filelayout_reset_write(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-        struct rpc_task *task = &data->task;
+        struct rpc_task *task = &hdr->task;
 
         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags)) {
                 dprintk("%s Reset task %5u for i/o through MDS "
                         "(req %s/%llu, %u bytes @ offset %llu)\n", __func__,
-                        data->task.tk_pid,
+                        hdr->task.tk_pid,
                         hdr->inode->i_sb->s_id,
                         (unsigned long long)NFS_FILEID(hdr->inode),
-                        data->args.count,
-                        (unsigned long long)data->args.offset);
+                        hdr->args.count,
+                        (unsigned long long)hdr->args.offset);
 
-                task->tk_status = pnfs_write_done_resend_to_mds(hdr->inode,
-                                                        &hdr->pages,
-                                                        hdr->completion_ops,
-                                                        hdr->dreq);
+                task->tk_status = pnfs_write_done_resend_to_mds(hdr);
         }
 }
 
-static void filelayout_reset_read(struct nfs_pgio_data *data)
+static void filelayout_reset_read(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-        struct rpc_task *task = &data->task;
+        struct rpc_task *task = &hdr->task;
 
         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags)) {
                 dprintk("%s Reset task %5u for i/o through MDS "
                         "(req %s/%llu, %u bytes @ offset %llu)\n", __func__,
-                        data->task.tk_pid,
+                        hdr->task.tk_pid,
                         hdr->inode->i_sb->s_id,
                         (unsigned long long)NFS_FILEID(hdr->inode),
-                        data->args.count,
-                        (unsigned long long)data->args.offset);
+                        hdr->args.count,
+                        (unsigned long long)hdr->args.offset);
 
-                task->tk_status = pnfs_read_done_resend_to_mds(hdr->inode,
-                                                        &hdr->pages,
-                                                        hdr->completion_ops,
-                                                        hdr->dreq);
+                task->tk_status = pnfs_read_done_resend_to_mds(hdr);
         }
 }
 
@@ -243,18 +235,17 @@ wait_on_recovery:
 /* NFS_PROTO call done callback routines */
 
 static int filelayout_read_done_cb(struct rpc_task *task,
-                                struct nfs_pgio_data *data)
+                                struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
         int err;
 
-        trace_nfs4_pnfs_read(data, task->tk_status);
-        err = filelayout_async_handle_error(task, data->args.context->state,
-                                            data->ds_clp, hdr->lseg);
+        trace_nfs4_pnfs_read(hdr, task->tk_status);
+        err = filelayout_async_handle_error(task, hdr->args.context->state,
+                                            hdr->ds_clp, hdr->lseg);
 
         switch (err) {
         case -NFS4ERR_RESET_TO_MDS:
-                filelayout_reset_read(data);
+                filelayout_reset_read(hdr);
                 return task->tk_status;
         case -EAGAIN:
                 rpc_restart_call_prepare(task);
@@ -270,15 +261,14 @@ static int filelayout_read_done_cb(struct rpc_task *task,
  * rfc5661 is not clear about which credential should be used.
  */
 static void
-filelayout_set_layoutcommit(struct nfs_pgio_data *wdata)
+filelayout_set_layoutcommit(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = wdata->header;
 
         if (FILELAYOUT_LSEG(hdr->lseg)->commit_through_mds ||
-            wdata->res.verf->committed == NFS_FILE_SYNC)
+            hdr->res.verf->committed == NFS_FILE_SYNC)
                 return;
 
-        pnfs_set_layoutcommit(wdata);
+        pnfs_set_layoutcommit(hdr);
         dprintk("%s inode %lu pls_end_pos %lu\n", __func__, hdr->inode->i_ino,
                 (unsigned long) NFS_I(hdr->inode)->layout->plh_lwb);
 }
@@ -305,83 +295,82 @@ filelayout_reset_to_mds(struct pnfs_layout_segment *lseg)
  */
 static void filelayout_read_prepare(struct rpc_task *task, void *data)
 {
-        struct nfs_pgio_data *rdata = data;
+        struct nfs_pgio_header *hdr = data;
 
-        if (unlikely(test_bit(NFS_CONTEXT_BAD, &rdata->args.context->flags))) {
+        if (unlikely(test_bit(NFS_CONTEXT_BAD, &hdr->args.context->flags))) {
                 rpc_exit(task, -EIO);
                 return;
         }
-        if (filelayout_reset_to_mds(rdata->header->lseg)) {
+        if (filelayout_reset_to_mds(hdr->lseg)) {
                 dprintk("%s task %u reset io to MDS\n", __func__, task->tk_pid);
-                filelayout_reset_read(rdata);
+                filelayout_reset_read(hdr);
                 rpc_exit(task, 0);
                 return;
         }
-        rdata->pgio_done_cb = filelayout_read_done_cb;
+        hdr->pgio_done_cb = filelayout_read_done_cb;
 
-        if (nfs41_setup_sequence(rdata->ds_clp->cl_session,
-                        &rdata->args.seq_args,
-                        &rdata->res.seq_res,
+        if (nfs41_setup_sequence(hdr->ds_clp->cl_session,
+                        &hdr->args.seq_args,
+                        &hdr->res.seq_res,
                         task))
                 return;
-        if (nfs4_set_rw_stateid(&rdata->args.stateid, rdata->args.context,
-                        rdata->args.lock_context, FMODE_READ) == -EIO)
+        if (nfs4_set_rw_stateid(&hdr->args.stateid, hdr->args.context,
+                        hdr->args.lock_context, FMODE_READ) == -EIO)
                 rpc_exit(task, -EIO); /* lost lock, terminate I/O */
 }
 
 static void filelayout_read_call_done(struct rpc_task *task, void *data)
 {
-        struct nfs_pgio_data *rdata = data;
+        struct nfs_pgio_header *hdr = data;
 
         dprintk("--> %s task->tk_status %d\n", __func__, task->tk_status);
 
-        if (test_bit(NFS_IOHDR_REDO, &rdata->header->flags) &&
+        if (test_bit(NFS_IOHDR_REDO, &hdr->flags) &&
             task->tk_status == 0) {
-                nfs41_sequence_done(task, &rdata->res.seq_res);
+                nfs41_sequence_done(task, &hdr->res.seq_res);
                 return;
         }
 
         /* Note this may cause RPC to be resent */
-        rdata->header->mds_ops->rpc_call_done(task, data);
+        hdr->mds_ops->rpc_call_done(task, data);
 }
 
 static void filelayout_read_count_stats(struct rpc_task *task, void *data)
 {
-        struct nfs_pgio_data *rdata = data;
+        struct nfs_pgio_header *hdr = data;
 
-        rpc_count_iostats(task, NFS_SERVER(rdata->header->inode)->client->cl_metrics);
+        rpc_count_iostats(task, NFS_SERVER(hdr->inode)->client->cl_metrics);
 }
 
 static void filelayout_read_release(void *data)
 {
-        struct nfs_pgio_data *rdata = data;
-        struct pnfs_layout_hdr *lo = rdata->header->lseg->pls_layout;
+        struct nfs_pgio_header *hdr = data;
+        struct pnfs_layout_hdr *lo = hdr->lseg->pls_layout;
 
         filelayout_fenceme(lo->plh_inode, lo);
-        nfs_put_client(rdata->ds_clp);
-        rdata->header->mds_ops->rpc_release(data);
+        nfs_put_client(hdr->ds_clp);
+        hdr->mds_ops->rpc_release(data);
 }
 
 static int filelayout_write_done_cb(struct rpc_task *task,
-                                struct nfs_pgio_data *data)
+                                struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
         int err;
 
-        trace_nfs4_pnfs_write(data, task->tk_status);
-        err = filelayout_async_handle_error(task, data->args.context->state,
-                                            data->ds_clp, hdr->lseg);
+        trace_nfs4_pnfs_write(hdr, task->tk_status);
+        err = filelayout_async_handle_error(task, hdr->args.context->state,
+                                            hdr->ds_clp, hdr->lseg);
 
         switch (err) {
         case -NFS4ERR_RESET_TO_MDS:
-                filelayout_reset_write(data);
+                filelayout_reset_write(hdr);
                 return task->tk_status;
         case -EAGAIN:
                 rpc_restart_call_prepare(task);
                 return -EAGAIN;
         }
 
-        filelayout_set_layoutcommit(data);
+        filelayout_set_layoutcommit(hdr);
         return 0;
 }
 
@@ -419,57 +408,57 @@ static int filelayout_commit_done_cb(struct rpc_task *task,
 
 static void filelayout_write_prepare(struct rpc_task *task, void *data)
 {
-        struct nfs_pgio_data *wdata = data;
+        struct nfs_pgio_header *hdr = data;
 
-        if (unlikely(test_bit(NFS_CONTEXT_BAD, &wdata->args.context->flags))) {
+        if (unlikely(test_bit(NFS_CONTEXT_BAD, &hdr->args.context->flags))) {
                 rpc_exit(task, -EIO);
                 return;
         }
-        if (filelayout_reset_to_mds(wdata->header->lseg)) {
+        if (filelayout_reset_to_mds(hdr->lseg)) {
                 dprintk("%s task %u reset io to MDS\n", __func__, task->tk_pid);
-                filelayout_reset_write(wdata);
+                filelayout_reset_write(hdr);
                 rpc_exit(task, 0);
                 return;
         }
-        if (nfs41_setup_sequence(wdata->ds_clp->cl_session,
-                        &wdata->args.seq_args,
-                        &wdata->res.seq_res,
+        if (nfs41_setup_sequence(hdr->ds_clp->cl_session,
+                        &hdr->args.seq_args,
+                        &hdr->res.seq_res,
                         task))
                 return;
-        if (nfs4_set_rw_stateid(&wdata->args.stateid, wdata->args.context,
-                        wdata->args.lock_context, FMODE_WRITE) == -EIO)
+        if (nfs4_set_rw_stateid(&hdr->args.stateid, hdr->args.context,
+                        hdr->args.lock_context, FMODE_WRITE) == -EIO)
                 rpc_exit(task, -EIO); /* lost lock, terminate I/O */
 }
 
 static void filelayout_write_call_done(struct rpc_task *task, void *data)
 {
-        struct nfs_pgio_data *wdata = data;
+        struct nfs_pgio_header *hdr = data;
 
-        if (test_bit(NFS_IOHDR_REDO, &wdata->header->flags) &&
+        if (test_bit(NFS_IOHDR_REDO, &hdr->flags) &&
             task->tk_status == 0) {
-                nfs41_sequence_done(task, &wdata->res.seq_res);
+                nfs41_sequence_done(task, &hdr->res.seq_res);
                 return;
         }
 
         /* Note this may cause RPC to be resent */
-        wdata->header->mds_ops->rpc_call_done(task, data);
+        hdr->mds_ops->rpc_call_done(task, data);
 }
 
 static void filelayout_write_count_stats(struct rpc_task *task, void *data)
 {
-        struct nfs_pgio_data *wdata = data;
+        struct nfs_pgio_header *hdr = data;
 
-        rpc_count_iostats(task, NFS_SERVER(wdata->header->inode)->client->cl_metrics);
+        rpc_count_iostats(task, NFS_SERVER(hdr->inode)->client->cl_metrics);
 }
 
 static void filelayout_write_release(void *data)
 {
-        struct nfs_pgio_data *wdata = data;
-        struct pnfs_layout_hdr *lo = wdata->header->lseg->pls_layout;
+        struct nfs_pgio_header *hdr = data;
+        struct pnfs_layout_hdr *lo = hdr->lseg->pls_layout;
 
         filelayout_fenceme(lo->plh_inode, lo);
-        nfs_put_client(wdata->ds_clp);
-        wdata->header->mds_ops->rpc_release(data);
+        nfs_put_client(hdr->ds_clp);
+        hdr->mds_ops->rpc_release(data);
 }
 
 static void filelayout_commit_prepare(struct rpc_task *task, void *data)
@@ -529,19 +518,18 @@ static const struct rpc_call_ops filelayout_commit_call_ops = {
 };
 
 static enum pnfs_try_status
-filelayout_read_pagelist(struct nfs_pgio_data *data)
+filelayout_read_pagelist(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
         struct pnfs_layout_segment *lseg = hdr->lseg;
         struct nfs4_pnfs_ds *ds;
         struct rpc_clnt *ds_clnt;
-        loff_t offset = data->args.offset;
+        loff_t offset = hdr->args.offset;
         u32 j, idx;
         struct nfs_fh *fh;
 
         dprintk("--> %s ino %lu pgbase %u req %Zu@%llu\n",
                 __func__, hdr->inode->i_ino,
-                data->args.pgbase, (size_t)data->args.count, offset);
+                hdr->args.pgbase, (size_t)hdr->args.count, offset);
 
         /* Retrieve the correct rpc_client for the byte range */
         j = nfs4_fl_calc_j_index(lseg, offset);
@@ -559,30 +547,29 @@ filelayout_read_pagelist(struct nfs_pgio_data *data)
 
         /* No multipath support. Use first DS */
         atomic_inc(&ds->ds_clp->cl_count);
-        data->ds_clp = ds->ds_clp;
-        data->ds_idx = idx;
+        hdr->ds_clp = ds->ds_clp;
+        hdr->ds_idx = idx;
         fh = nfs4_fl_select_ds_fh(lseg, j);
         if (fh)
-                data->args.fh = fh;
+                hdr->args.fh = fh;
 
-        data->args.offset = filelayout_get_dserver_offset(lseg, offset);
-        data->mds_offset = offset;
+        hdr->args.offset = filelayout_get_dserver_offset(lseg, offset);
+        hdr->mds_offset = offset;
 
         /* Perform an asynchronous read to ds */
-        nfs_initiate_pgio(ds_clnt, data,
+        nfs_initiate_pgio(ds_clnt, hdr,
                             &filelayout_read_call_ops, 0, RPC_TASK_SOFTCONN);
         return PNFS_ATTEMPTED;
 }
 
 /* Perform async writes. */
 static enum pnfs_try_status
-filelayout_write_pagelist(struct nfs_pgio_data *data, int sync)
+filelayout_write_pagelist(struct nfs_pgio_header *hdr, int sync)
 {
-        struct nfs_pgio_header *hdr = data->header;
         struct pnfs_layout_segment *lseg = hdr->lseg;
         struct nfs4_pnfs_ds *ds;
         struct rpc_clnt *ds_clnt;
-        loff_t offset = data->args.offset;
+        loff_t offset = hdr->args.offset;
         u32 j, idx;
         struct nfs_fh *fh;
 
@@ -598,21 +585,20 @@ filelayout_write_pagelist(struct nfs_pgio_data *data, int sync)
                 return PNFS_NOT_ATTEMPTED;
 
         dprintk("%s ino %lu sync %d req %Zu@%llu DS: %s cl_count %d\n",
-                __func__, hdr->inode->i_ino, sync, (size_t) data->args.count,
+                __func__, hdr->inode->i_ino, sync, (size_t) hdr->args.count,
                 offset, ds->ds_remotestr, atomic_read(&ds->ds_clp->cl_count));
 
-        data->pgio_done_cb = filelayout_write_done_cb;
+        hdr->pgio_done_cb = filelayout_write_done_cb;
         atomic_inc(&ds->ds_clp->cl_count);
-        data->ds_clp = ds->ds_clp;
-        data->ds_idx = idx;
+        hdr->ds_clp = ds->ds_clp;
+        hdr->ds_idx = idx;
         fh = nfs4_fl_select_ds_fh(lseg, j);
         if (fh)
-                data->args.fh = fh;
-
-        data->args.offset = filelayout_get_dserver_offset(lseg, offset);
+                hdr->args.fh = fh;
+        hdr->args.offset = filelayout_get_dserver_offset(lseg, offset);
 
         /* Perform an asynchronous write */
-        nfs_initiate_pgio(ds_clnt, data,
+        nfs_initiate_pgio(ds_clnt, hdr,
                                     &filelayout_write_call_ops, sync,
                                     RPC_TASK_SOFTCONN);
         return PNFS_ATTEMPTED;
@@ -1023,6 +1009,7 @@ static u32 select_bucket_index(struct nfs4_filelayout_segment *fl, u32 j)
 
 /* The generic layer is about to remove the req from the commit list.
  * If this will make the bucket empty, it will need to put the lseg reference.
+ * Note this is must be called holding the inode (/cinfo) lock
  */
 static void
 filelayout_clear_request_commit(struct nfs_page *req,
@@ -1030,7 +1017,6 @@ filelayout_clear_request_commit(struct nfs_page *req,
 {
         struct pnfs_layout_segment *freeme = NULL;
 
-        spin_lock(cinfo->lock);
         if (!test_and_clear_bit(PG_COMMIT_TO_DS, &req->wb_flags))
                 goto out;
         cinfo->ds->nwritten--;
@@ -1045,22 +1031,25 @@ filelayout_clear_request_commit(struct nfs_page *req,
         }
 out:
         nfs_request_remove_commit_list(req, cinfo);
-        spin_unlock(cinfo->lock);
-        pnfs_put_lseg(freeme);
+        pnfs_put_lseg_async(freeme);
 }
 
-static struct list_head *
-filelayout_choose_commit_list(struct nfs_page *req,
-                              struct pnfs_layout_segment *lseg,
-                              struct nfs_commit_info *cinfo)
+static void
+filelayout_mark_request_commit(struct nfs_page *req,
+                               struct pnfs_layout_segment *lseg,
+                               struct nfs_commit_info *cinfo)
+
 {
         struct nfs4_filelayout_segment *fl = FILELAYOUT_LSEG(lseg);
         u32 i, j;
         struct list_head *list;
         struct pnfs_commit_bucket *buckets;
 
-        if (fl->commit_through_mds)
-                return &cinfo->mds->list;
+        if (fl->commit_through_mds) {
+                list = &cinfo->mds->list;
+                spin_lock(cinfo->lock);
+                goto mds_commit;
+        }
 
         /* Note that we are calling nfs4_fl_calc_j_index on each page
          * that ends up being committed to a data server.  An attractive
@@ -1084,19 +1073,22 @@ filelayout_choose_commit_list(struct nfs_page *req,
         }
         set_bit(PG_COMMIT_TO_DS, &req->wb_flags);
         cinfo->ds->nwritten++;
-        spin_unlock(cinfo->lock);
-        return list;
-}
 
-static void
-filelayout_mark_request_commit(struct nfs_page *req,
-                               struct pnfs_layout_segment *lseg,
-                               struct nfs_commit_info *cinfo)
-{
-        struct list_head *list;
-
-        list = filelayout_choose_commit_list(req, lseg, cinfo);
-        nfs_request_add_commit_list(req, list, cinfo);
+mds_commit:
+        /* nfs_request_add_commit_list(). We need to add req to list without
+         * dropping cinfo lock.
+         */
+        set_bit(PG_CLEAN, &(req)->wb_flags);
+        nfs_list_add_request(req, list);
+        cinfo->mds->ncommit++;
+        spin_unlock(cinfo->lock);
+        if (!cinfo->dreq) {
+                inc_zone_page_state(req->wb_page, NR_UNSTABLE_NFS);
+                inc_bdi_stat(page_file_mapping(req->wb_page)->backing_dev_info,
+                             BDI_RECLAIMABLE);
+                __mark_inode_dirty(req->wb_context->dentry->d_inode,
+                                   I_DIRTY_DATASYNC);
+        }
 }
 
 static u32 calc_ds_index_from_commit(struct pnfs_layout_segment *lseg, u32 i)
@@ -1244,15 +1236,63 @@ restart:
         spin_unlock(cinfo->lock);
 }
 
+/* filelayout_search_commit_reqs - Search lists in @cinfo for the head reqest
+ *                                 for @page
+ * @cinfo - commit info for current inode
+ * @page - page to search for matching head request
+ *
+ * Returns a the head request if one is found, otherwise returns NULL.
+ */
+static struct nfs_page *
+filelayout_search_commit_reqs(struct nfs_commit_info *cinfo, struct page *page)
+{
+        struct nfs_page *freq, *t;
+        struct pnfs_commit_bucket *b;
+        int i;
+
+        /* Linearly search the commit lists for each bucket until a matching
+         * request is found */
+        for (i = 0, b = cinfo->ds->buckets; i < cinfo->ds->nbuckets; i++, b++) {
+                list_for_each_entry_safe(freq, t, &b->written, wb_list) {
+                        if (freq->wb_page == page)
+                                return freq->wb_head;
+                }
+                list_for_each_entry_safe(freq, t, &b->committing, wb_list) {
+                        if (freq->wb_page == page)
+                                return freq->wb_head;
+                }
+        }
+
+        return NULL;
+}
+
+static void filelayout_retry_commit(struct nfs_commit_info *cinfo, int idx)
+{
+        struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
+        struct pnfs_commit_bucket *bucket = fl_cinfo->buckets;
+        struct pnfs_layout_segment *freeme;
+        int i;
+
+        for (i = idx; i < fl_cinfo->nbuckets; i++, bucket++) {
+                if (list_empty(&bucket->committing))
+                        continue;
+                nfs_retry_commit(&bucket->committing, bucket->clseg, cinfo);
+                spin_lock(cinfo->lock);
+                freeme = bucket->clseg;
+                bucket->clseg = NULL;
+                spin_unlock(cinfo->lock);
+                pnfs_put_lseg(freeme);
+        }
+}
+
 static unsigned int
 alloc_ds_commits(struct nfs_commit_info *cinfo, struct list_head *list)
 {
         struct pnfs_ds_commit_info *fl_cinfo;
         struct pnfs_commit_bucket *bucket;
         struct nfs_commit_data *data;
-        int i, j;
+        int i;
         unsigned int nreq = 0;
-        struct pnfs_layout_segment *freeme;
 
         fl_cinfo = cinfo->ds;
         bucket = fl_cinfo->buckets;
@@ -1272,16 +1312,7 @@ alloc_ds_commits(struct nfs_commit_info *cinfo, struct list_head *list)
         }
 
         /* Clean up on error */
-        for (j = i; j < fl_cinfo->nbuckets; j++, bucket++) {
-                if (list_empty(&bucket->committing))
-                        continue;
-                nfs_retry_commit(&bucket->committing, bucket->clseg, cinfo);
-                spin_lock(cinfo->lock);
-                freeme = bucket->clseg;
-                bucket->clseg = NULL;
-                spin_unlock(cinfo->lock);
-                pnfs_put_lseg(freeme);
-        }
+        filelayout_retry_commit(cinfo, i);
         /* Caller will clean up entries put on list */
         return nreq;
 }
@@ -1301,8 +1332,12 @@ filelayout_commit_pagelist(struct inode *inode, struct list_head *mds_pages,
                         data->lseg = NULL;
                         list_add(&data->pages, &list);
                         nreq++;
-                } else
+                } else {
                         nfs_retry_commit(mds_pages, NULL, cinfo);
+                        filelayout_retry_commit(cinfo, 0);
+                        cinfo->completion_ops->error_cleanup(NFS_I(inode));
+                        return -ENOMEM;
+                }
         }
 
         nreq += alloc_ds_commits(cinfo, &list);
@@ -1380,6 +1415,7 @@ static struct pnfs_layoutdriver_type filelayout_type = {
         .clear_request_commit   = filelayout_clear_request_commit,
         .scan_commit_lists      = filelayout_scan_commit_lists,
         .recover_commit_reqs    = filelayout_recover_commit_reqs,
+        .search_commit_reqs     = filelayout_search_commit_reqs,
         .commit_pagelist        = filelayout_commit_pagelist,
         .read_pagelist          = filelayout_read_pagelist,
         .write_pagelist         = filelayout_write_pagelist,
diff --git a/fs/nfs/filelayout/filelayoutdev.c b/fs/nfs/filelayout/filelayoutdev.c
index e2a0361e24c6..8540516f4d71 100644
--- a/fs/nfs/filelayout/filelayoutdev.c
+++ b/fs/nfs/filelayout/filelayoutdev.c
@@ -695,7 +695,7 @@ filelayout_get_device_info(struct inode *inode,
         if (pdev == NULL)
                 return NULL;
 
-        pages = kzalloc(max_pages * sizeof(struct page *), gfp_flags);
+        pages = kcalloc(max_pages, sizeof(struct page *), gfp_flags);
         if (pages == NULL) {
                 kfree(pdev);
                 return NULL;
diff --git a/fs/nfs/inode.c b/fs/nfs/inode.c
index 68921b01b792..577a36f0a510 100644
--- a/fs/nfs/inode.c
+++ b/fs/nfs/inode.c
@@ -1002,6 +1002,15 @@ int nfs_revalidate_inode(struct nfs_server *server, struct inode *inode)
 }
 EXPORT_SYMBOL_GPL(nfs_revalidate_inode);
 
+int nfs_revalidate_inode_rcu(struct nfs_server *server, struct inode *inode)
+{
+        if (!(NFS_I(inode)->cache_validity &
+                        (NFS_INO_INVALID_ATTR|NFS_INO_INVALID_LABEL))
+                        && !nfs_attribute_cache_expired(inode))
+                return NFS_STALE(inode) ? -ESTALE : 0;
+        return -ECHILD;
+}
+
 static int nfs_invalidate_mapping(struct inode *inode, struct address_space *mapping)
 {
         struct nfs_inode *nfsi = NFS_I(inode);
diff --git a/fs/nfs/internal.h b/fs/nfs/internal.h
index e2a45ae5014e..9056622d2230 100644
--- a/fs/nfs/internal.h
+++ b/fs/nfs/internal.h
@@ -247,11 +247,11 @@ void nfs_set_pgio_error(struct nfs_pgio_header *hdr, int error, loff_t pos);
 int nfs_iocounter_wait(struct nfs_io_counter *c);
 
 extern const struct nfs_pageio_ops nfs_pgio_rw_ops;
-struct nfs_rw_header *nfs_rw_header_alloc(const struct nfs_rw_ops *);
-void nfs_rw_header_free(struct nfs_pgio_header *);
-void nfs_pgio_data_release(struct nfs_pgio_data *);
+struct nfs_pgio_header *nfs_pgio_header_alloc(const struct nfs_rw_ops *);
+void nfs_pgio_header_free(struct nfs_pgio_header *);
+void nfs_pgio_data_destroy(struct nfs_pgio_header *);
 int nfs_generic_pgio(struct nfs_pageio_descriptor *, struct nfs_pgio_header *);
-int nfs_initiate_pgio(struct rpc_clnt *, struct nfs_pgio_data *,
+int nfs_initiate_pgio(struct rpc_clnt *, struct nfs_pgio_header *,
                       const struct rpc_call_ops *, int, int);
 void nfs_free_request(struct nfs_page *req);
 
@@ -451,6 +451,7 @@ int nfs_scan_commit(struct inode *inode, struct list_head *dst,
 void nfs_mark_request_commit(struct nfs_page *req,
                              struct pnfs_layout_segment *lseg,
                              struct nfs_commit_info *cinfo);
+int nfs_write_need_commit(struct nfs_pgio_header *);
 int nfs_generic_commit_list(struct inode *inode, struct list_head *head,
                             int how, struct nfs_commit_info *cinfo);
 void nfs_retry_commit(struct list_head *page_list,
@@ -491,7 +492,7 @@ static inline void nfs_inode_dio_wait(struct inode *inode)
 extern ssize_t nfs_dreq_bytes_left(struct nfs_direct_req *dreq);
 
 /* nfs4proc.c */
-extern void __nfs4_read_done_cb(struct nfs_pgio_data *);
+extern void __nfs4_read_done_cb(struct nfs_pgio_header *);
 extern struct nfs_client *nfs4_init_client(struct nfs_client *clp,
                             const struct rpc_timeout *timeparms,
                             const char *ip_addr);
diff --git a/fs/nfs/nfs3acl.c b/fs/nfs/nfs3acl.c
index 8f854dde4150..d0fec260132a 100644
--- a/fs/nfs/nfs3acl.c
+++ b/fs/nfs/nfs3acl.c
@@ -256,7 +256,7 @@ nfs3_list_one_acl(struct inode *inode, int type, const char *name, void *data,
         char *p = data + *result;
 
         acl = get_acl(inode, type);
-        if (!acl)
+        if (IS_ERR_OR_NULL(acl))
                 return 0;
 
         posix_acl_release(acl);
diff --git a/fs/nfs/nfs3proc.c b/fs/nfs/nfs3proc.c
index f0afa291fd58..809670eba52a 100644
--- a/fs/nfs/nfs3proc.c
+++ b/fs/nfs/nfs3proc.c
@@ -795,41 +795,44 @@ nfs3_proc_pathconf(struct nfs_server *server, struct nfs_fh *fhandle,
         return status;
 }
 
-static int nfs3_read_done(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs3_read_done(struct rpc_task *task, struct nfs_pgio_header *hdr)
 {
-        struct inode *inode = data->header->inode;
+        struct inode *inode = hdr->inode;
 
         if (nfs3_async_handle_jukebox(task, inode))
                 return -EAGAIN;
 
         nfs_invalidate_atime(inode);
-        nfs_refresh_inode(inode, &data->fattr);
+        nfs_refresh_inode(inode, &hdr->fattr);
         return 0;
 }
 
-static void nfs3_proc_read_setup(struct nfs_pgio_data *data, struct rpc_message *msg)
+static void nfs3_proc_read_setup(struct nfs_pgio_header *hdr,
+                                 struct rpc_message *msg)
 {
         msg->rpc_proc = &nfs3_procedures[NFS3PROC_READ];
 }
 
-static int nfs3_proc_pgio_rpc_prepare(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs3_proc_pgio_rpc_prepare(struct rpc_task *task,
+                                      struct nfs_pgio_header *hdr)
 {
         rpc_call_start(task);
         return 0;
 }
 
-static int nfs3_write_done(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs3_write_done(struct rpc_task *task, struct nfs_pgio_header *hdr)
 {
-        struct inode *inode = data->header->inode;
+        struct inode *inode = hdr->inode;
 
         if (nfs3_async_handle_jukebox(task, inode))
                 return -EAGAIN;
         if (task->tk_status >= 0)
-                nfs_post_op_update_inode_force_wcc(inode, data->res.fattr);
+                nfs_post_op_update_inode_force_wcc(inode, hdr->res.fattr);
         return 0;
 }
 
-static void nfs3_proc_write_setup(struct nfs_pgio_data *data, struct rpc_message *msg)
+static void nfs3_proc_write_setup(struct nfs_pgio_header *hdr,
+                                  struct rpc_message *msg)
 {
         msg->rpc_proc = &nfs3_procedures[NFS3PROC_WRITE];
 }
diff --git a/fs/nfs/nfs4_fs.h b/fs/nfs/nfs4_fs.h
index ba2affa51941..92193eddb41d 100644
--- a/fs/nfs/nfs4_fs.h
+++ b/fs/nfs/nfs4_fs.h
@@ -54,7 +54,7 @@ struct nfs4_minor_version_ops {
                         const nfs4_stateid *);
         int     (*find_root_sec)(struct nfs_server *, struct nfs_fh *,
                         struct nfs_fsinfo *);
-        int     (*free_lock_state)(struct nfs_server *,
+        void    (*free_lock_state)(struct nfs_server *,
                         struct nfs4_lock_state *);
         const struct rpc_call_ops *call_sync_ops;
         const struct nfs4_state_recovery_ops *reboot_recovery_ops;
@@ -129,27 +129,17 @@ enum {
  * LOCK: one nfs4_state (LOCK) to hold the lock stateid nfs4_state(OPEN)
  */
 
-struct nfs4_lock_owner {
-        unsigned int lo_type;
-#define NFS4_ANY_LOCK_TYPE      (0U)
-#define NFS4_FLOCK_LOCK_TYPE    (1U << 0)
-#define NFS4_POSIX_LOCK_TYPE    (1U << 1)
-        union {
-                fl_owner_t posix_owner;
-                pid_t flock_owner;
-        } lo_u;
-};
-
 struct nfs4_lock_state {
-        struct list_head        ls_locks;       /* Other lock stateids */
-        struct nfs4_state *     ls_state;       /* Pointer to open state */
+        struct list_head                ls_locks;   /* Other lock stateids */
+        struct nfs4_state *             ls_state;   /* Pointer to open state */
 #define NFS_LOCK_INITIALIZED 0
 #define NFS_LOCK_LOST        1
-        unsigned long           ls_flags;
+        unsigned long                   ls_flags;
         struct nfs_seqid_counter        ls_seqid;
-        nfs4_stateid            ls_stateid;
-        atomic_t                ls_count;
-        struct nfs4_lock_owner  ls_owner;
+        nfs4_stateid                    ls_stateid;
+        atomic_t                        ls_count;
+        fl_owner_t                      ls_owner;
+        struct work_struct              ls_release;
 };
 
 /* bits for nfs4_state->flags */
@@ -337,11 +327,11 @@ nfs4_state_protect(struct nfs_client *clp, unsigned long sp4_mode,
  */
 static inline void
 nfs4_state_protect_write(struct nfs_client *clp, struct rpc_clnt **clntp,
-                         struct rpc_message *msg, struct nfs_pgio_data *wdata)
+                         struct rpc_message *msg, struct nfs_pgio_header *hdr)
 {
         if (_nfs4_state_protect(clp, NFS_SP4_MACH_CRED_WRITE, clntp, msg) &&
             !test_bit(NFS_SP4_MACH_CRED_COMMIT, &clp->cl_sp4_flags))
-                wdata->args.stable = NFS_FILE_SYNC;
+                hdr->args.stable = NFS_FILE_SYNC;
 }
 #else /* CONFIG_NFS_v4_1 */
 static inline struct nfs4_session *nfs4_get_session(const struct nfs_server *server)
@@ -369,7 +359,7 @@ nfs4_state_protect(struct nfs_client *clp, unsigned long sp4_flags,
 
 static inline void
 nfs4_state_protect_write(struct nfs_client *clp, struct rpc_clnt **clntp,
-                         struct rpc_message *msg, struct nfs_pgio_data *wdata)
+                         struct rpc_message *msg, struct nfs_pgio_header *hdr)
 {
 }
 #endif /* CONFIG_NFS_V4_1 */
diff --git a/fs/nfs/nfs4client.c b/fs/nfs/nfs4client.c
index aa9ef4876046..53e435a95260 100644
--- a/fs/nfs/nfs4client.c
+++ b/fs/nfs/nfs4client.c
@@ -855,6 +855,11 @@ struct nfs_client *nfs4_set_ds_client(struct nfs_client* mds_clp,
         };
         struct rpc_timeout ds_timeout;
         struct nfs_client *clp;
+        char buf[INET6_ADDRSTRLEN + 1];
+
+        if (rpc_ntop(ds_addr, buf, sizeof(buf)) <= 0)
+                return ERR_PTR(-EINVAL);
+        cl_init.hostname = buf;
 
         /*
          * Set an authflavor equual to the MDS value. Use the MDS nfs_client
diff --git a/fs/nfs/nfs4proc.c b/fs/nfs/nfs4proc.c
index 4bf3d97cc5a0..75ae8d22f067 100644
--- a/fs/nfs/nfs4proc.c
+++ b/fs/nfs/nfs4proc.c
@@ -1952,6 +1952,14 @@ static int _nfs4_recover_proc_open(struct nfs4_opendata *data)
         return status;
 }
 
+/*
+ * Additional permission checks in order to distinguish between an
+ * open for read, and an open for execute. This works around the
+ * fact that NFSv4 OPEN treats read and execute permissions as being
+ * the same.
+ * Note that in the non-execute case, we want to turn off permission
+ * checking if we just created a new file (POSIX open() semantics).
+ */
 static int nfs4_opendata_access(struct rpc_cred *cred,
                                 struct nfs4_opendata *opendata,
                                 struct nfs4_state *state, fmode_t fmode,
@@ -1966,14 +1974,14 @@ static int nfs4_opendata_access(struct rpc_cred *cred,
                 return 0;
 
         mask = 0;
-        /* don't check MAY_WRITE - a newly created file may not have
-         * write mode bits, but POSIX allows the creating process to write.
-         * use openflags to check for exec, because fmode won't
-         * always have FMODE_EXEC set when file open for exec. */
+        /*
+         * Use openflags to check for exec, because fmode won't
+         * always have FMODE_EXEC set when file open for exec.
+         */
         if (openflags & __FMODE_EXEC) {
                 /* ONLY check for exec rights */
                 mask = MAY_EXEC;
-        } else if (fmode & FMODE_READ)
+        } else if ((fmode & FMODE_READ) && !opendata->file_created)
                 mask = MAY_READ;
 
         cache.cred = cred;
@@ -2216,8 +2224,15 @@ static int _nfs4_open_and_get_state(struct nfs4_opendata *opendata,
         seq = raw_seqcount_begin(&sp->so_reclaim_seqcount);
 
         ret = _nfs4_proc_open(opendata);
-        if (ret != 0)
+        if (ret != 0) {
+                if (ret == -ENOENT) {
+                        d_drop(opendata->dentry);
+                        d_add(opendata->dentry, NULL);
+                        nfs_set_verifier(opendata->dentry,
+                                         nfs_save_change_attribute(opendata->dir->d_inode));
+                }
                 goto out;
+        }
 
         state = nfs4_opendata_to_nfs4_state(opendata);
         ret = PTR_ERR(state);
@@ -2647,6 +2662,48 @@ static const struct rpc_call_ops nfs4_close_ops = {
         .rpc_release = nfs4_free_closedata,
 };
 
+static bool nfs4_state_has_opener(struct nfs4_state *state)
+{
+        /* first check existing openers */
+        if (test_bit(NFS_O_RDONLY_STATE, &state->flags) != 0 &&
+            state->n_rdonly != 0)
+                return true;
+
+        if (test_bit(NFS_O_WRONLY_STATE, &state->flags) != 0 &&
+            state->n_wronly != 0)
+                return true;
+
+        if (test_bit(NFS_O_RDWR_STATE, &state->flags) != 0 &&
+            state->n_rdwr != 0)
+                return true;
+
+        return false;
+}
+
+static bool nfs4_roc(struct inode *inode)
+{
+        struct nfs_inode *nfsi = NFS_I(inode);
+        struct nfs_open_context *ctx;
+        struct nfs4_state *state;
+
+        spin_lock(&inode->i_lock);
+        list_for_each_entry(ctx, &nfsi->open_files, list) {
+                state = ctx->state;
+                if (state == NULL)
+                        continue;
+                if (nfs4_state_has_opener(state)) {
+                        spin_unlock(&inode->i_lock);
+                        return false;
+                }
+        }
+        spin_unlock(&inode->i_lock);
+
+        if (nfs4_check_delegation(inode, FMODE_READ))
+                return false;
+
+        return pnfs_roc(inode);
+}
+
 /* 
  * It is possible for data to be read/written from a mem-mapped file 
  * after the sys_close call (which hits the vfs layer as a flush).
@@ -2697,7 +2754,7 @@ int nfs4_do_close(struct nfs4_state *state, gfp_t gfp_mask, int wait)
         calldata->res.fattr = &calldata->fattr;
         calldata->res.seqid = calldata->arg.seqid;
         calldata->res.server = server;
-        calldata->roc = pnfs_roc(state->inode);
+        calldata->roc = nfs4_roc(state->inode);
         nfs_sb_active(calldata->inode->i_sb);
 
         msg.rpc_argp = &calldata->arg;
@@ -4033,24 +4090,25 @@ static bool nfs4_error_stateid_expired(int err)
         return false;
 }
 
-void __nfs4_read_done_cb(struct nfs_pgio_data *data)
+void __nfs4_read_done_cb(struct nfs_pgio_header *hdr)
 {
-        nfs_invalidate_atime(data->header->inode);
+        nfs_invalidate_atime(hdr->inode);
 }
 
-static int nfs4_read_done_cb(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs4_read_done_cb(struct rpc_task *task, struct nfs_pgio_header *hdr)
 {
-        struct nfs_server *server = NFS_SERVER(data->header->inode);
+        struct nfs_server *server = NFS_SERVER(hdr->inode);
 
-        trace_nfs4_read(data, task->tk_status);
-        if (nfs4_async_handle_error(task, server, data->args.context->state) == -EAGAIN) {
+        trace_nfs4_read(hdr, task->tk_status);
+        if (nfs4_async_handle_error(task, server,
+                                    hdr->args.context->state) == -EAGAIN) {
                 rpc_restart_call_prepare(task);
                 return -EAGAIN;
         }
 
-        __nfs4_read_done_cb(data);
+        __nfs4_read_done_cb(hdr);
         if (task->tk_status > 0)
-                renew_lease(server, data->timestamp);
+                renew_lease(server, hdr->timestamp);
         return 0;
 }
 
@@ -4068,54 +4126,59 @@ static bool nfs4_read_stateid_changed(struct rpc_task *task,
         return true;
 }
 
-static int nfs4_read_done(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs4_read_done(struct rpc_task *task, struct nfs_pgio_header *hdr)
 {
 
         dprintk("--> %s\n", __func__);
 
-        if (!nfs4_sequence_done(task, &data->res.seq_res))
+        if (!nfs4_sequence_done(task, &hdr->res.seq_res))
                 return -EAGAIN;
-        if (nfs4_read_stateid_changed(task, &data->args))
+        if (nfs4_read_stateid_changed(task, &hdr->args))
                 return -EAGAIN;
-        return data->pgio_done_cb ? data->pgio_done_cb(task, data) :
-                                    nfs4_read_done_cb(task, data);
+        return hdr->pgio_done_cb ? hdr->pgio_done_cb(task, hdr) :
+                                    nfs4_read_done_cb(task, hdr);
 }
 
-static void nfs4_proc_read_setup(struct nfs_pgio_data *data, struct rpc_message *msg)
+static void nfs4_proc_read_setup(struct nfs_pgio_header *hdr,
+                                 struct rpc_message *msg)
 {
-        data->timestamp   = jiffies;
-        data->pgio_done_cb = nfs4_read_done_cb;
+        hdr->timestamp   = jiffies;
+        hdr->pgio_done_cb = nfs4_read_done_cb;
         msg->rpc_proc = &nfs4_procedures[NFSPROC4_CLNT_READ];
-        nfs4_init_sequence(&data->args.seq_args, &data->res.seq_res, 0);
+        nfs4_init_sequence(&hdr->args.seq_args, &hdr->res.seq_res, 0);
 }
 
-static int nfs4_proc_pgio_rpc_prepare(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs4_proc_pgio_rpc_prepare(struct rpc_task *task,
+                                      struct nfs_pgio_header *hdr)
 {
-        if (nfs4_setup_sequence(NFS_SERVER(data->header->inode),
-                        &data->args.seq_args,
-                        &data->res.seq_res,
+        if (nfs4_setup_sequence(NFS_SERVER(hdr->inode),
+                        &hdr->args.seq_args,
+                        &hdr->res.seq_res,
                         task))
                 return 0;
-        if (nfs4_set_rw_stateid(&data->args.stateid, data->args.context,
-                                data->args.lock_context, data->header->rw_ops->rw_mode) == -EIO)
+        if (nfs4_set_rw_stateid(&hdr->args.stateid, hdr->args.context,
+                                hdr->args.lock_context,
+                                hdr->rw_ops->rw_mode) == -EIO)
                 return -EIO;
-        if (unlikely(test_bit(NFS_CONTEXT_BAD, &data->args.context->flags)))
+        if (unlikely(test_bit(NFS_CONTEXT_BAD, &hdr->args.context->flags)))
                 return -EIO;
         return 0;
 }
 
-static int nfs4_write_done_cb(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs4_write_done_cb(struct rpc_task *task,
+                              struct nfs_pgio_header *hdr)
 {
-        struct inode *inode = data->header->inode;
+        struct inode *inode = hdr->inode;
         
-        trace_nfs4_write(data, task->tk_status);
-        if (nfs4_async_handle_error(task, NFS_SERVER(inode), data->args.context->state) == -EAGAIN) {
+        trace_nfs4_write(hdr, task->tk_status);
+        if (nfs4_async_handle_error(task, NFS_SERVER(inode),
+                                    hdr->args.context->state) == -EAGAIN) {
                 rpc_restart_call_prepare(task);
                 return -EAGAIN;
         }
         if (task->tk_status >= 0) {
-                renew_lease(NFS_SERVER(inode), data->timestamp);
-                nfs_post_op_update_inode_force_wcc(inode, &data->fattr);
+                renew_lease(NFS_SERVER(inode), hdr->timestamp);
+                nfs_post_op_update_inode_force_wcc(inode, &hdr->fattr);
         }
         return 0;
 }
@@ -4134,23 +4197,21 @@ static bool nfs4_write_stateid_changed(struct rpc_task *task,
         return true;
 }
 
-static int nfs4_write_done(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs4_write_done(struct rpc_task *task, struct nfs_pgio_header *hdr)
 {
-        if (!nfs4_sequence_done(task, &data->res.seq_res))
+        if (!nfs4_sequence_done(task, &hdr->res.seq_res))
                 return -EAGAIN;
-        if (nfs4_write_stateid_changed(task, &data->args))
+        if (nfs4_write_stateid_changed(task, &hdr->args))
                 return -EAGAIN;
-        return data->pgio_done_cb ? data->pgio_done_cb(task, data) :
-                nfs4_write_done_cb(task, data);
+        return hdr->pgio_done_cb ? hdr->pgio_done_cb(task, hdr) :
+                nfs4_write_done_cb(task, hdr);
 }
 
 static
-bool nfs4_write_need_cache_consistency_data(const struct nfs_pgio_data *data)
+bool nfs4_write_need_cache_consistency_data(struct nfs_pgio_header *hdr)
 {
-        const struct nfs_pgio_header *hdr = data->header;
-
         /* Don't request attributes for pNFS or O_DIRECT writes */
-        if (data->ds_clp != NULL || hdr->dreq != NULL)
+        if (hdr->ds_clp != NULL || hdr->dreq != NULL)
                 return false;
         /* Otherwise, request attributes if and only if we don't hold
          * a delegation
@@ -4158,23 +4219,24 @@ bool nfs4_write_need_cache_consistency_data(const struct nfs_pgio_data *data)
         return nfs4_have_delegation(hdr->inode, FMODE_READ) == 0;
 }
 
-static void nfs4_proc_write_setup(struct nfs_pgio_data *data, struct rpc_message *msg)
+static void nfs4_proc_write_setup(struct nfs_pgio_header *hdr,
+                                  struct rpc_message *msg)
 {
-        struct nfs_server *server = NFS_SERVER(data->header->inode);
+        struct nfs_server *server = NFS_SERVER(hdr->inode);
 
-        if (!nfs4_write_need_cache_consistency_data(data)) {
-                data->args.bitmask = NULL;
-                data->res.fattr = NULL;
+        if (!nfs4_write_need_cache_consistency_data(hdr)) {
+                hdr->args.bitmask = NULL;
+                hdr->res.fattr = NULL;
         } else
-                data->args.bitmask = server->cache_consistency_bitmask;
+                hdr->args.bitmask = server->cache_consistency_bitmask;
 
-        if (!data->pgio_done_cb)
-                data->pgio_done_cb = nfs4_write_done_cb;
-        data->res.server = server;
-        data->timestamp   = jiffies;
+        if (!hdr->pgio_done_cb)
+                hdr->pgio_done_cb = nfs4_write_done_cb;
+        hdr->res.server = server;
+        hdr->timestamp   = jiffies;
 
         msg->rpc_proc = &nfs4_procedures[NFSPROC4_CLNT_WRITE];
-        nfs4_init_sequence(&data->args.seq_args, &data->res.seq_res, 1);
+        nfs4_init_sequence(&hdr->args.seq_args, &hdr->res.seq_res, 1);
 }
 
 static void nfs4_proc_commit_rpc_prepare(struct rpc_task *task, struct nfs_commit_data *data)
@@ -4881,6 +4943,18 @@ nfs4_init_callback_netid(const struct nfs_client *clp, char *buf, size_t len)
                 return scnprintf(buf, len, "tcp");
 }
 
+static void nfs4_setclientid_done(struct rpc_task *task, void *calldata)
+{
+        struct nfs4_setclientid *sc = calldata;
+
+        if (task->tk_status == 0)
+                sc->sc_cred = get_rpccred(task->tk_rqstp->rq_cred);
+}
+
+static const struct rpc_call_ops nfs4_setclientid_ops = {
+        .rpc_call_done = nfs4_setclientid_done,
+};
+
 /**
  * nfs4_proc_setclientid - Negotiate client ID
  * @clp: state data structure
@@ -4907,6 +4981,14 @@ int nfs4_proc_setclientid(struct nfs_client *clp, u32 program,
                 .rpc_resp = res,
                 .rpc_cred = cred,
         };
+        struct rpc_task *task;
+        struct rpc_task_setup task_setup_data = {
+                .rpc_client = clp->cl_rpcclient,
+                .rpc_message = &msg,
+                .callback_ops = &nfs4_setclientid_ops,
+                .callback_data = &setclientid,
+                .flags = RPC_TASK_TIMEOUT,
+        };
         int status;
 
         /* nfs_client_id4 */
@@ -4933,7 +5015,18 @@ int nfs4_proc_setclientid(struct nfs_client *clp, u32 program,
         dprintk("NFS call  setclientid auth=%s, '%.*s'\n",
                 clp->cl_rpcclient->cl_auth->au_ops->au_name,
                 setclientid.sc_name_len, setclientid.sc_name);
-        status = rpc_call_sync(clp->cl_rpcclient, &msg, RPC_TASK_TIMEOUT);
+        task = rpc_run_task(&task_setup_data);
+        if (IS_ERR(task)) {
+                status = PTR_ERR(task);
+                goto out;
+        }
+        status = task->tk_status;
+        if (setclientid.sc_cred) {
+                clp->cl_acceptor = rpcauth_stringify_acceptor(setclientid.sc_cred);
+                put_rpccred(setclientid.sc_cred);
+        }
+        rpc_put_task(task);
+out:
         trace_nfs4_setclientid(clp, status);
         dprintk("NFS reply setclientid: %d\n", status);
         return status;
@@ -4975,6 +5068,9 @@ struct nfs4_delegreturndata {
         unsigned long timestamp;
         struct nfs_fattr fattr;
         int rpc_status;
+        struct inode *inode;
+        bool roc;
+        u32 roc_barrier;
 };
 
 static void nfs4_delegreturn_done(struct rpc_task *task, void *calldata)
@@ -4988,7 +5084,6 @@ static void nfs4_delegreturn_done(struct rpc_task *task, void *calldata)
         switch (task->tk_status) {
         case 0:
                 renew_lease(data->res.server, data->timestamp);
-                break;
         case -NFS4ERR_ADMIN_REVOKED:
         case -NFS4ERR_DELEG_REVOKED:
         case -NFS4ERR_BAD_STATEID:
@@ -4996,6 +5091,8 @@ static void nfs4_delegreturn_done(struct rpc_task *task, void *calldata)
         case -NFS4ERR_STALE_STATEID:
         case -NFS4ERR_EXPIRED:
                 task->tk_status = 0;
+                if (data->roc)
+                        pnfs_roc_set_barrier(data->inode, data->roc_barrier);
                 break;
         default:
                 if (nfs4_async_handle_error(task, data->res.server, NULL) ==
@@ -5009,6 +5106,10 @@ static void nfs4_delegreturn_done(struct rpc_task *task, void *calldata)
 
 static void nfs4_delegreturn_release(void *calldata)
 {
+        struct nfs4_delegreturndata *data = calldata;
+
+        if (data->roc)
+                pnfs_roc_release(data->inode);
         kfree(calldata);
 }
 
@@ -5018,6 +5119,10 @@ static void nfs4_delegreturn_prepare(struct rpc_task *task, void *data)
 
         d_data = (struct nfs4_delegreturndata *)data;
 
+        if (d_data->roc &&
+            pnfs_roc_drain(d_data->inode, &d_data->roc_barrier, task))
+                return;
+
         nfs4_setup_sequence(d_data->res.server,
                         &d_data->args.seq_args,
                         &d_data->res.seq_res,
@@ -5061,6 +5166,9 @@ static int _nfs4_proc_delegreturn(struct inode *inode, struct rpc_cred *cred, co
         nfs_fattr_init(data->res.fattr);
         data->timestamp = jiffies;
         data->rpc_status = 0;
+        data->inode = inode;
+        data->roc = list_empty(&NFS_I(inode)->open_files) ?
+                    pnfs_roc(inode) : false;
 
         task_setup_data.callback_data = data;
         msg.rpc_argp = &data->args;
@@ -5834,8 +5942,10 @@ struct nfs_release_lockowner_data {
 static void nfs4_release_lockowner_prepare(struct rpc_task *task, void *calldata)
 {
         struct nfs_release_lockowner_data *data = calldata;
-        nfs40_setup_sequence(data->server,
-                                &data->args.seq_args, &data->res.seq_res, task);
+        struct nfs_server *server = data->server;
+        nfs40_setup_sequence(server, &data->args.seq_args,
+                                &data->res.seq_res, task);
+        data->args.lock_owner.clientid = server->nfs_client->cl_clientid;
         data->timestamp = jiffies;
 }
 
@@ -5852,6 +5962,8 @@ static void nfs4_release_lockowner_done(struct rpc_task *task, void *calldata)
                 break;
         case -NFS4ERR_STALE_CLIENTID:
         case -NFS4ERR_EXPIRED:
+                nfs4_schedule_lease_recovery(server->nfs_client);
+                break;
         case -NFS4ERR_LEASE_MOVED:
         case -NFS4ERR_DELAY:
                 if (nfs4_async_handle_error(task, server, NULL) == -EAGAIN)
@@ -5872,7 +5984,8 @@ static const struct rpc_call_ops nfs4_release_lockowner_ops = {
         .rpc_release = nfs4_release_lockowner_release,
 };
 
-static int nfs4_release_lockowner(struct nfs_server *server, struct nfs4_lock_state *lsp)
+static void
+nfs4_release_lockowner(struct nfs_server *server, struct nfs4_lock_state *lsp)
 {
         struct nfs_release_lockowner_data *data;
         struct rpc_message msg = {
@@ -5880,11 +5993,11 @@ static int nfs4_release_lockowner(struct nfs_server *server, struct nfs4_lock_st
         };
 
         if (server->nfs_client->cl_mvops->minor_version != 0)
-                return -EINVAL;
+                return;
 
         data = kmalloc(sizeof(*data), GFP_NOFS);
         if (!data)
-                return -ENOMEM;
+                return;
         data->lsp = lsp;
         data->server = server;
         data->args.lock_owner.clientid = server->nfs_client->cl_clientid;
@@ -5895,7 +6008,6 @@ static int nfs4_release_lockowner(struct nfs_server *server, struct nfs4_lock_st
         msg.rpc_resp = &data->res;
         nfs4_init_sequence(&data->args.seq_args, &data->res.seq_res, 0);
         rpc_call_async(server->client, &msg, 0, &nfs4_release_lockowner_ops, data);
-        return 0;
 }
 
 #define XATTR_NAME_NFSV4_ACL "system.nfs4_acl"
@@ -8182,7 +8294,8 @@ static int nfs41_free_stateid(struct nfs_server *server,
         return ret;
 }
 
-static int nfs41_free_lock_state(struct nfs_server *server, struct nfs4_lock_state *lsp)
+static void
+nfs41_free_lock_state(struct nfs_server *server, struct nfs4_lock_state *lsp)
 {
         struct rpc_task *task;
         struct rpc_cred *cred = lsp->ls_state->owner->so_cred;
@@ -8190,9 +8303,8 @@ static int nfs41_free_lock_state(struct nfs_server *server, struct nfs4_lock_sta
         task = _nfs41_free_stateid(server, &lsp->ls_stateid, cred, false);
         nfs4_free_lock_state(server, lsp);
         if (IS_ERR(task))
-                return PTR_ERR(task);
+                return;
         rpc_put_task(task);
-        return 0;
 }
 
 static bool nfs41_match_stateid(const nfs4_stateid *s1,
diff --git a/fs/nfs/nfs4state.c b/fs/nfs/nfs4state.c
index 42f121182167..a043f618cd5a 100644
--- a/fs/nfs/nfs4state.c
+++ b/fs/nfs/nfs4state.c
@@ -787,33 +787,36 @@ void nfs4_close_sync(struct nfs4_state *state, fmode_t fmode)
  * that is compatible with current->files
  */
 static struct nfs4_lock_state *
-__nfs4_find_lock_state(struct nfs4_state *state, fl_owner_t fl_owner, pid_t fl_pid, unsigned int type)
+__nfs4_find_lock_state(struct nfs4_state *state, fl_owner_t fl_owner)
 {
         struct nfs4_lock_state *pos;
         list_for_each_entry(pos, &state->lock_states, ls_locks) {
-                if (type != NFS4_ANY_LOCK_TYPE && pos->ls_owner.lo_type != type)
+                if (pos->ls_owner != fl_owner)
                         continue;
-                switch (pos->ls_owner.lo_type) {
-                case NFS4_POSIX_LOCK_TYPE:
-                        if (pos->ls_owner.lo_u.posix_owner != fl_owner)
-                                continue;
-                        break;
-                case NFS4_FLOCK_LOCK_TYPE:
-                        if (pos->ls_owner.lo_u.flock_owner != fl_pid)
-                                continue;
-                }
                 atomic_inc(&pos->ls_count);
                 return pos;
         }
         return NULL;
 }
 
+static void
+free_lock_state_work(struct work_struct *work)
+{
+        struct nfs4_lock_state *lsp = container_of(work,
+                                        struct nfs4_lock_state, ls_release);
+        struct nfs4_state *state = lsp->ls_state;
+        struct nfs_server *server = state->owner->so_server;
+        struct nfs_client *clp = server->nfs_client;
+
+        clp->cl_mvops->free_lock_state(server, lsp);
+}
+
 /*
  * Return a compatible lock_state. If no initialized lock_state structure
  * exists, return an uninitialized one.
  *
  */
-static struct nfs4_lock_state *nfs4_alloc_lock_state(struct nfs4_state *state, fl_owner_t fl_owner, pid_t fl_pid, unsigned int type)
+static struct nfs4_lock_state *nfs4_alloc_lock_state(struct nfs4_state *state, fl_owner_t fl_owner)
 {
         struct nfs4_lock_state *lsp;
         struct nfs_server *server = state->owner->so_server;
@@ -824,21 +827,12 @@ static struct nfs4_lock_state *nfs4_alloc_lock_state(struct nfs4_state *state, f
         nfs4_init_seqid_counter(&lsp->ls_seqid);
         atomic_set(&lsp->ls_count, 1);
         lsp->ls_state = state;
-        lsp->ls_owner.lo_type = type;
-        switch (lsp->ls_owner.lo_type) {
-        case NFS4_FLOCK_LOCK_TYPE:
-                lsp->ls_owner.lo_u.flock_owner = fl_pid;
-                break;
-        case NFS4_POSIX_LOCK_TYPE:
-                lsp->ls_owner.lo_u.posix_owner = fl_owner;
-                break;
-        default:
-                goto out_free;
-        }
+        lsp->ls_owner = fl_owner;
         lsp->ls_seqid.owner_id = ida_simple_get(&server->lockowner_id, 0, 0, GFP_NOFS);
         if (lsp->ls_seqid.owner_id < 0)
                 goto out_free;
         INIT_LIST_HEAD(&lsp->ls_locks);
+        INIT_WORK(&lsp->ls_release, free_lock_state_work);
         return lsp;
 out_free:
         kfree(lsp);
@@ -857,13 +851,13 @@ void nfs4_free_lock_state(struct nfs_server *server, struct nfs4_lock_state *lsp
  * exists, return an uninitialized one.
  *
  */
-static struct nfs4_lock_state *nfs4_get_lock_state(struct nfs4_state *state, fl_owner_t owner, pid_t pid, unsigned int type)
+static struct nfs4_lock_state *nfs4_get_lock_state(struct nfs4_state *state, fl_owner_t owner)
 {
         struct nfs4_lock_state *lsp, *new = NULL;
         
         for(;;) {
                 spin_lock(&state->state_lock);
-                lsp = __nfs4_find_lock_state(state, owner, pid, type);
+                lsp = __nfs4_find_lock_state(state, owner);
                 if (lsp != NULL)
                         break;
                 if (new != NULL) {
@@ -874,7 +868,7 @@ static struct nfs4_lock_state *nfs4_get_lock_state(struct nfs4_state *state, fl_
                         break;
                 }
                 spin_unlock(&state->state_lock);
-                new = nfs4_alloc_lock_state(state, owner, pid, type);
+                new = nfs4_alloc_lock_state(state, owner);
                 if (new == NULL)
                         return NULL;
         }
@@ -902,13 +896,12 @@ void nfs4_put_lock_state(struct nfs4_lock_state *lsp)
         if (list_empty(&state->lock_states))
                 clear_bit(LK_STATE_IN_USE, &state->flags);
         spin_unlock(&state->state_lock);
-        server = state->owner->so_server;
-        if (test_bit(NFS_LOCK_INITIALIZED, &lsp->ls_flags)) {
-                struct nfs_client *clp = server->nfs_client;
-
-                clp->cl_mvops->free_lock_state(server, lsp);
-        } else
+        if (test_bit(NFS_LOCK_INITIALIZED, &lsp->ls_flags))
+                queue_work(nfsiod_workqueue, &lsp->ls_release);
+        else {
+                server = state->owner->so_server;
                 nfs4_free_lock_state(server, lsp);
+        }
 }
 
 static void nfs4_fl_copy_lock(struct file_lock *dst, struct file_lock *src)
@@ -935,13 +928,7 @@ int nfs4_set_lock_state(struct nfs4_state *state, struct file_lock *fl)
 
         if (fl->fl_ops != NULL)
                 return 0;
-        if (fl->fl_flags & FL_POSIX)
-                lsp = nfs4_get_lock_state(state, fl->fl_owner, 0, NFS4_POSIX_LOCK_TYPE);
-        else if (fl->fl_flags & FL_FLOCK)
-                lsp = nfs4_get_lock_state(state, NULL, fl->fl_pid,
-                                NFS4_FLOCK_LOCK_TYPE);
-        else
-                return -EINVAL;
+        lsp = nfs4_get_lock_state(state, fl->fl_owner);
         if (lsp == NULL)
                 return -ENOMEM;
         fl->fl_u.nfs4_fl.owner = lsp;
@@ -955,7 +942,6 @@ static int nfs4_copy_lock_stateid(nfs4_stateid *dst,
 {
         struct nfs4_lock_state *lsp;
         fl_owner_t fl_owner;
-        pid_t fl_pid;
         int ret = -ENOENT;
 
 
@@ -966,9 +952,8 @@ static int nfs4_copy_lock_stateid(nfs4_stateid *dst,
                 goto out;
 
         fl_owner = lockowner->l_owner;
-        fl_pid = lockowner->l_pid;
         spin_lock(&state->state_lock);
-        lsp = __nfs4_find_lock_state(state, fl_owner, fl_pid, NFS4_ANY_LOCK_TYPE);
+        lsp = __nfs4_find_lock_state(state, fl_owner);
         if (lsp && test_bit(NFS_LOCK_LOST, &lsp->ls_flags))
                 ret = -EIO;
         else if (lsp != NULL && test_bit(NFS_LOCK_INITIALIZED, &lsp->ls_flags) != 0) {
diff --git a/fs/nfs/nfs4trace.h b/fs/nfs/nfs4trace.h
index 0a744f3a86f6..1c32adbe728d 100644
--- a/fs/nfs/nfs4trace.h
+++ b/fs/nfs/nfs4trace.h
@@ -932,11 +932,11 @@ DEFINE_NFS4_IDMAP_EVENT(nfs4_map_gid_to_group);
 
 DECLARE_EVENT_CLASS(nfs4_read_event,
                 TP_PROTO(
-                        const struct nfs_pgio_data *data,
+                        const struct nfs_pgio_header *hdr,
                         int error
                 ),
 
-                TP_ARGS(data, error),
+                TP_ARGS(hdr, error),
 
                 TP_STRUCT__entry(
                         __field(dev_t, dev)
@@ -948,12 +948,12 @@ DECLARE_EVENT_CLASS(nfs4_read_event,
                 ),
 
                 TP_fast_assign(
-                        const struct inode *inode = data->header->inode;
+                        const struct inode *inode = hdr->inode;
                         __entry->dev = inode->i_sb->s_dev;
                         __entry->fileid = NFS_FILEID(inode);
                         __entry->fhandle = nfs_fhandle_hash(NFS_FH(inode));
-                        __entry->offset = data->args.offset;
-                        __entry->count = data->args.count;
+                        __entry->offset = hdr->args.offset;
+                        __entry->count = hdr->args.count;
                         __entry->error = error;
                 ),
 
@@ -972,10 +972,10 @@ DECLARE_EVENT_CLASS(nfs4_read_event,
 #define DEFINE_NFS4_READ_EVENT(name) \
         DEFINE_EVENT(nfs4_read_event, name, \
                         TP_PROTO( \
-                                const struct nfs_pgio_data *data, \
+                                const struct nfs_pgio_header *hdr, \
                                 int error \
                         ), \
-                        TP_ARGS(data, error))
+                        TP_ARGS(hdr, error))
 DEFINE_NFS4_READ_EVENT(nfs4_read);
 #ifdef CONFIG_NFS_V4_1
 DEFINE_NFS4_READ_EVENT(nfs4_pnfs_read);
@@ -983,11 +983,11 @@ DEFINE_NFS4_READ_EVENT(nfs4_pnfs_read);
 
 DECLARE_EVENT_CLASS(nfs4_write_event,
                 TP_PROTO(
-                        const struct nfs_pgio_data *data,
+                        const struct nfs_pgio_header *hdr,
                         int error
                 ),
 
-                TP_ARGS(data, error),
+                TP_ARGS(hdr, error),
 
                 TP_STRUCT__entry(
                         __field(dev_t, dev)
@@ -999,12 +999,12 @@ DECLARE_EVENT_CLASS(nfs4_write_event,
                 ),
 
                 TP_fast_assign(
-                        const struct inode *inode = data->header->inode;
+                        const struct inode *inode = hdr->inode;
                         __entry->dev = inode->i_sb->s_dev;
                         __entry->fileid = NFS_FILEID(inode);
                         __entry->fhandle = nfs_fhandle_hash(NFS_FH(inode));
-                        __entry->offset = data->args.offset;
-                        __entry->count = data->args.count;
+                        __entry->offset = hdr->args.offset;
+                        __entry->count = hdr->args.count;
                         __entry->error = error;
                 ),
 
@@ -1024,10 +1024,10 @@ DECLARE_EVENT_CLASS(nfs4_write_event,
 #define DEFINE_NFS4_WRITE_EVENT(name) \
         DEFINE_EVENT(nfs4_write_event, name, \
                         TP_PROTO( \
-                                const struct nfs_pgio_data *data, \
+                                const struct nfs_pgio_header *hdr, \
                                 int error \
                         ), \
-                        TP_ARGS(data, error))
+                        TP_ARGS(hdr, error))
 DEFINE_NFS4_WRITE_EVENT(nfs4_write);
 #ifdef CONFIG_NFS_V4_1
 DEFINE_NFS4_WRITE_EVENT(nfs4_pnfs_write);
diff --git a/fs/nfs/nfs4xdr.c b/fs/nfs/nfs4xdr.c
index 939ae606cfa4..e13b59d8d9aa 100644
--- a/fs/nfs/nfs4xdr.c
+++ b/fs/nfs/nfs4xdr.c
@@ -7092,7 +7092,7 @@ static int nfs4_xdr_dec_reclaim_complete(struct rpc_rqst *rqstp,
         if (!status)
                 status = decode_sequence(xdr, &res->seq_res, rqstp);
         if (!status)
-                status = decode_reclaim_complete(xdr, (void *)NULL);
+                status = decode_reclaim_complete(xdr, NULL);
         return status;
 }
 
diff --git a/fs/nfs/objlayout/objio_osd.c b/fs/nfs/objlayout/objio_osd.c
index 611320753db2..ae05278b3761 100644
--- a/fs/nfs/objlayout/objio_osd.c
+++ b/fs/nfs/objlayout/objio_osd.c
@@ -439,22 +439,21 @@ static void _read_done(struct ore_io_state *ios, void *private)
         objlayout_read_done(&objios->oir, status, objios->sync);
 }
 
-int objio_read_pagelist(struct nfs_pgio_data *rdata)
+int objio_read_pagelist(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = rdata->header;
         struct objio_state *objios;
         int ret;
 
         ret = objio_alloc_io_state(NFS_I(hdr->inode)->layout, true,
-                        hdr->lseg, rdata->args.pages, rdata->args.pgbase,
-                        rdata->args.offset, rdata->args.count, rdata,
+                        hdr->lseg, hdr->args.pages, hdr->args.pgbase,
+                        hdr->args.offset, hdr->args.count, hdr,
                         GFP_KERNEL, &objios);
         if (unlikely(ret))
                 return ret;
 
         objios->ios->done = _read_done;
         dprintk("%s: offset=0x%llx length=0x%x\n", __func__,
-                rdata->args.offset, rdata->args.count);
+                hdr->args.offset, hdr->args.count);
         ret = ore_read(objios->ios);
         if (unlikely(ret))
                 objio_free_result(&objios->oir);
@@ -487,11 +486,11 @@ static void _write_done(struct ore_io_state *ios, void *private)
 static struct page *__r4w_get_page(void *priv, u64 offset, bool *uptodate)
 {
         struct objio_state *objios = priv;
-        struct nfs_pgio_data *wdata = objios->oir.rpcdata;
-        struct address_space *mapping = wdata->header->inode->i_mapping;
+        struct nfs_pgio_header *hdr = objios->oir.rpcdata;
+        struct address_space *mapping = hdr->inode->i_mapping;
         pgoff_t index = offset / PAGE_SIZE;
         struct page *page;
-        loff_t i_size = i_size_read(wdata->header->inode);
+        loff_t i_size = i_size_read(hdr->inode);
 
         if (offset >= i_size) {
                 *uptodate = true;
@@ -531,15 +530,14 @@ static const struct _ore_r4w_op _r4w_op = {
         .put_page = &__r4w_put_page,
 };
 
-int objio_write_pagelist(struct nfs_pgio_data *wdata, int how)
+int objio_write_pagelist(struct nfs_pgio_header *hdr, int how)
 {
-        struct nfs_pgio_header *hdr = wdata->header;
         struct objio_state *objios;
         int ret;
 
         ret = objio_alloc_io_state(NFS_I(hdr->inode)->layout, false,
-                        hdr->lseg, wdata->args.pages, wdata->args.pgbase,
-                        wdata->args.offset, wdata->args.count, wdata, GFP_NOFS,
+                        hdr->lseg, hdr->args.pages, hdr->args.pgbase,
+                        hdr->args.offset, hdr->args.count, hdr, GFP_NOFS,
                         &objios);
         if (unlikely(ret))
                 return ret;
@@ -551,7 +549,7 @@ int objio_write_pagelist(struct nfs_pgio_data *wdata, int how)
                 objios->ios->done = _write_done;
 
         dprintk("%s: offset=0x%llx length=0x%x\n", __func__,
-                wdata->args.offset, wdata->args.count);
+                hdr->args.offset, hdr->args.count);
         ret = ore_write(objios->ios);
         if (unlikely(ret)) {
                 objio_free_result(&objios->oir);
diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index 765d3f54e986..697a16d11fac 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -229,36 +229,36 @@ objlayout_io_set_result(struct objlayout_io_res *oir, unsigned index,
 static void _rpc_read_complete(struct work_struct *work)
 {
         struct rpc_task *task;
-        struct nfs_pgio_data *rdata;
+        struct nfs_pgio_header *hdr;
 
         dprintk("%s enter\n", __func__);
         task = container_of(work, struct rpc_task, u.tk_work);
-        rdata = container_of(task, struct nfs_pgio_data, task);
+        hdr = container_of(task, struct nfs_pgio_header, task);
 
-        pnfs_ld_read_done(rdata);
+        pnfs_ld_read_done(hdr);
 }
 
 void
 objlayout_read_done(struct objlayout_io_res *oir, ssize_t status, bool sync)
 {
-        struct nfs_pgio_data *rdata = oir->rpcdata;
+        struct nfs_pgio_header *hdr = oir->rpcdata;
 
-        oir->status = rdata->task.tk_status = status;
+        oir->status = hdr->task.tk_status = status;
         if (status >= 0)
-                rdata->res.count = status;
+                hdr->res.count = status;
         else
-                rdata->header->pnfs_error = status;
+                hdr->pnfs_error = status;
         objlayout_iodone(oir);
         /* must not use oir after this point */
 
         dprintk("%s: Return status=%zd eof=%d sync=%d\n", __func__,
-                status, rdata->res.eof, sync);
+                status, hdr->res.eof, sync);
 
         if (sync)
-                pnfs_ld_read_done(rdata);
+                pnfs_ld_read_done(hdr);
         else {
-                INIT_WORK(&rdata->task.u.tk_work, _rpc_read_complete);
-                schedule_work(&rdata->task.u.tk_work);
+                INIT_WORK(&hdr->task.u.tk_work, _rpc_read_complete);
+                schedule_work(&hdr->task.u.tk_work);
         }
 }
 
@@ -266,12 +266,11 @@ objlayout_read_done(struct objlayout_io_res *oir, ssize_t status, bool sync)
  * Perform sync or async reads.
  */
 enum pnfs_try_status
-objlayout_read_pagelist(struct nfs_pgio_data *rdata)
+objlayout_read_pagelist(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = rdata->header;
         struct inode *inode = hdr->inode;
-        loff_t offset = rdata->args.offset;
-        size_t count = rdata->args.count;
+        loff_t offset = hdr->args.offset;
+        size_t count = hdr->args.count;
         int err;
         loff_t eof;
 
@@ -279,23 +278,23 @@ objlayout_read_pagelist(struct nfs_pgio_data *rdata)
         if (unlikely(offset + count > eof)) {
                 if (offset >= eof) {
                         err = 0;
-                        rdata->res.count = 0;
-                        rdata->res.eof = 1;
+                        hdr->res.count = 0;
+                        hdr->res.eof = 1;
                         /*FIXME: do we need to call pnfs_ld_read_done() */
                         goto out;
                 }
                 count = eof - offset;
         }
 
-        rdata->res.eof = (offset + count) >= eof;
-        _fix_verify_io_params(hdr->lseg, &rdata->args.pages,
-                              &rdata->args.pgbase,
-                              rdata->args.offset, rdata->args.count);
+        hdr->res.eof = (offset + count) >= eof;
+        _fix_verify_io_params(hdr->lseg, &hdr->args.pages,
+                              &hdr->args.pgbase,
+                              hdr->args.offset, hdr->args.count);
 
         dprintk("%s: inode(%lx) offset 0x%llx count 0x%Zx eof=%d\n",
-                __func__, inode->i_ino, offset, count, rdata->res.eof);
+                __func__, inode->i_ino, offset, count, hdr->res.eof);
 
-        err = objio_read_pagelist(rdata);
+        err = objio_read_pagelist(hdr);
  out:
         if (unlikely(err)) {
                 hdr->pnfs_error = err;
@@ -312,38 +311,38 @@ objlayout_read_pagelist(struct nfs_pgio_data *rdata)
 static void _rpc_write_complete(struct work_struct *work)
 {
         struct rpc_task *task;
-        struct nfs_pgio_data *wdata;
+        struct nfs_pgio_header *hdr;
 
         dprintk("%s enter\n", __func__);
         task = container_of(work, struct rpc_task, u.tk_work);
-        wdata = container_of(task, struct nfs_pgio_data, task);
+        hdr = container_of(task, struct nfs_pgio_header, task);
 
-        pnfs_ld_write_done(wdata);
+        pnfs_ld_write_done(hdr);
 }
 
 void
 objlayout_write_done(struct objlayout_io_res *oir, ssize_t status, bool sync)
 {
-        struct nfs_pgio_data *wdata = oir->rpcdata;
+        struct nfs_pgio_header *hdr = oir->rpcdata;
 
-        oir->status = wdata->task.tk_status = status;
+        oir->status = hdr->task.tk_status = status;
         if (status >= 0) {
-                wdata->res.count = status;
-                wdata->verf.committed = oir->committed;
+                hdr->res.count = status;
+                hdr->verf.committed = oir->committed;
         } else {
-                wdata->header->pnfs_error = status;
+                hdr->pnfs_error = status;
         }
         objlayout_iodone(oir);
         /* must not use oir after this point */
 
         dprintk("%s: Return status %zd committed %d sync=%d\n", __func__,
-                status, wdata->verf.committed, sync);
+                status, hdr->verf.committed, sync);
 
         if (sync)
-                pnfs_ld_write_done(wdata);
+                pnfs_ld_write_done(hdr);
         else {
-                INIT_WORK(&wdata->task.u.tk_work, _rpc_write_complete);
-                schedule_work(&wdata->task.u.tk_work);
+                INIT_WORK(&hdr->task.u.tk_work, _rpc_write_complete);
+                schedule_work(&hdr->task.u.tk_work);
         }
 }
 
@@ -351,17 +350,15 @@ objlayout_write_done(struct objlayout_io_res *oir, ssize_t status, bool sync)
  * Perform sync or async writes.
  */
 enum pnfs_try_status
-objlayout_write_pagelist(struct nfs_pgio_data *wdata,
-                         int how)
+objlayout_write_pagelist(struct nfs_pgio_header *hdr, int how)
 {
-        struct nfs_pgio_header *hdr = wdata->header;
         int err;
 
-        _fix_verify_io_params(hdr->lseg, &wdata->args.pages,
-                              &wdata->args.pgbase,
-                              wdata->args.offset, wdata->args.count);
+        _fix_verify_io_params(hdr->lseg, &hdr->args.pages,
+                              &hdr->args.pgbase,
+                              hdr->args.offset, hdr->args.count);
 
-        err = objio_write_pagelist(wdata, how);
+        err = objio_write_pagelist(hdr, how);
         if (unlikely(err)) {
                 hdr->pnfs_error = err;
                 dprintk("%s: Returned Error %d\n", __func__, err);
diff --git a/fs/nfs/objlayout/objlayout.h b/fs/nfs/objlayout/objlayout.h
index 01e041029a6c..fd13f1d2f136 100644
--- a/fs/nfs/objlayout/objlayout.h
+++ b/fs/nfs/objlayout/objlayout.h
@@ -119,8 +119,8 @@ extern void objio_free_lseg(struct pnfs_layout_segment *lseg);
  */
 extern void objio_free_result(struct objlayout_io_res *oir);
 
-extern int objio_read_pagelist(struct nfs_pgio_data *rdata);
-extern int objio_write_pagelist(struct nfs_pgio_data *wdata, int how);
+extern int objio_read_pagelist(struct nfs_pgio_header *rdata);
+extern int objio_write_pagelist(struct nfs_pgio_header *wdata, int how);
 
 /*
  * callback API
@@ -168,10 +168,10 @@ extern struct pnfs_layout_segment *objlayout_alloc_lseg(
 extern void objlayout_free_lseg(struct pnfs_layout_segment *);
 
 extern enum pnfs_try_status objlayout_read_pagelist(
-        struct nfs_pgio_data *);
+        struct nfs_pgio_header *);
 
 extern enum pnfs_try_status objlayout_write_pagelist(
-        struct nfs_pgio_data *,
+        struct nfs_pgio_header *,
         int how);
 
 extern void objlayout_encode_layoutcommit(
diff --git a/fs/nfs/pagelist.c b/fs/nfs/pagelist.c
index 0be5050638f7..ba491926df5f 100644
--- a/fs/nfs/pagelist.c
+++ b/fs/nfs/pagelist.c
@@ -141,16 +141,24 @@ nfs_iocounter_wait(struct nfs_io_counter *c)
  * @req - request in group that is to be locked
  *
  * this lock must be held if modifying the page group list
+ *
+ * returns result from wait_on_bit_lock: 0 on success, < 0 on error
  */
-void
-nfs_page_group_lock(struct nfs_page *req)
+int
+nfs_page_group_lock(struct nfs_page *req, bool wait)
 {
         struct nfs_page *head = req->wb_head;
+        int ret;
 
         WARN_ON_ONCE(head != head->wb_head);
 
-        wait_on_bit_lock(&head->wb_flags, PG_HEADLOCK,
+        do {
+                ret = wait_on_bit_lock(&head->wb_flags, PG_HEADLOCK,
                         TASK_UNINTERRUPTIBLE);
+        } while (wait && ret != 0);
+
+        WARN_ON_ONCE(ret > 0);
+        return ret;
 }
 
 /*
@@ -211,7 +219,7 @@ bool nfs_page_group_sync_on_bit(struct nfs_page *req, unsigned int bit)
 {
         bool ret;
 
-        nfs_page_group_lock(req);
+        nfs_page_group_lock(req, true);
         ret = nfs_page_group_sync_on_bit_locked(req, bit);
         nfs_page_group_unlock(req);
 
@@ -454,123 +462,72 @@ size_t nfs_generic_pg_test(struct nfs_pageio_descriptor *desc,
 }
 EXPORT_SYMBOL_GPL(nfs_generic_pg_test);
 
-static inline struct nfs_rw_header *NFS_RW_HEADER(struct nfs_pgio_header *hdr)
-{
-        return container_of(hdr, struct nfs_rw_header, header);
-}
-
-/**
- * nfs_rw_header_alloc - Allocate a header for a read or write
- * @ops: Read or write function vector
- */
-struct nfs_rw_header *nfs_rw_header_alloc(const struct nfs_rw_ops *ops)
+struct nfs_pgio_header *nfs_pgio_header_alloc(const struct nfs_rw_ops *ops)
 {
-        struct nfs_rw_header *header = ops->rw_alloc_header();
-
-        if (header) {
-                struct nfs_pgio_header *hdr = &header->header;
+        struct nfs_pgio_header *hdr = ops->rw_alloc_header();
 
+        if (hdr) {
                 INIT_LIST_HEAD(&hdr->pages);
                 spin_lock_init(&hdr->lock);
-                atomic_set(&hdr->refcnt, 0);
                 hdr->rw_ops = ops;
         }
-        return header;
+        return hdr;
 }
-EXPORT_SYMBOL_GPL(nfs_rw_header_alloc);
+EXPORT_SYMBOL_GPL(nfs_pgio_header_alloc);
 
 /*
- * nfs_rw_header_free - Free a read or write header
+ * nfs_pgio_header_free - Free a read or write header
  * @hdr: The header to free
  */
-void nfs_rw_header_free(struct nfs_pgio_header *hdr)
+void nfs_pgio_header_free(struct nfs_pgio_header *hdr)
 {
-        hdr->rw_ops->rw_free_header(NFS_RW_HEADER(hdr));
+        hdr->rw_ops->rw_free_header(hdr);
 }
-EXPORT_SYMBOL_GPL(nfs_rw_header_free);
+EXPORT_SYMBOL_GPL(nfs_pgio_header_free);
 
 /**
- * nfs_pgio_data_alloc - Allocate pageio data
- * @hdr: The header making a request
- * @pagecount: Number of pages to create
- */
-static struct nfs_pgio_data *nfs_pgio_data_alloc(struct nfs_pgio_header *hdr,
-                                                 unsigned int pagecount)
-{
-        struct nfs_pgio_data *data, *prealloc;
-
-        prealloc = &NFS_RW_HEADER(hdr)->rpc_data;
-        if (prealloc->header == NULL)
-                data = prealloc;
-        else
-                data = kzalloc(sizeof(*data), GFP_KERNEL);
-        if (!data)
-                goto out;
-
-        if (nfs_pgarray_set(&data->pages, pagecount)) {
-                data->header = hdr;
-                atomic_inc(&hdr->refcnt);
-        } else {
-                if (data != prealloc)
-                        kfree(data);
-                data = NULL;
-        }
-out:
-        return data;
-}
-
-/**
- * nfs_pgio_data_release - Properly free pageio data
- * @data: The data to release
+ * nfs_pgio_data_destroy - make @hdr suitable for reuse
+ *
+ * Frees memory and releases refs from nfs_generic_pgio, so that it may
+ * be called again.
+ *
+ * @hdr: A header that has had nfs_generic_pgio called
  */
-void nfs_pgio_data_release(struct nfs_pgio_data *data)
+void nfs_pgio_data_destroy(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-        struct nfs_rw_header *pageio_header = NFS_RW_HEADER(hdr);
-
-        put_nfs_open_context(data->args.context);
-        if (data->pages.pagevec != data->pages.page_array)
-                kfree(data->pages.pagevec);
-        if (data == &pageio_header->rpc_data) {
-                data->header = NULL;
-                data = NULL;
-        }
-        if (atomic_dec_and_test(&hdr->refcnt))
-                hdr->completion_ops->completion(hdr);
-        /* Note: we only free the rpc_task after callbacks are done.
-         * See the comment in rpc_free_task() for why
-         */
-        kfree(data);
+        put_nfs_open_context(hdr->args.context);
+        if (hdr->page_array.pagevec != hdr->page_array.page_array)
+                kfree(hdr->page_array.pagevec);
 }
-EXPORT_SYMBOL_GPL(nfs_pgio_data_release);
+EXPORT_SYMBOL_GPL(nfs_pgio_data_destroy);
 
 /**
  * nfs_pgio_rpcsetup - Set up arguments for a pageio call
- * @data: The pageio data
+ * @hdr: The pageio hdr
  * @count: Number of bytes to read
  * @offset: Initial offset
  * @how: How to commit data (writes only)
  * @cinfo: Commit information for the call (writes only)
  */
-static void nfs_pgio_rpcsetup(struct nfs_pgio_data *data,
+static void nfs_pgio_rpcsetup(struct nfs_pgio_header *hdr,
                               unsigned int count, unsigned int offset,
                               int how, struct nfs_commit_info *cinfo)
 {
-        struct nfs_page *req = data->header->req;
+        struct nfs_page *req = hdr->req;
 
         /* Set up the RPC argument and reply structs
-         * NB: take care not to mess about with data->commit et al. */
+         * NB: take care not to mess about with hdr->commit et al. */
 
-        data->args.fh     = NFS_FH(data->header->inode);
-        data->args.offset = req_offset(req) + offset;
+        hdr->args.fh     = NFS_FH(hdr->inode);
+        hdr->args.offset = req_offset(req) + offset;
         /* pnfs_set_layoutcommit needs this */
-        data->mds_offset = data->args.offset;
-        data->args.pgbase = req->wb_pgbase + offset;
-        data->args.pages  = data->pages.pagevec;
-        data->args.count  = count;
-        data->args.context = get_nfs_open_context(req->wb_context);
-        data->args.lock_context = req->wb_lock_context;
-        data->args.stable  = NFS_UNSTABLE;
+        hdr->mds_offset = hdr->args.offset;
+        hdr->args.pgbase = req->wb_pgbase + offset;
+        hdr->args.pages  = hdr->page_array.pagevec;
+        hdr->args.count  = count;
+        hdr->args.context = get_nfs_open_context(req->wb_context);
+        hdr->args.lock_context = req->wb_lock_context;
+        hdr->args.stable  = NFS_UNSTABLE;
         switch (how & (FLUSH_STABLE | FLUSH_COND_STABLE)) {
         case 0:
                 break;
@@ -578,59 +535,59 @@ static void nfs_pgio_rpcsetup(struct nfs_pgio_data *data,
                 if (nfs_reqs_to_commit(cinfo))
                         break;
         default:
-                data->args.stable = NFS_FILE_SYNC;
+                hdr->args.stable = NFS_FILE_SYNC;
         }
 
-        data->res.fattr   = &data->fattr;
-        data->res.count   = count;
-        data->res.eof     = 0;
-        data->res.verf    = &data->verf;
-        nfs_fattr_init(&data->fattr);
+        hdr->res.fattr   = &hdr->fattr;
+        hdr->res.count   = count;
+        hdr->res.eof     = 0;
+        hdr->res.verf    = &hdr->verf;
+        nfs_fattr_init(&hdr->fattr);
 }
 
 /**
- * nfs_pgio_prepare - Prepare pageio data to go over the wire
+ * nfs_pgio_prepare - Prepare pageio hdr to go over the wire
  * @task: The current task
- * @calldata: pageio data to prepare
+ * @calldata: pageio header to prepare
  */
 static void nfs_pgio_prepare(struct rpc_task *task, void *calldata)
 {
-        struct nfs_pgio_data *data = calldata;
+        struct nfs_pgio_header *hdr = calldata;
         int err;
-        err = NFS_PROTO(data->header->inode)->pgio_rpc_prepare(task, data);
+        err = NFS_PROTO(hdr->inode)->pgio_rpc_prepare(task, hdr);
         if (err)
                 rpc_exit(task, err);
 }
 
-int nfs_initiate_pgio(struct rpc_clnt *clnt, struct nfs_pgio_data *data,
+int nfs_initiate_pgio(struct rpc_clnt *clnt, struct nfs_pgio_header *hdr,
                       const struct rpc_call_ops *call_ops, int how, int flags)
 {
         struct rpc_task *task;
         struct rpc_message msg = {
-                .rpc_argp = &data->args,
-                .rpc_resp = &data->res,
-                .rpc_cred = data->header->cred,
+                .rpc_argp = &hdr->args,
+                .rpc_resp = &hdr->res,
+                .rpc_cred = hdr->cred,
         };
         struct rpc_task_setup task_setup_data = {
                 .rpc_client = clnt,
-                .task = &data->task,
+                .task = &hdr->task,
                 .rpc_message = &msg,
                 .callback_ops = call_ops,
-                .callback_data = data,
+                .callback_data = hdr,
                 .workqueue = nfsiod_workqueue,
                 .flags = RPC_TASK_ASYNC | flags,
         };
         int ret = 0;
 
-        data->header->rw_ops->rw_initiate(data, &msg, &task_setup_data, how);
+        hdr->rw_ops->rw_initiate(hdr, &msg, &task_setup_data, how);
 
         dprintk("NFS: %5u initiated pgio call "
                 "(req %s/%llu, %u bytes @ offset %llu)\n",
-                data->task.tk_pid,
-                data->header->inode->i_sb->s_id,
-                (unsigned long long)NFS_FILEID(data->header->inode),
-                data->args.count,
-                (unsigned long long)data->args.offset);
+                hdr->task.tk_pid,
+                hdr->inode->i_sb->s_id,
+                (unsigned long long)NFS_FILEID(hdr->inode),
+                hdr->args.count,
+                (unsigned long long)hdr->args.offset);
 
         task = rpc_run_task(&task_setup_data);
         if (IS_ERR(task)) {
@@ -657,22 +614,23 @@ static int nfs_pgio_error(struct nfs_pageio_descriptor *desc,
                           struct nfs_pgio_header *hdr)
 {
         set_bit(NFS_IOHDR_REDO, &hdr->flags);
-        nfs_pgio_data_release(hdr->data);
-        hdr->data = NULL;
+        nfs_pgio_data_destroy(hdr);
+        hdr->completion_ops->completion(hdr);
         desc->pg_completion_ops->error_cleanup(&desc->pg_list);
         return -ENOMEM;
 }
 
 /**
  * nfs_pgio_release - Release pageio data
- * @calldata: The pageio data to release
+ * @calldata: The pageio header to release
  */
 static void nfs_pgio_release(void *calldata)
 {
-        struct nfs_pgio_data *data = calldata;
-        if (data->header->rw_ops->rw_release)
-                data->header->rw_ops->rw_release(data);
-        nfs_pgio_data_release(data);
+        struct nfs_pgio_header *hdr = calldata;
+        if (hdr->rw_ops->rw_release)
+                hdr->rw_ops->rw_release(hdr);
+        nfs_pgio_data_destroy(hdr);
+        hdr->completion_ops->completion(hdr);
 }
 
 /**
@@ -713,22 +671,22 @@ EXPORT_SYMBOL_GPL(nfs_pageio_init);
 /**
  * nfs_pgio_result - Basic pageio error handling
  * @task: The task that ran
- * @calldata: Pageio data to check
+ * @calldata: Pageio header to check
  */
 static void nfs_pgio_result(struct rpc_task *task, void *calldata)
 {
-        struct nfs_pgio_data *data = calldata;
-        struct inode *inode = data->header->inode;
+        struct nfs_pgio_header *hdr = calldata;
+        struct inode *inode = hdr->inode;
 
         dprintk("NFS: %s: %5u, (status %d)\n", __func__,
                 task->tk_pid, task->tk_status);
 
-        if (data->header->rw_ops->rw_done(task, data, inode) != 0)
+        if (hdr->rw_ops->rw_done(task, hdr, inode) != 0)
                 return;
         if (task->tk_status < 0)
-                nfs_set_pgio_error(data->header, task->tk_status, data->args.offset);
+                nfs_set_pgio_error(hdr, task->tk_status, hdr->args.offset);
         else
-                data->header->rw_ops->rw_result(task, data);
+                hdr->rw_ops->rw_result(task, hdr);
 }
 
 /*
@@ -744,17 +702,16 @@ int nfs_generic_pgio(struct nfs_pageio_descriptor *desc,
 {
         struct nfs_page         *req;
         struct page             **pages;
-        struct nfs_pgio_data    *data;
         struct list_head *head = &desc->pg_list;
         struct nfs_commit_info cinfo;
+        unsigned int pagecount;
 
-        data = nfs_pgio_data_alloc(hdr, nfs_page_array_len(desc->pg_base,
-                                                           desc->pg_count));
-        if (!data)
+        pagecount = nfs_page_array_len(desc->pg_base, desc->pg_count);
+        if (!nfs_pgarray_set(&hdr->page_array, pagecount))
                 return nfs_pgio_error(desc, hdr);
 
         nfs_init_cinfo(&cinfo, desc->pg_inode, desc->pg_dreq);
-        pages = data->pages.pagevec;
+        pages = hdr->page_array.pagevec;
         while (!list_empty(head)) {
                 req = nfs_list_entry(head->next);
                 nfs_list_remove_request(req);
@@ -767,8 +724,7 @@ int nfs_generic_pgio(struct nfs_pageio_descriptor *desc,
                 desc->pg_ioflags &= ~FLUSH_COND_STABLE;
 
         /* Set up the argument struct */
-        nfs_pgio_rpcsetup(data, desc->pg_count, 0, desc->pg_ioflags, &cinfo);
-        hdr->data = data;
+        nfs_pgio_rpcsetup(hdr, desc->pg_count, 0, desc->pg_ioflags, &cinfo);
         desc->pg_rpc_callops = &nfs_pgio_common_ops;
         return 0;
 }
@@ -776,25 +732,20 @@ EXPORT_SYMBOL_GPL(nfs_generic_pgio);
 
 static int nfs_generic_pg_pgios(struct nfs_pageio_descriptor *desc)
 {
-        struct nfs_rw_header *rw_hdr;
         struct nfs_pgio_header *hdr;
         int ret;
 
-        rw_hdr = nfs_rw_header_alloc(desc->pg_rw_ops);
-        if (!rw_hdr) {
+        hdr = nfs_pgio_header_alloc(desc->pg_rw_ops);
+        if (!hdr) {
                 desc->pg_completion_ops->error_cleanup(&desc->pg_list);
                 return -ENOMEM;
         }
-        hdr = &rw_hdr->header;
-        nfs_pgheader_init(desc, hdr, nfs_rw_header_free);
-        atomic_inc(&hdr->refcnt);
+        nfs_pgheader_init(desc, hdr, nfs_pgio_header_free);
         ret = nfs_generic_pgio(desc, hdr);
         if (ret == 0)
                 ret = nfs_initiate_pgio(NFS_CLIENT(hdr->inode),
-                                        hdr->data, desc->pg_rpc_callops,
+                                        hdr, desc->pg_rpc_callops,
                                         desc->pg_ioflags, 0);
-        if (atomic_dec_and_test(&hdr->refcnt))
-                hdr->completion_ops->completion(hdr);
         return ret;
 }
 
@@ -907,8 +858,13 @@ static int __nfs_pageio_add_request(struct nfs_pageio_descriptor *desc,
         struct nfs_page *subreq;
         unsigned int bytes_left = 0;
         unsigned int offset, pgbase;
+        int ret;
 
-        nfs_page_group_lock(req);
+        ret = nfs_page_group_lock(req, false);
+        if (ret < 0) {
+                desc->pg_error = ret;
+                return 0;
+        }
 
         subreq = req;
         bytes_left = subreq->wb_bytes;
@@ -930,7 +886,11 @@ static int __nfs_pageio_add_request(struct nfs_pageio_descriptor *desc,
                         if (desc->pg_recoalesce)
                                 return 0;
                         /* retry add_request for this subreq */
-                        nfs_page_group_lock(req);
+                        ret = nfs_page_group_lock(req, false);
+                        if (ret < 0) {
+                                desc->pg_error = ret;
+                                return 0;
+                        }
                         continue;
                 }
 
@@ -1005,7 +965,38 @@ int nfs_pageio_add_request(struct nfs_pageio_descriptor *desc,
         } while (ret);
         return ret;
 }
-EXPORT_SYMBOL_GPL(nfs_pageio_add_request);
+
+/*
+ * nfs_pageio_resend - Transfer requests to new descriptor and resend
+ * @hdr - the pgio header to move request from
+ * @desc - the pageio descriptor to add requests to
+ *
+ * Try to move each request (nfs_page) from @hdr to @desc then attempt
+ * to send them.
+ *
+ * Returns 0 on success and < 0 on error.
+ */
+int nfs_pageio_resend(struct nfs_pageio_descriptor *desc,
+                      struct nfs_pgio_header *hdr)
+{
+        LIST_HEAD(failed);
+
+        desc->pg_dreq = hdr->dreq;
+        while (!list_empty(&hdr->pages)) {
+                struct nfs_page *req = nfs_list_entry(hdr->pages.next);
+
+                nfs_list_remove_request(req);
+                if (!nfs_pageio_add_request(desc, req))
+                        nfs_list_add_request(req, &failed);
+        }
+        nfs_pageio_complete(desc);
+        if (!list_empty(&failed)) {
+                list_move(&failed, &hdr->pages);
+                return -EIO;
+        }
+        return 0;
+}
+EXPORT_SYMBOL_GPL(nfs_pageio_resend);
 
 /**
  * nfs_pageio_complete - Complete I/O on an nfs_pageio_descriptor
@@ -1021,7 +1012,6 @@ void nfs_pageio_complete(struct nfs_pageio_descriptor *desc)
                         break;
         }
 }
-EXPORT_SYMBOL_GPL(nfs_pageio_complete);
 
 /**
  * nfs_pageio_cond_complete - Conditional I/O completion
diff --git a/fs/nfs/pnfs.c b/fs/nfs/pnfs.c
index a8914b335617..a3851debf8a2 100644
--- a/fs/nfs/pnfs.c
+++ b/fs/nfs/pnfs.c
@@ -361,6 +361,23 @@ pnfs_put_lseg(struct pnfs_layout_segment *lseg)
 }
 EXPORT_SYMBOL_GPL(pnfs_put_lseg);
 
+static void pnfs_put_lseg_async_work(struct work_struct *work)
+{
+        struct pnfs_layout_segment *lseg;
+
+        lseg = container_of(work, struct pnfs_layout_segment, pls_work);
+
+        pnfs_put_lseg(lseg);
+}
+
+void
+pnfs_put_lseg_async(struct pnfs_layout_segment *lseg)
+{
+        INIT_WORK(&lseg->pls_work, pnfs_put_lseg_async_work);
+        schedule_work(&lseg->pls_work);
+}
+EXPORT_SYMBOL_GPL(pnfs_put_lseg_async);
+
 static u64
 end_offset(u64 start, u64 len)
 {
@@ -1470,41 +1487,19 @@ pnfs_generic_pg_test(struct nfs_pageio_descriptor *pgio, struct nfs_page *prev,
 }
 EXPORT_SYMBOL_GPL(pnfs_generic_pg_test);
 
-int pnfs_write_done_resend_to_mds(struct inode *inode,
-                                struct list_head *head,
-                                const struct nfs_pgio_completion_ops *compl_ops,
-                                struct nfs_direct_req *dreq)
+int pnfs_write_done_resend_to_mds(struct nfs_pgio_header *hdr)
 {
         struct nfs_pageio_descriptor pgio;
-        LIST_HEAD(failed);
 
         /* Resend all requests through the MDS */
-        nfs_pageio_init_write(&pgio, inode, FLUSH_STABLE, true, compl_ops);
-        pgio.pg_dreq = dreq;
-        while (!list_empty(head)) {
-                struct nfs_page *req = nfs_list_entry(head->next);
-
-                nfs_list_remove_request(req);
-                if (!nfs_pageio_add_request(&pgio, req))
-                        nfs_list_add_request(req, &failed);
-        }
-        nfs_pageio_complete(&pgio);
-
-        if (!list_empty(&failed)) {
-                /* For some reason our attempt to resend pages. Mark the
-                 * overall send request as having failed, and let
-                 * nfs_writeback_release_full deal with the error.
-                 */
-                list_move(&failed, head);
-                return -EIO;
-        }
-        return 0;
+        nfs_pageio_init_write(&pgio, hdr->inode, FLUSH_STABLE, true,
+                              hdr->completion_ops);
+        return nfs_pageio_resend(&pgio, hdr);
 }
 EXPORT_SYMBOL_GPL(pnfs_write_done_resend_to_mds);
 
-static void pnfs_ld_handle_write_error(struct nfs_pgio_data *data)
+static void pnfs_ld_handle_write_error(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
 
         dprintk("pnfs write error = %d\n", hdr->pnfs_error);
         if (NFS_SERVER(hdr->inode)->pnfs_curr_ld->flags &
@@ -1512,50 +1507,42 @@ static void pnfs_ld_handle_write_error(struct nfs_pgio_data *data)
                 pnfs_return_layout(hdr->inode);
         }
         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags))
-                data->task.tk_status = pnfs_write_done_resend_to_mds(hdr->inode,
-                                                        &hdr->pages,
-                                                        hdr->completion_ops,
-                                                        hdr->dreq);
+                hdr->task.tk_status = pnfs_write_done_resend_to_mds(hdr);
 }
 
 /*
  * Called by non rpc-based layout drivers
  */
-void pnfs_ld_write_done(struct nfs_pgio_data *data)
+void pnfs_ld_write_done(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-
-        trace_nfs4_pnfs_write(data, hdr->pnfs_error);
+        trace_nfs4_pnfs_write(hdr, hdr->pnfs_error);
         if (!hdr->pnfs_error) {
-                pnfs_set_layoutcommit(data);
-                hdr->mds_ops->rpc_call_done(&data->task, data);
+                pnfs_set_layoutcommit(hdr);
+                hdr->mds_ops->rpc_call_done(&hdr->task, hdr);
         } else
-                pnfs_ld_handle_write_error(data);
-        hdr->mds_ops->rpc_release(data);
+                pnfs_ld_handle_write_error(hdr);
+        hdr->mds_ops->rpc_release(hdr);
 }
 EXPORT_SYMBOL_GPL(pnfs_ld_write_done);
 
 static void
 pnfs_write_through_mds(struct nfs_pageio_descriptor *desc,
-                struct nfs_pgio_data *data)
+                struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-
         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags)) {
                 list_splice_tail_init(&hdr->pages, &desc->pg_list);
                 nfs_pageio_reset_write_mds(desc);
                 desc->pg_recoalesce = 1;
         }
-        nfs_pgio_data_release(data);
+        nfs_pgio_data_destroy(hdr);
 }
 
 static enum pnfs_try_status
-pnfs_try_to_write_data(struct nfs_pgio_data *wdata,
+pnfs_try_to_write_data(struct nfs_pgio_header *hdr,
                         const struct rpc_call_ops *call_ops,
                         struct pnfs_layout_segment *lseg,
                         int how)
 {
-        struct nfs_pgio_header *hdr = wdata->header;
         struct inode *inode = hdr->inode;
         enum pnfs_try_status trypnfs;
         struct nfs_server *nfss = NFS_SERVER(inode);
@@ -1563,8 +1550,8 @@ pnfs_try_to_write_data(struct nfs_pgio_data *wdata,
         hdr->mds_ops = call_ops;
 
         dprintk("%s: Writing ino:%lu %u@%llu (how %d)\n", __func__,
-                inode->i_ino, wdata->args.count, wdata->args.offset, how);
-        trypnfs = nfss->pnfs_curr_ld->write_pagelist(wdata, how);
+                inode->i_ino, hdr->args.count, hdr->args.offset, how);
+        trypnfs = nfss->pnfs_curr_ld->write_pagelist(hdr, how);
         if (trypnfs != PNFS_NOT_ATTEMPTED)
                 nfs_inc_stats(inode, NFSIOS_PNFS_WRITE);
         dprintk("%s End (trypnfs:%d)\n", __func__, trypnfs);
@@ -1575,139 +1562,105 @@ static void
 pnfs_do_write(struct nfs_pageio_descriptor *desc,
               struct nfs_pgio_header *hdr, int how)
 {
-        struct nfs_pgio_data *data = hdr->data;
         const struct rpc_call_ops *call_ops = desc->pg_rpc_callops;
         struct pnfs_layout_segment *lseg = desc->pg_lseg;
         enum pnfs_try_status trypnfs;
 
         desc->pg_lseg = NULL;
-        trypnfs = pnfs_try_to_write_data(data, call_ops, lseg, how);
+        trypnfs = pnfs_try_to_write_data(hdr, call_ops, lseg, how);
         if (trypnfs == PNFS_NOT_ATTEMPTED)
-                pnfs_write_through_mds(desc, data);
+                pnfs_write_through_mds(desc, hdr);
         pnfs_put_lseg(lseg);
 }
 
 static void pnfs_writehdr_free(struct nfs_pgio_header *hdr)
 {
         pnfs_put_lseg(hdr->lseg);
-        nfs_rw_header_free(hdr);
+        nfs_pgio_header_free(hdr);
 }
 EXPORT_SYMBOL_GPL(pnfs_writehdr_free);
 
 int
 pnfs_generic_pg_writepages(struct nfs_pageio_descriptor *desc)
 {
-        struct nfs_rw_header *whdr;
         struct nfs_pgio_header *hdr;
         int ret;
 
-        whdr = nfs_rw_header_alloc(desc->pg_rw_ops);
-        if (!whdr) {
+        hdr = nfs_pgio_header_alloc(desc->pg_rw_ops);
+        if (!hdr) {
                 desc->pg_completion_ops->error_cleanup(&desc->pg_list);
                 pnfs_put_lseg(desc->pg_lseg);
                 desc->pg_lseg = NULL;
                 return -ENOMEM;
         }
-        hdr = &whdr->header;
         nfs_pgheader_init(desc, hdr, pnfs_writehdr_free);
         hdr->lseg = pnfs_get_lseg(desc->pg_lseg);
-        atomic_inc(&hdr->refcnt);
         ret = nfs_generic_pgio(desc, hdr);
         if (ret != 0) {
                 pnfs_put_lseg(desc->pg_lseg);
                 desc->pg_lseg = NULL;
         } else
                 pnfs_do_write(desc, hdr, desc->pg_ioflags);
-        if (atomic_dec_and_test(&hdr->refcnt))
-                hdr->completion_ops->completion(hdr);
         return ret;
 }
 EXPORT_SYMBOL_GPL(pnfs_generic_pg_writepages);
 
-int pnfs_read_done_resend_to_mds(struct inode *inode,
-                                struct list_head *head,
-                                const struct nfs_pgio_completion_ops *compl_ops,
-                                struct nfs_direct_req *dreq)
+int pnfs_read_done_resend_to_mds(struct nfs_pgio_header *hdr)
 {
         struct nfs_pageio_descriptor pgio;
-        LIST_HEAD(failed);
 
         /* Resend all requests through the MDS */
-        nfs_pageio_init_read(&pgio, inode, true, compl_ops);
-        pgio.pg_dreq = dreq;
-        while (!list_empty(head)) {
-                struct nfs_page *req = nfs_list_entry(head->next);
-
-                nfs_list_remove_request(req);
-                if (!nfs_pageio_add_request(&pgio, req))
-                        nfs_list_add_request(req, &failed);
-        }
-        nfs_pageio_complete(&pgio);
-
-        if (!list_empty(&failed)) {
-                list_move(&failed, head);
-                return -EIO;
-        }
-        return 0;
+        nfs_pageio_init_read(&pgio, hdr->inode, true, hdr->completion_ops);
+        return nfs_pageio_resend(&pgio, hdr);
 }
 EXPORT_SYMBOL_GPL(pnfs_read_done_resend_to_mds);
 
-static void pnfs_ld_handle_read_error(struct nfs_pgio_data *data)
+static void pnfs_ld_handle_read_error(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-
         dprintk("pnfs read error = %d\n", hdr->pnfs_error);
         if (NFS_SERVER(hdr->inode)->pnfs_curr_ld->flags &
             PNFS_LAYOUTRET_ON_ERROR) {
                 pnfs_return_layout(hdr->inode);
         }
         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags))
-                data->task.tk_status = pnfs_read_done_resend_to_mds(hdr->inode,
-                                                        &hdr->pages,
-                                                        hdr->completion_ops,
-                                                        hdr->dreq);
+                hdr->task.tk_status = pnfs_read_done_resend_to_mds(hdr);
 }
 
 /*
  * Called by non rpc-based layout drivers
  */
-void pnfs_ld_read_done(struct nfs_pgio_data *data)
+void pnfs_ld_read_done(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-
-        trace_nfs4_pnfs_read(data, hdr->pnfs_error);
+        trace_nfs4_pnfs_read(hdr, hdr->pnfs_error);
         if (likely(!hdr->pnfs_error)) {
-                __nfs4_read_done_cb(data);
-                hdr->mds_ops->rpc_call_done(&data->task, data);
+                __nfs4_read_done_cb(hdr);
+                hdr->mds_ops->rpc_call_done(&hdr->task, hdr);
         } else
-                pnfs_ld_handle_read_error(data);
-        hdr->mds_ops->rpc_release(data);
+                pnfs_ld_handle_read_error(hdr);
+        hdr->mds_ops->rpc_release(hdr);
 }
 EXPORT_SYMBOL_GPL(pnfs_ld_read_done);
 
 static void
 pnfs_read_through_mds(struct nfs_pageio_descriptor *desc,
-                struct nfs_pgio_data *data)
+                struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-
         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags)) {
                 list_splice_tail_init(&hdr->pages, &desc->pg_list);
                 nfs_pageio_reset_read_mds(desc);
                 desc->pg_recoalesce = 1;
         }
-        nfs_pgio_data_release(data);
+        nfs_pgio_data_destroy(hdr);
 }
 
 /*
  * Call the appropriate parallel I/O subsystem read function.
  */
 static enum pnfs_try_status
-pnfs_try_to_read_data(struct nfs_pgio_data *rdata,
+pnfs_try_to_read_data(struct nfs_pgio_header *hdr,
                        const struct rpc_call_ops *call_ops,
                        struct pnfs_layout_segment *lseg)
 {
-        struct nfs_pgio_header *hdr = rdata->header;
         struct inode *inode = hdr->inode;
         struct nfs_server *nfss = NFS_SERVER(inode);
         enum pnfs_try_status trypnfs;
@@ -1715,9 +1668,9 @@ pnfs_try_to_read_data(struct nfs_pgio_data *rdata,
         hdr->mds_ops = call_ops;
 
         dprintk("%s: Reading ino:%lu %u@%llu\n",
-                __func__, inode->i_ino, rdata->args.count, rdata->args.offset);
+                __func__, inode->i_ino, hdr->args.count, hdr->args.offset);
 
-        trypnfs = nfss->pnfs_curr_ld->read_pagelist(rdata);
+        trypnfs = nfss->pnfs_curr_ld->read_pagelist(hdr);
         if (trypnfs != PNFS_NOT_ATTEMPTED)
                 nfs_inc_stats(inode, NFSIOS_PNFS_READ);
         dprintk("%s End (trypnfs:%d)\n", __func__, trypnfs);
@@ -1727,52 +1680,46 @@ pnfs_try_to_read_data(struct nfs_pgio_data *rdata,
 static void
 pnfs_do_read(struct nfs_pageio_descriptor *desc, struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_data *data = hdr->data;
         const struct rpc_call_ops *call_ops = desc->pg_rpc_callops;
         struct pnfs_layout_segment *lseg = desc->pg_lseg;
         enum pnfs_try_status trypnfs;
 
         desc->pg_lseg = NULL;
-        trypnfs = pnfs_try_to_read_data(data, call_ops, lseg);
+        trypnfs = pnfs_try_to_read_data(hdr, call_ops, lseg);
         if (trypnfs == PNFS_NOT_ATTEMPTED)
-                pnfs_read_through_mds(desc, data);
+                pnfs_read_through_mds(desc, hdr);
         pnfs_put_lseg(lseg);
 }
 
 static void pnfs_readhdr_free(struct nfs_pgio_header *hdr)
 {
         pnfs_put_lseg(hdr->lseg);
-        nfs_rw_header_free(hdr);
+        nfs_pgio_header_free(hdr);
 }
 EXPORT_SYMBOL_GPL(pnfs_readhdr_free);
 
 int
 pnfs_generic_pg_readpages(struct nfs_pageio_descriptor *desc)
 {
-        struct nfs_rw_header *rhdr;
         struct nfs_pgio_header *hdr;
         int ret;
 
-        rhdr = nfs_rw_header_alloc(desc->pg_rw_ops);
-        if (!rhdr) {
+        hdr = nfs_pgio_header_alloc(desc->pg_rw_ops);
+        if (!hdr) {
                 desc->pg_completion_ops->error_cleanup(&desc->pg_list);
                 ret = -ENOMEM;
                 pnfs_put_lseg(desc->pg_lseg);
                 desc->pg_lseg = NULL;
                 return ret;
         }
-        hdr = &rhdr->header;
         nfs_pgheader_init(desc, hdr, pnfs_readhdr_free);
         hdr->lseg = pnfs_get_lseg(desc->pg_lseg);
-        atomic_inc(&hdr->refcnt);
         ret = nfs_generic_pgio(desc, hdr);
         if (ret != 0) {
                 pnfs_put_lseg(desc->pg_lseg);
                 desc->pg_lseg = NULL;
         } else
                 pnfs_do_read(desc, hdr);
-        if (atomic_dec_and_test(&hdr->refcnt))
-                hdr->completion_ops->completion(hdr);
         return ret;
 }
 EXPORT_SYMBOL_GPL(pnfs_generic_pg_readpages);
@@ -1820,12 +1767,11 @@ void pnfs_set_lo_fail(struct pnfs_layout_segment *lseg)
 EXPORT_SYMBOL_GPL(pnfs_set_lo_fail);
 
 void
-pnfs_set_layoutcommit(struct nfs_pgio_data *wdata)
+pnfs_set_layoutcommit(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = wdata->header;
         struct inode *inode = hdr->inode;
         struct nfs_inode *nfsi = NFS_I(inode);
-        loff_t end_pos = wdata->mds_offset + wdata->res.count;
+        loff_t end_pos = hdr->mds_offset + hdr->res.count;
         bool mark_as_dirty = false;
 
         spin_lock(&inode->i_lock);
diff --git a/fs/nfs/pnfs.h b/fs/nfs/pnfs.h
index 4fb309a2b4c4..aca3dff5dae6 100644
--- a/fs/nfs/pnfs.h
+++ b/fs/nfs/pnfs.h
@@ -32,6 +32,7 @@
 
 #include <linux/nfs_fs.h>
 #include <linux/nfs_page.h>
+#include <linux/workqueue.h>
 
 enum {
         NFS_LSEG_VALID = 0,     /* cleared when lseg is recalled/returned */
@@ -46,6 +47,7 @@ struct pnfs_layout_segment {
         atomic_t pls_refcount;
         unsigned long pls_flags;
         struct pnfs_layout_hdr *pls_layout;
+        struct work_struct pls_work;
 };
 
 enum pnfs_try_status {
@@ -104,6 +106,8 @@ struct pnfs_layoutdriver_type {
                                   int max);
         void (*recover_commit_reqs) (struct list_head *list,
                                      struct nfs_commit_info *cinfo);
+        struct nfs_page * (*search_commit_reqs)(struct nfs_commit_info *cinfo,
+                                                struct page *page);
         int (*commit_pagelist)(struct inode *inode,
                                struct list_head *mds_pages,
                                int how,
@@ -113,8 +117,8 @@ struct pnfs_layoutdriver_type {
          * Return PNFS_ATTEMPTED to indicate the layout code has attempted
          * I/O, else return PNFS_NOT_ATTEMPTED to fall back to normal NFS
          */
-        enum pnfs_try_status (*read_pagelist) (struct nfs_pgio_data *nfs_data);
-        enum pnfs_try_status (*write_pagelist) (struct nfs_pgio_data *nfs_data, int how);
+        enum pnfs_try_status (*read_pagelist)(struct nfs_pgio_header *);
+        enum pnfs_try_status (*write_pagelist)(struct nfs_pgio_header *, int);
 
         void (*free_deviceid_node) (struct nfs4_deviceid_node *);
 
@@ -179,6 +183,7 @@ extern int nfs4_proc_layoutreturn(struct nfs4_layoutreturn *lrp);
 /* pnfs.c */
 void pnfs_get_layout_hdr(struct pnfs_layout_hdr *lo);
 void pnfs_put_lseg(struct pnfs_layout_segment *lseg);
+void pnfs_put_lseg_async(struct pnfs_layout_segment *lseg);
 
 void set_pnfs_layoutdriver(struct nfs_server *, const struct nfs_fh *, u32);
 void unset_pnfs_layoutdriver(struct nfs_server *);
@@ -213,13 +218,13 @@ bool pnfs_roc(struct inode *ino);
 void pnfs_roc_release(struct inode *ino);
 void pnfs_roc_set_barrier(struct inode *ino, u32 barrier);
 bool pnfs_roc_drain(struct inode *ino, u32 *barrier, struct rpc_task *task);
-void pnfs_set_layoutcommit(struct nfs_pgio_data *wdata);
+void pnfs_set_layoutcommit(struct nfs_pgio_header *);
 void pnfs_cleanup_layoutcommit(struct nfs4_layoutcommit_data *data);
 int pnfs_layoutcommit_inode(struct inode *inode, bool sync);
 int _pnfs_return_layout(struct inode *);
 int pnfs_commit_and_return_layout(struct inode *);
-void pnfs_ld_write_done(struct nfs_pgio_data *);
-void pnfs_ld_read_done(struct nfs_pgio_data *);
+void pnfs_ld_write_done(struct nfs_pgio_header *);
+void pnfs_ld_read_done(struct nfs_pgio_header *);
 struct pnfs_layout_segment *pnfs_update_layout(struct inode *ino,
                                                struct nfs_open_context *ctx,
                                                loff_t pos,
@@ -228,12 +233,8 @@ struct pnfs_layout_segment *pnfs_update_layout(struct inode *ino,
                                                gfp_t gfp_flags);
 
 void nfs4_deviceid_mark_client_invalid(struct nfs_client *clp);
-int pnfs_read_done_resend_to_mds(struct inode *inode, struct list_head *head,
-                        const struct nfs_pgio_completion_ops *compl_ops,
-                        struct nfs_direct_req *dreq);
-int pnfs_write_done_resend_to_mds(struct inode *inode, struct list_head *head,
-                        const struct nfs_pgio_completion_ops *compl_ops,
-                        struct nfs_direct_req *dreq);
+int pnfs_read_done_resend_to_mds(struct nfs_pgio_header *);
+int pnfs_write_done_resend_to_mds(struct nfs_pgio_header *);
 struct nfs4_threshold *pnfs_mdsthreshold_alloc(void);
 
 /* nfs4_deviceid_flags */
@@ -345,6 +346,17 @@ pnfs_recover_commit_reqs(struct inode *inode, struct list_head *list,
         NFS_SERVER(inode)->pnfs_curr_ld->recover_commit_reqs(list, cinfo);
 }
 
+static inline struct nfs_page *
+pnfs_search_commit_reqs(struct inode *inode, struct nfs_commit_info *cinfo,
+                        struct page *page)
+{
+        struct pnfs_layoutdriver_type *ld = NFS_SERVER(inode)->pnfs_curr_ld;
+
+        if (ld == NULL || ld->search_commit_reqs == NULL)
+                return NULL;
+        return ld->search_commit_reqs(cinfo, page);
+}
+
 /* Should the pNFS client commit and return the layout upon a setattr */
 static inline bool
 pnfs_ld_layoutret_on_setattr(struct inode *inode)
@@ -410,6 +422,10 @@ static inline void pnfs_put_lseg(struct pnfs_layout_segment *lseg)
 {
 }
 
+static inline void pnfs_put_lseg_async(struct pnfs_layout_segment *lseg)
+{
+}
+
 static inline int pnfs_return_layout(struct inode *ino)
 {
         return 0;
@@ -496,6 +512,13 @@ pnfs_recover_commit_reqs(struct inode *inode, struct list_head *list,
 {
 }
 
+static inline struct nfs_page *
+pnfs_search_commit_reqs(struct inode *inode, struct nfs_commit_info *cinfo,
+                        struct page *page)
+{
+        return NULL;
+}
+
 static inline int pnfs_layoutcommit_inode(struct inode *inode, bool sync)
 {
         return 0;
diff --git a/fs/nfs/proc.c b/fs/nfs/proc.c
index c171ce1a8a30..b09cc23d6f43 100644
--- a/fs/nfs/proc.c
+++ b/fs/nfs/proc.c
@@ -578,46 +578,49 @@ nfs_proc_pathconf(struct nfs_server *server, struct nfs_fh *fhandle,
         return 0;
 }
 
-static int nfs_read_done(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs_read_done(struct rpc_task *task, struct nfs_pgio_header *hdr)
 {
-        struct inode *inode = data->header->inode;
+        struct inode *inode = hdr->inode;
 
         nfs_invalidate_atime(inode);
         if (task->tk_status >= 0) {
-                nfs_refresh_inode(inode, data->res.fattr);
+                nfs_refresh_inode(inode, hdr->res.fattr);
                 /* Emulate the eof flag, which isn't normally needed in NFSv2
                  * as it is guaranteed to always return the file attributes
                  */
-                if (data->args.offset + data->res.count >= data->res.fattr->size)
-                        data->res.eof = 1;
+                if (hdr->args.offset + hdr->res.count >= hdr->res.fattr->size)
+                        hdr->res.eof = 1;
         }
         return 0;
 }
 
-static void nfs_proc_read_setup(struct nfs_pgio_data *data, struct rpc_message *msg)
+static void nfs_proc_read_setup(struct nfs_pgio_header *hdr,
+                                struct rpc_message *msg)
 {
         msg->rpc_proc = &nfs_procedures[NFSPROC_READ];
 }
 
-static int nfs_proc_pgio_rpc_prepare(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs_proc_pgio_rpc_prepare(struct rpc_task *task,
+                                     struct nfs_pgio_header *hdr)
 {
         rpc_call_start(task);
         return 0;
 }
 
-static int nfs_write_done(struct rpc_task *task, struct nfs_pgio_data *data)
+static int nfs_write_done(struct rpc_task *task, struct nfs_pgio_header *hdr)
 {
-        struct inode *inode = data->header->inode;
+        struct inode *inode = hdr->inode;
 
         if (task->tk_status >= 0)
-                nfs_post_op_update_inode_force_wcc(inode, data->res.fattr);
+                nfs_post_op_update_inode_force_wcc(inode, hdr->res.fattr);
         return 0;
 }
 
-static void nfs_proc_write_setup(struct nfs_pgio_data *data, struct rpc_message *msg)
+static void nfs_proc_write_setup(struct nfs_pgio_header *hdr,
+                                 struct rpc_message *msg)
 {
         /* Note: NFSv2 ignores @stable and always uses NFS_FILE_SYNC */
-        data->args.stable = NFS_FILE_SYNC;
+        hdr->args.stable = NFS_FILE_SYNC;
         msg->rpc_proc = &nfs_procedures[NFSPROC_WRITE];
 }
 
diff --git a/fs/nfs/read.c b/fs/nfs/read.c
index e818a475ca64..beff2769c5c5 100644
--- a/fs/nfs/read.c
+++ b/fs/nfs/read.c
@@ -33,12 +33,12 @@ static const struct nfs_rw_ops nfs_rw_read_ops;
 
 static struct kmem_cache *nfs_rdata_cachep;
 
-static struct nfs_rw_header *nfs_readhdr_alloc(void)
+static struct nfs_pgio_header *nfs_readhdr_alloc(void)
 {
         return kmem_cache_zalloc(nfs_rdata_cachep, GFP_KERNEL);
 }
 
-static void nfs_readhdr_free(struct nfs_rw_header *rhdr)
+static void nfs_readhdr_free(struct nfs_pgio_header *rhdr)
 {
         kmem_cache_free(nfs_rdata_cachep, rhdr);
 }
@@ -115,12 +115,6 @@ static void nfs_readpage_release(struct nfs_page *req)
 
                 unlock_page(req->wb_page);
         }
-
-        dprintk("NFS: read done (%s/%Lu %d@%Ld)\n",
-                        req->wb_context->dentry->d_inode->i_sb->s_id,
-                        (unsigned long long)NFS_FILEID(req->wb_context->dentry->d_inode),
-                        req->wb_bytes,
-                        (long long)req_offset(req));
         nfs_release_request(req);
 }
 
@@ -172,14 +166,15 @@ out:
         hdr->release(hdr);
 }
 
-static void nfs_initiate_read(struct nfs_pgio_data *data, struct rpc_message *msg,
+static void nfs_initiate_read(struct nfs_pgio_header *hdr,
+                              struct rpc_message *msg,
                               struct rpc_task_setup *task_setup_data, int how)
 {
-        struct inode *inode = data->header->inode;
+        struct inode *inode = hdr->inode;
         int swap_flags = IS_SWAPFILE(inode) ? NFS_RPC_SWAPFLAGS : 0;
 
         task_setup_data->flags |= swap_flags;
-        NFS_PROTO(inode)->read_setup(data, msg);
+        NFS_PROTO(inode)->read_setup(hdr, msg);
 }
 
 static void
@@ -203,14 +198,15 @@ static const struct nfs_pgio_completion_ops nfs_async_read_completion_ops = {
  * This is the callback from RPC telling us whether a reply was
  * received or some error occurred (timeout or socket shutdown).
  */
-static int nfs_readpage_done(struct rpc_task *task, struct nfs_pgio_data *data,
+static int nfs_readpage_done(struct rpc_task *task,
+                             struct nfs_pgio_header *hdr,
                              struct inode *inode)
 {
-        int status = NFS_PROTO(inode)->read_done(task, data);
+        int status = NFS_PROTO(inode)->read_done(task, hdr);
         if (status != 0)
                 return status;
 
-        nfs_add_stats(inode, NFSIOS_SERVERREADBYTES, data->res.count);
+        nfs_add_stats(inode, NFSIOS_SERVERREADBYTES, hdr->res.count);
 
         if (task->tk_status == -ESTALE) {
                 set_bit(NFS_INO_STALE, &NFS_I(inode)->flags);
@@ -219,34 +215,34 @@ static int nfs_readpage_done(struct rpc_task *task, struct nfs_pgio_data *data,
         return 0;
 }
 
-static void nfs_readpage_retry(struct rpc_task *task, struct nfs_pgio_data *data)
+static void nfs_readpage_retry(struct rpc_task *task,
+                               struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_args *argp = &data->args;
-        struct nfs_pgio_res  *resp = &data->res;
+        struct nfs_pgio_args *argp = &hdr->args;
+        struct nfs_pgio_res  *resp = &hdr->res;
 
         /* This is a short read! */
-        nfs_inc_stats(data->header->inode, NFSIOS_SHORTREAD);
+        nfs_inc_stats(hdr->inode, NFSIOS_SHORTREAD);
         /* Has the server at least made some progress? */
         if (resp->count == 0) {
-                nfs_set_pgio_error(data->header, -EIO, argp->offset);
+                nfs_set_pgio_error(hdr, -EIO, argp->offset);
                 return;
         }
-        /* Yes, so retry the read at the end of the data */
-        data->mds_offset += resp->count;
+        /* Yes, so retry the read at the end of the hdr */
+        hdr->mds_offset += resp->count;
         argp->offset += resp->count;
         argp->pgbase += resp->count;
         argp->count -= resp->count;
         rpc_restart_call_prepare(task);
 }
 
-static void nfs_readpage_result(struct rpc_task *task, struct nfs_pgio_data *data)
+static void nfs_readpage_result(struct rpc_task *task,
+                                struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-
-        if (data->res.eof) {
+        if (hdr->res.eof) {
                 loff_t bound;
 
-                bound = data->args.offset + data->res.count;
+                bound = hdr->args.offset + hdr->res.count;
                 spin_lock(&hdr->lock);
                 if (bound < hdr->io_start + hdr->good_bytes) {
                         set_bit(NFS_IOHDR_EOF, &hdr->flags);
@@ -254,8 +250,8 @@ static void nfs_readpage_result(struct rpc_task *task, struct nfs_pgio_data *dat
                         hdr->good_bytes = bound - hdr->io_start;
                 }
                 spin_unlock(&hdr->lock);
-        } else if (data->res.count != data->args.count)
-                nfs_readpage_retry(task, data);
+        } else if (hdr->res.count != hdr->args.count)
+                nfs_readpage_retry(task, hdr);
 }
 
 /*
@@ -404,7 +400,7 @@ out:
 int __init nfs_init_readpagecache(void)
 {
         nfs_rdata_cachep = kmem_cache_create("nfs_read_data",
-                                             sizeof(struct nfs_rw_header),
+                                             sizeof(struct nfs_pgio_header),
                                              0, SLAB_HWCACHE_ALIGN,
                                              NULL);
         if (nfs_rdata_cachep == NULL)
diff --git a/fs/nfs/super.c b/fs/nfs/super.c
index 084af1060d79..e4499d5b51e8 100644
--- a/fs/nfs/super.c
+++ b/fs/nfs/super.c
@@ -1027,8 +1027,7 @@ static bool nfs_auth_info_add(struct nfs_auth_info *auth_info,
                               rpc_authflavor_t flavor)
 {
         unsigned int i;
-        unsigned int max_flavor_len = (sizeof(auth_info->flavors) /
-                                       sizeof(auth_info->flavors[0]));
+        unsigned int max_flavor_len = ARRAY_SIZE(auth_info->flavors);
 
         /* make sure this flavor isn't already in the list */
         for (i = 0; i < auth_info->flavor_len; i++) {
@@ -2180,7 +2179,7 @@ out_no_address:
         return -EINVAL;
 }
 
-#define NFS_MOUNT_CMP_FLAGMASK ~(NFS_MOUNT_INTR \
+#define NFS_REMOUNT_CMP_FLAGMASK ~(NFS_MOUNT_INTR \
                 | NFS_MOUNT_SECURE \
                 | NFS_MOUNT_TCP \
                 | NFS_MOUNT_VER3 \
@@ -2188,15 +2187,16 @@ out_no_address:
                 | NFS_MOUNT_NONLM \
                 | NFS_MOUNT_BROKEN_SUID \
                 | NFS_MOUNT_STRICTLOCK \
-                | NFS_MOUNT_UNSHARED \
-                | NFS_MOUNT_NORESVPORT \
                 | NFS_MOUNT_LEGACY_INTERFACE)
 
+#define NFS_MOUNT_CMP_FLAGMASK (NFS_REMOUNT_CMP_FLAGMASK & \
+                ~(NFS_MOUNT_UNSHARED | NFS_MOUNT_NORESVPORT))
+
 static int
 nfs_compare_remount_data(struct nfs_server *nfss,
                          struct nfs_parsed_mount_data *data)
 {
-        if ((data->flags ^ nfss->flags) & NFS_MOUNT_CMP_FLAGMASK ||
+        if ((data->flags ^ nfss->flags) & NFS_REMOUNT_CMP_FLAGMASK ||
             data->rsize != nfss->rsize ||
             data->wsize != nfss->wsize ||
             data->version != nfss->nfs_client->rpc_ops->version ||
diff --git a/fs/nfs/write.c b/fs/nfs/write.c
index 962c9ee758be..e3b5cf28bdc5 100644
--- a/fs/nfs/write.c
+++ b/fs/nfs/write.c
@@ -47,6 +47,8 @@ static const struct nfs_pgio_completion_ops nfs_async_write_completion_ops;
 static const struct nfs_commit_completion_ops nfs_commit_completion_ops;
 static const struct nfs_rw_ops nfs_rw_write_ops;
 static void nfs_clear_request_commit(struct nfs_page *req);
+static void nfs_init_cinfo_from_inode(struct nfs_commit_info *cinfo,
+                                      struct inode *inode);
 
 static struct kmem_cache *nfs_wdata_cachep;
 static mempool_t *nfs_wdata_mempool;
@@ -71,18 +73,18 @@ void nfs_commit_free(struct nfs_commit_data *p)
 }
 EXPORT_SYMBOL_GPL(nfs_commit_free);
 
-static struct nfs_rw_header *nfs_writehdr_alloc(void)
+static struct nfs_pgio_header *nfs_writehdr_alloc(void)
 {
-        struct nfs_rw_header *p = mempool_alloc(nfs_wdata_mempool, GFP_NOIO);
+        struct nfs_pgio_header *p = mempool_alloc(nfs_wdata_mempool, GFP_NOIO);
 
         if (p)
                 memset(p, 0, sizeof(*p));
         return p;
 }
 
-static void nfs_writehdr_free(struct nfs_rw_header *whdr)
+static void nfs_writehdr_free(struct nfs_pgio_header *hdr)
 {
-        mempool_free(whdr, nfs_wdata_mempool);
+        mempool_free(hdr, nfs_wdata_mempool);
 }
 
 static void nfs_context_set_write_error(struct nfs_open_context *ctx, int error)
@@ -93,6 +95,38 @@ static void nfs_context_set_write_error(struct nfs_open_context *ctx, int error)
 }
 
 /*
+ * nfs_page_search_commits_for_head_request_locked
+ *
+ * Search through commit lists on @inode for the head request for @page.
+ * Must be called while holding the inode (which is cinfo) lock.
+ *
+ * Returns the head request if found, or NULL if not found.
+ */
+static struct nfs_page *
+nfs_page_search_commits_for_head_request_locked(struct nfs_inode *nfsi,
+                                                struct page *page)
+{
+        struct nfs_page *freq, *t;
+        struct nfs_commit_info cinfo;
+        struct inode *inode = &nfsi->vfs_inode;
+
+        nfs_init_cinfo_from_inode(&cinfo, inode);
+
+        /* search through pnfs commit lists */
+        freq = pnfs_search_commit_reqs(inode, &cinfo, page);
+        if (freq)
+                return freq->wb_head;
+
+        /* Linearly search the commit list for the correct request */
+        list_for_each_entry_safe(freq, t, &cinfo.mds->list, wb_list) {
+                if (freq->wb_page == page)
+                        return freq->wb_head;
+        }
+
+        return NULL;
+}
+
+/*
  * nfs_page_find_head_request_locked - find head request associated with @page
  *
  * must be called while holding the inode lock.
@@ -106,21 +140,12 @@ nfs_page_find_head_request_locked(struct nfs_inode *nfsi, struct page *page)
 
         if (PagePrivate(page))
                 req = (struct nfs_page *)page_private(page);
-        else if (unlikely(PageSwapCache(page))) {
-                struct nfs_page *freq, *t;
-
-                /* Linearly search the commit list for the correct req */
-                list_for_each_entry_safe(freq, t, &nfsi->commit_info.list, wb_list) {
-                        if (freq->wb_page == page) {
-                                req = freq->wb_head;
-                                break;
-                        }
-                }
-        }
+        else if (unlikely(PageSwapCache(page)))
+                req = nfs_page_search_commits_for_head_request_locked(nfsi,
+                        page);
 
         if (req) {
                 WARN_ON_ONCE(req->wb_head != req);
-
                 kref_get(&req->wb_kref);
         }
 
@@ -216,7 +241,7 @@ static bool nfs_page_group_covers_page(struct nfs_page *req)
         unsigned int pos = 0;
         unsigned int len = nfs_page_length(req->wb_page);
 
-        nfs_page_group_lock(req);
+        nfs_page_group_lock(req, true);
 
         do {
                 tmp = nfs_page_group_search_locked(req->wb_head, pos);
@@ -379,8 +404,6 @@ nfs_destroy_unlinked_subrequests(struct nfs_page *destroy_list,
                 subreq->wb_head = subreq;
                 subreq->wb_this_page = subreq;
 
-                nfs_clear_request_commit(subreq);
-
                 /* subreq is now totally disconnected from page group or any
                  * write / commit lists. last chance to wake any waiters */
                 nfs_unlock_request(subreq);
@@ -456,7 +479,9 @@ try_again:
         }
 
         /* lock each request in the page group */
-        nfs_page_group_lock(head);
+        ret = nfs_page_group_lock(head, false);
+        if (ret < 0)
+                return ERR_PTR(ret);
         subreq = head;
         do {
                 /*
@@ -488,7 +513,7 @@ try_again:
          * Commit list removal accounting is done after locks are dropped */
         subreq = head;
         do {
-                nfs_list_remove_request(subreq);
+                nfs_clear_request_commit(subreq);
                 subreq = subreq->wb_this_page;
         } while (subreq != head);
 
@@ -518,15 +543,11 @@ try_again:
 
         nfs_page_group_unlock(head);
 
-        /* drop lock to clear_request_commit the head req and clean up
-         * requests on destroy list */
+        /* drop lock to clean uprequests on destroy list */
         spin_unlock(&inode->i_lock);
 
         nfs_destroy_unlinked_subrequests(destroy_list, head);
 
-        /* clean up commit list state */
-        nfs_clear_request_commit(head);
-
         /* still holds ref on head from nfs_page_find_head_request_locked
          * and still has lock on head from lock loop */
         return head;
@@ -705,6 +726,8 @@ static void nfs_inode_remove_request(struct nfs_page *req)
 
         if (test_and_clear_bit(PG_INODE_REF, &req->wb_flags))
                 nfs_release_request(req);
+        else
+                WARN_ON_ONCE(1);
 }
 
 static void
@@ -808,6 +831,7 @@ nfs_clear_page_commit(struct page *page)
         dec_bdi_stat(page_file_mapping(page)->backing_dev_info, BDI_RECLAIMABLE);
 }
 
+/* Called holding inode (/cinfo) lock */
 static void
 nfs_clear_request_commit(struct nfs_page *req)
 {
@@ -817,20 +841,17 @@ nfs_clear_request_commit(struct nfs_page *req)
 
                 nfs_init_cinfo_from_inode(&cinfo, inode);
                 if (!pnfs_clear_request_commit(req, &cinfo)) {
-                        spin_lock(cinfo.lock);
                         nfs_request_remove_commit_list(req, &cinfo);
-                        spin_unlock(cinfo.lock);
                 }
                 nfs_clear_page_commit(req->wb_page);
         }
 }
 
-static inline
-int nfs_write_need_commit(struct nfs_pgio_data *data)
+int nfs_write_need_commit(struct nfs_pgio_header *hdr)
 {
-        if (data->verf.committed == NFS_DATA_SYNC)
-                return data->header->lseg == NULL;
-        return data->verf.committed != NFS_FILE_SYNC;
+        if (hdr->verf.committed == NFS_DATA_SYNC)
+                return hdr->lseg == NULL;
+        return hdr->verf.committed != NFS_FILE_SYNC;
 }
 
 #else
@@ -856,8 +877,7 @@ nfs_clear_request_commit(struct nfs_page *req)
 {
 }
 
-static inline
-int nfs_write_need_commit(struct nfs_pgio_data *data)
+int nfs_write_need_commit(struct nfs_pgio_header *hdr)
 {
         return 0;
 }
@@ -883,11 +903,7 @@ static void nfs_write_completion(struct nfs_pgio_header *hdr)
                         nfs_context_set_write_error(req->wb_context, hdr->error);
                         goto remove_req;
                 }
-                if (test_bit(NFS_IOHDR_NEED_RESCHED, &hdr->flags)) {
-                        nfs_mark_request_dirty(req);
-                        goto next;
-                }
-                if (test_bit(NFS_IOHDR_NEED_COMMIT, &hdr->flags)) {
+                if (nfs_write_need_commit(hdr)) {
                         memcpy(&req->wb_verf, &hdr->verf.verifier, sizeof(req->wb_verf));
                         nfs_mark_request_commit(req, hdr->lseg, &cinfo);
                         goto next;
@@ -1038,9 +1054,9 @@ static struct nfs_page *nfs_try_to_update_request(struct inode *inode,
         else
                 req->wb_bytes = rqend - req->wb_offset;
 out_unlock:
-        spin_unlock(&inode->i_lock);
         if (req)
                 nfs_clear_request_commit(req);
+        spin_unlock(&inode->i_lock);
         return req;
 out_flushme:
         spin_unlock(&inode->i_lock);
@@ -1241,17 +1257,18 @@ static int flush_task_priority(int how)
         return RPC_PRIORITY_NORMAL;
 }
 
-static void nfs_initiate_write(struct nfs_pgio_data *data, struct rpc_message *msg,
+static void nfs_initiate_write(struct nfs_pgio_header *hdr,
+                               struct rpc_message *msg,
                                struct rpc_task_setup *task_setup_data, int how)
 {
-        struct inode *inode = data->header->inode;
+        struct inode *inode = hdr->inode;
         int priority = flush_task_priority(how);
 
         task_setup_data->priority = priority;
-        NFS_PROTO(inode)->write_setup(data, msg);
+        NFS_PROTO(inode)->write_setup(hdr, msg);
 
         nfs4_state_protect_write(NFS_SERVER(inode)->nfs_client,
-                                 &task_setup_data->rpc_client, msg, data);
+                                 &task_setup_data->rpc_client, msg, hdr);
 }
 
 /* If a nfs_flush_* function fails, it should remove reqs from @head and
@@ -1313,21 +1330,9 @@ void nfs_commit_prepare(struct rpc_task *task, void *calldata)
         NFS_PROTO(data->inode)->commit_rpc_prepare(task, data);
 }
 
-static void nfs_writeback_release_common(struct nfs_pgio_data *data)
+static void nfs_writeback_release_common(struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_header *hdr = data->header;
-        int status = data->task.tk_status;
-
-        if ((status >= 0) && nfs_write_need_commit(data)) {
-                spin_lock(&hdr->lock);
-                if (test_bit(NFS_IOHDR_NEED_RESCHED, &hdr->flags))
-                        ; /* Do nothing */
-                else if (!test_and_set_bit(NFS_IOHDR_NEED_COMMIT, &hdr->flags))
-                        memcpy(&hdr->verf, &data->verf, sizeof(hdr->verf));
-                else if (memcmp(&hdr->verf, &data->verf, sizeof(hdr->verf)))
-                        set_bit(NFS_IOHDR_NEED_RESCHED, &hdr->flags);
-                spin_unlock(&hdr->lock);
-        }
+        /* do nothing! */
 }
 
 /*
@@ -1358,7 +1363,8 @@ static int nfs_should_remove_suid(const struct inode *inode)
 /*
  * This function is called when the WRITE call is complete.
  */
-static int nfs_writeback_done(struct rpc_task *task, struct nfs_pgio_data *data,
+static int nfs_writeback_done(struct rpc_task *task,
+                              struct nfs_pgio_header *hdr,
                               struct inode *inode)
 {
         int status;
@@ -1370,13 +1376,14 @@ static int nfs_writeback_done(struct rpc_task *task, struct nfs_pgio_data *data,
          * another writer had changed the file, but some applications
          * depend on tighter cache coherency when writing.
          */
-        status = NFS_PROTO(inode)->write_done(task, data);
+        status = NFS_PROTO(inode)->write_done(task, hdr);
         if (status != 0)
                 return status;
-        nfs_add_stats(inode, NFSIOS_SERVERWRITTENBYTES, data->res.count);
+        nfs_add_stats(inode, NFSIOS_SERVERWRITTENBYTES, hdr->res.count);
 
 #if IS_ENABLED(CONFIG_NFS_V3) || IS_ENABLED(CONFIG_NFS_V4)
-        if (data->res.verf->committed < data->args.stable && task->tk_status >= 0) {
+        if (hdr->res.verf->committed < hdr->args.stable &&
+            task->tk_status >= 0) {
                 /* We tried a write call, but the server did not
                  * commit data to stable storage even though we
                  * requested it.
@@ -1392,7 +1399,7 @@ static int nfs_writeback_done(struct rpc_task *task, struct nfs_pgio_data *data,
                         dprintk("NFS:       faulty NFS server %s:"
                                 " (committed = %d) != (stable = %d)\n",
                                 NFS_SERVER(inode)->nfs_client->cl_hostname,
-                                data->res.verf->committed, data->args.stable);
+                                hdr->res.verf->committed, hdr->args.stable);
                         complain = jiffies + 300 * HZ;
                 }
         }
@@ -1407,16 +1414,17 @@ static int nfs_writeback_done(struct rpc_task *task, struct nfs_pgio_data *data,
 /*
  * This function is called when the WRITE call is complete.
  */
-static void nfs_writeback_result(struct rpc_task *task, struct nfs_pgio_data *data)
+static void nfs_writeback_result(struct rpc_task *task,
+                                 struct nfs_pgio_header *hdr)
 {
-        struct nfs_pgio_args    *argp = &data->args;
-        struct nfs_pgio_res     *resp = &data->res;
+        struct nfs_pgio_args    *argp = &hdr->args;
+        struct nfs_pgio_res     *resp = &hdr->res;
 
         if (resp->count < argp->count) {
                 static unsigned long    complain;
 
                 /* This a short write! */
-                nfs_inc_stats(data->header->inode, NFSIOS_SHORTWRITE);
+                nfs_inc_stats(hdr->inode, NFSIOS_SHORTWRITE);
 
                 /* Has the server at least made some progress? */
                 if (resp->count == 0) {
@@ -1426,14 +1434,14 @@ static void nfs_writeback_result(struct rpc_task *task, struct nfs_pgio_data *da
                                        argp->count);
                                 complain = jiffies + 300 * HZ;
                         }
-                        nfs_set_pgio_error(data->header, -EIO, argp->offset);
+                        nfs_set_pgio_error(hdr, -EIO, argp->offset);
                         task->tk_status = -EIO;
                         return;
                 }
                 /* Was this an NFSv2 write or an NFSv3 stable write? */
                 if (resp->verf->committed != NFS_UNSTABLE) {
                         /* Resend from where the server left off */
-                        data->mds_offset += resp->count;
+                        hdr->mds_offset += resp->count;
                         argp->offset += resp->count;
                         argp->pgbase += resp->count;
                         argp->count -= resp->count;
@@ -1884,7 +1892,7 @@ int nfs_migrate_page(struct address_space *mapping, struct page *newpage,
 int __init nfs_init_writepagecache(void)
 {
         nfs_wdata_cachep = kmem_cache_create("nfs_write_data",
-                                             sizeof(struct nfs_rw_header),
+                                             sizeof(struct nfs_pgio_header),
                                              0, SLAB_HWCACHE_ALIGN,
                                              NULL);
         if (nfs_wdata_cachep == NULL)
diff --git a/fs/nfs_common/nfsacl.c b/fs/nfs_common/nfsacl.c
index ed628f71274c..538f142935ea 100644
--- a/fs/nfs_common/nfsacl.c
+++ b/fs/nfs_common/nfsacl.c
@@ -30,9 +30,6 @@
 
 MODULE_LICENSE("GPL");
 
-EXPORT_SYMBOL_GPL(nfsacl_encode);
-EXPORT_SYMBOL_GPL(nfsacl_decode);
-
 struct nfsacl_encode_desc {
         struct xdr_array2_desc desc;
         unsigned int count;
@@ -136,6 +133,7 @@ int nfsacl_encode(struct xdr_buf *buf, unsigned int base, struct inode *inode,
                           nfsacl_desc.desc.array_len;
         return err;
 }
+EXPORT_SYMBOL_GPL(nfsacl_encode);
 
 struct nfsacl_decode_desc {
         struct xdr_array2_desc desc;
@@ -295,3 +293,4 @@ int nfsacl_decode(struct xdr_buf *buf, unsigned int base, unsigned int *aclcnt,
         return 8 + nfsacl_desc.desc.elem_size *
                    nfsacl_desc.desc.array_len;
 }
+EXPORT_SYMBOL_GPL(nfsacl_decode);