[PATCH RFC 2/3] SUNRPC: dispatch idle transports ahead of backlogged ones

A pool dispatches ready transports in FIFO order, so a connection that
already has requests in flight sits in the same queue, on equal terms,
as one that has been idle.  A client driving many concurrent requests
therefore delays the next request of an interactive client whose
previous reply has already completed: the interactive request waits
behind the busy client's backlog even though servicing it costs a
single round trip.

Route a transport with no requests in flight onto the new high-priority
queue, sp_xprts_hi, and drain that queue ahead of sp_xprts.  xpt_nr_rqsts
is incremented when a thread reserves a request slot and decremented when
the reply completes, so at enqueue time it counts the transport's prior
in-flight work, not the request that just arrived: a flow whose last
reply has completed reads zero and jumps the queue, while a flow with
requests still in flight stays in the bulk queue.  The classification
needs no client identity and no lock -- each queue is an independent lwq
with its own ordering, so svc_thread_should_sleep() simply tests both.

This gives an idle flow's request a latency floor that a backlogged flow
cannot push it below.  It is starvation avoidance, not proportional
fairness: under sustained load across many idle flows the bulk queue can
wait, and bounding that is left to later work.

Signed-off-by: Benjamin Coddington <bcodding-F/[email protected]>
---
 net/sunrpc/svc_xprt.c | 44 +++++++++++++++++++++++++++----------------
 1 file changed, 28 insertions(+), 16 deletions(-)

diff --git a/net/sunrpc/svc_xprt.c b/net/sunrpc/svc_xprt.c
index 63d1002e63e7..ec4c05094e9a 100644
--- a/net/sunrpc/svc_xprt.c
+++ b/net/sunrpc/svc_xprt.c
@@ -523,7 +523,10 @@ void svc_xprt_enqueue(struct svc_xprt *xprt)
 
 	percpu_counter_inc(&pool->sp_sockets_queued);
 	xprt->xpt_qtime = ktime_get();
-	lwq_enqueue(&xprt->xpt_ready, &pool->sp_xprts);
+	if (atomic_read(&xprt->xpt_nr_rqsts))
+		lwq_enqueue(&xprt->xpt_ready, &pool->sp_xprts);
+	else
+		lwq_enqueue(&xprt->xpt_ready, &pool->sp_xprts_hi);
 
 	svc_pool_wake_idle_thread(pool);
 }
@@ -536,7 +539,9 @@ static struct svc_xprt *svc_xprt_dequeue(struct svc_pool *pool)
 {
 	struct svc_xprt	*xprt = NULL;
 
-	xprt = lwq_dequeue(&pool->sp_xprts, struct svc_xprt, xpt_ready);
+	xprt = lwq_dequeue(&pool->sp_xprts_hi, struct svc_xprt, xpt_ready);
+	if (!xprt)
+		xprt = lwq_dequeue(&pool->sp_xprts, struct svc_xprt, xpt_ready);
 	if (xprt)
 		svc_xprt_get(xprt);
 	return xprt;
@@ -759,7 +764,7 @@ svc_thread_should_sleep(struct svc_rqst *rqstp)
 		return false;
 
 	/* was a socket queued? */
-	if (!lwq_empty(&pool->sp_xprts))
+	if (!lwq_empty(&pool->sp_xprts_hi) || !lwq_empty(&pool->sp_xprts))
 		return false;
 
 	/* are we shutting down? */
@@ -1183,26 +1188,33 @@ static int svc_close_list(struct svc_serv *serv, struct list_head *xprt_list, st
 	return ret;
 }
 
-static void svc_clean_up_xprts(struct svc_serv *serv, struct net *net)
+static void svc_clean_up_queue(struct lwq *queue, struct net *net)
 {
 	struct svc_xprt *xprt;
+	struct llist_node *q, **t1, *t2;
+
+	q = lwq_dequeue_all(queue);
+	lwq_for_each_safe(xprt, t1, t2, &q, xpt_ready) {
+		if (xprt->xpt_net == net) {
+			set_bit(XPT_CLOSE, &xprt->xpt_flags);
+			svc_delete_xprt(xprt);
+			xprt = NULL;
+		}
+	}
+
+	if (q)
+		lwq_enqueue_batch(q, queue);
+}
+
+static void svc_clean_up_xprts(struct svc_serv *serv, struct net *net)
+{
 	int i;
 
 	for (i = 0; i < serv->sv_nrpools; i++) {
 		struct svc_pool *pool = &serv->sv_pools[i];
-		struct llist_node *q, **t1, *t2;
-
-		q = lwq_dequeue_all(&pool->sp_xprts);
-		lwq_for_each_safe(xprt, t1, t2, &q, xpt_ready) {
-			if (xprt->xpt_net == net) {
-				set_bit(XPT_CLOSE, &xprt->xpt_flags);
-				svc_delete_xprt(xprt);
-				xprt = NULL;
-			}
-		}
 
-		if (q)
-			lwq_enqueue_batch(q, &pool->sp_xprts);
+		svc_clean_up_queue(&pool->sp_xprts_hi, net);
+		svc_clean_up_queue(&pool->sp_xprts, net);
 	}
 }
 
-- 
2.53.0