* [PATCH v3 2/4] SUNRPC: Split recv_lock out of xprt->queue_lock
2026-09-15 14:20 [PATCH v3 0/4] Reduce lock contention in the NFS client Chuck Lever
2026-09-15 14:20 ` [PATCH v3 1/4] SUNRPC: Use atomic_t for XID allocation Chuck Lever
@ 2026-09-15 14:20 ` Chuck Lever
2026-09-15 14:20 ` [PATCH v3 3/4] SUNRPC: Set WQ_SYSFS on rpciod and xprtiod Chuck Lever
` (3 subsequent siblings)
5 siblings, 0 replies; 7+ messages in thread
From: Chuck Lever @ 2026-09-15 14:20 UTC (permalink / raw)
To: Trond Myklebust, Anna Schumaker
Cc: Jeff Layton, NeilBrown, Tim Menninger, linux-nfs, linux-kernel,
Chuck Lever
xprt->queue_lock protects two independent structures: the recv_queue
rb-tree for reply matching and the xmit_queue list for transmit
draining. No hot path touches both in one critical section, yet every
RPC submit and completion contends on the same lock. Under a 4KB
NFSv3 READ workload on 100GbE RDMA, 53% of non-idle CPU cycles are
spent in native_queued_spin_lock_slowpath: the CQ completion worker
running rpcrdma_reply_handler serializes against ~150 kworker threads
enqueuing receives and transmits.
Introduce xprt->recv_lock for the receive path -- recv_queue
operations, request lookup, receive-side pinning, and completion --
leaving queue_lock to the xmit_queue and the xprt_transmit drain
loop.
A request is pinned under the lock of the queue it was found through,
so xprt_request_dequeue_xprt() drains pins once under each lock
before it dequeues from that queue. The transmit dequeue has to see a
zero pin count under queue_lock: it frees the send buffer's bvec that
a transmitter is iterating, and it aborts a partial send only if the
request is still first in the queue. A receive-side unpin runs under
recv_lock rather than the lock that publishes RPC_TASK_MSG_PIN_WAIT,
so xprt_unpin_rqst() wakes the waiter whenever the count reaches
zero instead of testing the flag.
Also move the rq_private_buf memcpy in xprt_request_enqueue_receive
above the lock acquisition: until the rb-tree insert publishes the
request, the reply handler cannot see it, so the copy is safe
unlocked and the submitter's critical section shrinks to the insert
alone.
Signed-off-by: Chuck Lever <cel@kernel.org>
---
include/linux/sunrpc/xprt.h | 6 ++-
net/sunrpc/svcsock.c | 6 +--
net/sunrpc/xprt.c | 76 +++++++++++++++++++-----------
net/sunrpc/xprtrdma/rpc_rdma.c | 14 +++---
net/sunrpc/xprtrdma/svc_rdma_backchannel.c | 8 ++--
net/sunrpc/xprtsock.c | 18 +++----
6 files changed, 77 insertions(+), 51 deletions(-)
diff --git a/include/linux/sunrpc/xprt.h b/include/linux/sunrpc/xprt.h
index 0d6c3f6bf97e..ed1e28b74f02 100644
--- a/include/linux/sunrpc/xprt.h
+++ b/include/linux/sunrpc/xprt.h
@@ -272,7 +272,7 @@ struct rpc_xprt {
atomic_long_t queuelen;
spinlock_t transport_lock; /* lock transport info */
spinlock_t reserve_lock; /* lock slot table */
- spinlock_t queue_lock; /* send/receive queue lock */
+ spinlock_t queue_lock; /* send queue lock */
atomic_t xid; /* Most recently issued XID */
struct rpc_task * snd_task; /* Task blocked in send */
@@ -292,6 +292,10 @@ struct rpc_xprt {
* backchannel rpc_rqst's */
#endif /* CONFIG_SUNRPC_BACKCHANNEL */
+ /*
+ * Receive stuff
+ */
+ spinlock_t recv_lock; /* receive queue lock */
struct rb_root recv_queue; /* Receive queue */
struct {
diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c
index 50e5e7f5b762..8939ba604385 100644
--- a/net/sunrpc/svcsock.c
+++ b/net/sunrpc/svcsock.c
@@ -1102,7 +1102,7 @@ static int receive_cb_reply(struct svc_sock *svsk, struct svc_rqst *rqstp)
if (!bc_xprt)
return -EAGAIN;
- spin_lock(&bc_xprt->queue_lock);
+ spin_lock(&bc_xprt->recv_lock);
req = xprt_lookup_rqst(bc_xprt, xid);
if (!req)
goto unlock_eagain;
@@ -1120,10 +1120,10 @@ static int receive_cb_reply(struct svc_sock *svsk, struct svc_rqst *rqstp)
memcpy(dst->iov_base, src->iov_base, src->iov_len);
xprt_complete_rqst(req->rq_task, rqstp->rq_arg.len);
rqstp->rq_arg.len = 0;
- spin_unlock(&bc_xprt->queue_lock);
+ spin_unlock(&bc_xprt->recv_lock);
return 0;
unlock_eagain:
- spin_unlock(&bc_xprt->queue_lock);
+ spin_unlock(&bc_xprt->recv_lock);
return -EAGAIN;
}
diff --git a/net/sunrpc/xprt.c b/net/sunrpc/xprt.c
index 186c14f0f928..883123ec70b0 100644
--- a/net/sunrpc/xprt.c
+++ b/net/sunrpc/xprt.c
@@ -1061,7 +1061,7 @@ xprt_request_rb_remove(struct rpc_xprt *xprt, struct rpc_rqst *req)
* @xprt: transport on which the original request was transmitted
* @xid: RPC XID of incoming reply
*
- * Caller holds xprt->queue_lock.
+ * Caller holds xprt->recv_lock.
*/
struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid)
{
@@ -1092,8 +1092,9 @@ xprt_is_pinned_rqst(struct rpc_rqst *req)
* xprt_pin_rqst - Pin a request on the transport receive list
* @req: Request to pin
*
- * Caller must ensure this is atomic with the call to xprt_lookup_rqst()
- * so should be holding xprt->queue_lock.
+ * Caller must hold the lock that protects the queue through which
+ * it found the request: xprt->recv_lock for the receive path,
+ * xprt->queue_lock for the transmit drain path.
*/
void xprt_pin_rqst(struct rpc_rqst *req)
{
@@ -1105,14 +1106,10 @@ EXPORT_SYMBOL_GPL(xprt_pin_rqst);
* xprt_unpin_rqst - Unpin a request on the transport receive list
* @req: Request to pin
*
- * Caller should be holding xprt->queue_lock.
+ * Caller holds the lock it held for the matching xprt_pin_rqst().
*/
void xprt_unpin_rqst(struct rpc_rqst *req)
{
- if (!test_bit(RPC_TASK_MSG_PIN_WAIT, &req->rq_task->tk_runstate)) {
- atomic_dec(&req->rq_pin);
- return;
- }
if (atomic_dec_and_test(&req->rq_pin))
wake_up_var(&req->rq_pin);
}
@@ -1123,6 +1120,26 @@ static void xprt_wait_on_pinned_rqst(struct rpc_rqst *req)
wait_var_event(&req->rq_pin, !xprt_is_pinned_rqst(req));
}
+/*
+ * A pin is taken under the lock of the queue the request was found
+ * through, so a zero count observed under @lock rules out any pinner
+ * that came through that queue. The lock is dropped to wait, and the
+ * re-test under it catches a pinner that arrived in the gap.
+ */
+static void xprt_request_drain_pins(struct rpc_task *task, spinlock_t *lock)
+ __must_hold(lock)
+{
+ struct rpc_rqst *req = task->tk_rqstp;
+
+ while (xprt_is_pinned_rqst(req)) {
+ set_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
+ spin_unlock(lock);
+ xprt_wait_on_pinned_rqst(req);
+ spin_lock(lock);
+ clear_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
+ }
+}
+
static bool
xprt_request_data_received(struct rpc_task *task)
{
@@ -1155,16 +1172,16 @@ xprt_request_enqueue_receive(struct rpc_task *task)
ret = xprt_request_prepare(task->tk_rqstp, &req->rq_rcv_buf);
if (ret)
return ret;
- spin_lock(&xprt->queue_lock);
-
- /* Update the softirq receive buffer */
+ /* Reply handlers cannot find the request until the rb-tree
+ * insert below publishes it, so the copy needs no lock.
+ */
memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
sizeof(req->rq_private_buf));
- /* Add request to the receive list */
+ spin_lock(&xprt->recv_lock);
xprt_request_rb_insert(xprt, req);
set_bit(RPC_TASK_NEED_RECV, &task->tk_runstate);
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
/* Turn off autodisconnect */
timer_delete_sync(&xprt->timer);
@@ -1175,7 +1192,7 @@ xprt_request_enqueue_receive(struct rpc_task *task)
* xprt_request_dequeue_receive_locked - Remove a request from the receive queue
* @task: RPC task
*
- * Caller must hold xprt->queue_lock.
+ * Caller must hold xprt->recv_lock.
*/
static void
xprt_request_dequeue_receive_locked(struct rpc_task *task)
@@ -1190,7 +1207,7 @@ xprt_request_dequeue_receive_locked(struct rpc_task *task)
* xprt_update_rtt - Update RPC RTT statistics
* @task: RPC request that recently completed
*
- * Caller holds xprt->queue_lock.
+ * Caller holds xprt->recv_lock.
*/
void xprt_update_rtt(struct rpc_task *task)
{
@@ -1212,7 +1229,7 @@ EXPORT_SYMBOL_GPL(xprt_update_rtt);
* @task: RPC request that recently completed
* @copied: actual number of bytes received from the transport
*
- * Caller holds xprt->queue_lock.
+ * Caller holds xprt->recv_lock.
*/
void xprt_complete_rqst(struct rpc_task *task, int copied)
{
@@ -1309,7 +1326,7 @@ void xprt_request_wait_receive(struct rpc_task *task)
* The spinlock ensures atomicity between the test of
* req->rq_reply_bytes_recvd, and the call to rpc_sleep_on().
*/
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
if (test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate)) {
xprt->ops->wait_for_reply_request(task);
/*
@@ -1321,7 +1338,7 @@ void xprt_request_wait_receive(struct rpc_task *task)
rpc_wake_up_queued_task_set_status(&xprt->pending,
task, -ENOTCONN);
}
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
}
static bool
@@ -1439,7 +1456,12 @@ xprt_request_dequeue_transmit(struct rpc_task *task)
* @task: pointer to rpc_task
*
* Remove a task from the transmit and receive queues, and ensure that
- * it is not pinned by the receive work item.
+ * it is not pinned by any concurrent work item.
+ *
+ * The transmit dequeue frees the send buffer's bvec and may abort a
+ * partial send, so it must not run while a transmitter holds a pin.
+ * Drain pins under each queue's lock before leaving that queue; once
+ * the request is off both, no new pin can be taken.
*/
void
xprt_request_dequeue_xprt(struct rpc_task *task)
@@ -1451,16 +1473,15 @@ xprt_request_dequeue_xprt(struct rpc_task *task)
test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) ||
xprt_is_pinned_rqst(req)) {
spin_lock(&xprt->queue_lock);
- while (xprt_is_pinned_rqst(req)) {
- set_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
- spin_unlock(&xprt->queue_lock);
- xprt_wait_on_pinned_rqst(req);
- spin_lock(&xprt->queue_lock);
- clear_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
- }
+ xprt_request_drain_pins(task, &xprt->queue_lock);
xprt_request_dequeue_transmit_locked(task);
- xprt_request_dequeue_receive_locked(task);
spin_unlock(&xprt->queue_lock);
+
+ spin_lock(&xprt->recv_lock);
+ xprt_request_drain_pins(task, &xprt->recv_lock);
+ xprt_request_dequeue_receive_locked(task);
+ spin_unlock(&xprt->recv_lock);
+
xdr_free_bvec(&req->rq_rcv_buf);
}
}
@@ -2038,6 +2059,7 @@ static void xprt_init(struct rpc_xprt *xprt, struct net *net)
spin_lock_init(&xprt->transport_lock);
spin_lock_init(&xprt->reserve_lock);
spin_lock_init(&xprt->queue_lock);
+ spin_lock_init(&xprt->recv_lock);
INIT_LIST_HEAD(&xprt->free);
xprt->recv_queue = RB_ROOT;
diff --git a/net/sunrpc/xprtrdma/rpc_rdma.c b/net/sunrpc/xprtrdma/rpc_rdma.c
index 1285f04cdac1..a82d3d9bc7ae 100644
--- a/net/sunrpc/xprtrdma/rpc_rdma.c
+++ b/net/sunrpc/xprtrdma/rpc_rdma.c
@@ -1321,9 +1321,9 @@ void rpcrdma_unpin_rqst(struct rpcrdma_rep *rep)
req->rl_reply = NULL;
rep->rr_rqst = NULL;
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
xprt_unpin_rqst(rqst);
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
}
/**
@@ -1363,10 +1363,10 @@ void rpcrdma_complete_rqst(struct rpcrdma_rep *rep)
goto out_badheader;
out:
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
xprt_complete_rqst(rqst->rq_task, status);
xprt_unpin_rqst(rqst);
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
return;
out_badheader:
@@ -1492,12 +1492,12 @@ void rpcrdma_reply_handler(struct rpcrdma_rep *rep)
/* Match incoming rpcrdma_rep to an rpcrdma_req to
* get context for handling any incoming chunks.
*/
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
rqst = xprt_lookup_rqst(xprt, rep->rr_xid);
if (!rqst)
goto out_norqst;
xprt_pin_rqst(rqst);
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
if (buf->rb_credits != credits)
rpcrdma_update_cwnd(r_xprt, credits);
@@ -1524,7 +1524,7 @@ void rpcrdma_reply_handler(struct rpcrdma_rep *rep)
return;
out_norqst:
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
trace_xprtrdma_reply_rqst_err(rep);
rpcrdma_rep_put(buf, rep);
goto out_post;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
index e5a78b761012..3c7b85427f33 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
@@ -28,7 +28,7 @@ void svc_rdma_handle_bc_reply(struct svc_rqst *rqstp,
struct rpc_rqst *req;
u32 credits;
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
req = xprt_lookup_rqst(xprt, *rdma_resp);
if (!req)
goto out_unlock;
@@ -39,7 +39,7 @@ void svc_rdma_handle_bc_reply(struct svc_rqst *rqstp,
goto out_unlock;
memcpy(dst->iov_base, src->iov_base, src->iov_len);
xprt_pin_rqst(req);
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
credits = be32_to_cpup(rdma_resp + 2);
if (credits == 0)
@@ -50,13 +50,13 @@ void svc_rdma_handle_bc_reply(struct svc_rqst *rqstp,
xprt->cwnd = credits << RPC_CWNDSHIFT;
spin_unlock(&xprt->transport_lock);
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
xprt_complete_rqst(req->rq_task, rcvbuf->len);
xprt_unpin_rqst(req);
rcvbuf->len = 0;
out_unlock:
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
}
/* Send a reverse-direction RPC Call.
diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 7f60723fa64d..1454da9575b3 100644
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -673,25 +673,25 @@ xs_read_stream_reply(struct sock_xprt *transport, struct msghdr *msg, int flags)
ssize_t ret = 0;
/* Look up and lock the request corresponding to the given XID */
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
req = xprt_lookup_rqst(xprt, transport->recv.xid);
if (!req || (transport->recv.copied && !req->rq_private_buf.len)) {
msg->msg_flags |= MSG_TRUNC;
goto out;
}
xprt_pin_rqst(req);
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
ret = xs_read_stream_request(transport, msg, flags, req);
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
if (msg->msg_flags & (MSG_EOR|MSG_TRUNC))
xprt_complete_rqst(req->rq_task, transport->recv.copied);
else
req->rq_private_buf.len = transport->recv.copied;
xprt_unpin_rqst(req);
out:
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
return ret;
}
@@ -1398,13 +1398,13 @@ static void xs_udp_data_read_skb(struct rpc_xprt *xprt,
return;
/* Look up and lock the request corresponding to the given XID */
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
rovr = xprt_lookup_rqst(xprt, *xp);
if (!rovr)
goto out_unlock;
xprt_pin_rqst(rovr);
xprt_update_rtt(rovr->rq_task);
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
task = rovr->rq_task;
if ((copied = rovr->rq_private_buf.buflen) > repsize)
@@ -1412,7 +1412,7 @@ static void xs_udp_data_read_skb(struct rpc_xprt *xprt,
/* Suck it into the iovec, verify checksum if not done by hw. */
if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
__UDPX_INC_STATS(sk, UDP_MIB_INERRORS);
goto out_unpin;
}
@@ -1421,13 +1421,13 @@ static void xs_udp_data_read_skb(struct rpc_xprt *xprt,
spin_lock(&xprt->transport_lock);
xprt_adjust_cwnd(xprt, task, copied);
spin_unlock(&xprt->transport_lock);
- spin_lock(&xprt->queue_lock);
+ spin_lock(&xprt->recv_lock);
xprt_complete_rqst(task, copied);
__UDPX_INC_STATS(sk, UDP_MIB_INDATAGRAMS);
out_unpin:
xprt_unpin_rqst(rovr);
out_unlock:
- spin_unlock(&xprt->queue_lock);
+ spin_unlock(&xprt->recv_lock);
}
static void xs_udp_data_receive(struct sock_xprt *transport)
--
2.54.0
^ permalink raw reply [flat|nested] 7+ messages in thread