[PATCH] srcu: switch work func to allow concurrent gp

[Pingfan Liu <kernelfans@gmail.com>]

ssp->srcu_cb_mutex is introduced to allow the other srcu state machine
to advance as soon as possible. But according to the implement of
workqueue, the same work_struct is serialized and can not run
concurrently in fact.

Quoting from Documentation/core-api/workqueue.rst
"
Non-reentrance Conditions
=========================

Workqueue guarantees that a work item cannot be re-entrant if the following
conditions hold after a work item gets queued:

        1. The work function hasn't been changed.
        2. No one queues the work item to another workqueue.
        3. The work item hasn't been reinitiated.
"

To allow the concurrence to some extent, it can be achieved by changing
the work function to break the conditions. As a result, when
srcu_gp_end() releases srcu_gp_mutex, a new state machine can begin.

Signed-off-by: Pingfan Liu <kernelfans@gmail.com>
Cc: Lai Jiangshan <jiangshanlai@gmail.com>
Cc: "Paul E. McKenney" <paulmck@kernel.org>
Cc: Frederic Weisbecker <frederic@kernel.org>
Cc: Josh Triplett <josh@joshtriplett.org>
Cc: Steven Rostedt <rostedt@goodmis.org>
Cc: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Cc: "Zhang, Qiang1" <qiang1.zhang@intel.com>
To: rcu@vger.kernel.org
---
 kernel/rcu/srcutree.c | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

diff --git a/kernel/rcu/srcutree.c b/kernel/rcu/srcutree.c
index 1c304fec89c0..56dd9bb2c8b8 100644
--- a/kernel/rcu/srcutree.c
+++ b/kernel/rcu/srcutree.c
@@ -75,6 +75,7 @@ static bool __read_mostly srcu_init_done;
 static void srcu_invoke_callbacks(struct work_struct *work);
 static void srcu_reschedule(struct srcu_struct *ssp, unsigned long delay);
 static void process_srcu(struct work_struct *work);
+static void process_srcu_wrap(struct work_struct *work);
 static void srcu_delay_timer(struct timer_list *t);
 
 /* Wrappers for lock acquisition and release, see raw_spin_lock_rcu_node(). */
@@ -763,6 +764,11 @@ static void srcu_gp_end(struct srcu_struct *ssp)
 		cbdelay = 0;
 
 	WRITE_ONCE(ssp->srcu_last_gp_end, ktime_get_mono_fast_ns());
+	/* Change work func so work can be concurrent */
+	if (ssp->work.work.func == process_srcu_wrap)
+		ssp->work.work.func = process_srcu;
+	else
+		ssp->work.work.func = process_srcu_wrap;
 	rcu_seq_end(&ssp->srcu_gp_seq);
 	gpseq = rcu_seq_current(&ssp->srcu_gp_seq);
 	if (ULONG_CMP_LT(ssp->srcu_gp_seq_needed_exp, gpseq))
@@ -1637,6 +1643,19 @@ static void process_srcu(struct work_struct *work)
 	srcu_reschedule(ssp, curdelay);
 }
 
+/*
+ * The ssp->work is expected to be concurrent to some extent, but the current
+ * workqueue does not support the concurrence on the same work. (Refer to the
+ * section "Non-reentrance Conditions" in the file workqueue.rst)
+ * Resolving it by changing the work func.
+ *
+ * Prevent compilering from optimizing out it.
+ */
+static __used void process_srcu_wrap(struct work_struct *work)
+{
+	process_srcu(work);
+}
+
 void srcutorture_get_gp_data(enum rcutorture_type test_type,
 			     struct srcu_struct *ssp, int *flags,
 			     unsigned long *gp_seq)
-- 
2.31.1