[PATCH] xprtrdma: serialize unmap_sync with xprt_disconnect

[PATCH] xprtrdma: serialize unmap_sync with xprt_disconnect

[tmenninger]

From: Tim Menninger <tmenninger@purestorage.com>

frwr_unmap_sync() posts LOCAL_INV Work Requests to the QP and waits
for the matching CQE. This can race with rpcrdma_xprt_disconnect():

  1. frwr_unmap_sync() snapshots r_xprt->rx_ep
  2. rpcrdma_xprt_disconnect() drains the QP
  3. frwr_unmap_sync() posts LOCAL_INV WRs after the drain
  4. No CQE arrives and wait_for_completion() blocks indefinitely

The hung task appears in the rpciod workqueue:

  Workqueue: rpciod rpc_async_schedule [sunrpc]
  wait_for_completion()
  frwr_unmap_sync()
  xprt_rdma_free()
  xprt_release()

Serialize synchronous MR invalidation against QP teardown with
rx_unmap_rwsem. frwr_unmap_sync() holds the read side through
ib_post_send(), then drops it before waiting for completion so that
disconnect can acquire the write side and drain the QP.

Track admitted synchronous unmaps with rx_unmap_active. Disconnect
waits for the count to reach zero after draining the QP and before
destroying transport resources. This ensures that no synchronous
unmap is still accessing the endpoint or MR completion state when
those resources are freed.

Tested on v7.2-rc6 under pNFS load with repeated data server restarts
to provoke rapid QP disconnect/reconnect cycling. Without the fix, a
hung task in frwr_unmap_sync() reproduced twice in six attempts. With
the fix, no hang occurred across 300 cycles.

The issue has also been observed on v6.8 and v6.18.34. The race has
been present since synchronous FRWR invalidation was introduced.

Fixes: c9918ff56dfb ("xprtrdma: Add ro_unmap_sync method for FRWR")
Cc: stable@vger.kernel.org
Signed-off-by: Tim Menninger <tmenninger@purestorage.com>
---
 net/sunrpc/xprtrdma/frwr_ops.c  | 23 +++++++++++++++++++++--
 net/sunrpc/xprtrdma/transport.c |  3 +++
 net/sunrpc/xprtrdma/verbs.c     | 13 +++++++++++--
 net/sunrpc/xprtrdma/xprt_rdma.h |  5 +++++
 4 files changed, 40 insertions(+), 4 deletions(-)

diff --git a/net/sunrpc/xprtrdma/frwr_ops.c b/net/sunrpc/xprtrdma/frwr_ops.c
index e5c71cf705a3..a4733d75664f 100644
--- a/net/sunrpc/xprtrdma/frwr_ops.c
+++ b/net/sunrpc/xprtrdma/frwr_ops.c
@@ -567,11 +567,20 @@ static void frwr_wc_localinv_wake(struct ib_cq *cq, struct ib_wc *wc)
 void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 {
 	struct ib_send_wr *first, **prev, *last;
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
 	const struct ib_send_wr *bad_wr;
+	struct rpcrdma_ep *ep;
 	struct rpcrdma_mr *mr;
 	int rc;
 
+	/* serialize against rpcrdma_xprt_disconnect() */
+	down_read(&r_xprt->rx_unmap_rwsem);
+
+	ep = r_xprt->rx_ep;
+	if (!ep)
+		goto out_unlock;
+
+	atomic_inc(&r_xprt->rx_unmap_active);
+
 	/* ORDER: Invalidate all of the MRs first
 	 *
 	 * Chain the LOCAL_INV Work Requests and post them with
@@ -614,6 +623,8 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 	bad_wr = NULL;
 	rc = ib_post_send(ep->re_id->qp, first, &bad_wr);
 
+	up_read(&r_xprt->rx_unmap_rwsem);
+
 	/* The final LOCAL_INV WR in the chain is supposed to
 	 * do the wake. If it was never posted, the wake will
 	 * not happen, so don't wait in that case.
@@ -621,7 +632,7 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 	if (bad_wr != first)
 		wait_for_completion(&mr->mr_linv_done);
 	if (!rc)
-		return;
+		goto out_atomic_dec;
 
 	/* On error, the MRs get destroyed once the QP has drained. */
 	trace_xprtrdma_post_linv_err(req, rc);
@@ -629,6 +640,14 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 	/* Force a connection loss to ensure complete recovery.
 	 */
 	rpcrdma_force_disconnect(ep);
+
+out_atomic_dec:
+	if (atomic_dec_and_test(&r_xprt->rx_unmap_active))
+		wake_up_var(&r_xprt->rx_unmap_active);
+	return;
+
+out_unlock:
+	up_read(&r_xprt->rx_unmap_rwsem);
 }
 
 /**
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index d4e6746d8ecd..e2b6287511a4 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -364,6 +364,9 @@ xprt_setup_rdma(struct xprt_create *args)
 	INIT_DELAYED_WORK(&new_xprt->rx_connect_worker,
 			  xprt_rdma_connect_worker);
 
+	init_rwsem(&new_xprt->rx_unmap_rwsem);
+	atomic_set(&new_xprt->rx_unmap_active, 0);
+
 	xprt->max_payload = RPCRDMA_MAX_DATA_SEGS << PAGE_SHIFT;
 
 	return xprt;
diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 04b286223b24..ae8d06d1cf29 100644
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -580,18 +580,24 @@ int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt)
  */
 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 {
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
+	struct rpcrdma_ep *ep;
 	struct rdma_cm_id *id;
 	int rc;
 
+	down_write(&r_xprt->rx_unmap_rwsem);
+
+	ep = r_xprt->rx_ep;
 	if (!ep)
-		return;
+		goto out_unlock;
 
 	id = ep->re_id;
 	rc = rdma_disconnect(id);
 	trace_xprtrdma_disconnect(r_xprt, rc);
 
 	rpcrdma_xprt_drain(r_xprt);
+	wait_var_event(&r_xprt->rx_unmap_active,
+		       !atomic_read(&r_xprt->rx_unmap_active));
+
 	rpcrdma_reps_unmap(r_xprt);
 	rpcrdma_sendctxs_destroy(r_xprt);
 	rpcrdma_reqs_reset(r_xprt);
@@ -601,6 +607,9 @@ void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 		rdma_destroy_id(id);
 
 	r_xprt->rx_ep = NULL;
+
+out_unlock:
+	up_write(&r_xprt->rx_unmap_rwsem);
 }
 
 /* Fixed-size circular FIFO queue. This implementation is wait-free and
diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
index 4cbc941e4a3e..419d5952150a 100644
--- a/net/sunrpc/xprtrdma/xprt_rdma.h
+++ b/net/sunrpc/xprtrdma/xprt_rdma.h
@@ -46,6 +46,7 @@
 #include <linux/spinlock.h> 		/* spinlock_t, etc */
 #include <linux/atomic.h>		/* atomic_t, etc */
 #include <linux/kref.h>			/* struct kref */
+#include <linux/rwsem.h>		/* struct rw_semaphore */
 #include <linux/workqueue.h>		/* struct work_struct */
 #include <linux/llist.h>
 
@@ -449,6 +450,10 @@ struct rpcrdma_xprt {
 	struct delayed_work	rx_connect_worker;
 	struct rpc_timeout	rx_timeout;
 	struct rpcrdma_stats	rx_stats;
+
+	/* for serializing xprt disconnect and sync MR unmap */
+	struct rw_semaphore	rx_unmap_rwsem;
+	atomic_t                rx_unmap_active;
 };
 
 #define rpcx_to_rdmax(x) container_of(x, struct rpcrdma_xprt, rx_xprt)
-- 
2.34.1

Re: [PATCH] xprtrdma: serialize unmap_sync with xprt_disconnect

[Chuck Lever]

Hi Tim -

Reviewer hat on.

I went over this patch with several LLM reviewers and my own
recollection of the code. The combined findings are below.

Summary: I agree there is a bug. The proposed fix introduces
technical debt that I would rather not see in-tree. Importantly,
there are areas where you can reuse existing code instead of
open-coding, the Fixes: tag might be wrong, and a few code
comments need updating as part of the fix.


On Tue, Aug 18, 2026, at 8:28 AM, tmenninger@purestorage.com wrote:
>     xprtrdma: serialize unmap_sync with xprt_disconnect
>
>     frwr_unmap_sync() posts LOCAL_INV Work Requests to the QP and waits
>     for the matching CQE. This can race with rpcrdma_xprt_disconnect():
>
>       1. frwr_unmap_sync() snapshots r_xprt->rx_ep
>       2. rpcrdma_xprt_disconnect() drains the QP
>       3. frwr_unmap_sync() posts LOCAL_INV WRs after the drain
>       4. No CQE arrives and wait_for_completion() blocks indefinitely
>
>     The hung task appears in the rpciod workqueue:
>
>       Workqueue: rpciod rpc_async_schedule [sunrpc]
>       wait_for_completion()
>       frwr_unmap_sync()
>       xprt_rdma_free()
>       xprt_release()
>
>     Serialize synchronous MR invalidation against QP teardown with
>     rx_unmap_rwsem. frwr_unmap_sync() holds the read side through
>     ib_post_send(), then drops it before waiting for completion so that
>     disconnect can acquire the write side and drain the QP.
>
>     Track admitted synchronous unmaps with rx_unmap_active. Disconnect
>     waits for the count to reach zero after draining the QP and before
>     destroying transport resources. This ensures that no synchronous
>     unmap is still accessing the endpoint or MR completion state when
>     those resources are freed.
>
>     Tested on v7.2-rc6 under pNFS load with repeated data server restarts
>     to provoke rapid QP disconnect/reconnect cycling. Without the fix, a
>     hung task in frwr_unmap_sync() reproduced twice in six attempts. With
>     the fix, no hang occurred across 300 cycles.
>
>     The issue has also been observed on v6.8 and v6.18.34. The race has
>     been present since synchronous FRWR invalidation was introduced.
>
>     Fixes: c9918ff56dfb ("xprtrdma: Add ro_unmap_sync method for FRWR")
>     Signed-off-by: Tim Menninger <tmenninger@purestorage.com>
>

Nice find, and the read side around the rx_ep sample and ib_post_send()
does close the window.

