sunrpc: skip svc_xprt_enqueue when transport is busy

svc_xprt_resource_released() calls svc_xprt_enqueue()
whenever XPT_DATA or XPT_DEFERRED is set. During RPC
processing, svc_reserve_auth() reduces the reservation
counter and triggers this path while the current thread
still holds XPT_BUSY. The enqueue enters svc_xprt_ready(),
executes an smp_rmb(), READ_ONCE(), and tracepoint, then
returns false on seeing XPT_BUSY.

Trace data from a 256KB NFSv3 WRITE workload over TCP
shows this pattern generates roughly 195,000 wasted
enqueue calls -- approximately one per RPC -- each
paying the full svc_xprt_ready() cost for no benefit.

Add a BUSY check alongside the existing DATA|DEFERRED
check in svc_xprt_resource_released(). When the
transport is BUSY, the holder will call
svc_xprt_received() upon completion, which already
checks for pending work flags and re-enqueues.

Reviewed-by: Jeff Layton <jlayton@kernel.org>
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>

diff --git a/net/sunrpc/svc_xprt.c b/net/sunrpc/svc_xprt.c
index 377f5d7490aa403e8c80bc712989e661455357f2..63d1002e63e7476f6361e52eb76701610934f9c8 100644 (file)
--- a/net/sunrpc/svc_xprt.c
+++ b/net/sunrpc/svc_xprt.c
@@ -440,16 +440,23 @@ static bool svc_xprt_reserve_slot(struct svc_rqst *rqstp, struct svc_xprt *xprt)
 /*
  * After a caller releases write-space or a request slot,
  * re-enqueue the transport only when there is pending
- * work that a thread could act on. The smp_mb() pairs
+ * work that a thread could act on.  The smp_mb() pairs
  * with the smp_rmb() in svc_xprt_ready() and orders the
  * preceding counter update before the flags read so a
  * concurrent set_bit(XPT_DATA) is visible here.
+ *
+ * When the transport is BUSY, the thread holding it will
+ * call svc_xprt_received() upon completion, which checks
+ * for pending work and re-enqueues as needed.
  */
 static void svc_xprt_resource_released(struct svc_xprt *xprt)
 {
+       unsigned long xpt_flags;
+
        smp_mb();
-       if (READ_ONCE(xprt->xpt_flags) &
-           (BIT(XPT_DATA) | BIT(XPT_DEFERRED)))
+       xpt_flags = READ_ONCE(xprt->xpt_flags);
+       if (xpt_flags & (BIT(XPT_DATA) | BIT(XPT_DEFERRED)) &&
+           !(xpt_flags & BIT(XPT_BUSY)))
                svc_xprt_enqueue(xprt);
 }