Merge tag 'nfs-for-4.1-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs
authorLinus Torvalds <torvalds@linux-foundation.org>
Mon, 27 Apr 2015 00:33:59 +0000 (17:33 -0700)
committerLinus Torvalds <torvalds@linux-foundation.org>
Mon, 27 Apr 2015 00:33:59 +0000 (17:33 -0700)
Pull NFS client updates from Trond Myklebust:
 "Another set of mainly bugfixes and a couple of cleanups.  No new
  functionality in this round.

  Highlights include:

  Stable patches:
   - Fix a regression in /proc/self/mountstats
   - Fix the pNFS flexfiles O_DIRECT support
   - Fix high load average due to callback thread sleeping

  Bugfixes:
   - Various patches to fix the pNFS layoutcommit support
   - Do not cache pNFS deviceids unless server notifications are enabled
   - Fix a SUNRPC transport reconnection regression
   - make debugfs file creation failure non-fatal in SUNRPC
   - Another fix for circular directory warnings on NFSv4 "junctioned"
     mountpoints
   - Fix locking around NFSv4.2 fallocate() support
   - Truncating NFSv4 file opens should also sync O_DIRECT writes
   - Prevent infinite loop in rpcrdma_ep_create()

  Features:
   - Various improvements to the RDMA transport code's handling of
     memory registration
   - Various code cleanups"

* tag 'nfs-for-4.1-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs: (55 commits)
  fs/nfs: fix new compiler warning about boolean in switch
  nfs: Remove unneeded casts in nfs
  NFS: Don't attempt to decode missing directory entries
  Revert "nfs: replace nfs_add_stats with nfs_inc_stats when add one"
  NFS: Rename idmap.c to nfs4idmap.c
  NFS: Move nfs_idmap.h into fs/nfs/
  NFS: Remove CONFIG_NFS_V4 checks from nfs_idmap.h
  NFS: Add a stub for GETDEVICELIST
  nfs: remove WARN_ON_ONCE from nfs_direct_good_bytes
  nfs: fix DIO good bytes calculation
  nfs: Fetch MOUNTED_ON_FILEID when updating an inode
  sunrpc: make debugfs file creation failure non-fatal
  nfs: fix high load average due to callback thread sleeping
  NFS: Reduce time spent holding the i_mutex during fallocate()
  NFS: Don't zap caches on fallocate()
  xprtrdma: Make rpcrdma_{un}map_one() into inline functions
  xprtrdma: Handle non-SEND completions via a callout
  xprtrdma: Add "open" memreg op
  xprtrdma: Add "destroy MRs" memreg op
  xprtrdma: Add "reset MRs" memreg op
  ...

15 files changed:
1  2 
fs/nfs/dir.c
fs/nfs/direct.c
fs/nfs/file.c
fs/nfs/inode.c
fs/nfs/nfs4client.c
fs/nfs/nfs4file.c
fs/nfs/nfs4proc.c
fs/nfs/nfs4state.c
fs/nfs/pnfs.h
fs/nfs/read.c
fs/nfs/super.c
fs/nfs/write.c
include/linux/nfs_fs.h
net/sunrpc/xprtrdma/fmr_ops.c
net/sunrpc/xprtrdma/frwr_ops.c

diff --cc fs/nfs/dir.c
Simple merge
diff --cc fs/nfs/direct.c
index b2cbc3a6cdd9dd2702c97872f8caa14038dbe88a,eeb52b434e6fb0e40ed81a905e1574968d395858..38678d9a5cc4a64838e70ff3c915107b828bb65d
@@@ -258,18 -262,11 +261,11 @@@ ssize_t nfs_direct_IO(struct kiocb *ioc
        if (!IS_SWAPFILE(inode))
                return 0;
  
- #ifndef CONFIG_NFS_SWAP
-       dprintk("NFS: nfs_direct_IO (%pD) off/no(%Ld/%lu) EINVAL\n",
-                       iocb->ki_filp, (long long) pos, iter->nr_segs);
-       return -EINVAL;
- #else
 -      VM_BUG_ON(iocb->ki_nbytes != PAGE_SIZE);
 +      VM_BUG_ON(iov_iter_count(iter) != PAGE_SIZE);
  
 -      if (rw == READ)
 +      if (iov_iter_rw(iter) == READ)
                return nfs_file_direct_read(iocb, iter, pos);
 -      return nfs_file_direct_write(iocb, iter, pos);
 +      return nfs_file_direct_write(iocb, iter);
- #endif /* CONFIG_NFS_SWAP */
  }
  
  static void nfs_direct_release_pages(struct page **pages, unsigned int npages)