On the tag: c9918ff56dfb landed in v4.5.  There rx_ep was an embedded
struct rpcrdma_ep rather than a pointer, frwr_op_unmap_sync() posted to
ia->ri_id->qp, and rpcrdma_ep_disconnect() called rpcrdma_flush_cqs()
without draining the QP.  ib_drain_qp() reached that path in v4.7 with
550d7502cf66 ("xprtrdma: Use core ib_drain_qp() API"), and the sync unmap
moved to xprt_rdma_free() in v5.3 with 0ab115237025 ("xprtrdma: Wake RPCs
directly in rpcrdma_wc_send path").

The race as described needs a freeable rx_ep and the "r_xprt->rx_ep =
NULL" store, and both arrive in v5.7.  Should the tag be:

  Fixes: e28ce90083f0 ("xprtrdma: kmalloc rpcrdma_ep separate from rpcrdma_xprt")

That would also make this sentence inaccurate:

>     The race has been present since synchronous FRWR invalidation was
>     introduced.

The changelog does not say why only the sync path needs this.
xprt_rdma_free() runs from xprt_release() after ->release_xprt has
dropped the transport send lock, so it is the one LOCAL_INV poster that
disconnect is not serialized against.  frwr_unmap_async() runs inline in
rpcrdma_wc_receive()->rpcrdma_reply_handler(), which ib_drain_rq() orders
ahead of ib_drain_sq().  Worth a sentence?

> diff --git a/net/sunrpc/xprtrdma/frwr_ops.c b/net/sunrpc/xprtrdma/frwr_ops.c
> index e5c71cf705a3..a4733d75664f 100644
> --- a/net/sunrpc/xprtrdma/frwr_ops.c
> +++ b/net/sunrpc/xprtrdma/frwr_ops.c
> @@ -567,11 +567,20 @@ static void frwr_wc_localinv_wake(struct ib_cq *cq, struct ib_wc *wc)
>  void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
>  {
>  	struct ib_send_wr *first, **prev, *last;
> -	struct rpcrdma_ep *ep = r_xprt->rx_ep;
>  	const struct ib_send_wr *bad_wr;
> +	struct rpcrdma_ep *ep;
>  	struct rpcrdma_mr *mr;
>  	int rc;
>  
> +	/* serialize against rpcrdma_xprt_disconnect() */
> +	down_read(&r_xprt->rx_unmap_rwsem);
> +
> +	ep = r_xprt->rx_ep;
> +	if (!ep)
> +		goto out_unlock;

The kernel-doc above frwr_unmap_sync() says it "Sleeps until it is safe
for the host CPU to access the previously mapped memory regions."  With
this early return that is now conditional.  Would it help to note there
that rpcrdma_reqs_reset() has already released this req's MRs whenever
rx_ep is NULL?  The correctness of returning here depends on that, and it
happens in another file.

> +
> +	atomic_inc(&r_xprt->rx_unmap_active);
> +
>  	/* ORDER: Invalidate all of the MRs first
>  	 *
>  	 * Chain the LOCAL_INV Work Requests and post them with
> @@ -614,6 +623,8 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
>  	bad_wr = NULL;
>  	rc = ib_post_send(ep->re_id->qp, first, &bad_wr);

The comment just above this hunk still reads:

	/* Transport disconnect drains the receive CQ before it
	 * replaces the QP. The RPC reply handler won't call us
	 * unless re_id->qp is a valid pointer.
	 */

Is that still the rationale here?  frwr_unmap_sync()'s only caller is
xprt_rdma_free(), not the reply handler, and the new down_read() is there
because the receive CQ drain does not cover this path.  The same comment
sits above frwr_unmap_async(), where it does still hold.

>  
> +	up_read(&r_xprt->rx_unmap_rwsem);
> +
>  	/* The final LOCAL_INV WR in the chain is supposed to
>  	 * do the wake. If it was never posted, the wake will
>  	 * not happen, so don't wait in that case.
> @@ -621,7 +632,7 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
>  	if (bad_wr != first)
>  		wait_for_completion(&mr->mr_linv_done);
>  	if (!rc)
> -		return;
> +		goto out_atomic_dec;
>  
>  	/* On error, the MRs get destroyed once the QP has drained. */
>  	trace_xprtrdma_post_linv_err(req, rc);
> @@ -629,6 +640,14 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
>  	/* Force a connection loss to ensure complete recovery.
>  	 */
>  	rpcrdma_force_disconnect(ep);
> +
> +out_atomic_dec:
> +	if (atomic_dec_and_test(&r_xprt->rx_unmap_active))
> +		wake_up_var(&r_xprt->rx_unmap_active);
> +	return;
> +
> +out_unlock:
> +	up_read(&r_xprt->rx_unmap_rwsem);
>  }
>  
>  /**

Could rx_unmap_active come out entirely?

It is needed because the waiter sleeps on mr->mr_linv_done, and
rpcrdma_mrs_destroy() frwr_mr_release()s that mr once the count reaches
zero.  Would moving that completion into struct rpcrdma_req decouple the
two?  A req is freed only by rpcrdma_req_destroy() from
rpcrdma_buffer_destroy(), and xprt_rdma_free() is running on this one, so
it cannot be freed under the wait.  rpcrdma_req_reset() leaves it alone
as well.

frwr_wc_localinv_wake() can reach it without a new field.  mr->mr_req is
already there, and frwr_wc_localinv_done() dereferences it the same way
for rl_reply.

The other use is keeping ep alive for the rpcrdma_force_disconnect(ep)
above, after up_read().  Would rpcrdma_ep_get() and rpcrdma_ep_put()
across the read section cover that one?

Together those would drop the atomic_t, the wait_var_event() in
rpcrdma_xprt_disconnect(), and the atomic_set() in xprt_setup_rdma(),
and leave the rwsem with one job.

This isn't a bug, but if the counter stays, is this
atomic_dec_and_wake_up() from linux/wait_bit.h?  nfsd41_cb_inflight_end()
uses it for the same inflight-counter pattern.

> diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
> index d4e6746d8ecd..e2b6287511a4 100644
> --- a/net/sunrpc/xprtrdma/transport.c
> +++ b/net/sunrpc/xprtrdma/transport.c
> @@ -364,6 +364,9 @@ xprt_setup_rdma(struct xprt_create *args)
>  	INIT_DELAYED_WORK(&new_xprt->rx_connect_worker,
>  			  xprt_rdma_connect_worker);
>  
> +	init_rwsem(&new_xprt->rx_unmap_rwsem);
> +	atomic_set(&new_xprt->rx_unmap_active, 0);
> +
>  	xprt->max_payload = RPCRDMA_MAX_DATA_SEGS << PAGE_SHIFT;
>  
>  	return xprt;

This isn't a bug, but is the atomic_set() needed?  xprt_alloc() kzallocs
the whole struct rpcrdma_xprt.

> diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
> index 04b286223b24..ae8d06d1cf29 100644
> --- a/net/sunrpc/xprtrdma/verbs.c
> +++ b/net/sunrpc/xprtrdma/verbs.c
> @@ -580,18 +580,24 @@ int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt)
>   */
>  void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
>  {
> -	struct rpcrdma_ep *ep = r_xprt->rx_ep;
> +	struct rpcrdma_ep *ep;
>  	struct rdma_cm_id *id;
>  	int rc;
>  
> +	down_write(&r_xprt->rx_unmap_rwsem);
> +
> +	ep = r_xprt->rx_ep;
>  	if (!ep)
> -		return;
> +		goto out_unlock;
>  
>  	id = ep->re_id;
>  	rc = rdma_disconnect(id);
>  	trace_xprtrdma_disconnect(r_xprt, rc);
>  
>  	rpcrdma_xprt_drain(r_xprt);
> +	wait_var_event(&r_xprt->rx_unmap_active,
> +		       !atomic_read(&r_xprt->rx_unmap_active));
> +
>  	rpcrdma_reps_unmap(r_xprt);
>  	rpcrdma_sendctxs_destroy(r_xprt);
>  	rpcrdma_reqs_reset(r_xprt);
> @@ -601,6 +607,9 @@ void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
>  		rdma_destroy_id(id);
>  
>  	r_xprt->rx_ep = NULL;
> +
> +out_unlock:
> +	up_write(&r_xprt->rx_unmap_rwsem);
>  }
>  
>  /* Fixed-size circular FIFO queue. This implementation is wait-free and

This wait is the half of rx_unmap_active that a req-resident completion
would remove, leaving drain-then-teardown as before.

> diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
> index 4cbc941e4a3e..419d5952150a 100644
> --- a/net/sunrpc/xprtrdma/xprt_rdma.h
> +++ b/net/sunrpc/xprtrdma/xprt_rdma.h

[ ... ]

> @@ -449,6 +450,10 @@ struct rpcrdma_xprt {
>  	struct delayed_work	rx_connect_worker;
>  	struct rpc_timeout	rx_timeout;
>  	struct rpcrdma_stats	rx_stats;
> +
> +	/* for serializing xprt disconnect and sync MR unmap */
> +	struct rw_semaphore	rx_unmap_rwsem;
> +	atomic_t                rx_unmap_active;
>  };
>  
>  #define rpcx_to_rdmax(x) container_of(x, struct rpcrdma_xprt, rx_xprt)

Have you considered putting these on struct rpcrdma_ep instead?
rpcrdma_xprt_drain() already fences the receive side against the drain
with ep->re_receiving and ep->re_done, and an rwsem on the transport
outlives every ep it guards.

Would the comment be more useful naming the rule rather than the two
functions?  Something like: read side held across sampling rx_ep and
posting to its QP.

This isn't a bug, but rx_unmap_active is space-aligned where the rest of
the struct uses a tab.

-- 
Chuck Lever

[PATCH v2] xprtrdma: serialize unmap_sync with xprt_disconnect

[Tim Menninger]

frwr_unmap_sync() posts LOCAL_INV Work Requests to the QP and waits
for the matching CQE. This can race with rpcrdma_xprt_disconnect():

    1. frwr_unmap_sync() samples r_xprt->rx_ep
    2. rpcrdma_xprt_disconnect() drains the QP
    3. frwr_unmap_sync() posts LOCAL_INV WRs after the drain
    4. No CQE arrives, so wait_for_completion() blocks indefinitely

The hung task appears in the rpciod workqueue:

Workqueue: rpciod rpc_async_schedule [sunrpc]
    wait_for_completion()
    frwr_unmap_sync()
    xprt_rdma_free()
    xprt_release()

