Rather than searching a list of threads to find an idle one, having a
list of idle threads allows an idle thread to be found immediately.
This adds some spin_lock calls which is not ideal, but as the hold-time
is tiny it is still faster than searching a list. A future patch will
remove them using llist.h. This involves some subtlety and so is left
to a separate patch.
Signed-off-by: NeilBrown <neilb@xxxxxxx>
---
include/linux/sunrpc/svc.h | 25 ++++++++++++++++++++++++-
include/trace/events/sunrpc.h | 1 -
net/sunrpc/svc.c | 13 ++++++++-----
net/sunrpc/svc_xprt.c | 17 ++++++++++++-----
4 files changed, 44 insertions(+), 12 deletions(-)
diff --git a/include/linux/sunrpc/svc.h b/include/linux/sunrpc/svc.h
index b39c613fbe06..ec8088d5b57f 100644
--- a/include/linux/sunrpc/svc.h
+++ b/include/linux/sunrpc/svc.h
@@ -36,6 +36,7 @@ struct svc_pool {
struct list_head sp_sockets; /* pending sockets */
unsigned int sp_nrthreads; /* # of threads in pool */
struct list_head sp_all_threads; /* all server threads */
+ struct list_head sp_idle_threads; /* idle server threads */
/* statistics on pool operation */
struct percpu_counter sp_messages_arrived;
@@ -199,6 +200,7 @@ extern u32 svc_max_payload(const struct svc_rqst *rqstp);
*/
struct svc_rqst {
struct list_head rq_all; /* all threads list */
+ struct list_head rq_idle; /* On the idle list */
struct rcu_head rq_rcu_head; /* for RCU deferred kfree */
struct svc_xprt * rq_xprt; /* transport ptr */
@@ -278,10 +280,31 @@ enum {
RQ_SPLICE_OK, /* turned off in gss privacy to prevent encrypting page
* cache pages */
RQ_VICTIM, /* Have agreed to shut down */
- RQ_BUSY, /* request is busy */
RQ_DATA, /* request has data */
};
+/**
+ * svc_thread_set_busy - mark a thread as busy
+ * @rqstp: the thread which is now busy
+ *
+ * If rq_idle is "empty", the thread must be busy.
+ */
+static inline void svc_thread_set_busy(struct svc_rqst *rqstp)
+{
+ INIT_LIST_HEAD(&rqstp->rq_idle);
+}
+
+/**
+ * svc_thread_busy - check if a thread as busy
+ * @rqstp: the thread which might be busy
+ *
+ * If rq_idle is "empty", the thread must be busy.
+ */
+static inline bool svc_thread_busy(struct svc_rqst *rqstp)
+{
+ return list_empty(&rqstp->rq_idle);
+}
+
#define SVC_NET(rqst) (rqst->rq_xprt ? rqst->rq_xprt->xpt_net : rqst->rq_bc_net)
/*
diff --git a/include/trace/events/sunrpc.h b/include/trace/events/sunrpc.h
index c79375e37dc2..dd0323c40ef5 100644
--- a/include/trace/events/sunrpc.h
+++ b/include/trace/events/sunrpc.h
@@ -1677,7 +1677,6 @@ DEFINE_SVCXDRBUF_EVENT(sendto);
svc_rqst_flag(DROPME) \
svc_rqst_flag(SPLICE_OK) \
svc_rqst_flag(VICTIM) \
- svc_rqst_flag(BUSY) \
svc_rqst_flag_end(DATA)
#undef svc_rqst_flag
diff --git a/net/sunrpc/svc.c b/net/sunrpc/svc.c
index fd49e7b12c94..028de8f662a8 100644
--- a/net/sunrpc/svc.c
+++ b/net/sunrpc/svc.c
@@ -644,7 +644,7 @@ svc_rqst_alloc(struct svc_serv *serv, struct svc_pool *pool, int node)
folio_batch_init(&rqstp->rq_fbatch);
- __set_bit(RQ_BUSY, &rqstp->rq_flags);
+ svc_thread_set_busy(rqstp);
rqstp->rq_server = serv;
rqstp->rq_pool = pool;
@@ -704,10 +704,13 @@ void svc_pool_wake_idle_thread(struct svc_serv *serv,
struct svc_rqst *rqstp;
rcu_read_lock();
- list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {
- if (test_and_set_bit(RQ_BUSY, &rqstp->rq_flags))
- continue;
-
+ spin_lock_bh(&pool->sp_lock);
+ rqstp = list_first_entry_or_null(&pool->sp_idle_threads,
+ struct svc_rqst, rq_idle);
+ if (rqstp)
+ list_del_init(&rqstp->rq_idle);
+ spin_unlock_bh(&pool->sp_lock);
+ if (rqstp) {
WRITE_ONCE(rqstp->rq_qtime, ktime_get());
wake_up_process(rqstp->rq_task);
rcu_read_unlock();
diff --git a/net/sunrpc/svc_xprt.c b/net/sunrpc/svc_xprt.c
index 964c97dbb36c..1d134415ae3f 100644
--- a/net/sunrpc/svc_xprt.c
+++ b/net/sunrpc/svc_xprt.c
@@ -736,18 +736,25 @@ static void svc_wait_for_work(struct svc_rqst *rqstp)
set_current_state(TASK_IDLE);
smp_mb__before_atomic();
clear_bit(SP_CONGESTED, &pool->sp_flags);
- clear_bit(RQ_BUSY, &rqstp->rq_flags);
- smp_mb__after_atomic();
+ spin_lock_bh(&pool->sp_lock);
+ list_add(&rqstp->rq_idle, &pool->sp_idle_threads);
+ spin_unlock_bh(&pool->sp_lock);
if (likely(rqst_should_sleep(rqstp)))
schedule();
else
__set_current_state(TASK_RUNNING);
- try_to_freeze();
+ /* We *must* be removed from the list before we can continue.
+ * If we were woken, this is already done
+ */
+ if (!svc_thread_busy(rqstp)) {
+ spin_lock_bh(&pool->sp_lock);
+ list_del_init(&rqstp->rq_idle);
+ spin_unlock_bh(&pool->sp_lock);
+ }
- set_bit(RQ_BUSY, &rqstp->rq_flags);
- smp_mb__after_atomic();
+ try_to_freeze();
rqstp->rq_xprt = svc_xprt_dequeue(pool);
if (rqstp->rq_xprt) {