diff --cc fs/nfs/file.c
Simple merge
diff --cc fs/nfs/inode.c
Simple merge
Simple merge
Simple merge
Simple merge
Simple merge
diff --cc fs/nfs/pnfs.h
Simple merge
diff --cc fs/nfs/read.c
Simple merge
diff --cc fs/nfs/super.c
Simple merge
diff --cc fs/nfs/write.c
Simple merge
Simple merge
index 0000000000000000000000000000000000000000,a91ba2c8ef1e74c228bb045d50e1b56846d95085..302d4ebf6fbfb2a2c15780ef966b52bb473b91e4
mode 000000,100644..100644
--- /dev/null
@@@ -1,0 -1,208 +1,208 @@@
 -      dprintk("RPC:       %s: initalizing %d FMRs\n", __func__, i);
+ /*
+  * Copyright (c) 2015 Oracle.  All rights reserved.
+  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
+  */
+ /* Lightweight memory registration using Fast Memory Regions (FMR).
+  * Referred to sometimes as MTHCAFMR mode.
+  *
+  * FMR uses synchronous memory registration and deregistration.
+  * FMR registration is known to be fast, but FMR deregistration
+  * can take tens of usecs to complete.
+  */
+ #include "xprt_rdma.h"
+ #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
+ # define RPCDBG_FACILITY      RPCDBG_TRANS
+ #endif
+ /* Maximum scatter/gather per FMR */
+ #define RPCRDMA_MAX_FMR_SGES  (64)
+ static int
+ fmr_op_open(struct rpcrdma_ia *ia, struct rpcrdma_ep *ep,
+           struct rpcrdma_create_data_internal *cdata)
+ {
+       return 0;
+ }
+ /* FMR mode conveys up to 64 pages of payload per chunk segment.
+  */
+ static size_t
+ fmr_op_maxpages(struct rpcrdma_xprt *r_xprt)
+ {
+       return min_t(unsigned int, RPCRDMA_MAX_DATA_SEGS,
+                    rpcrdma_max_segments(r_xprt) * RPCRDMA_MAX_FMR_SGES);
+ }
+ static int
+ fmr_op_init(struct rpcrdma_xprt *r_xprt)
+ {
+       struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
+       int mr_access_flags = IB_ACCESS_REMOTE_WRITE | IB_ACCESS_REMOTE_READ;
+       struct ib_fmr_attr fmr_attr = {
+               .max_pages      = RPCRDMA_MAX_FMR_SGES,
+               .max_maps       = 1,
+               .page_shift     = PAGE_SHIFT
+       };
+       struct ib_pd *pd = r_xprt->rx_ia.ri_pd;
+       struct rpcrdma_mw *r;
+       int i, rc;
+       INIT_LIST_HEAD(&buf->rb_mws);
+       INIT_LIST_HEAD(&buf->rb_all);
+       i = (buf->rb_max_requests + 1) * RPCRDMA_MAX_SEGS;
++      dprintk("RPC:       %s: initializing %d FMRs\n", __func__, i);
+       while (i--) {
+               r = kzalloc(sizeof(*r), GFP_KERNEL);
+               if (!r)
+                       return -ENOMEM;
+               r->r.fmr = ib_alloc_fmr(pd, mr_access_flags, &fmr_attr);
+               if (IS_ERR(r->r.fmr))
+                       goto out_fmr_err;
+               list_add(&r->mw_list, &buf->rb_mws);
+               list_add(&r->mw_all, &buf->rb_all);
+       }
+       return 0;
+ out_fmr_err:
+       rc = PTR_ERR(r->r.fmr);
+       dprintk("RPC:       %s: ib_alloc_fmr status %i\n", __func__, rc);
+       kfree(r);
+       return rc;
+ }
+ /* Use the ib_map_phys_fmr() verb to register a memory region
+  * for remote access via RDMA READ or RDMA WRITE.
+  */
+ static int
+ fmr_op_map(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr_seg *seg,
+          int nsegs, bool writing)
+ {
+       struct rpcrdma_ia *ia = &r_xprt->rx_ia;
+       struct ib_device *device = ia->ri_id->device;
+       enum dma_data_direction direction = rpcrdma_data_dir(writing);
+       struct rpcrdma_mr_seg *seg1 = seg;
+       struct rpcrdma_mw *mw = seg1->rl_mw;
+       u64 physaddrs[RPCRDMA_MAX_DATA_SEGS];
+       int len, pageoff, i, rc;
+       pageoff = offset_in_page(seg1->mr_offset);
+       seg1->mr_offset -= pageoff;     /* start of page */
+       seg1->mr_len += pageoff;
+       len = -pageoff;
+       if (nsegs > RPCRDMA_MAX_FMR_SGES)
+               nsegs = RPCRDMA_MAX_FMR_SGES;
+       for (i = 0; i < nsegs;) {
+               rpcrdma_map_one(device, seg, direction);
+               physaddrs[i] = seg->mr_dma;
+               len += seg->mr_len;
+               ++seg;
+               ++i;
+               /* Check for holes */
+               if ((i < nsegs && offset_in_page(seg->mr_offset)) ||
+                   offset_in_page((seg-1)->mr_offset + (seg-1)->mr_len))
+                       break;
+       }
+       rc = ib_map_phys_fmr(mw->r.fmr, physaddrs, i, seg1->mr_dma);
+       if (rc)
+               goto out_maperr;
+       seg1->mr_rkey = mw->r.fmr->rkey;
+       seg1->mr_base = seg1->mr_dma + pageoff;
+       seg1->mr_nsegs = i;
+       seg1->mr_len = len;
+       return i;
+ out_maperr:
+       dprintk("RPC:       %s: ib_map_phys_fmr %u@0x%llx+%i (%d) status %i\n",
+               __func__, len, (unsigned long long)seg1->mr_dma,
+               pageoff, i, rc);
+       while (i--)
+               rpcrdma_unmap_one(device, --seg);
+       return rc;
+ }
+ /* Use the ib_unmap_fmr() verb to prevent further remote
+  * access via RDMA READ or RDMA WRITE.
+  */
+ static int
+ fmr_op_unmap(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr_seg *seg)
+ {
+       struct rpcrdma_ia *ia = &r_xprt->rx_ia;
+       struct rpcrdma_mr_seg *seg1 = seg;
+       struct ib_device *device;
+       int rc, nsegs = seg->mr_nsegs;
+       LIST_HEAD(l);
+       list_add(&seg1->rl_mw->r.fmr->list, &l);
+       rc = ib_unmap_fmr(&l);
+       read_lock(&ia->ri_qplock);
+       device = ia->ri_id->device;
+       while (seg1->mr_nsegs--)
+               rpcrdma_unmap_one(device, seg++);
+       read_unlock(&ia->ri_qplock);
+       if (rc)
+               goto out_err;
+       return nsegs;
+ out_err:
+       dprintk("RPC:       %s: ib_unmap_fmr status %i\n", __func__, rc);
+       return nsegs;
+ }
+ /* After a disconnect, unmap all FMRs.
+  *
+  * This is invoked only in the transport connect worker in order
+  * to serialize with rpcrdma_register_fmr_external().
+  */
+ static void
+ fmr_op_reset(struct rpcrdma_xprt *r_xprt)
+ {
+       struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
+       struct rpcrdma_mw *r;
+       LIST_HEAD(list);
+       int rc;
+       list_for_each_entry(r, &buf->rb_all, mw_all)
+               list_add(&r->r.fmr->list, &list);
+       rc = ib_unmap_fmr(&list);
+       if (rc)
+               dprintk("RPC:       %s: ib_unmap_fmr failed %i\n",
+                       __func__, rc);
+ }
+ static void
+ fmr_op_destroy(struct rpcrdma_buffer *buf)
+ {
+       struct rpcrdma_mw *r;
+       int rc;
+       while (!list_empty(&buf->rb_all)) {
+               r = list_entry(buf->rb_all.next, struct rpcrdma_mw, mw_all);
+               list_del(&r->mw_all);
+               rc = ib_dealloc_fmr(r->r.fmr);
+               if (rc)
+                       dprintk("RPC:       %s: ib_dealloc_fmr failed %i\n",
+                               __func__, rc);
+               kfree(r);
+       }
+ }
+ const struct rpcrdma_memreg_ops rpcrdma_fmr_memreg_ops = {
+       .ro_map                         = fmr_op_map,
+       .ro_unmap                       = fmr_op_unmap,
+       .ro_open                        = fmr_op_open,
+       .ro_maxpages                    = fmr_op_maxpages,
+       .ro_init                        = fmr_op_init,
+       .ro_reset                       = fmr_op_reset,
+       .ro_destroy                     = fmr_op_destroy,
+       .ro_displayname                 = "fmr",
+ };
index 0000000000000000000000000000000000000000,0a7b9df701333c0fcc399ad3648cc374a2666698..dff0481dbcf8011dbbdf3050dd4ff9e67f0a6df3
mode 000000,100644..100644
--- /dev/null
@@@ -1,0 -1,353 +1,353 @@@
 -      dprintk("RPC:       %s: initalizing %d FRMRs\n", __func__, i);