Serialize synchronous MR invalidation against QP teardown with
rx_unmap_rwsem. frwr_unmap_sync() holds the read side while it samples
rx_ep, builds the LOCAL_INV chain, and submits it. Disconnect holds the
write side across QP drain and resource teardown. Drop the read side
before waiting so disconnect can drain the submitted Work Requests and
deliver their CQEs.

Disconnect can reset a request while frwr_unmap_sync() waits for the
read side. Recheck rl_registered after acquiring the rwsem and return
without posting if the request has already been reset.

Move the synchronous invalidation completion from the final MR to the
request, whose lifetime covers xprt_rdma_free(), and hold a reference
to the sampled endpoint through the completion wait and error path.
This allows disconnect to destroy MRs after draining the QP and drop
its endpoint reference without a separate active-unmap counter.

frwr_unmap_async() runs inline from Receive completion processing.
rpcrdma_xprt_drain() drains the Receive Queue before the Send Queue, so
asynchronous LOCAL_INV Work Requests are submitted before Send Queue
drain. frwr_unmap_sync(), however, runs from xprt_rdma_free() after the
transport send lock has been released, so it is not ordered against
disconnect.

Tested on v7.2-rc6 under pNFS load with repeated data server restarts
to provoke rapid QP disconnect/reconnect cycling. Without the fix, a
hung task in frwr_unmap_sync() reproduced twice in six attempts. With
the fix, no hang occurred across 200 cycles.

Also observed on v6.8 and v6.18.34.

Fixes: e28ce90083f0 ("xprtrdma: kmalloc rpcrdma_ep separate from rpcrdma_xprt")
Cc: stable@vger.kernel.org
Assisted-by: Claude:opus-4-7
Signed-off-by: Tim Menninger <tmenninger@everpuredata.com>
---
Changes since v1:

- Change the Fixes tag to e28ce90083f0, which made rx_ep a separately
  allocated, kref-managed pointer.
- Explain why Receive-before-Send Queue drain orders
  frwr_unmap_async(), but does not order frwr_unmap_sync().
- Move the synchronous LOCAL_INV completion from struct rpcrdma_mr to
  struct rpcrdma_req.
- Hold a reference to the sampled endpoint through the completion wait
  and error path.
- Remove rx_unmap_active and its disconnect-side wait.
- Recheck rl_registered after acquiring the rwsem in case disconnect
  reset the request while the reader was blocked.
- Update the transport-recovery comments, LOCAL_INV wake kernel-doc,
  frwr_unmap_sync() kernel-doc, and the QP-posting locking comment.
- Keep rx_unmap_rwsem in struct rpcrdma_xprt; an endpoint-resident lock
  cannot protect the initial rx_ep lookup.

v1: https://lore.kernel.org/all/20260818122825.2347594-1-tmenninger@purestorage.com/

 net/sunrpc/xprtrdma/frwr_ops.c  | 98 ++++++++++++++++++++++-----------
 net/sunrpc/xprtrdma/transport.c |  2 +
 net/sunrpc/xprtrdma/verbs.c     | 18 ++++--
 net/sunrpc/xprtrdma/xprt_rdma.h | 11 +++-
 4 files changed, 89 insertions(+), 40 deletions(-)

diff --git a/net/sunrpc/xprtrdma/frwr_ops.c b/net/sunrpc/xprtrdma/frwr_ops.c
index e5c71cf705a3..77aff9c0533a 100644
--- a/net/sunrpc/xprtrdma/frwr_ops.c
+++ b/net/sunrpc/xprtrdma/frwr_ops.c
@@ -27,13 +27,17 @@
 
 /* Transport recovery
  *
- * frwr_map and frwr_unmap_* cannot run at the same time the transport
- * connect worker is running. The connect worker holds the transport
- * send lock, just as ->send_request does. This prevents frwr_map and
- * the connect worker from running concurrently. When a connection is
- * closed, the Receive completion queue is drained before the allowing
- * the connect worker to get control. This prevents frwr_unmap and the
- * connect worker from running concurrently.
+ * frwr_map() is serialized with the connect worker by the transport
+ * send lock.
+ *
+ * frwr_unmap_async() runs from Receive completion processing.
+ * rpcrdma_xprt_drain() drains the Receive Queue before the Send Queue,
+ * so asynchronous LOCAL_INV Work Requests are submitted before Send
+ * Queue drain.
+ *
+ * frwr_unmap_sync() runs after the transport send lock is released.
+ * rx_unmap_rwsem orders its LOCAL_INV submission before disconnect
+ * drains the QP.
  *
  * When the underlying transport disconnects, MRs that are in flight
  * are flushed and are likely unusable. Thus all MRs are destroyed.
@@ -139,7 +143,6 @@ int frwr_mr_init(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr *mr)
 	mr->mr_ibmr = frmr;
 	mr->mr_device = NULL;
 	INIT_LIST_HEAD(&mr->mr_list);
-	init_completion(&mr->mr_linv_done);
 	frwr_cid_init(ep, mr);
 
 	sg_init_table(sg, depth);
@@ -538,40 +541,62 @@ static void frwr_wc_localinv(struct ib_cq *cq, struct ib_wc *wc)
  * @cq: completion queue
  * @wc: WCE for a completed LocalInv WR
  *
- * Awaken anyone waiting for an MR to finish being fenced.
+ * Wake the request waiting for the final LOCAL_INV completion.
  */
 static void frwr_wc_localinv_wake(struct ib_cq *cq, struct ib_wc *wc)
 {
 	struct ib_cqe *cqe = wc->wr_cqe;
 	struct rpcrdma_mr *mr = container_of(cqe, struct rpcrdma_mr, mr_cqe);
+	struct rpcrdma_req *req = mr->mr_req;
 
 	/* WARNING: Only wr_cqe and status are reliable at this point */
 	trace_xprtrdma_wc_li_wake(wc, &mr->mr_cid);
 	frwr_mr_done(wc, mr);
-	complete(&mr->mr_linv_done);
+	complete(&req->rl_linv_done);
 
 	rpcrdma_flush_disconnect(cq->cq_context, wc);
 }
 
 /**
- * frwr_unmap_sync - invalidate memory regions that were registered for @req
+ * frwr_unmap_sync - synchronously invalidate MRs registered for @req
  * @r_xprt: controlling transport instance
- * @req: rpcrdma_req with a non-empty list of MRs to process
+ * @req: request whose registered MRs are to be invalidated
  *
- * Sleeps until it is safe for the host CPU to access the previously mapped
- * memory regions. This guarantees that registered MRs are properly fenced
- * from the server before the RPC consumer accesses the data in them. It
- * also ensures proper Send flow control: waking the next RPC waits until
- * this RPC has relinquished all its Send Queue entries.
+ * If @req still owns registered MRs after synchronizing with transport
+ * disconnect, post a chain of LOCAL_INV Work Requests and wait for the
+ * final completion. This fences the mapped regions from remote access
+ * and preserves Send Queue flow control.
+ *
+ * A concurrent disconnect can reset @req while this function waits for
+ * the read side. In that case the MRs have already been released and no
+ * LOCAL_INV Work Requests are posted.
  */
 void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 {
 	struct ib_send_wr *first, **prev, *last;
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
 	const struct ib_send_wr *bad_wr;
+	struct rpcrdma_ep *ep;
+	struct rdma_cm_id *id;
 	struct rpcrdma_mr *mr;
 	int rc;
 
+	/* serialize against rpcrdma_xprt_disconnect() */
+	down_read(&r_xprt->rx_unmap_rwsem);
+
+	/*
+	 * Disconnect can reset this request while we wait for the read side.
+	 * Recheck rl_registered in case rx_ep now refers to a new connection.
+	 */
+	ep = r_xprt->rx_ep;
+	if (!ep || list_empty(&req->rl_registered))
+		goto out_unlock;
+
+	/* Hold an endpoint reference so ep and its CM ID remain valid
+	 * after we drop the rwsem and while we wait for completion.
+	 */
+	rpcrdma_ep_get(ep);
+	id = ep->re_id;
+
 	/* ORDER: Invalidate all of the MRs first
 	 *
 	 * Chain the LOCAL_INV Work Requests and post them with
@@ -598,37 +623,44 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 		prev = &last->next;
 	} while ((mr = rpcrdma_mr_pop(&req->rl_registered)));
 
-	mr = container_of(last, struct rpcrdma_mr, mr_invwr);
-
 	/* Strong send queue ordering guarantees that when the
 	 * last WR in the chain completes, all WRs in the chain
 	 * are complete.
 	 */
 	last->wr_cqe->done = frwr_wc_localinv_wake;
-	reinit_completion(&mr->mr_linv_done);
+	reinit_completion(&req->rl_linv_done);
 
-	/* Transport disconnect drains the receive CQ before it
-	 * replaces the QP. The RPC reply handler won't call us
-	 * unless re_id->qp is a valid pointer.
+	/*
+	 * The read side prevents disconnect from draining this QP until the
+	 * LOCAL_INV chain has been submitted.
 	 */
 	bad_wr = NULL;
-	rc = ib_post_send(ep->re_id->qp, first, &bad_wr);
+	rc = ib_post_send(id->qp, first, &bad_wr);
+
+	up_read(&r_xprt->rx_unmap_rwsem);
 
 	/* The final LOCAL_INV WR in the chain is supposed to
 	 * do the wake. If it was never posted, the wake will
 	 * not happen, so don't wait in that case.
 	 */
 	if (bad_wr != first)
-		wait_for_completion(&mr->mr_linv_done);
-	if (!rc)
-		return;
+		wait_for_completion(&req->rl_linv_done);
 
-	/* On error, the MRs get destroyed once the QP has drained. */
-	trace_xprtrdma_post_linv_err(req, rc);
+	if (rc) {
+		/* On error, the MRs get destroyed once the QP has drained. */
+		trace_xprtrdma_post_linv_err(req, rc);
 
-	/* Force a connection loss to ensure complete recovery.
-	 */
-	rpcrdma_force_disconnect(ep);
+		/* Force a connection loss to ensure complete recovery.
+		 */
+		rpcrdma_force_disconnect(ep);
+	}
+
+	if (rpcrdma_ep_put(ep))
+		rdma_destroy_id(id);
+	return;
+
+out_unlock:
+	up_read(&r_xprt->rx_unmap_rwsem);
 }
 
 /**
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index d4e6746d8ecd..e38637c2ec78 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -364,6 +364,8 @@ xprt_setup_rdma(struct xprt_create *args)
 	INIT_DELAYED_WORK(&new_xprt->rx_connect_worker,
 			  xprt_rdma_connect_worker);
 
+	init_rwsem(&new_xprt->rx_unmap_rwsem);
+
 	xprt->max_payload = RPCRDMA_MAX_DATA_SEGS << PAGE_SHIFT;
 
 	return xprt;
diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 04b286223b24..193a90c97f7a 100644
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -74,8 +74,6 @@ static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_reps_unmap(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt);
-static void rpcrdma_ep_get(struct rpcrdma_ep *ep);
-static int rpcrdma_ep_put(struct rpcrdma_ep *ep);
 static struct rpcrdma_regbuf *
 rpcrdma_regbuf_alloc_node(size_t size, enum dma_data_direction direction,
 			  int node);
@@ -374,7 +372,7 @@ static void rpcrdma_ep_destroy(struct kref *kref)
 	module_put(THIS_MODULE);
 }
 
-static noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
+noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
 {
 	kref_get(&ep->re_kref);
 }
@@ -383,7 +381,7 @@ static noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
  *     %0 if @ep still has a positive kref count, or
  *     %1 if @ep was destroyed successfully.
  */
