sunrpc: skip svc_xprt_enqueue when no work is pending

svc_reserve() and svc_xprt_release_slot() call
svc_xprt_enqueue() after modifying xpt_reserved or
xpt_nr_rqsts. The purpose is to re-dispatch the
transport when write-space or a slot becomes available.
However, when neither XPT_DATA nor XPT_DEFERRED is
set, no thread can make progress on the transport and
the enqueue accomplishes nothing.

Trace data from a 256KB NFSv3 WRITE workload over RDMA
shows 11.2 svc_xprt_enqueue() calls per RPC. Of these,
6.9 per RPC lack XPT_DATA and exit svc_xprt_ready()
immediately after executing the smp_rmb(), READ_ONCE(),
and tracepoint. svc_reserve() and svc_xprt_release_slot()
account for roughly five of these per RPC.

A new helper, svc_xprt_resource_released(), checks
XPT_DATA | XPT_DEFERRED before calling
svc_xprt_enqueue(). The existing smp_wmb() barriers
are upgraded to smp_mb() to ensure the flags check
observes a concurrent producer's set_bit(XPT_DATA).
Each producer (svc_rdma_wc_receive, etc.) both sets
XPT_DATA and calls svc_xprt_enqueue(), so even if the
check reads a stale value, the producer's own enqueue
provides a fallback path.

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 b16e710926c1ee3c9cfac4283c341dd358ebb71f..b8f3ac29e7186d82d67fc27e0e425b630d050e50 100644 (file)
--- a/net/sunrpc/svc_xprt.c
+++ b/net/sunrpc/svc_xprt.c
@@ -425,13 +425,28 @@ static bool svc_xprt_reserve_slot(struct svc_rqst *rqstp, struct svc_xprt *xprt)
        return true;
 }
 
+/*
+ * 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
+ * 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.
+ */
+static void svc_xprt_resource_released(struct svc_xprt *xprt)
+{
+       smp_mb();
+       if (READ_ONCE(xprt->xpt_flags) &
+           (BIT(XPT_DATA) | BIT(XPT_DEFERRED)))
+               svc_xprt_enqueue(xprt);
+}
+
 static void svc_xprt_release_slot(struct svc_rqst *rqstp)
 {
        struct svc_xprt *xprt = rqstp->rq_xprt;
        if (test_and_clear_bit(RQ_DATA, &rqstp->rq_flags)) {
                atomic_dec(&xprt->xpt_nr_rqsts);
-               smp_wmb(); /* See smp_rmb() in svc_xprt_ready() */
-               svc_xprt_enqueue(xprt);
+               svc_xprt_resource_released(xprt);
        }
 }
 
@@ -525,10 +540,10 @@ void svc_reserve(struct svc_rqst *rqstp, int space)
        space += rqstp->rq_res.head[0].iov_len;
 
        if (xprt && space < rqstp->rq_reserved) {
-               atomic_sub((rqstp->rq_reserved - space), &xprt->xpt_reserved);
+               atomic_sub((rqstp->rq_reserved - space),
+                          &xprt->xpt_reserved);
                rqstp->rq_reserved = space;
-               smp_wmb(); /* See smp_rmb() in svc_xprt_ready() */
-               svc_xprt_enqueue(xprt);
+               svc_xprt_resource_released(xprt);
        }
 }
 EXPORT_SYMBOL_GPL(svc_reserve);