+ /*
+  * Copyright (c) 2015 Oracle.  All rights reserved.
+  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
+  */
+ /* Lightweight memory registration using Fast Registration Work
+  * Requests (FRWR). Also referred to sometimes as FRMR mode.
+  *
+  * FRWR features ordered asynchronous registration and deregistration
+  * of arbitrarily sized memory regions. This is the fastest and safest
+  * but most complex memory registration mode.
+  */
+ #include "xprt_rdma.h"
+ #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
+ # define RPCDBG_FACILITY      RPCDBG_TRANS
+ #endif
+ static int
+ __frwr_init(struct rpcrdma_mw *r, struct ib_pd *pd, struct ib_device *device,
+           unsigned int depth)
+ {
+       struct rpcrdma_frmr *f = &r->r.frmr;
+       int rc;
+       f->fr_mr = ib_alloc_fast_reg_mr(pd, depth);
+       if (IS_ERR(f->fr_mr))
+               goto out_mr_err;
+       f->fr_pgl = ib_alloc_fast_reg_page_list(device, depth);
+       if (IS_ERR(f->fr_pgl))
+               goto out_list_err;
+       return 0;
+ out_mr_err:
+       rc = PTR_ERR(f->fr_mr);
+       dprintk("RPC:       %s: ib_alloc_fast_reg_mr status %i\n",
+               __func__, rc);
+       return rc;
+ out_list_err:
+       rc = PTR_ERR(f->fr_pgl);
+       dprintk("RPC:       %s: ib_alloc_fast_reg_page_list status %i\n",
+               __func__, rc);
+       ib_dereg_mr(f->fr_mr);
+       return rc;
+ }
+ static void
+ __frwr_release(struct rpcrdma_mw *r)
+ {
+       int rc;
+       rc = ib_dereg_mr(r->r.frmr.fr_mr);
+       if (rc)
+               dprintk("RPC:       %s: ib_dereg_mr status %i\n",
+                       __func__, rc);
+       ib_free_fast_reg_page_list(r->r.frmr.fr_pgl);
+ }
+ static int
+ frwr_op_open(struct rpcrdma_ia *ia, struct rpcrdma_ep *ep,
+            struct rpcrdma_create_data_internal *cdata)
+ {
+       struct ib_device_attr *devattr = &ia->ri_devattr;
+       int depth, delta;
+       ia->ri_max_frmr_depth =
+                       min_t(unsigned int, RPCRDMA_MAX_DATA_SEGS,
+                             devattr->max_fast_reg_page_list_len);
+       dprintk("RPC:       %s: device's max FR page list len = %u\n",
+               __func__, ia->ri_max_frmr_depth);
+       /* Add room for frmr register and invalidate WRs.
+        * 1. FRMR reg WR for head
+        * 2. FRMR invalidate WR for head
+        * 3. N FRMR reg WRs for pagelist
+        * 4. N FRMR invalidate WRs for pagelist
+        * 5. FRMR reg WR for tail
+        * 6. FRMR invalidate WR for tail
+        * 7. The RDMA_SEND WR
+        */
+       depth = 7;
+       /* Calculate N if the device max FRMR depth is smaller than
+        * RPCRDMA_MAX_DATA_SEGS.
+        */
+       if (ia->ri_max_frmr_depth < RPCRDMA_MAX_DATA_SEGS) {
+               delta = RPCRDMA_MAX_DATA_SEGS - ia->ri_max_frmr_depth;
+               do {
+                       depth += 2; /* FRMR reg + invalidate */
+                       delta -= ia->ri_max_frmr_depth;
+               } while (delta > 0);
+       }
+       ep->rep_attr.cap.max_send_wr *= depth;
+       if (ep->rep_attr.cap.max_send_wr > devattr->max_qp_wr) {
+               cdata->max_requests = devattr->max_qp_wr / depth;
+               if (!cdata->max_requests)
+                       return -EINVAL;
+               ep->rep_attr.cap.max_send_wr = cdata->max_requests *
+                                              depth;
+       }
+       return 0;
+ }
+ /* FRWR mode conveys a list of pages per chunk segment. The
+  * maximum length of that list is the FRWR page list depth.
+  */
+ static size_t
+ frwr_op_maxpages(struct rpcrdma_xprt *r_xprt)
+ {
+       struct rpcrdma_ia *ia = &r_xprt->rx_ia;
+       return min_t(unsigned int, RPCRDMA_MAX_DATA_SEGS,
+                    rpcrdma_max_segments(r_xprt) * ia->ri_max_frmr_depth);
+ }
+ /* If FAST_REG or LOCAL_INV failed, indicate the frmr needs to be reset. */
+ static void
+ frwr_sendcompletion(struct ib_wc *wc)
+ {
+       struct rpcrdma_mw *r;
+       if (likely(wc->status == IB_WC_SUCCESS))
+               return;
+       /* WARNING: Only wr_id and status are reliable at this point */
+       r = (struct rpcrdma_mw *)(unsigned long)wc->wr_id;
+       dprintk("RPC:       %s: frmr %p (stale), status %d\n",
+               __func__, r, wc->status);
+       r->r.frmr.fr_state = FRMR_IS_STALE;
+ }
+ static int
+ frwr_op_init(struct rpcrdma_xprt *r_xprt)
+ {
+       struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
+       struct ib_device *device = r_xprt->rx_ia.ri_id->device;
+       unsigned int depth = r_xprt->rx_ia.ri_max_frmr_depth;
+       struct ib_pd *pd = r_xprt->rx_ia.ri_pd;
+       int i;
+       INIT_LIST_HEAD(&buf->rb_mws);
+       INIT_LIST_HEAD(&buf->rb_all);
+       i = (buf->rb_max_requests + 1) * RPCRDMA_MAX_SEGS;
++      dprintk("RPC:       %s: initializing %d FRMRs\n", __func__, i);
+       while (i--) {
+               struct rpcrdma_mw *r;
+               int rc;
+               r = kzalloc(sizeof(*r), GFP_KERNEL);
+               if (!r)
+                       return -ENOMEM;
+               rc = __frwr_init(r, pd, device, depth);
+               if (rc) {
+                       kfree(r);
+                       return rc;
+               }
+               list_add(&r->mw_list, &buf->rb_mws);
+               list_add(&r->mw_all, &buf->rb_all);
+               r->mw_sendcompletion = frwr_sendcompletion;
+       }
+       return 0;
+ }
+ /* Post a FAST_REG Work Request to register a memory region
+  * for remote access via RDMA READ or RDMA WRITE.
+  */
+ static int
+ frwr_op_map(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr_seg *seg,
+           int nsegs, bool writing)
+ {
+       struct rpcrdma_ia *ia = &r_xprt->rx_ia;
+       struct ib_device *device = ia->ri_id->device;
+       enum dma_data_direction direction = rpcrdma_data_dir(writing);
+       struct rpcrdma_mr_seg *seg1 = seg;
+       struct rpcrdma_mw *mw = seg1->rl_mw;
+       struct rpcrdma_frmr *frmr = &mw->r.frmr;
+       struct ib_mr *mr = frmr->fr_mr;
+       struct ib_send_wr fastreg_wr, *bad_wr;
+       u8 key;
+       int len, pageoff;
+       int i, rc;
+       int seg_len;
+       u64 pa;
+       int page_no;
+       pageoff = offset_in_page(seg1->mr_offset);
+       seg1->mr_offset -= pageoff;     /* start of page */
+       seg1->mr_len += pageoff;
+       len = -pageoff;
+       if (nsegs > ia->ri_max_frmr_depth)
+               nsegs = ia->ri_max_frmr_depth;
+       for (page_no = i = 0; i < nsegs;) {
+               rpcrdma_map_one(device, seg, direction);
+               pa = seg->mr_dma;
+               for (seg_len = seg->mr_len; seg_len > 0; seg_len -= PAGE_SIZE) {
+                       frmr->fr_pgl->page_list[page_no++] = pa;
+                       pa += PAGE_SIZE;
+               }
+               len += seg->mr_len;
+               ++seg;
+               ++i;
+               /* Check for holes */
+               if ((i < nsegs && offset_in_page(seg->mr_offset)) ||
+                   offset_in_page((seg-1)->mr_offset + (seg-1)->mr_len))
+                       break;
+       }
+       dprintk("RPC:       %s: Using frmr %p to map %d segments (%d bytes)\n",
+               __func__, mw, i, len);
+       frmr->fr_state = FRMR_IS_VALID;
+       memset(&fastreg_wr, 0, sizeof(fastreg_wr));
+       fastreg_wr.wr_id = (unsigned long)(void *)mw;
+       fastreg_wr.opcode = IB_WR_FAST_REG_MR;
+       fastreg_wr.wr.fast_reg.iova_start = seg1->mr_dma + pageoff;
+       fastreg_wr.wr.fast_reg.page_list = frmr->fr_pgl;
+       fastreg_wr.wr.fast_reg.page_shift = PAGE_SHIFT;
+       fastreg_wr.wr.fast_reg.page_list_len = page_no;
+       fastreg_wr.wr.fast_reg.length = len;
+       fastreg_wr.wr.fast_reg.access_flags = writing ?
+                               IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE :
+                               IB_ACCESS_REMOTE_READ;
+       key = (u8)(mr->rkey & 0x000000FF);
+       ib_update_fast_reg_key(mr, ++key);
+       fastreg_wr.wr.fast_reg.rkey = mr->rkey;
+       DECR_CQCOUNT(&r_xprt->rx_ep);
+       rc = ib_post_send(ia->ri_id->qp, &fastreg_wr, &bad_wr);
+       if (rc)
+               goto out_senderr;
+       seg1->mr_rkey = mr->rkey;
+       seg1->mr_base = seg1->mr_dma + pageoff;
+       seg1->mr_nsegs = i;
+       seg1->mr_len = len;
+       return i;
+ out_senderr:
+       dprintk("RPC:       %s: ib_post_send status %i\n", __func__, rc);
+       ib_update_fast_reg_key(mr, --key);
+       frmr->fr_state = FRMR_IS_INVALID;
+       while (i--)
+               rpcrdma_unmap_one(device, --seg);
+       return rc;
+ }
+ /* Post a LOCAL_INV Work Request to prevent further remote access
+  * via RDMA READ or RDMA WRITE.
+  */
+ static int
+ frwr_op_unmap(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr_seg *seg)
+ {
+       struct rpcrdma_mr_seg *seg1 = seg;
+       struct rpcrdma_ia *ia = &r_xprt->rx_ia;
+       struct ib_send_wr invalidate_wr, *bad_wr;
+       int rc, nsegs = seg->mr_nsegs;
+       struct ib_device *device;
+       seg1->rl_mw->r.frmr.fr_state = FRMR_IS_INVALID;
+       memset(&invalidate_wr, 0, sizeof(invalidate_wr));
+       invalidate_wr.wr_id = (unsigned long)(void *)seg1->rl_mw;
+       invalidate_wr.opcode = IB_WR_LOCAL_INV;
+       invalidate_wr.ex.invalidate_rkey = seg1->rl_mw->r.frmr.fr_mr->rkey;
+       DECR_CQCOUNT(&r_xprt->rx_ep);
+       read_lock(&ia->ri_qplock);
+       device = ia->ri_id->device;
+       while (seg1->mr_nsegs--)
+               rpcrdma_unmap_one(device, seg++);
+       rc = ib_post_send(ia->ri_id->qp, &invalidate_wr, &bad_wr);
+       read_unlock(&ia->ri_qplock);
+       if (rc)
+               goto out_err;
+       return nsegs;
+ out_err:
+       /* Force rpcrdma_buffer_get() to retry */
+       seg1->rl_mw->r.frmr.fr_state = FRMR_IS_STALE;
+       dprintk("RPC:       %s: ib_post_send status %i\n", __func__, rc);
+       return nsegs;
+ }
+ /* After a disconnect, a flushed FAST_REG_MR can leave an FRMR in
+  * an unusable state. Find FRMRs in this state and dereg / reg
+  * each.  FRMRs that are VALID and attached to an rpcrdma_req are
+  * also torn down.
+  *
+  * This gives all in-use FRMRs a fresh rkey and leaves them INVALID.
+  *
+  * This is invoked only in the transport connect worker in order
+  * to serialize with rpcrdma_register_frmr_external().
+  */
+ static void
+ frwr_op_reset(struct rpcrdma_xprt *r_xprt)
+ {
+       struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
+       struct ib_device *device = r_xprt->rx_ia.ri_id->device;
+       unsigned int depth = r_xprt->rx_ia.ri_max_frmr_depth;
+       struct ib_pd *pd = r_xprt->rx_ia.ri_pd;
+       struct rpcrdma_mw *r;
+       int rc;
+       list_for_each_entry(r, &buf->rb_all, mw_all) {
+               if (r->r.frmr.fr_state == FRMR_IS_INVALID)
+                       continue;
+               __frwr_release(r);
+               rc = __frwr_init(r, pd, device, depth);
+               if (rc) {
+                       dprintk("RPC:       %s: mw %p left %s\n",
+                               __func__, r,
+                               (r->r.frmr.fr_state == FRMR_IS_STALE ?
+                                       "stale" : "valid"));
+                       continue;
+               }
+               r->r.frmr.fr_state = FRMR_IS_INVALID;
+       }
+ }
+ static void
+ frwr_op_destroy(struct rpcrdma_buffer *buf)
+ {
+       struct rpcrdma_mw *r;
+       while (!list_empty(&buf->rb_all)) {
+               r = list_entry(buf->rb_all.next, struct rpcrdma_mw, mw_all);
+               list_del(&r->mw_all);
+               __frwr_release(r);
+               kfree(r);
+       }
+ }
+ const struct rpcrdma_memreg_ops rpcrdma_frwr_memreg_ops = {
+       .ro_map                         = frwr_op_map,
+       .ro_unmap                       = frwr_op_unmap,
+       .ro_open                        = frwr_op_open,
+       .ro_maxpages                    = frwr_op_maxpages,
+       .ro_init                        = frwr_op_init,
+       .ro_reset                       = frwr_op_reset,
+       .ro_destroy                     = frwr_op_destroy,
+       .ro_displayname                 = "frwr",
+ };