-static noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep)
+noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep)
 {
 	return kref_put(&ep->re_kref, rpcrdma_ep_destroy);
 }
@@ -580,18 +578,22 @@ int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt)
  */
 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 {
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
+	struct rpcrdma_ep *ep;
 	struct rdma_cm_id *id;
 	int rc;
 
+	down_write(&r_xprt->rx_unmap_rwsem);
+
+	ep = r_xprt->rx_ep;
 	if (!ep)
-		return;
+		goto out_unlock;
 
 	id = ep->re_id;
 	rc = rdma_disconnect(id);
 	trace_xprtrdma_disconnect(r_xprt, rc);
 
 	rpcrdma_xprt_drain(r_xprt);
+
 	rpcrdma_reps_unmap(r_xprt);
 	rpcrdma_sendctxs_destroy(r_xprt);
 	rpcrdma_reqs_reset(r_xprt);
@@ -601,6 +603,9 @@ void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 		rdma_destroy_id(id);
 
 	r_xprt->rx_ep = NULL;
+
+out_unlock:
+	up_write(&r_xprt->rx_unmap_rwsem);
 }
 
 /* Fixed-size circular FIFO queue. This implementation is wait-free and
@@ -924,6 +929,7 @@ struct rpcrdma_req *rpcrdma_req_create(struct rpcrdma_xprt *r_xprt,
 
 	INIT_LIST_HEAD(&req->rl_free_mrs);
 	INIT_LIST_HEAD(&req->rl_registered);
+	init_completion(&req->rl_linv_done);
 	spin_lock(&buffer->rb_lock);
 	list_add(&req->rl_all, &buffer->rb_allreqs);
 	spin_unlock(&buffer->rb_lock);
diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
index 4cbc941e4a3e..e50b0af5adf8 100644
--- a/net/sunrpc/xprtrdma/xprt_rdma.h
+++ b/net/sunrpc/xprtrdma/xprt_rdma.h
@@ -46,6 +46,7 @@
 #include <linux/spinlock.h> 		/* spinlock_t, etc */
 #include <linux/atomic.h>		/* atomic_t, etc */
 #include <linux/kref.h>			/* struct kref */
+#include <linux/rwsem.h>		/* struct rw_semaphore */
 #include <linux/workqueue.h>		/* struct work_struct */
 #include <linux/llist.h>
 
@@ -249,7 +250,6 @@ struct rpcrdma_mr {
 	int			mr_nents;
 	enum dma_data_direction	mr_dir;
 	struct ib_cqe		mr_cqe;
-	struct completion	mr_linv_done;
 	union {
 		struct ib_reg_wr	mr_regwr;
 		struct ib_send_wr	mr_invwr;
@@ -348,6 +348,7 @@ struct rpcrdma_req {
 
 	struct list_head	rl_free_mrs;
 	struct list_head	rl_registered;
+	struct completion	rl_linv_done;
 };
 
 static inline struct rpcrdma_req *
@@ -449,6 +450,12 @@ struct rpcrdma_xprt {
 	struct delayed_work	rx_connect_worker;
 	struct rpc_timeout	rx_timeout;
 	struct rpcrdma_stats	rx_stats;
+
+	/*
+	 * Read side protects sampling rx_ep and posting synchronous LOCAL_INV
+	 * WRs; disconnect holds the write side across QP drain and teardown.
+	 */
+	struct rw_semaphore	rx_unmap_rwsem;
 };
 
 #define rpcx_to_rdmax(x) container_of(x, struct rpcrdma_xprt, rx_xprt)
@@ -479,6 +486,8 @@ extern unsigned int xprt_rdma_memreg_strategy;
  * Endpoint calls - xprtrdma/verbs.c
  */
 void rpcrdma_force_disconnect(struct rpcrdma_ep *ep);
+void rpcrdma_ep_get(struct rpcrdma_ep *ep);
+int rpcrdma_ep_put(struct rpcrdma_ep *ep);
 void rpcrdma_flush_disconnect(struct rpcrdma_xprt *r_xprt, struct ib_wc *wc);
 int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt);
 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt);
-- 
2.34.1

Re: [PATCH v2] xprtrdma: serialize unmap_sync with xprt_disconnect

[Chuck Lever]

Hey Tim -

I didn't find any correctness issues in this one.

Nits: A few comments were made stale, and a layering change.


On Thu, Aug 20, 2026, at 5:17 PM, Tim Menninger wrote:
> diff --git a/net/sunrpc/xprtrdma/frwr_ops.c b/net/sunrpc/xprtrdma/frwr_ops.c
> index e5c71cf705a3..77aff9c0533a 100644
> --- a/net/sunrpc/xprtrdma/frwr_ops.c
> +++ b/net/sunrpc/xprtrdma/frwr_ops.c

[ ... ]

> @@ -538,40 +541,62 @@ static void frwr_wc_localinv(struct ib_cq *cq, struct ib_wc *wc)

[ ... ]

>  /**
> - * frwr_unmap_sync - invalidate memory regions that were registered for @req
> + * frwr_unmap_sync - synchronously invalidate MRs registered for @req
>   * @r_xprt: controlling transport instance
> - * @req: rpcrdma_req with a non-empty list of MRs to process
> + * @req: request whose registered MRs are to be invalidated
>   *
> - * Sleeps until it is safe for the host CPU to access the previously mapped
> - * memory regions. This guarantees that registered MRs are properly fenced
> - * from the server before the RPC consumer accesses the data in them. It
> - * also ensures proper Send flow control: waking the next RPC waits until
> - * this RPC has relinquished all its Send Queue entries.
> + * If @req still owns registered MRs after synchronizing with transport
> + * disconnect, post a chain of LOCAL_INV Work Requests and wait for the
> + * final completion. This fences the mapped regions from remote access
> + * and preserves Send Queue flow control.
> + *
> + * A concurrent disconnect can reset @req while this function waits for
> + * the read side. In that case the MRs have already been released and no
> + * LOCAL_INV Work Requests are posted.
>   */

The rewritten block drops the sleeping behavior the old text carried:

  * Sleeps until it is safe for the host CPU to access the previously mapped
  * memory regions.

frwr_unmap_sync() now also takes and releases a transport-wide rwsem.
Both facts belong in a Context: section.

	 * Context: Process context. Takes and releases
	 *	    @r_xprt->rx_unmap_rwsem for read. May sleep.

[ ... ]

> @@ -598,37 +623,44 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)

[ ... ]

>  	if (bad_wr != first)
> -		wait_for_completion(&mr->mr_linv_done);
> -	if (!rc)
> -		return;
> +		wait_for_completion(&req->rl_linv_done);
>
> -	/* On error, the MRs get destroyed once the QP has drained. */
> -	trace_xprtrdma_post_linv_err(req, rc);
> +	if (rc) {
> +		/* On error, the MRs get destroyed once the QP has drained. */
> +		trace_xprtrdma_post_linv_err(req, rc);
>
> -	/* Force a connection loss to ensure complete recovery.
> -	 */
> -	rpcrdma_force_disconnect(ep);
> +		/* Force a connection loss to ensure complete recovery.
> +		 */
> +		rpcrdma_force_disconnect(ep);
> +	}
> +
> +	if (rpcrdma_ep_put(ep))
> +		rdma_destroy_id(id);

This is not a defect, but it makes frwr_ops.c the only file outside
verbs.c that destroys a CM ID.  The other three teardowns are all in
verbs.c: rpcrdma_create_id() on its error path, rpcrdma_ep_create() on
its error path, and rpcrdma_xprt_disconnect().

The rule that a nonzero return obliges the caller to destroy the CM ID
lives in a comment above rpcrdma_ep_put():

net/sunrpc/xprtrdma/verbs.c {
	/*
	 *     %0 if @ep still has a positive kref count, or
	 *     %1 if @ep was destroyed successfully.
	 */
	noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep)
	{
		return kref_put(&ep->re_kref, rpcrdma_ep_destroy);
	}
}

A small helper in verbs.c would carry that rule instead of open coding
the idiom in a second file.

	void rpcrdma_ep_release(struct rpcrdma_ep *ep)
	{
		struct rdma_cm_id *id = ep->re_id;

		if (rpcrdma_ep_put(ep))
			rdma_destroy_id(id);
	}

That exports rpcrdma_ep_get() and rpcrdma_ep_release() rather than the
get/put pair.  The local id then goes away, because ib_post_send() can
read ep->re_id->qp directly while the read side is still held.

[ ... ]

> diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
> index 04b286223b24..193a90c97f7a 100644
> --- a/net/sunrpc/xprtrdma/verbs.c
> +++ b/net/sunrpc/xprtrdma/verbs.c

[ ... ]

> @@ -580,18 +578,22 @@ int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt)
>   */
>  void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
>  {
> -	struct rpcrdma_ep *ep = r_xprt->rx_ep;
> +	struct rpcrdma_ep *ep;
>  	struct rdma_cm_id *id;
>  	int rc;
>
> +	down_write(&r_xprt->rx_unmap_rwsem);
> +
> +	ep = r_xprt->rx_ep;

rpcrdma_xprt_disconnect() now holds rx_unmap_rwsem for write across the
QP drain and all of the resource teardown.  Its kernel-doc still records
only the caller-side rule:

  * Caller serializes. Either the transport send lock is held,
  * or we're being called to destroy the transport.

The rwsem belongs there as well.

The same block goes on to say:

  * On return, @r_xprt is completely divested of all hardware
  * resources and prepared for the next ->connect operation.

The first half of that no longer holds.  A concurrent frwr_unmap_sync()
holds an endpoint reference across its completion wait, so
rpcrdma_ep_put() below returns 0.  The QP, both completion queues, the
PD and the CM ID then outlive this return.

[ ... ]

> diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
> index 4cbc941e4a3e..e50b0af5adf8 100644
> --- a/net/sunrpc/xprtrdma/xprt_rdma.h
> +++ b/net/sunrpc/xprtrdma/xprt_rdma.h

[ ... ]

> @@ -449,6 +450,12 @@ struct rpcrdma_xprt {
>  	struct delayed_work	rx_connect_worker;
>  	struct rpc_timeout	rx_timeout;
>  	struct rpcrdma_stats	rx_stats;
> +
> +	/*
> +	 * Read side protects sampling rx_ep and posting synchronous LOCAL_INV
> +	 * WRs; disconnect holds the write side across QP drain and teardown.
> +	 */
> +	struct rw_semaphore	rx_unmap_rwsem;
>  };

This is not a defect, but the comment claims more than the lock
delivers.  rx_ep is sampled without this rwsem in frwr_mr_init(),
frwr_map(), frwr_send(), frwr_unmap_async(), frwr_wp_create(),
rpcrdma_xprt_drain(), rpcrdma_mrs_create(), rpcrdma_mrs_refresh(),
rpcrdma_rep_create(), rpcrdma_rep_resize() and rpcrdma_post_recvs().

What the read side establishes is narrower: it orders
frwr_unmap_sync()'s LOCAL_INV submission ahead of the QP drain in
rpcrdma_xprt_disconnect().  Something closer to:

	/*
	 * Orders frwr_unmap_sync()'s LOCAL_INV submission ahead of the
	 * QP drain.  Disconnect holds the write side across the drain
	 * and resource teardown.
	 */

> @@ -479,6 +486,8 @@ extern unsigned int xprt_rdma_memreg_strategy;
>   * Endpoint calls - xprtrdma/verbs.c
>   */
>  void rpcrdma_force_disconnect(struct rpcrdma_ep *ep);
> +void rpcrdma_ep_get(struct rpcrdma_ep *ep);
> +int rpcrdma_ep_put(struct rpcrdma_ep *ep);

Nothing here says that a return of 1 leaves the CM ID for the caller to
destroy, which is the rule the frwr_unmap_sync() hunk above open codes.

-- 
Chuck Lever

[PATCH v3] xprtrdma: serialize unmap_sync with xprt_disconnect

[tmenninger]

From: Tim Menninger <tmenninger@everpuredata.com>

frwr_unmap_sync() posts LOCAL_INV Work Requests to the QP and waits
for the matching CQE. This can race with rpcrdma_xprt_disconnect():

    1. frwr_unmap_sync() samples r_xprt->rx_ep
    2. rpcrdma_xprt_disconnect() drains the QP
    3. frwr_unmap_sync() posts LOCAL_INV WRs after the drain
    4. No CQE arrives, so wait_for_completion() blocks indefinitely

The hung task appears in the rpciod workqueue:

Workqueue: rpciod rpc_async_schedule [sunrpc]
    wait_for_completion()
    frwr_unmap_sync()
    xprt_rdma_free()
    xprt_release()

Serialize synchronous MR invalidation against QP teardown with
rx_unmap_rwsem. frwr_unmap_sync() holds the read side while it builds
and submits the LOCAL_INV chain. Disconnect holds the write side across
QP drain and resource teardown. Drop the read side before waiting so
disconnect can drain the submitted Work Requests and deliver their CQEs.

Disconnect can reset a request while frwr_unmap_sync() waits for the
read side. Recheck rl_registered after acquiring the rwsem and return
without posting if the request has already been reset.

Move the synchronous invalidation completion from the final MR to the
request, whose lifetime covers xprt_rdma_free(), and hold a reference
to the sampled endpoint through the completion wait and error path.
This allows disconnect to destroy MRs after draining the QP and release
the transport-owned endpoint reference without a separate active-unmap
counter.

frwr_unmap_async() runs inline from Receive completion processing.
rpcrdma_xprt_drain() drains the Receive Queue before the Send Queue, so
asynchronous LOCAL_INV Work Requests are submitted before Send Queue
drain. frwr_unmap_sync(), however, runs from xprt_rdma_free() after the
transport send lock has been released, so it is not ordered against
disconnect.

Tested on v7.2-rc6 under pNFS load with repeated data server restarts
to provoke rapid QP disconnect/reconnect cycling. Without the fix, a
hung task in frwr_unmap_sync() reproduced twice in six attempts. With
the fix, no hang occurred across 200 cycles.

Also observed on v6.8 and v6.18.34.

Fixes: e28ce90083f0 ("xprtrdma: kmalloc rpcrdma_ep separate from rpcrdma_xprt")
Cc: stable@vger.kernel.org
Assisted-by: Claude:opus-4-7
Signed-off-by: Tim Menninger <tmenninger@everpuredata.com>
---
Changes in v3:
- Document the sleepable rx_unmap_rwsem locking context in
  frwr_unmap_sync() and rpcrdma_xprt_disconnect().
- Add rpcrdma_ep_release() so CM ID destruction remains encapsulated
  in verbs.c; export the endpoint get/release pair instead of get/put.
- Use ep->re_id->qp directly while frwr_unmap_sync() holds the rwsem
  read side.
- Correct rpcrdma_xprt_disconnect() documentation to account for
  endpoint resources that can outlive the disconnect operation.
- Narrow the rx_unmap_rwsem comment to describe the LOCAL_INV versus
  QP-drain ordering it provides.

v2: https://lore.kernel.org/all/20260820211700.318824-1-tmenninger@everpuredata.com/

Changes in v2:
- Change the Fixes tag to e28ce90083f0, which made rx_ep a separately
  allocated, kref-managed pointer.
- Explain why Receive-before-Send Queue drain orders
  frwr_unmap_async(), but does not order frwr_unmap_sync().
- Move the synchronous LOCAL_INV completion from struct rpcrdma_mr to
  struct rpcrdma_req.
- Hold a reference to the sampled endpoint through the completion wait
  and error path.
- Remove rx_unmap_active and its disconnect-side wait.
- Recheck rl_registered after acquiring the rwsem in case disconnect
  reset the request while the reader was blocked.
- Update the transport-recovery comments, LOCAL_INV wake kernel-doc,
  frwr_unmap_sync() kernel-doc, and the QP-posting locking comment.
- Keep rx_unmap_rwsem in struct rpcrdma_xprt so the reader can
  synchronize with disconnect before retaining the current endpoint.

v1: https://lore.kernel.org/all/20260818122825.2347594-1-tmenninger@purestorage.com/

 net/sunrpc/xprtrdma/frwr_ops.c  | 97 ++++++++++++++++++++++-----------
 net/sunrpc/xprtrdma/transport.c |  2 +
 net/sunrpc/xprtrdma/verbs.c     | 33 ++++++++---
 net/sunrpc/xprtrdma/xprt_rdma.h | 12 +++-
 4 files changed, 103 insertions(+), 41 deletions(-)

diff --git a/net/sunrpc/xprtrdma/frwr_ops.c b/net/sunrpc/xprtrdma/frwr_ops.c
index e5c71cf705a3..cfe6339577fe 100644
--- a/net/sunrpc/xprtrdma/frwr_ops.c
+++ b/net/sunrpc/xprtrdma/frwr_ops.c
@@ -27,13 +27,17 @@
 
 /* Transport recovery
  *
- * frwr_map and frwr_unmap_* cannot run at the same time the transport
- * connect worker is running. The connect worker holds the transport
- * send lock, just as ->send_request does. This prevents frwr_map and
- * the connect worker from running concurrently. When a connection is
- * closed, the Receive completion queue is drained before the allowing
- * the connect worker to get control. This prevents frwr_unmap and the
- * connect worker from running concurrently.
+ * frwr_map() is serialized with the connect worker by the transport
+ * send lock.
+ *
+ * frwr_unmap_async() runs from Receive completion processing.
+ * rpcrdma_xprt_drain() drains the Receive Queue before the Send Queue,
+ * so asynchronous LOCAL_INV Work Requests are submitted before Send
+ * Queue drain.
+ *
+ * frwr_unmap_sync() runs after the transport send lock is released.
+ * rx_unmap_rwsem orders its LOCAL_INV submission before disconnect
+ * drains the QP.
  *
  * When the underlying transport disconnects, MRs that are in flight
  * are flushed and are likely unusable. Thus all MRs are destroyed.
@@ -139,7 +143,6 @@ int frwr_mr_init(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr *mr)
 	mr->mr_ibmr = frmr;
 	mr->mr_device = NULL;
 	INIT_LIST_HEAD(&mr->mr_list);
-	init_completion(&mr->mr_linv_done);
 	frwr_cid_init(ep, mr);
 
 	sg_init_table(sg, depth);
@@ -538,40 +541,64 @@ static void frwr_wc_localinv(struct ib_cq *cq, struct ib_wc *wc)
  * @cq: completion queue
  * @wc: WCE for a completed LocalInv WR
  *
- * Awaken anyone waiting for an MR to finish being fenced.
+ * Wake frwr_unmap_sync() after the final LOCAL_INV completion.
  */
 static void frwr_wc_localinv_wake(struct ib_cq *cq, struct ib_wc *wc)
 {
 	struct ib_cqe *cqe = wc->wr_cqe;
 	struct rpcrdma_mr *mr = container_of(cqe, struct rpcrdma_mr, mr_cqe);
+	struct rpcrdma_req *req = mr->mr_req;
 
 	/* WARNING: Only wr_cqe and status are reliable at this point */
 	trace_xprtrdma_wc_li_wake(wc, &mr->mr_cid);
 	frwr_mr_done(wc, mr);
-	complete(&mr->mr_linv_done);
+	complete(&req->rl_linv_done);
 
 	rpcrdma_flush_disconnect(cq->cq_context, wc);
 }
 
 /**
- * frwr_unmap_sync - invalidate memory regions that were registered for @req
+ * frwr_unmap_sync - synchronously invalidate MRs registered for @req
  * @r_xprt: controlling transport instance
- * @req: rpcrdma_req with a non-empty list of MRs to process
+ * @req: request whose registered MRs are to be invalidated
  *
- * Sleeps until it is safe for the host CPU to access the previously mapped
- * memory regions. This guarantees that registered MRs are properly fenced
- * from the server before the RPC consumer accesses the data in them. It
- * also ensures proper Send flow control: waking the next RPC waits until
- * this RPC has relinquished all its Send Queue entries.
+ * If @req still owns registered MRs after synchronizing with transport
+ * disconnect, post a chain of LOCAL_INV Work Requests and wait for the
+ * final completion. On successful completion, this fences the mapped
+ * regions from remote access and preserves Send Queue flow control.
+ *
+ * A concurrent disconnect can reset @req while this function waits for
+ * the read side. In that case the MRs have already been released and no
+ * LOCAL_INV Work Requests are posted.
+ *
+ * Context: Process context. Takes and releases
+ *	    @r_xprt->rx_unmap_rwsem for read. May sleep.
  */
 void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 {
 	struct ib_send_wr *first, **prev, *last;
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
 	const struct ib_send_wr *bad_wr;
+	struct rpcrdma_ep *ep;
 	struct rpcrdma_mr *mr;
 	int rc;
 
+	/* serialize against rpcrdma_xprt_disconnect() */
+	down_read(&r_xprt->rx_unmap_rwsem);
+
+	/*
+	 * Disconnect can reset this request while we wait for the read side.
+	 * Recheck rl_registered in case rx_ep now refers to a new connection.
+	 */
+	ep = r_xprt->rx_ep;
+	if (!ep || list_empty(&req->rl_registered))
+		goto out_unlock;
+
+	/*
+	 * Hold an endpoint reference so ep, its QP, and its CM ID remain
+	 * valid after the rwsem is dropped and through the completion wait.
+	 */
+	rpcrdma_ep_get(ep);
+
 	/* ORDER: Invalidate all of the MRs first
 	 *
 	 * Chain the LOCAL_INV Work Requests and post them with
@@ -598,37 +625,43 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 		prev = &last->next;
 	} while ((mr = rpcrdma_mr_pop(&req->rl_registered)));
 
-	mr = container_of(last, struct rpcrdma_mr, mr_invwr);
-
 	/* Strong send queue ordering guarantees that when the
 	 * last WR in the chain completes, all WRs in the chain
 	 * are complete.
 	 */
 	last->wr_cqe->done = frwr_wc_localinv_wake;
-	reinit_completion(&mr->mr_linv_done);
+	reinit_completion(&req->rl_linv_done);
 
-	/* Transport disconnect drains the receive CQ before it
-	 * replaces the QP. The RPC reply handler won't call us
-	 * unless re_id->qp is a valid pointer.
+	/*
+	 * The read side prevents disconnect from draining this QP until the
+	 * LOCAL_INV chain has been submitted.
 	 */
 	bad_wr = NULL;
 	rc = ib_post_send(ep->re_id->qp, first, &bad_wr);
 
+	up_read(&r_xprt->rx_unmap_rwsem);
+
 	/* The final LOCAL_INV WR in the chain is supposed to
 	 * do the wake. If it was never posted, the wake will
 	 * not happen, so don't wait in that case.
 	 */
 	if (bad_wr != first)
-		wait_for_completion(&mr->mr_linv_done);
-	if (!rc)
-		return;
+		wait_for_completion(&req->rl_linv_done);
 
-	/* On error, the MRs get destroyed once the QP has drained. */
-	trace_xprtrdma_post_linv_err(req, rc);
+	if (rc) {
+		/* On error, the MRs get destroyed once the QP has drained. */
+		trace_xprtrdma_post_linv_err(req, rc);
 
-	/* Force a connection loss to ensure complete recovery.
-	 */
-	rpcrdma_force_disconnect(ep);
+		/* Force a connection loss to ensure complete recovery.
+		 */
+		rpcrdma_force_disconnect(ep);
+	}
+
+	rpcrdma_ep_release(ep);
+	return;
+
+out_unlock:
+	up_read(&r_xprt->rx_unmap_rwsem);
 }
 
 /**
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index d4e6746d8ecd..e38637c2ec78 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -364,6 +364,8 @@ xprt_setup_rdma(struct xprt_create *args)
 	INIT_DELAYED_WORK(&new_xprt->rx_connect_worker,
 			  xprt_rdma_connect_worker);
 
+	init_rwsem(&new_xprt->rx_unmap_rwsem);
+
 	xprt->max_payload = RPCRDMA_MAX_DATA_SEGS << PAGE_SHIFT;
 
 	return xprt;
diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 04b286223b24..b9ebfcbcabc5 100644
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -74,7 +74,6 @@ static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_reps_unmap(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt);
-static void rpcrdma_ep_get(struct rpcrdma_ep *ep);
 static int rpcrdma_ep_put(struct rpcrdma_ep *ep);
 static struct rpcrdma_regbuf *
 rpcrdma_regbuf_alloc_node(size_t size, enum dma_data_direction direction,
@@ -374,7 +373,7 @@ static void rpcrdma_ep_destroy(struct kref *kref)
 	module_put(THIS_MODULE);
 }
 
-static noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
+noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
 {
 	kref_get(&ep->re_kref);
 }
@@ -388,6 +387,14 @@ static noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep)
 	return kref_put(&ep->re_kref, rpcrdma_ep_destroy);
 }
 
+void rpcrdma_ep_release(struct rpcrdma_ep *ep)
+{
+	struct rdma_cm_id *id = ep->re_id;
+
+	if (rpcrdma_ep_put(ep))
+		rdma_destroy_id(id);
+}
+
 static int rpcrdma_ep_create(struct rpcrdma_xprt *r_xprt)
 {
 	struct rpcrdma_connect_private *pmsg;
@@ -572,26 +579,32 @@ int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt)
  * rpcrdma_xprt_disconnect - Disconnect underlying transport
  * @r_xprt: controlling transport instance
  *
- * Caller serializes. Either the transport send lock is held,
- * or we're being called to destroy the transport.
+ * Context: Caller serializes. Either the transport send lock is held,
+ *	    or the transport is being destroyed. Takes and releases
+ *	    @r_xprt->rx_unmap_rwsem for write. May sleep.
  *
- * On return, @r_xprt is completely divested of all hardware
- * resources and prepared for the next ->connect operation.
+ * On return, @r_xprt is prepared for the next ->connect operation.
+ * The detached endpoint and its resources can outlive this function
+ * until the final endpoint reference is released.
  */
 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 {
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
+	struct rpcrdma_ep *ep;
 	struct rdma_cm_id *id;
 	int rc;
 
+	down_write(&r_xprt->rx_unmap_rwsem);
+
+	ep = r_xprt->rx_ep;
 	if (!ep)
-		return;
+		goto out_unlock;
 
 	id = ep->re_id;
 	rc = rdma_disconnect(id);
 	trace_xprtrdma_disconnect(r_xprt, rc);
 
 	rpcrdma_xprt_drain(r_xprt);
+
 	rpcrdma_reps_unmap(r_xprt);
 	rpcrdma_sendctxs_destroy(r_xprt);
 	rpcrdma_reqs_reset(r_xprt);
@@ -601,6 +614,9 @@ void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 		rdma_destroy_id(id);
 
 	r_xprt->rx_ep = NULL;
+
+out_unlock:
+	up_write(&r_xprt->rx_unmap_rwsem);
 }
 
 /* Fixed-size circular FIFO queue. This implementation is wait-free and
@@ -924,6 +940,7 @@ struct rpcrdma_req *rpcrdma_req_create(struct rpcrdma_xprt *r_xprt,
 
 	INIT_LIST_HEAD(&req->rl_free_mrs);
 	INIT_LIST_HEAD(&req->rl_registered);
+	init_completion(&req->rl_linv_done);
 	spin_lock(&buffer->rb_lock);
 	list_add(&req->rl_all, &buffer->rb_allreqs);
 	spin_unlock(&buffer->rb_lock);
diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
index 4cbc941e4a3e..08dfe9414657 100644
--- a/net/sunrpc/xprtrdma/xprt_rdma.h
+++ b/net/sunrpc/xprtrdma/xprt_rdma.h
@@ -46,6 +46,7 @@
 #include <linux/spinlock.h> 		/* spinlock_t, etc */
 #include <linux/atomic.h>		/* atomic_t, etc */
 #include <linux/kref.h>			/* struct kref */
+#include <linux/rwsem.h>		/* struct rw_semaphore */
 #include <linux/workqueue.h>		/* struct work_struct */
 #include <linux/llist.h>
 
@@ -249,7 +250,6 @@ struct rpcrdma_mr {
 	int			mr_nents;
 	enum dma_data_direction	mr_dir;
 	struct ib_cqe		mr_cqe;
-	struct completion	mr_linv_done;
 	union {
 		struct ib_reg_wr	mr_regwr;
 		struct ib_send_wr	mr_invwr;
@@ -348,6 +348,7 @@ struct rpcrdma_req {
 
 	struct list_head	rl_free_mrs;
 	struct list_head	rl_registered;
+	struct completion	rl_linv_done;
 };
 
 static inline struct rpcrdma_req *
@@ -449,6 +450,13 @@ struct rpcrdma_xprt {
 	struct delayed_work	rx_connect_worker;
 	struct rpc_timeout	rx_timeout;
 	struct rpcrdma_stats	rx_stats;
+
+	/*
+	 * Orders frwr_unmap_sync()'s LOCAL_INV submission ahead of the
+	 * QP drain. Disconnect holds the write side across the drain
+	 * and resource teardown.
+	 */
+	struct rw_semaphore	rx_unmap_rwsem;
 };
 
 #define rpcx_to_rdmax(x) container_of(x, struct rpcrdma_xprt, rx_xprt)
@@ -479,6 +487,8 @@ extern unsigned int xprt_rdma_memreg_strategy;
  * Endpoint calls - xprtrdma/verbs.c
  */
 void rpcrdma_force_disconnect(struct rpcrdma_ep *ep);
+void rpcrdma_ep_get(struct rpcrdma_ep *ep);
+void rpcrdma_ep_release(struct rpcrdma_ep *ep);
 void rpcrdma_flush_disconnect(struct rpcrdma_xprt *r_xprt, struct ib_wc *wc);
 int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt);
 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt);
-- 
2.34.1

Re: [PATCH v3] xprtrdma: serialize unmap_sync with xprt_disconnect

[Chuck Lever]

On Fri, Aug 21, 2026, at 7:48 PM, tmenninger@purestorage.com wrote:

> diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
> index 04b286223b24..b9ebfcbcabc5 100644
> --- a/net/sunrpc/xprtrdma/verbs.c
> +++ b/net/sunrpc/xprtrdma/verbs.c

> @@ -388,6 +387,14 @@ static noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep)
>  	return kref_put(&ep->re_kref, rpcrdma_ep_destroy);
>  }
> 
> +void rpcrdma_ep_release(struct rpcrdma_ep *ep)
> +{
> +	struct rdma_cm_id *id = ep->re_id;
> +
> +	if (rpcrdma_ep_put(ep))
> +		rdma_destroy_id(id);
> +}
> +

Nit: The new externally-visible functions need kdoc comments.
rpcrdma_ep_put() is now static, so the kdoc comment there can
be converted to non-kdoc-style.


> @@ -601,6 +614,9 @@ void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
>  		rdma_destroy_id(id);
> 
>  	r_xprt->rx_ep = NULL;
> +
> +out_unlock:
> +	up_write(&r_xprt->rx_unmap_rwsem);
>  }

Nit: rpcrdma_xprt_disconnect() retains

    if (rpcrdma_ep_put(ep)) 
        rdma_destroy_id(id);

but should call the new rpcrdma_ep_release() helper.

The new block comments are wordy, and could be tightened. But no
show-stoppers here:

Reviewed-by: Chuck Lever <cel@kernel.org>

-- 
Chuck Lever

[PATCH v4] xprtrdma: serialize unmap_sync with xprt_disconnect

[Tim Menninger]

frwr_unmap_sync() posts LOCAL_INV Work Requests to the QP and waits
for the matching CQE. This can race with rpcrdma_xprt_disconnect():

    1. frwr_unmap_sync() samples r_xprt->rx_ep
    2. rpcrdma_xprt_disconnect() drains the QP
    3. frwr_unmap_sync() posts LOCAL_INV WRs after the drain
    4. No CQE arrives, so wait_for_completion() blocks indefinitely

The hung task appears in the rpciod workqueue:

Workqueue: rpciod rpc_async_schedule [sunrpc]
    wait_for_completion()
    frwr_unmap_sync()
    xprt_rdma_free()
    xprt_release()

Serialize synchronous MR invalidation against QP teardown with
rx_unmap_rwsem. frwr_unmap_sync() holds the read side while it builds
and submits the LOCAL_INV chain. Disconnect holds the write side across
QP drain and resource teardown. Drop the read side before waiting so
disconnect can drain the submitted Work Requests and deliver their CQEs.

Disconnect can reset a request while frwr_unmap_sync() waits for the
read side. Recheck rl_registered after acquiring the rwsem and return
without posting if the request has already been reset.

Move the synchronous invalidation completion from the final MR to the
request, whose lifetime covers xprt_rdma_free(), and hold a reference
to the sampled endpoint through the completion wait and error path.
This allows disconnect to destroy MRs after draining the QP and release
the transport-owned endpoint reference without a separate active-unmap
counter.

frwr_unmap_async() runs inline from Receive completion processing.
rpcrdma_xprt_drain() drains the Receive Queue before the Send Queue, so
asynchronous LOCAL_INV Work Requests are submitted before Send Queue
drain. frwr_unmap_sync(), however, runs from xprt_rdma_free() after the
transport send lock has been released, so it is not ordered against
disconnect.

Tested on v7.2-rc6 under pNFS load with repeated data server restarts
to provoke rapid QP disconnect/reconnect cycling. Without the fix, a
hung task in frwr_unmap_sync() reproduced twice in six attempts. With
the fix, no hang occurred across 200 cycles.

Also observed on v6.8 and v6.18.34.

Fixes: e28ce90083f0 ("xprtrdma: kmalloc rpcrdma_ep separate from rpcrdma_xprt")
Cc: stable@vger.kernel.org
Assisted-by: Claude:opus-4-7
Reviewed-by: Chuck Lever <cel@kernel.org>
Signed-off-by: Tim Menninger <tmenninger@everpuredata.com>
---
Changes in v4:
- Add kernel-doc comments for rpcrdma_ep_get() and
  rpcrdma_ep_release().
- Convert the rpcrdma_ep_put() comment to ordinary comment style.

v3: https://lore.kernel.org/all/20260821234827.1560856-1-tmenninger@purestorage.com/

Changes in v3:
- Document the sleepable rx_unmap_rwsem locking context in
  frwr_unmap_sync() and rpcrdma_xprt_disconnect().
- Add rpcrdma_ep_release() so CM ID destruction remains encapsulated
  in verbs.c; export the endpoint get/release pair instead of get/put.
- Use ep->re_id->qp directly while frwr_unmap_sync() holds the rwsem
  read side.
- Correct rpcrdma_xprt_disconnect() documentation to account for
  endpoint resources that can outlive the disconnect operation.
- Narrow the rx_unmap_rwsem comment to describe the LOCAL_INV versus
  QP-drain ordering it provides.

v2: https://lore.kernel.org/all/20260820211700.318824-1-tmenninger@everpuredata.com/

Changes in v2:
- Change the Fixes tag to e28ce90083f0, which made rx_ep a separately
  allocated, kref-managed pointer.
- Explain why Receive-before-Send Queue drain orders
  frwr_unmap_async(), but does not order frwr_unmap_sync().
- Move the synchronous LOCAL_INV completion from struct rpcrdma_mr to
  struct rpcrdma_req.
- Hold a reference to the sampled endpoint through the completion wait
  and error path.
- Remove rx_unmap_active and its disconnect-side wait.
- Recheck rl_registered after acquiring the rwsem in case disconnect
  reset the request while the reader was blocked.
- Update the transport-recovery comments, LOCAL_INV wake kernel-doc,
  frwr_unmap_sync() kernel-doc, and the QP-posting locking comment.
- Keep rx_unmap_rwsem in struct rpcrdma_xprt so the reader can
  synchronize with disconnect before retaining the current endpoint.

v1: https://lore.kernel.org/all/20260818122825.2347594-1-tmenninger@purestorage.com/

 net/sunrpc/xprtrdma/frwr_ops.c  | 97 ++++++++++++++++++++++-----------
 net/sunrpc/xprtrdma/transport.c |  2 +
 net/sunrpc/xprtrdma/verbs.c     | 52 ++++++++++++++----
 net/sunrpc/xprtrdma/xprt_rdma.h | 12 +++-
 4 files changed, 118 insertions(+), 45 deletions(-)

diff --git a/net/sunrpc/xprtrdma/frwr_ops.c b/net/sunrpc/xprtrdma/frwr_ops.c
index e5c71cf705a3..cfe6339577fe 100644
--- a/net/sunrpc/xprtrdma/frwr_ops.c
+++ b/net/sunrpc/xprtrdma/frwr_ops.c
@@ -27,13 +27,17 @@
 
 /* Transport recovery
  *
- * frwr_map and frwr_unmap_* cannot run at the same time the transport
- * connect worker is running. The connect worker holds the transport
- * send lock, just as ->send_request does. This prevents frwr_map and
- * the connect worker from running concurrently. When a connection is
- * closed, the Receive completion queue is drained before the allowing
- * the connect worker to get control. This prevents frwr_unmap and the
- * connect worker from running concurrently.
+ * frwr_map() is serialized with the connect worker by the transport
+ * send lock.
+ *
+ * frwr_unmap_async() runs from Receive completion processing.
+ * rpcrdma_xprt_drain() drains the Receive Queue before the Send Queue,
+ * so asynchronous LOCAL_INV Work Requests are submitted before Send
+ * Queue drain.
+ *
+ * frwr_unmap_sync() runs after the transport send lock is released.
+ * rx_unmap_rwsem orders its LOCAL_INV submission before disconnect
+ * drains the QP.
  *
  * When the underlying transport disconnects, MRs that are in flight
  * are flushed and are likely unusable. Thus all MRs are destroyed.
@@ -139,7 +143,6 @@ int frwr_mr_init(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr *mr)
 	mr->mr_ibmr = frmr;
 	mr->mr_device = NULL;
 	INIT_LIST_HEAD(&mr->mr_list);
-	init_completion(&mr->mr_linv_done);
 	frwr_cid_init(ep, mr);
 
 	sg_init_table(sg, depth);
@@ -538,40 +541,64 @@ static void frwr_wc_localinv(struct ib_cq *cq, struct ib_wc *wc)
  * @cq: completion queue
  * @wc: WCE for a completed LocalInv WR
  *
- * Awaken anyone waiting for an MR to finish being fenced.
+ * Wake frwr_unmap_sync() after the final LOCAL_INV completion.
  */
 static void frwr_wc_localinv_wake(struct ib_cq *cq, struct ib_wc *wc)
 {
 	struct ib_cqe *cqe = wc->wr_cqe;
 	struct rpcrdma_mr *mr = container_of(cqe, struct rpcrdma_mr, mr_cqe);
+	struct rpcrdma_req *req = mr->mr_req;
 
 	/* WARNING: Only wr_cqe and status are reliable at this point */
 	trace_xprtrdma_wc_li_wake(wc, &mr->mr_cid);
 	frwr_mr_done(wc, mr);
-	complete(&mr->mr_linv_done);
+	complete(&req->rl_linv_done);
 
 	rpcrdma_flush_disconnect(cq->cq_context, wc);
 }
 
 /**
- * frwr_unmap_sync - invalidate memory regions that were registered for @req
+ * frwr_unmap_sync - synchronously invalidate MRs registered for @req
  * @r_xprt: controlling transport instance
- * @req: rpcrdma_req with a non-empty list of MRs to process
+ * @req: request whose registered MRs are to be invalidated
  *
- * Sleeps until it is safe for the host CPU to access the previously mapped
- * memory regions. This guarantees that registered MRs are properly fenced
- * from the server before the RPC consumer accesses the data in them. It
- * also ensures proper Send flow control: waking the next RPC waits until
- * this RPC has relinquished all its Send Queue entries.
+ * If @req still owns registered MRs after synchronizing with transport
+ * disconnect, post a chain of LOCAL_INV Work Requests and wait for the
+ * final completion. On successful completion, this fences the mapped
+ * regions from remote access and preserves Send Queue flow control.
+ *
+ * A concurrent disconnect can reset @req while this function waits for
+ * the read side. In that case the MRs have already been released and no
+ * LOCAL_INV Work Requests are posted.
+ *
+ * Context: Process context. Takes and releases
+ *	    @r_xprt->rx_unmap_rwsem for read. May sleep.
  */
 void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 {
 	struct ib_send_wr *first, **prev, *last;
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
 	const struct ib_send_wr *bad_wr;
+	struct rpcrdma_ep *ep;
 	struct rpcrdma_mr *mr;
 	int rc;
 
+	/* serialize against rpcrdma_xprt_disconnect() */
+	down_read(&r_xprt->rx_unmap_rwsem);
+
+	/*
+	 * Disconnect can reset this request while we wait for the read side.
+	 * Recheck rl_registered in case rx_ep now refers to a new connection.
+	 */
+	ep = r_xprt->rx_ep;
+	if (!ep || list_empty(&req->rl_registered))
+		goto out_unlock;
+
+	/*
+	 * Hold an endpoint reference so ep, its QP, and its CM ID remain
+	 * valid after the rwsem is dropped and through the completion wait.
+	 */
+	rpcrdma_ep_get(ep);
+
 	/* ORDER: Invalidate all of the MRs first
 	 *
 	 * Chain the LOCAL_INV Work Requests and post them with
@@ -598,37 +625,43 @@ void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
 		prev = &last->next;
 	} while ((mr = rpcrdma_mr_pop(&req->rl_registered)));
 
-	mr = container_of(last, struct rpcrdma_mr, mr_invwr);
-
 	/* Strong send queue ordering guarantees that when the
 	 * last WR in the chain completes, all WRs in the chain
 	 * are complete.
 	 */
 	last->wr_cqe->done = frwr_wc_localinv_wake;
-	reinit_completion(&mr->mr_linv_done);
+	reinit_completion(&req->rl_linv_done);
 
-	/* Transport disconnect drains the receive CQ before it
-	 * replaces the QP. The RPC reply handler won't call us
-	 * unless re_id->qp is a valid pointer.
+	/*
+	 * The read side prevents disconnect from draining this QP until the
+	 * LOCAL_INV chain has been submitted.
 	 */
 	bad_wr = NULL;
 	rc = ib_post_send(ep->re_id->qp, first, &bad_wr);
 
+	up_read(&r_xprt->rx_unmap_rwsem);
+
 	/* The final LOCAL_INV WR in the chain is supposed to
 	 * do the wake. If it was never posted, the wake will
 	 * not happen, so don't wait in that case.
 	 */
 	if (bad_wr != first)
-		wait_for_completion(&mr->mr_linv_done);
-	if (!rc)
-		return;
+		wait_for_completion(&req->rl_linv_done);
 
-	/* On error, the MRs get destroyed once the QP has drained. */
-	trace_xprtrdma_post_linv_err(req, rc);
+	if (rc) {
+		/* On error, the MRs get destroyed once the QP has drained. */
+		trace_xprtrdma_post_linv_err(req, rc);
 
-	/* Force a connection loss to ensure complete recovery.
-	 */
-	rpcrdma_force_disconnect(ep);
+		/* Force a connection loss to ensure complete recovery.
+		 */
+		rpcrdma_force_disconnect(ep);
+	}
+
+	rpcrdma_ep_release(ep);
+	return;
+
+out_unlock:
+	up_read(&r_xprt->rx_unmap_rwsem);
 }
 
 /**
diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c
index d4e6746d8ecd..e38637c2ec78 100644
--- a/net/sunrpc/xprtrdma/transport.c
+++ b/net/sunrpc/xprtrdma/transport.c
@@ -364,6 +364,8 @@ xprt_setup_rdma(struct xprt_create *args)
 	INIT_DELAYED_WORK(&new_xprt->rx_connect_worker,
 			  xprt_rdma_connect_worker);
 
+	init_rwsem(&new_xprt->rx_unmap_rwsem);
+
 	xprt->max_payload = RPCRDMA_MAX_DATA_SEGS << PAGE_SHIFT;
 
 	return xprt;
diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 04b286223b24..b9f7fbc67b74 100644
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -74,7 +74,6 @@ static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_reps_unmap(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt);
-static void rpcrdma_ep_get(struct rpcrdma_ep *ep);
 static int rpcrdma_ep_put(struct rpcrdma_ep *ep);
 static struct rpcrdma_regbuf *
 rpcrdma_regbuf_alloc_node(size_t size, enum dma_data_direction direction,
@@ -374,20 +373,39 @@ static void rpcrdma_ep_destroy(struct kref *kref)
 	module_put(THIS_MODULE);
 }
 
-static noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
+/**
+ * rpcrdma_ep_get - Acquire an endpoint reference
+ * @ep: RPC/RDMA endpoint to retain
+ *
+ * Context: Any context.
+ */
+noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
 {
 	kref_get(&ep->re_kref);
 }
 
-/* Returns:
- *     %0 if @ep still has a positive kref count, or
- *     %1 if @ep was destroyed successfully.
- */
+/* Return 1 if ep was destroyed, otherwise 0. */
 static noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep)
 {
 	return kref_put(&ep->re_kref, rpcrdma_ep_destroy);
 }
 
+/**
+ * rpcrdma_ep_release - Release an endpoint reference
+ * @ep: RPC/RDMA endpoint to release
+ *
+ * The final release destroys @ep and its RDMA CM ID.
+ *
+ * Context: Process context. May sleep.
+ */
+void rpcrdma_ep_release(struct rpcrdma_ep *ep)
+{
+	struct rdma_cm_id *id = ep->re_id;
+
+	if (rpcrdma_ep_put(ep))
+		rdma_destroy_id(id);
+}
+
 static int rpcrdma_ep_create(struct rpcrdma_xprt *r_xprt)
 {
 	struct rpcrdma_connect_private *pmsg;
@@ -572,26 +590,32 @@ int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt)
  * rpcrdma_xprt_disconnect - Disconnect underlying transport
  * @r_xprt: controlling transport instance
  *
- * Caller serializes. Either the transport send lock is held,
- * or we're being called to destroy the transport.
+ * Context: Caller serializes. Either the transport send lock is held,
+ *	    or the transport is being destroyed. Takes and releases
+ *	    @r_xprt->rx_unmap_rwsem for write. May sleep.
  *
- * On return, @r_xprt is completely divested of all hardware
- * resources and prepared for the next ->connect operation.
+ * On return, @r_xprt is prepared for the next ->connect operation.
+ * The detached endpoint and its resources can outlive this function
+ * until the final endpoint reference is released.
  */
 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 {
-	struct rpcrdma_ep *ep = r_xprt->rx_ep;
+	struct rpcrdma_ep *ep;
 	struct rdma_cm_id *id;
 	int rc;
 
+	down_write(&r_xprt->rx_unmap_rwsem);
+
+	ep = r_xprt->rx_ep;
 	if (!ep)
-		return;
+		goto out_unlock;
 
 	id = ep->re_id;
 	rc = rdma_disconnect(id);
 	trace_xprtrdma_disconnect(r_xprt, rc);
 
 	rpcrdma_xprt_drain(r_xprt);
+
 	rpcrdma_reps_unmap(r_xprt);
 	rpcrdma_sendctxs_destroy(r_xprt);
 	rpcrdma_reqs_reset(r_xprt);
@@ -601,6 +625,9 @@ void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
 		rdma_destroy_id(id);
 
 	r_xprt->rx_ep = NULL;
+
+out_unlock:
+	up_write(&r_xprt->rx_unmap_rwsem);
 }
 
 /* Fixed-size circular FIFO queue. This implementation is wait-free and
@@ -924,6 +951,7 @@ struct rpcrdma_req *rpcrdma_req_create(struct rpcrdma_xprt *r_xprt,
 
 	INIT_LIST_HEAD(&req->rl_free_mrs);
 	INIT_LIST_HEAD(&req->rl_registered);
+	init_completion(&req->rl_linv_done);
 	spin_lock(&buffer->rb_lock);
 	list_add(&req->rl_all, &buffer->rb_allreqs);
 	spin_unlock(&buffer->rb_lock);
diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
index 4cbc941e4a3e..08dfe9414657 100644
--- a/net/sunrpc/xprtrdma/xprt_rdma.h
+++ b/net/sunrpc/xprtrdma/xprt_rdma.h
@@ -46,6 +46,7 @@
 #include <linux/spinlock.h> 		/* spinlock_t, etc */
 #include <linux/atomic.h>		/* atomic_t, etc */
 #include <linux/kref.h>			/* struct kref */
+#include <linux/rwsem.h>		/* struct rw_semaphore */
 #include <linux/workqueue.h>		/* struct work_struct */
 #include <linux/llist.h>
 
@@ -249,7 +250,6 @@ struct rpcrdma_mr {
 	int			mr_nents;
 	enum dma_data_direction	mr_dir;
 	struct ib_cqe		mr_cqe;
-	struct completion	mr_linv_done;
 	union {
 		struct ib_reg_wr	mr_regwr;
 		struct ib_send_wr	mr_invwr;
@@ -348,6 +348,7 @@ struct rpcrdma_req {
 
 	struct list_head	rl_free_mrs;
 	struct list_head	rl_registered;
+	struct completion	rl_linv_done;
 };
 
 static inline struct rpcrdma_req *
@@ -449,6 +450,13 @@ struct rpcrdma_xprt {
 	struct delayed_work	rx_connect_worker;
 	struct rpc_timeout	rx_timeout;
 	struct rpcrdma_stats	rx_stats;
+
+	/*
+	 * Orders frwr_unmap_sync()'s LOCAL_INV submission ahead of the
+	 * QP drain. Disconnect holds the write side across the drain
+	 * and resource teardown.
+	 */
+	struct rw_semaphore	rx_unmap_rwsem;
 };
 
 #define rpcx_to_rdmax(x) container_of(x, struct rpcrdma_xprt, rx_xprt)
@@ -479,6 +487,8 @@ extern unsigned int xprt_rdma_memreg_strategy;
  * Endpoint calls - xprtrdma/verbs.c
  */
 void rpcrdma_force_disconnect(struct rpcrdma_ep *ep);
+void rpcrdma_ep_get(struct rpcrdma_ep *ep);
+void rpcrdma_ep_release(struct rpcrdma_ep *ep);
 void rpcrdma_flush_disconnect(struct rpcrdma_xprt *r_xprt, struct ib_wc *wc);
 int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt);
 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt);
-- 
2.34.1