* [PATCH 1/7] rcu: Make call_rcu() safe to call from any context
2026-09-19 0:32 [PATCH 0/7] Allow call_{s,}rcu() from NMI and BPF environments Paul E. McKenney
@ 2026-09-19 0:32 ` Paul E. McKenney
2026-09-19 0:32 ` [PATCH 2/7] rcu: Make Tiny " Paul E. McKenney
` (5 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Paul E. McKenney @ 2026-09-19 0:32 UTC (permalink / raw)
To: rcu; +Cc: linux-kernel, kernel-team, rostedt, Puranjay Mohan, Paul E . McKenney
From: Puranjay Mohan <puranjay@kernel.org>
RCU's per-CPU callback list is only touched with interrupts disabled: the
enqueue runs under local_irq_save() (and the nocb locks when offloaded),
as do callback invocation and grace-period work. A call_rcu() that
arrives with interrupts already disabled, whether from an NMI or from
instrumentation that re-enters RCU, can interrupt one of those and corrupt
the list or deadlock.
Defer instead: stage the callback on a per-CPU llist and raise an irq_work
that re-issues it once interrupts are on, straight to the enqueue so it
cannot defer again. The gate is bare irqs_disabled(), so callers that
merely hold interrupts off are deferred too and pay one irq_work hop.
Skip it while the scheduler is down (RCU_SCHEDULER_INACTIVE): irq_work is
not usable that early, rcu_init() already calls call_rcu(), and the per-CPU
deferral state is not initialised until rcu_init_one() runs later in it.
rcu_barrier() drains every CPU's ->defer_head before it scans the lists,
and rcutree_migrate_callbacks() drains an outgoing CPU's. A drain
re-issues onto the draining CPU, so a barrier moves other CPUs' staged
callbacks onto
its own ->cblist; call_rcu() promises no CPU affinity for invocation.
->defer_lock is held across llist_del_all() and the whole re-issue so the
drainers
serialize: one that finds the list empty can conclude that everything
staged before it is already on a callback list. Interrupts stay off for
the batch. Where the arch has an irq_work self-IPI that is what one
interrupts-disabled region could stage, normally a single callback; where
arch_irq_work_has_interrupt() is false the drain waits for the tick, so
several regions can accumulate first.
The drain clears ->next before re-issuing. A double call_rcu() on a head
that is already debug-object-active self-links the staged node, and
rcu_do_enqueue()'s duplicate path returns without clearing it, so the
drain would spin. A re-add behind other staged callbacks makes a longer
cycle, which that does not bound; a double call_rcu() stays undefined.
llist_del_all() yields newest-first, so a batch is re-issued in reverse
call order; nothing depends on call_rcu() ordering. The re-issue drops
the lazy hint, since staging records only ->func, so a deferred callback
loses its batching on CONFIG_RCU_LAZY. kasan_record_aux_stack() moves to
__call_rcu_common() so a use-after-free report names the caller rather
than the irq_work.
The re-issue runs with interrupts disabled, so instrumentation on the
enqueue path can re-enter call_rcu(), stage another callback and re-raise
the irq_work, livelocking the drain. A per-CPU flag guards it: a deferral
that arrives while this CPU is draining, and is not from an NMI, is
dropped. The WARN_ONCE() is under CONFIG_PROVE_RCU, so a production kernel
drops it silently. That leaks the callback and can strand state
the caller tied to it, since a one-shot flag only the callback clears
never resets, but the alternative is an unbounded loop.
A callback deferred past the CPUHP_AP_SMPCFD_DYING irq_work flush leaves
->defer_work claimed with its self-IPI lost. rcutree_migrate_callbacks()
still re-issues the callback, but the first deferral after that CPU comes
back raises no IPI and waits for the next irq_work there, or for
rcu_barrier(). Unqueueing an irq_work is not something the API offers.
The irq_work is IRQ_WORK_INIT_HARD so the re-issue stays prompt on
PREEMPT_RT, where a non-HARD irq_work runs in a kthread that can be
delayed under load. A hidden CONFIG_RCU_DEFER gates the deferral code and
its IRQ_WORK dependency, though the rcu_data members are unconditional;
without it call_rcu() enqueues directly as before. Under
CONFIG_PROVE_RCU, warn if the direct path is reached from an NMI.
Suggested-by: Paul E. McKenney <paulmck@kernel.org>
Signed-off-by: Puranjay Mohan <puranjay@kernel.org>
Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
---
kernel/rcu/Kconfig | 6 +++
kernel/rcu/rcu.h | 11 ++++
kernel/rcu/tree.c | 131 +++++++++++++++++++++++++++++++++++++++++----
kernel/rcu/tree.h | 6 +++
4 files changed, 143 insertions(+), 11 deletions(-)
diff --git a/kernel/rcu/Kconfig b/kernel/rcu/Kconfig
index 332df7a7a634..bc9483c0b5d8 100644
--- a/kernel/rcu/Kconfig
+++ b/kernel/rcu/Kconfig
@@ -175,6 +175,12 @@ config RCU_STALL_COMMON
config RCU_NEED_SEGCBLIST
def_bool ( TREE_RCU || TREE_SRCU || TASKS_RCU_GENERIC )
+# The deferral (and the IRQ_WORK it uses) is only needed where call_rcu() /
+# call_srcu() can be invoked while a callback-list operation is in flight.
+config RCU_DEFER
+ def_bool HAVE_NMI || KPROBES || FUNCTION_TRACER || TRACEPOINTS
+ select IRQ_WORK
+
config RCU_FANOUT
int "Tree-based hierarchical RCU fanout value"
range 2 64 if 64BIT
diff --git a/kernel/rcu/rcu.h b/kernel/rcu/rcu.h
index 39a9f6fa9a7b..91e33571a554 100644
--- a/kernel/rcu/rcu.h
+++ b/kernel/rcu/rcu.h
@@ -572,6 +572,17 @@ static inline void tasks_cblist_init_generic(void) { }
#define RCU_SCHEDULER_INIT 1
#define RCU_SCHEDULER_RUNNING 2
+/*
+ * Defer whenever interrupts are disabled, since a callback-list operation may
+ * be in flight on this CPU. Not before the scheduler is up: irq_work is not
+ * usable that early, and rcu_init() itself calls call_rcu().
+ */
+static inline bool should_rcu_defer(void)
+{
+ return IS_ENABLED(CONFIG_RCU_DEFER) && irqs_disabled() &&
+ rcu_scheduler_active != RCU_SCHEDULER_INACTIVE;
+}
+
enum rcutorture_type {
RCU_FLAVOR,
RCU_TASKS_FLAVOR,
diff --git a/kernel/rcu/tree.c b/kernel/rcu/tree.c
index 96848fc1f02b..ff9a2395c9e8 100644
--- a/kernel/rcu/tree.c
+++ b/kernel/rcu/tree.c
@@ -24,6 +24,7 @@
#include <linux/smp.h>
#include <linux/rcupdate_wait.h>
#include <linux/interrupt.h>
+#include <linux/llist.h>
#include <linux/sched.h>
#include <linux/sched/debug.h>
#include <linux/nmi.h>
@@ -3148,21 +3149,19 @@ static void check_cb_ovld(struct rcu_data *rdp)
raw_spin_unlock_rcu_node(rnp);
}
-static void
-__call_rcu_common(struct rcu_head *head, rcu_callback_t func, bool lazy_in)
+/*
+ * Also called by __rcu_defer_drain() to re-issue a deferred callback, so it
+ * must not re-check the deferral condition. Either caller may have interrupts
+ * already disabled, and a drain of a remote CPU re-issues onto the draining
+ * CPU.
+ */
+static void rcu_do_enqueue(struct rcu_head *head, rcu_callback_t func, bool lazy_in)
{
static atomic_t doublefrees;
unsigned long flags;
bool lazy;
struct rcu_data *rdp;
- /* Misaligned rcu_head! */
- WARN_ON_ONCE((unsigned long)head & (sizeof(void *) - 1));
-
- /* Avoid NULL dereference if callback is NULL. */
- if (WARN_ON_ONCE(!func))
- return;
-
if (debug_rcu_head_queue(head)) {
/*
* Probable double call_rcu(), so leak the callback.
@@ -3178,7 +3177,6 @@ __call_rcu_common(struct rcu_head *head, rcu_callback_t func, bool lazy_in)
}
head->func = func;
head->next = NULL;
- kasan_record_aux_stack(head);
local_irq_save(flags);
rdp = this_cpu_ptr(&rcu_data);
@@ -3206,6 +3204,103 @@ __call_rcu_common(struct rcu_head *head, rcu_callback_t func, bool lazy_in)
local_irq_restore(flags);
}
+/*
+ * Re-issue deferred callbacks straight to the enqueue so they cannot defer
+ * again. ->defer_lock serializes the drainers: this CPU's irq_work,
+ * rcu_defer_flush() and rcutree_migrate_callbacks().
+ */
+static void __rcu_defer_drain(struct rcu_data *rdp)
+{
+ struct llist_node *node, *next;
+ unsigned long flags;
+
+ if (!IS_ENABLED(CONFIG_RCU_DEFER))
+ return;
+
+ raw_spin_lock_irqsave(&rdp->defer_lock, flags);
+ llist_for_each_safe(node, next, llist_del_all(&rdp->defer_head)) {
+ struct rcu_head *head = (struct rcu_head *)node;
+
+ /* Bounds a node self-linked by a double call_rcu(). */
+ head->next = NULL;
+ rcu_do_enqueue(head, head->func, false);
+ }
+ raw_spin_unlock_irqrestore(&rdp->defer_lock, flags);
+}
+
+/*
+ * Only the irq_work drain can be re-fed by its own re-issue, so only it sets
+ * ->defer_draining. Anything staged during a direct drain is picked up by the
+ * staging CPU's own irq_work. Every caller of irq_work_run_list() has
+ * interrupts disabled, so the flag is never visible with them enabled.
+ */
+static void rcu_defer_drain(struct irq_work *iw)
+{
+ struct rcu_data *rdp = container_of(iw, struct rcu_data, defer_work);
+
+ WRITE_ONCE(rdp->defer_draining, true);
+ __rcu_defer_drain(rdp);
+ WRITE_ONCE(rdp->defer_draining, false);
+}
+
+/*
+ * Stage @head for this CPU's irq_work to re-issue once interrupts are on. Only
+ * the drain side takes a lock, so this stays safe from NMI.
+ */
+static void call_rcu_defer(struct rcu_head *head, rcu_callback_t func)
+{
+ struct rcu_data *rdp = this_cpu_ptr(&rcu_data);
+
+ /*
+ * Instrumentation on the enqueue path can re-enter here from inside the
+ * drain. Re-queuing would livelock it, so drop the callback; an NMI
+ * cannot loop, so let it through.
+ */
+ if (READ_ONCE(rdp->defer_draining) && !in_nmi()) {
+ WARN_ONCE(IS_ENABLED(CONFIG_PROVE_RCU),
+ "call_rcu() re-entered during callback drain; leaking callback\n");
+ return;
+ }
+ head->func = func;
+ if (llist_add((struct llist_node *)head, &rdp->defer_head))
+ irq_work_queue(&rdp->defer_work);
+}
+
+static void rcu_defer_flush(void)
+{
+ int cpu;
+
+ for_each_possible_cpu(cpu)
+ __rcu_defer_drain(per_cpu_ptr(&rcu_data, cpu));
+}
+
+static void
+__call_rcu_common(struct rcu_head *head, rcu_callback_t func, bool lazy_in)
+{
+ /* Misaligned rcu_head! */
+ WARN_ON_ONCE((unsigned long)head & (sizeof(void *) - 1));
+
+ /* Avoid NULL dereference if callback is NULL. */
+ if (WARN_ON_ONCE(!func))
+ return;
+
+ /* Record the caller: the irq_work's stack says nothing about it. */
+ kasan_record_aux_stack(head);
+
+ if (should_rcu_defer()) {
+ call_rcu_defer(head, func);
+ return;
+ }
+
+ /*
+ * Only reachable from an NMI when deferral is off: before the scheduler
+ * is up, or with CONFIG_RCU_DEFER=n. The enqueue can then race.
+ */
+ WARN_ON_ONCE(IS_ENABLED(CONFIG_PROVE_RCU) && in_nmi());
+
+ rcu_do_enqueue(head, func, lazy_in);
+}
+
#ifdef CONFIG_RCU_LAZY
static bool enable_rcu_lazy __read_mostly = !IS_ENABLED(CONFIG_RCU_LAZY_DEFAULT_OFF);
module_param(enable_rcu_lazy, bool, 0444);
@@ -3896,8 +3991,12 @@ void rcu_barrier(void)
unsigned long flags;
unsigned long gseq;
struct rcu_data *rdp;
- unsigned long s = rcu_seq_snap(&rcu_state.barrier_sequence);
+ unsigned long s;
+ /* Register any deferred callbacks before snapshotting the sequence. */
+ rcu_defer_flush();
+
+ s = rcu_seq_snap(&rcu_state.barrier_sequence);
rcu_barrier_trace(TPS("Begin"), -1, s);
/* Take mutex to serialize concurrent rcu_barrier() requests. */
@@ -4231,6 +4330,9 @@ rcu_boot_init_percpu_data(int cpu)
rdp->rcu_onl_gp_state = RCU_GP_CLEANED;
rdp->last_sched_clock = jiffies;
rdp->cpu = cpu;
+ init_llist_head(&rdp->defer_head);
+ raw_spin_lock_init(&rdp->defer_lock);
+ rdp->defer_work = IRQ_WORK_INIT_HARD(rcu_defer_drain);
rcu_boot_init_nocb_percpu_data(rdp);
}
@@ -4528,6 +4630,13 @@ void rcutree_migrate_callbacks(int cpu)
struct rcu_data *rdp = per_cpu_ptr(&rcu_data, cpu);
bool needwake;
+ /*
+ * Callbacks deferred past the point the outgoing CPU's irq_work can run
+ * sit on ->defer_head, which the ->cblist migration below does not
+ * cover. Drain them here, before the early returns.
+ */
+ __rcu_defer_drain(rdp);
+
if (rcu_rdp_is_offloaded(rdp))
return;
diff --git a/kernel/rcu/tree.h b/kernel/rcu/tree.h
index eedfa43059e8..b7cac7a13b4f 100644
--- a/kernel/rcu/tree.h
+++ b/kernel/rcu/tree.h
@@ -229,6 +229,12 @@ struct rcu_data {
struct rcu_head barrier_head;
int exp_watching_snap; /* Double-check need for IPI. */
+ /* Deferral of an NMI/reentrant call_rcu(); see __call_rcu_common(). */
+ struct llist_head defer_head;
+ struct irq_work defer_work;
+ raw_spinlock_t defer_lock;
+ bool defer_draining;
+
/* 5) Callback offloading. */
#ifdef CONFIG_RCU_NOCB_CPU
struct swait_queue_head nocb_cb_wq; /* For nocb kthreads to sleep on. */
--
2.40.1
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 2/7] rcu: Make Tiny call_rcu() safe to call from any context
2026-09-19 0:32 [PATCH 0/7] Allow call_{s,}rcu() from NMI and BPF environments Paul E. McKenney
2026-09-19 0:32 ` [PATCH 1/7] rcu: Make call_rcu() safe to call from any context Paul E. McKenney
@ 2026-09-19 0:32 ` Paul E. McKenney
2026-09-19 0:32 ` [PATCH 3/7] srcu: Make call_srcu() " Paul E. McKenney
` (4 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Paul E. McKenney @ 2026-09-19 0:32 UTC (permalink / raw)
To: rcu; +Cc: linux-kernel, kernel-team, rostedt, Puranjay Mohan, Paul E . McKenney
From: Puranjay Mohan <puranjay@kernel.org>
Give Tiny call_rcu() the same treatment as Tree RCU. When interrupts are
disabled and the scheduler is up, stage the callback on a lockless list
that an irq_work re-issues later. One global list and irq_work suffice
since Tiny RCU is uniprocessor, and there is no CPU-offline drain.
The re-issue runs with interrupts disabled and can be re-entered by
instrumentation, so a draining flag drops a deferring call_rcu() seen
mid-drain (unless from an NMI), as in Tree RCU. Gated by CONFIG_RCU_DEFER,
though the deferral state is unconditional.
Interrupts stay off for the whole batch, but the re-issue is a tail append
with no locks. TINY_RCU implies !SMP, where arch_irq_work_has_interrupt()
is false, so the drain always waits for the tick and a batch is whatever
one tick's worth of interrupts-disabled call_rcu()s staged. As in Tree
RCU the drain clears ->next before re-issuing, which bounds a node
self-linked by a double call_rcu(): rcu_do_enqueue()'s duplicate path
returns without clearing it. A longer cycle is not bounded; a double
call_rcu() stays undefined.
The idle-task reschedule moves out of the enqueue helper so that a drain
does it once for the batch rather than once per callback, which would
otherwise take the runqueue lock N times with interrupts disabled.
Suggested-by: Paul E. McKenney <paulmck@kernel.org>
Signed-off-by: Puranjay Mohan <puranjay@kernel.org>
Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
---
kernel/rcu/tiny.c | 127 +++++++++++++++++++++++++++++++++++++---------
1 file changed, 104 insertions(+), 23 deletions(-)
diff --git a/kernel/rcu/tiny.c b/kernel/rcu/tiny.c
index dccccd6be941..656b6a682e31 100644
--- a/kernel/rcu/tiny.c
+++ b/kernel/rcu/tiny.c
@@ -11,6 +11,8 @@
*/
#include <linux/completion.h>
#include <linux/interrupt.h>
+#include <linux/irq_work.h>
+#include <linux/llist.h>
#include <linux/notifier.h>
#include <linux/rcupdate_wait.h>
#include <linux/kernel.h>
@@ -42,8 +44,100 @@ static struct rcu_ctrlblk rcu_ctrlblk = {
.gp_seq = 0 - 300UL,
};
+/*
+ * The callback list is only accessed with interrupts disabled, so a call_rcu()
+ * that arrives with interrupts off stages the callback on a lockless list that
+ * an irq_work re-issues later. One global list and irq_work suffice, as Tiny
+ * RCU is uniprocessor.
+ */
+static void rcu_defer_drain(struct irq_work *iw);
+static LLIST_HEAD(rcu_defer_list);
+static struct irq_work rcu_defer_iw = IRQ_WORK_INIT_HARD(rcu_defer_drain);
+static bool rcu_defer_draining;
+
+/*
+ * Also called by __rcu_defer_drain() to re-issue a deferred callback, so it
+ * must not re-check the deferral condition.
+ */
+static void rcu_do_enqueue(struct rcu_head *head, rcu_callback_t func)
+{
+ static atomic_t doublefrees;
+ unsigned long flags;
+
+ if (debug_rcu_head_queue(head)) {
+ if (atomic_inc_return(&doublefrees) < 4) {
+ pr_err("%s(): Double-freed CB %p->%pS()!!! ", __func__, head, head->func);
+ mem_dump_obj(head);
+ }
+ return;
+ }
+
+ head->func = func;
+ head->next = NULL;
+
+ local_irq_save(flags);
+ *rcu_ctrlblk.curtail = head;
+ rcu_ctrlblk.curtail = &head->next;
+ local_irq_restore(flags);
+}
+
+/* Force scheduling for rcu_qs() when enqueuing from the idle task. */
+static void rcu_resched_if_idle(void)
+{
+ if (unlikely(is_idle_task(current)))
+ resched_cpu(0);
+}
+
+static void __rcu_defer_drain(void)
+{
+ struct llist_node *node, *next;
+ bool drained = false;
+ unsigned long flags;
+
+ if (!IS_ENABLED(CONFIG_RCU_DEFER))
+ return;
+
+ /* Re-issued newest-first; nothing depends on call_rcu() ordering. */
+ local_irq_save(flags);
+ llist_for_each_safe(node, next, llist_del_all(&rcu_defer_list)) {
+ struct rcu_head *head = (struct rcu_head *)node;
+
+ /* Bounds a node self-linked by a double call_rcu(). */
+ head->next = NULL;
+ rcu_do_enqueue(head, head->func);
+ drained = true;
+ }
+ local_irq_restore(flags);
+
+ if (drained)
+ rcu_resched_if_idle();
+}
+
+/* Only the irq_work drain can be re-fed by its own re-issue; see Tree RCU. */
+static void rcu_defer_drain(struct irq_work *iw)
+{
+ WRITE_ONCE(rcu_defer_draining, true);
+ __rcu_defer_drain();
+ WRITE_ONCE(rcu_defer_draining, false);
+}
+
+static void call_rcu_defer(struct rcu_head *head, rcu_callback_t func)
+{
+ /* A re-entrant call_rcu() during the drain would livelock it; drop it. */
+ if (READ_ONCE(rcu_defer_draining) && !in_nmi()) {
+ WARN_ONCE(IS_ENABLED(CONFIG_PROVE_RCU),
+ "call_rcu() re-entered during callback drain; leaking callback\n");
+ return;
+ }
+ head->func = func;
+ if (llist_add((struct llist_node *)head, &rcu_defer_list))
+ irq_work_queue(&rcu_defer_iw);
+}
+
void rcu_barrier(void)
{
+ /* Register any deferred callbacks so the wait below covers them. */
+ __rcu_defer_drain();
wait_rcu_gp(call_rcu_hurry);
}
EXPORT_SYMBOL(rcu_barrier);
@@ -157,29 +251,19 @@ EXPORT_SYMBOL_GPL(synchronize_rcu);
*/
void call_rcu(struct rcu_head *head, rcu_callback_t func)
{
- static atomic_t doublefrees;
- unsigned long flags;
-
- if (debug_rcu_head_queue(head)) {
- if (atomic_inc_return(&doublefrees) < 4) {
- pr_err("%s(): Double-freed CB %p->%pS()!!! ", __func__, head, head->func);
- mem_dump_obj(head);
- }
+ if (should_rcu_defer()) {
+ call_rcu_defer(head, func);
return;
}
- head->func = func;
- head->next = NULL;
+ /*
+ * Only reachable from an NMI when deferral is off: before the scheduler
+ * is up, or with CONFIG_RCU_DEFER=n. The enqueue can then race.
+ */
+ WARN_ON_ONCE(IS_ENABLED(CONFIG_PROVE_RCU) && in_nmi());
- local_irq_save(flags);
- *rcu_ctrlblk.curtail = head;
- rcu_ctrlblk.curtail = &head->next;
- local_irq_restore(flags);
-
- if (unlikely(is_idle_task(current))) {
- /* force scheduling for rcu_qs() */
- resched_cpu(0);
- }
+ rcu_do_enqueue(head, func);
+ rcu_resched_if_idle();
}
EXPORT_SYMBOL_GPL(call_rcu);
@@ -211,10 +295,7 @@ unsigned long start_poll_synchronize_rcu(void)
{
unsigned long gp_seq = get_state_synchronize_rcu();
- if (unlikely(is_idle_task(current))) {
- /* force scheduling for rcu_qs() */
- resched_cpu(0);
- }
+ rcu_resched_if_idle();
return gp_seq;
}
EXPORT_SYMBOL_GPL(start_poll_synchronize_rcu);
--
2.40.1
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 3/7] srcu: Make call_srcu() safe to call from any context
2026-09-19 0:32 [PATCH 0/7] Allow call_{s,}rcu() from NMI and BPF environments Paul E. McKenney
2026-09-19 0:32 ` [PATCH 1/7] rcu: Make call_rcu() safe to call from any context Paul E. McKenney
2026-09-19 0:32 ` [PATCH 2/7] rcu: Make Tiny " Paul E. McKenney
@ 2026-09-19 0:32 ` Paul E. McKenney
2026-09-19 0:32 ` [PATCH 4/7] srcu: Make Tiny " Paul E. McKenney
` (3 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Paul E. McKenney @ 2026-09-19 0:32 UTC (permalink / raw)
To: rcu; +Cc: linux-kernel, kernel-team, rostedt, Puranjay Mohan, Paul E . McKenney
From: Puranjay Mohan <puranjay@kernel.org>
call_srcu() has the same constraint as call_rcu():
srcu_gp_start_if_needed() enqueues under raw_spin_lock_irqsave() and may
walk the srcu_node tree, as do callback invocation and grace-period work,
so a call_srcu() with interrupts already disabled can race an operation in
flight on this CPU. call_rcu_tasks_trace() is call_srcu() under the hood,
so a sleepable BPF program freeing an object can reach this.
Defer as call_rcu() does: stage the callback on the srcu_data's
->defer_cbs, chain that srcu_data onto a per-CPU list, and raise a per-CPU
irq_work that re-issues it straight to the enqueue helper, never back
through __call_srcu(). The irqs-enabled path is unchanged; as for
call_rcu() the gate is bare irqs_disabled(), so callers that merely hold
interrupts off are deferred too and pay one irq_work hop, including
call_rcu_tasks_trace() from the BPF memalloc irq_work.
The irq_work is per-CPU rather than per-srcu_struct and statically
initialized, so deferral never runs check_init_srcu_struct(); it is
IRQ_WORK_INIT_HARD as for call_rcu(). srcu_barrier() flushes it first, and
rcutree_migrate_callbacks() calls srcu_offline_drain() for an outgoing CPU.
cleanup_srcu_struct() drains before its "just leak it" early returns, and
srcu_module_going() before freeing any ->sda, since a staged srcu_data left
chained on a per-CPU list would dangle. Staging is two steps, the callback
onto
->defer_cbs and then the srcu_data onto the per-CPU list, so a flusher can
find the per-CPU list empty while a callback whose call_srcu() has not
returned sits on ->defer_cbs; the staging CPU's own irq_work takes that
one.
The per-CPU srcu_defer ->lock, not any srcu_data's, is held with interrupts
off across the whole nested drain: the chain of srcu_datas staged on that
CPU and, for each, its callbacks, with srcu_do_enqueue() taking that
srcu_data's ->lock and possibly starting a grace period for every one.
That is what serializes the drainers. The bound is as for call_rcu(): what
one interrupts-disabled region could stage, normally a single callback, or
whatever accumulates before the tick where arch_irq_work_has_interrupt() is
false. Both lists are drained newest-first; nothing depends on call_srcu()
ordering.
The drain clears ->next before re-issuing, which bounds a node self-linked
by a double call_srcu(); a longer cycle is not bounded, and a double
call_srcu() stays undefined, as for call_rcu().
A callback deferred past the CPUHP_AP_SMPCFD_DYING irq_work flush leaves
that CPU's srcu_defer ->iw claimed with its self-IPI lost, as for
call_rcu(). srcu_offline_drain() still re-issues the callback, but the
irq_work cannot be un-queued, and here the claim is shared by every
srcu_struct on the CPU.
As in call_rcu(), the re-issue runs with interrupts disabled and can be
re-entered by instrumentation, so a per-CPU flag, set only while that CPU
is inside its own irq_work drain, drops a deferring call_srcu() seen
mid-drain unless it comes from an NMI. Such a drop can strand state the
caller associated with the callback, not just the callback itself.
Staging records only the callback, so a deferred expedited call_srcu()
completes as a normal grace period. Only srcu_expedite_current() can hit
that, and only when invoked with interrupts already disabled.
Gated by CONFIG_RCU_DEFER, though the srcu_data members and the per-CPU
srcu_defer are unconditional. Under CONFIG_PROVE_RCU, warn if the direct
path is reached from an NMI.
Suggested-by: Paul E. McKenney <paulmck@kernel.org>
Signed-off-by: Puranjay Mohan <puranjay@kernel.org>
Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
---
include/linux/srcutree.h | 4 +
kernel/rcu/rcu.h | 3 +
kernel/rcu/srcutree.c | 170 ++++++++++++++++++++++++++++++++++++++-
kernel/rcu/tree.c | 2 +
4 files changed, 175 insertions(+), 4 deletions(-)
diff --git a/include/linux/srcutree.h b/include/linux/srcutree.h
index 75e54e4f963f..1ce759fb7094 100644
--- a/include/linux/srcutree.h
+++ b/include/linux/srcutree.h
@@ -13,6 +13,8 @@
#include <linux/rcu_node_tree.h>
#include <linux/completion.h>
+#include <linux/irq_work_types.h>
+#include <linux/llist.h>
struct srcu_node;
struct srcu_struct;
@@ -41,6 +43,8 @@ struct srcu_data {
bool srcu_cblist_invoking; /* Invoking these CBs? */
struct timer_list delay_work; /* Delay for CB invoking */
struct work_struct work; /* Context for CB invoking. */
+ struct llist_head defer_cbs; /* Callbacks deferred on re-entry. */
+ struct llist_node defer_link; /* Links onto the per-CPU deferral drain list */
struct rcu_head srcu_barrier_head; /* For srcu_barrier() use. */
struct rcu_head srcu_ec_head; /* For srcu_expedite_current() use. */
int srcu_ec_state; /* State for srcu_expedite_current(). */
diff --git a/kernel/rcu/rcu.h b/kernel/rcu/rcu.h
index 91e33571a554..d60444bf3a02 100644
--- a/kernel/rcu/rcu.h
+++ b/kernel/rcu/rcu.h
@@ -583,6 +583,9 @@ static inline bool should_rcu_defer(void)
rcu_scheduler_active != RCU_SCHEDULER_INACTIVE;
}
+/* Drain an outgoing CPU's deferred SRCU callbacks; see rcutree_migrate_callbacks(). */
+void srcu_offline_drain(int cpu);
+
enum rcutorture_type {
RCU_FLAVOR,
RCU_TASKS_FLAVOR,
diff --git a/kernel/rcu/srcutree.c b/kernel/rcu/srcutree.c
index ed204b3f4b84..d32c374ee72a 100644
--- a/kernel/rcu/srcutree.c
+++ b/kernel/rcu/srcutree.c
@@ -20,6 +20,7 @@
#include <linux/percpu.h>
#include <linux/preempt.h>
#include <linux/irq_work.h>
+#include <linux/llist.h>
#include <linux/rcupdate_wait.h>
#include <linux/sched.h>
#include <linux/smp.h>
@@ -79,6 +80,38 @@ static void process_srcu(struct work_struct *work);
static void srcu_irq_work(struct irq_work *work);
static void srcu_delay_timer(struct timer_list *t);
+struct srcu_defer;
+static void srcu_defer_drain(struct irq_work *iw);
+static void __srcu_defer_drain(struct srcu_defer *sndp);
+
+/*
+ * Per-CPU call_srcu() deferral state, shared by every srcu_struct. A deferred
+ * callback is staged on its srcu_data's ->defer_cbs; that srcu_data is chained
+ * via ->defer_link onto ->list, which the irq_work walks.
+ */
+struct srcu_defer {
+ struct llist_head list;
+ struct irq_work iw;
+ raw_spinlock_t lock;
+ bool draining;
+};
+
+static DEFINE_PER_CPU(struct srcu_defer, srcu_defer) = {
+ .lock = __RAW_SPIN_LOCK_UNLOCKED(srcu_defer.lock),
+ .iw = IRQ_WORK_INIT_HARD(srcu_defer_drain),
+};
+
+/*
+ * Flush pending deferred callbacks so a following srcu_barrier() waits for them.
+ */
+static void srcu_defer_flush(void)
+{
+ int cpu;
+
+ for_each_possible_cpu(cpu)
+ __srcu_defer_drain(&per_cpu(srcu_defer, cpu));
+}
+
/*
* Initialize SRCU per-CPU data. Note that statically allocated
* srcu_struct structures might already have srcu_read_lock() and
@@ -107,6 +140,11 @@ static void init_srcu_struct_data(struct srcu_struct *ssp)
sdp->cpu = cpu;
INIT_WORK(&sdp->work, srcu_invoke_callbacks);
timer_setup(&sdp->delay_work, srcu_delay_timer, 0);
+ /*
+ * ->defer_cbs and ->defer_link are valid when zeroed and are not
+ * reinitialized here: that would clobber callbacks a reentrant
+ * call_srcu() already staged. See __call_srcu().
+ */
sdp->ssp = ssp;
}
}
@@ -688,6 +726,14 @@ void cleanup_srcu_struct(struct srcu_struct *ssp)
unsigned long delay;
struct srcu_usage *sup = ssp->srcu_sup;
+ /*
+ * Drain before the early returns below: they leak the srcu_struct, but
+ * srcu_module_going() frees ->sda regardless, and a staged srcu_data
+ * left chained on a per-CPU list would then dangle. Draining first also
+ * has to precede the ->irq_work sync, since re-issuing a callback can
+ * start a grace period and re-queue ->irq_work, which schedules ->work.
+ */
+ srcu_defer_flush();
raw_spin_lock_irq_rcu_node(ssp->srcu_sup);
delay = srcu_get_delay(ssp);
raw_spin_unlock_irq_rcu_node(ssp->srcu_sup);
@@ -695,7 +741,6 @@ void cleanup_srcu_struct(struct srcu_struct *ssp)
return; /* Just leak it! */
if (WARN_ON(srcu_readers_active(ssp)))
return; /* Just leak it! */
- /* Wait for irq_work to finish first as it may queue a new work. */
irq_work_sync(&sup->irq_work);
flush_delayed_work(&sup->work);
for_each_possible_cpu(cpu) {
@@ -1411,8 +1456,8 @@ static unsigned long srcu_gp_start_if_needed(struct srcu_struct *ssp,
* srcu_read_lock(), and srcu_read_unlock() that are all passed the same
* srcu_struct structure.
*/
-static void __call_srcu(struct srcu_struct *ssp, struct rcu_head *rhp,
- rcu_callback_t func, bool do_norm)
+static void srcu_do_enqueue(struct srcu_struct *ssp, struct rcu_head *rhp,
+ rcu_callback_t func, bool do_norm)
{
if (debug_rcu_head_queue(rhp)) {
/* Probable double call_srcu(), so leak the callback. */
@@ -1424,6 +1469,108 @@ static void __call_srcu(struct srcu_struct *ssp, struct rcu_head *rhp,
(void)srcu_gp_start_if_needed(ssp, rhp, do_norm);
}
+/*
+ * The srcu_cblist and srcu_node tree are only accessed with interrupts
+ * disabled, so defer when interrupts are already off rather than enqueue into
+ * an operation that may be in flight on this CPU.
+ */
+static void __call_srcu(struct srcu_struct *ssp, struct rcu_head *rhp,
+ rcu_callback_t func, bool do_norm)
+{
+ if (should_rcu_defer()) {
+ struct srcu_defer *sndp = this_cpu_ptr(&srcu_defer);
+ struct srcu_data *sdp;
+
+ /*
+ * Instrumentation on the enqueue path can re-enter here from
+ * inside the drain. Re-queuing would livelock it, so drop the
+ * callback; an NMI cannot loop, so let it through.
+ */
+ if (READ_ONCE(sndp->draining) && !in_nmi()) {
+ WARN_ONCE(IS_ENABLED(CONFIG_PROVE_RCU),
+ "call_srcu() re-entered during callback drain; leaking callback\n");
+ return;
+ }
+ sdp = this_cpu_ptr(ssp->sda);
+ rhp->func = func;
+ if (llist_add((struct llist_node *)rhp, &sdp->defer_cbs)) {
+ /*
+ * Chain this srcu_data for the drain. ->ssp must be
+ * published here: deferral skips
+ * check_init_srcu_struct(), so on a never-initialized
+ * static srcu_struct the srcu_data are still zeroed and
+ * the drain would read a NULL ->ssp.
+ */
+ sdp->ssp = ssp;
+ if (llist_add(&sdp->defer_link, &sndp->list))
+ irq_work_queue(&sndp->iw);
+ }
+ return;
+ }
+
+ /*
+ * Only reachable from an NMI when deferral is off: before the scheduler
+ * is up, or with CONFIG_RCU_DEFER=n. The enqueue can then race.
+ */
+ WARN_ON_ONCE(IS_ENABLED(CONFIG_PROVE_RCU) && in_nmi());
+
+ srcu_do_enqueue(ssp, rhp, func, do_norm);
+}
+
+/*
+ * Re-issue deferred callbacks straight to srcu_do_enqueue() so they cannot defer
+ * again. ->lock serializes the drainers: the irq_work, srcu_defer_flush() and
+ * srcu_offline_drain().
+ */
+static void __srcu_defer_drain(struct srcu_defer *sndp)
+{
+ struct llist_node *snode, *snext;
+ unsigned long flags;
+
+ if (!IS_ENABLED(CONFIG_RCU_DEFER))
+ return;
+
+ raw_spin_lock_irqsave(&sndp->lock, flags);
+ llist_for_each_safe(snode, snext, llist_del_all(&sndp->list)) {
+ struct srcu_data *sdp = container_of(snode, struct srcu_data, defer_link);
+ struct srcu_struct *ssp = sdp->ssp;
+ struct llist_node *cnode, *cnext;
+
+ cnode = llist_del_all(&sdp->defer_cbs);
+ llist_for_each_safe(cnode, cnext, cnode) {
+ struct rcu_head *rhp = (struct rcu_head *)cnode;
+
+ /* Bounds a node self-linked by a double call_srcu(). */
+ rhp->next = NULL;
+ srcu_do_enqueue(ssp, rhp, rhp->func, true);
+ }
+ }
+ raw_spin_unlock_irqrestore(&sndp->lock, flags);
+}
+
+/*
+ * Only the irq_work drain can be re-fed by its own re-issue, so only it sets
+ * ->draining. A direct drain re-issues onto this CPU, and anything staged
+ * during it is picked up by that CPU's own irq_work.
+ */
+static void srcu_defer_drain(struct irq_work *iw)
+{
+ struct srcu_defer *sndp = container_of(iw, struct srcu_defer, iw);
+
+ WRITE_ONCE(sndp->draining, true);
+ __srcu_defer_drain(sndp);
+ WRITE_ONCE(sndp->draining, false);
+}
+
+/*
+ * Drain @cpu's deferred call_srcu() callbacks once @cpu is dead. One pass
+ * covers every srcu_struct; the re-issue lands on the current CPU.
+ */
+void srcu_offline_drain(int cpu)
+{
+ __srcu_defer_drain(&per_cpu(srcu_defer, cpu));
+}
+
/**
* call_srcu() - Queue a callback for invocation after an SRCU grace period
* @ssp: srcu_struct in queue the callback
@@ -1678,9 +1825,18 @@ void srcu_barrier(struct srcu_struct *ssp)
{
int cpu;
int idx;
- unsigned long s = rcu_seq_snap(&ssp->srcu_sup->srcu_barrier_seq);
+ unsigned long s;
check_init_srcu_struct(ssp);
+
+ /*
+ * Register any deferred callbacks before snapshotting the sequence. The
+ * staging list is per-CPU, not per-srcu_struct, so this also drains
+ * other srcu_structs'.
+ */
+ srcu_defer_flush();
+
+ s = rcu_seq_snap(&ssp->srcu_sup->srcu_barrier_seq);
mutex_lock(&ssp->srcu_sup->srcu_barrier_mutex);
if (rcu_seq_done(&ssp->srcu_sup->srcu_barrier_seq, s)) {
smp_mb(); /* Force ordering following return. */
@@ -2135,6 +2291,12 @@ static void srcu_module_going(struct module *mod)
struct srcu_struct *ssp;
struct srcu_struct **sspp = mod->srcu_struct_ptrs;
+ /*
+ * Deferral skips check_init_srcu_struct(), so cleanup_srcu_struct()
+ * below can be skipped for an srcu_struct that has staged callbacks.
+ * Drain them before any ->sda is freed.
+ */
+ srcu_defer_flush();
for (i = 0; i < mod->num_srcu_structs; i++) {
ssp = *(sspp++);
if (!rcu_seq_state(smp_load_acquire(&ssp->srcu_sup->srcu_gp_seq_needed)) &&
diff --git a/kernel/rcu/tree.c b/kernel/rcu/tree.c
index ff9a2395c9e8..e363e1a6a33c 100644
--- a/kernel/rcu/tree.c
+++ b/kernel/rcu/tree.c
@@ -4636,6 +4636,8 @@ void rcutree_migrate_callbacks(int cpu)
* cover. Drain them here, before the early returns.
*/
__rcu_defer_drain(rdp);
+ /* Likewise for the outgoing CPU's deferred call_srcu() callbacks. */
+ srcu_offline_drain(cpu);
if (rcu_rdp_is_offloaded(rdp))
return;
--
2.40.1
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 4/7] srcu: Make Tiny call_srcu() safe to call from any context
2026-09-19 0:32 [PATCH 0/7] Allow call_{s,}rcu() from NMI and BPF environments Paul E. McKenney
` (2 preceding siblings ...)
2026-09-19 0:32 ` [PATCH 3/7] srcu: Make call_srcu() " Paul E. McKenney
@ 2026-09-19 0:32 ` Paul E. McKenney
2026-09-19 0:32 ` [PATCH 5/7] rcutorture: Disable fragile readers during overload testing Paul E. McKenney
` (2 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Paul E. McKenney @ 2026-09-19 0:32 UTC (permalink / raw)
To: rcu; +Cc: linux-kernel, kernel-team, rostedt, Puranjay Mohan, Paul E . McKenney
From: Puranjay Mohan <puranjay@kernel.org>
Give Tiny call_srcu() the same treatment as Tree SRCU. When interrupts
are disabled and the scheduler is up, stage the callback on the
srcu_struct's lockless list for an irq_work to re-issue later. Tiny SRCU
is uniprocessor, so there is no CPU-offline drain. A draining flag drops
a deferring call_srcu() that re-enters mid-drain (unless from an NMI), as
in Tree SRCU; such a drop can strand state the caller tied to the callback,
not just the callback itself.
Interrupts stay off for the whole batch. TINY_SRCU implies !SMP, where
arch_irq_work_has_interrupt() is false, so the drain always waits for the
tick and a batch is whatever one tick's worth of interrupts-disabled
call_srcu()s staged. Unlike the other three flavors srcu_do_enqueue() here
has no debug_rcu_head_queue(), so nothing reports a double call_srcu();
termination of the drain rests on srcu_do_enqueue() clearing ->next, and
the callback list self-links at the tail exactly as a double call_srcu()
made it before.
srcu_barrier() (now out of line) and cleanup_srcu_struct() drain the
deferred list first, so a deferred callback is re-issued onto the callback
list and invoked by the grace-period work that cleanup_srcu_struct()
flushes, rather than stranded on a soon-to-be-freed srcu_struct.
cleanup_srcu_struct() also syncs ->defer_iw, since that irq_work is
embedded in the srcu_struct the caller is about to free.
The draining flag is global rather than per-srcu_struct: a re-entrant
call_srcu(B) inside a drain of A raises B's own ->defer_iw, whose drain can
stage back onto A, so a per-srcu_struct flag would not break the chain.
The cost is that a drain of A also drops a non-NMI call_srcu() to any
other srcu_struct for its duration.
Gated by CONFIG_RCU_DEFER like Tree SRCU, though the srcu_struct members
are unconditional.
Suggested-by: Paul E. McKenney <paulmck@kernel.org>
Signed-off-by: Puranjay Mohan <puranjay@kernel.org>
Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
---
include/linux/srcutiny.h | 12 ++++--
kernel/rcu/srcutiny.c | 93 ++++++++++++++++++++++++++++++++++++++--
2 files changed, 97 insertions(+), 8 deletions(-)
diff --git a/include/linux/srcutiny.h b/include/linux/srcutiny.h
index fbcf13bc12d1..85b5de438450 100644
--- a/include/linux/srcutiny.h
+++ b/include/linux/srcutiny.h
@@ -12,6 +12,7 @@
#define _LINUX_SRCU_TINY_H
#include <linux/irq_work_types.h>
+#include <linux/llist.h>
#include <linux/swait.h>
struct srcu_struct {
@@ -26,6 +27,8 @@ struct srcu_struct {
struct rcu_head **srcu_cb_tail; /* Pending callbacks: Tail. */
struct work_struct srcu_work; /* For driving grace periods. */
struct irq_work srcu_irq_work; /* Defer schedule_work() to irq work. */
+ struct llist_head defer_cbs; /* Callbacks deferred on re-entry. */
+ struct irq_work defer_iw; /* Re-issues defer_cbs later. */
#ifdef CONFIG_DEBUG_LOCK_ALLOC
struct lockdep_map dep_map;
#endif /* #ifdef CONFIG_DEBUG_LOCK_ALLOC */
@@ -33,6 +36,7 @@ struct srcu_struct {
void srcu_drive_gp(struct work_struct *wp);
void srcu_tiny_irq_work(struct irq_work *irq_work);
+void srcu_defer_drain(struct irq_work *irq_work);
#define __SRCU_STRUCT_INIT(name, __ignored, ___ignored, ____ignored) \
{ \
@@ -40,6 +44,9 @@ void srcu_tiny_irq_work(struct irq_work *irq_work);
.srcu_cb_tail = &name.srcu_cb_head, \
.srcu_work = __WORK_INITIALIZER(name.srcu_work, srcu_drive_gp), \
.srcu_irq_work = { .func = srcu_tiny_irq_work }, \
+ .defer_cbs = LLIST_HEAD_INIT(name.defer_cbs), \
+ .defer_iw = { .node = { .u_flags = IRQ_WORK_HARD_IRQ }, \
+ .func = srcu_defer_drain }, \
__SRCU_DEP_MAP_INIT(name) \
}
@@ -131,10 +138,7 @@ static inline void synchronize_srcu_expedited(struct srcu_struct *ssp)
synchronize_srcu(ssp);
}
-static inline void srcu_barrier(struct srcu_struct *ssp)
-{
- synchronize_srcu(ssp);
-}
+void srcu_barrier(struct srcu_struct *ssp);
static inline void srcu_expedite_current(struct srcu_struct *ssp) { }
#define srcu_check_read_flavor(ssp, read_flavor) do { } while (0)
diff --git a/kernel/rcu/srcutiny.c b/kernel/rcu/srcutiny.c
index 558ba8d316db..5de9a6905838 100644
--- a/kernel/rcu/srcutiny.c
+++ b/kernel/rcu/srcutiny.c
@@ -10,6 +10,7 @@
#include <linux/export.h>
#include <linux/irq_work.h>
+#include <linux/llist.h>
#include <linux/mutex.h>
#include <linux/preempt.h>
#include <linux/rcupdate_wait.h>
@@ -29,6 +30,8 @@ extern int rcu_scheduler_active;
static LIST_HEAD(srcu_boot_list);
static bool srcu_init_done;
+static void __srcu_defer_drain(struct srcu_struct *ssp);
+
static int init_srcu_struct_fields(struct srcu_struct *ssp)
{
ssp->srcu_lock_nesting[0] = 0;
@@ -43,6 +46,8 @@ static int init_srcu_struct_fields(struct srcu_struct *ssp)
INIT_WORK(&ssp->srcu_work, srcu_drive_gp);
INIT_LIST_HEAD(&ssp->srcu_work.entry);
init_irq_work(&ssp->srcu_irq_work, srcu_tiny_irq_work);
+ init_llist_head(&ssp->defer_cbs);
+ ssp->defer_iw = IRQ_WORK_INIT_HARD(srcu_defer_drain);
return 0;
}
@@ -86,6 +91,16 @@ EXPORT_SYMBOL_GPL(init_srcu_struct_generic);
void cleanup_srcu_struct(struct srcu_struct *ssp)
{
WARN_ON(srcu_readers_active(ssp));
+ /*
+ * Re-issue any deferred callbacks, then wait out ->defer_iw before it is
+ * freed. Skipped entirely with CONFIG_RCU_DEFER=n: irq_work_sync() ends
+ * in an unconditional synchronize_rcu() wherever
+ * arch_irq_work_has_interrupt() is false, which is every !SMP target.
+ */
+ if (IS_ENABLED(CONFIG_RCU_DEFER)) {
+ __srcu_defer_drain(ssp);
+ irq_work_sync(&ssp->defer_iw);
+ }
irq_work_sync(&ssp->srcu_irq_work);
flush_work(&ssp->srcu_work);
WARN_ON(ssp->srcu_gp_running);
@@ -213,11 +228,11 @@ static void srcu_gp_start_if_needed(struct srcu_struct *ssp)
}
/*
- * Enqueue an SRCU callback on the specified srcu_struct structure,
- * initiating grace-period processing if it is not already running.
+ * Also called by __srcu_defer_drain() to re-issue a deferred callback, so it
+ * must not re-check the deferral condition.
*/
-void call_srcu(struct srcu_struct *ssp, struct rcu_head *rhp,
- rcu_callback_t func)
+static void srcu_do_enqueue(struct srcu_struct *ssp, struct rcu_head *rhp,
+ rcu_callback_t func)
{
unsigned long flags;
@@ -231,6 +246,68 @@ void call_srcu(struct srcu_struct *ssp, struct rcu_head *rhp,
srcu_gp_start_if_needed(ssp);
preempt_enable();
}
+
+/*
+ * Set only by the irq_work drain, the one drain its own re-issue can re-feed;
+ * a callback staged during a direct drain is taken by ->defer_iw afterwards.
+ * Global rather than per-srcu_struct: a re-entrant call_srcu(B) inside a drain
+ * of A raises B's own ->defer_iw, whose drain can stage back onto A.
+ */
+static bool srcu_defer_draining;
+
+static void __srcu_defer_drain(struct srcu_struct *ssp)
+{
+ struct llist_node *node, *next;
+ unsigned long flags;
+
+ if (!IS_ENABLED(CONFIG_RCU_DEFER))
+ return;
+
+ /* Re-issued newest-first; nothing depends on call_srcu() ordering. */
+ local_irq_save(flags);
+ llist_for_each_safe(node, next, llist_del_all(&ssp->defer_cbs)) {
+ struct rcu_head *rhp = (struct rcu_head *)node;
+
+ srcu_do_enqueue(ssp, rhp, rhp->func);
+ }
+ local_irq_restore(flags);
+}
+
+/* Only the irq_work drain can be re-fed by its own re-issue; see Tree SRCU. */
+void srcu_defer_drain(struct irq_work *iw)
+{
+ struct srcu_struct *ssp = container_of(iw, struct srcu_struct, defer_iw);
+
+ WRITE_ONCE(srcu_defer_draining, true);
+ __srcu_defer_drain(ssp);
+ WRITE_ONCE(srcu_defer_draining, false);
+}
+EXPORT_SYMBOL_GPL(srcu_defer_drain);
+
+void call_srcu(struct srcu_struct *ssp, struct rcu_head *rhp,
+ rcu_callback_t func)
+{
+ if (should_rcu_defer()) {
+ /* A re-entrant call_srcu() during the drain would livelock it. */
+ if (READ_ONCE(srcu_defer_draining) && !in_nmi()) {
+ WARN_ONCE(IS_ENABLED(CONFIG_PROVE_RCU),
+ "call_srcu() re-entered during callback drain; leaking callback\n");
+ return;
+ }
+ rhp->func = func;
+ if (llist_add((struct llist_node *)rhp, &ssp->defer_cbs))
+ irq_work_queue(&ssp->defer_iw);
+ return;
+ }
+
+ /*
+ * Only reachable from an NMI when deferral is off: before the scheduler
+ * is up, or with CONFIG_RCU_DEFER=n. The enqueue can then race.
+ */
+ WARN_ON_ONCE(IS_ENABLED(CONFIG_PROVE_RCU) && in_nmi());
+
+ srcu_do_enqueue(ssp, rhp, func);
+}
EXPORT_SYMBOL_GPL(call_srcu);
/*
@@ -260,6 +337,14 @@ void synchronize_srcu(struct srcu_struct *ssp)
}
EXPORT_SYMBOL_GPL(synchronize_srcu);
+/* Register any deferred callbacks, then wait for all in-flight ones. */
+void srcu_barrier(struct srcu_struct *ssp)
+{
+ __srcu_defer_drain(ssp);
+ synchronize_srcu(ssp);
+}
+EXPORT_SYMBOL_GPL(srcu_barrier);
+
/*
* get_state_synchronize_srcu - Provide an end-of-grace-period cookie
*/
--
2.40.1
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 5/7] rcutorture: Disable fragile readers during overload testing
2026-09-19 0:32 [PATCH 0/7] Allow call_{s,}rcu() from NMI and BPF environments Paul E. McKenney
` (3 preceding siblings ...)
2026-09-19 0:32 ` [PATCH 4/7] srcu: Make Tiny " Paul E. McKenney
@ 2026-09-19 0:32 ` Paul E. McKenney
2026-09-19 0:32 ` [PATCH 6/7] rcutorture: Exercise ->call() from NMI context Paul E. McKenney
2026-09-19 0:32 ` [PATCH 7/7] selftests/bpf: Add a call_srcu() re-entry reproducer Paul E. McKenney
6 siblings, 0 replies; 8+ messages in thread
From: Paul E. McKenney @ 2026-09-19 0:32 UTC (permalink / raw)
To: rcu; +Cc: linux-kernel, kernel-team, rostedt, Paul E. McKenney
Vanilla RCU readers, when either non-preemptible or preemptible with
priority-boosting enabled, can handle heavy overload conditions.
Other RCU implementations are more fragile. This commit adds an
rcu_torture_ops structure field named ->rdrs_handle_load that is set
for non-fragile configurations of vanilla RCU, but cleared otherwise.
It also refrains from running fragile readers concurrently with overload
testing, thus avoiding false-positive overload-testing failures.
Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
---
kernel/rcu/rcutorture.c | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/kernel/rcu/rcutorture.c b/kernel/rcu/rcutorture.c
index 794937e13e7c..7e08be857f01 100644
--- a/kernel/rcu/rcutorture.c
+++ b/kernel/rcu/rcutorture.c
@@ -440,6 +440,7 @@ struct rcu_torture_ops {
int debug_objects;
int start_poll_irqsoff;
int have_up_down;
+ int rdrs_handle_load;
const char *name;
};
@@ -648,6 +649,7 @@ static struct rcu_torture_ops rcu_ops = {
.extendables = RCUTORTURE_MAX_EXTEND,
.debug_objects = 1,
.start_poll_irqsoff = 1,
+ .rdrs_handle_load = !IS_ENABLED(CONFIG_PREEMPT_RCU) || IS_ENABLED(CONFIG_RCU_BOOST),
.name = "rcu"
};
@@ -2534,8 +2536,10 @@ static bool rcu_torture_one_read_start(struct rcu_torture_one_read_state *rtorsp
rtorsp->p = rcu_dereference_check(rcu_torture_current,
!cur_ops->readlock_held || cur_ops->readlock_held() ||
(rtorsp->readstate & RCUTORTURE_RDR_UPDOWN));
- if (rtorsp->p == NULL) {
- /* Wait for rcu_torture_writer to get underway */
+ if ((!cur_ops->rdrs_handle_load && atomic_read(&rcu_fwd_cb_nodelay)) || rtorsp->p == NULL) {
+ // Wait for rcu_torture_writer to get underway and
+ // (if readers cannot handle heavy loads) for any
+ // forward-progress testing to complete.
rcutorture_one_extend(&rtorsp->readstate, 0, trsp, rtorsp->rtrsp);
return false;
}
--
2.40.1
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 6/7] rcutorture: Exercise ->call() from NMI context
2026-09-19 0:32 [PATCH 0/7] Allow call_{s,}rcu() from NMI and BPF environments Paul E. McKenney
` (4 preceding siblings ...)
2026-09-19 0:32 ` [PATCH 5/7] rcutorture: Disable fragile readers during overload testing Paul E. McKenney
@ 2026-09-19 0:32 ` Paul E. McKenney
2026-09-19 0:32 ` [PATCH 7/7] selftests/bpf: Add a call_srcu() re-entry reproducer Paul E. McKenney
6 siblings, 0 replies; 8+ messages in thread
From: Paul E. McKenney @ 2026-09-19 0:32 UTC (permalink / raw)
To: rcu; +Cc: linux-kernel, kernel-team, rostedt, Puranjay Mohan, Paul E . McKenney
From: Puranjay Mohan <puranjay@kernel.org>
call_rcu() and call_srcu() are now safe to invoke from NMI, but
rcutorture never does, leaving the deferral path untested.
Add an ->nmi_capable flag to rcu_torture_ops. For flavors that set it,
arm a per-CPU hardware perf counter whose overflow handler submits a
callback via ->call(). The handler acts only when in_nmi(), so only a
genuine NMI exercises the deferral path. One preallocated callback per
CPU is kept in flight, guarded by an atomic, to avoid allocating in NMI.
The counter uses a fixed sample period rather than a frequency: a
frequency-based event sets TICK_DEP_BIT_PERF_EVENTS and would pin the tick
for the whole run on NO_HZ_FULL kernels.
Report the count issued from NMI ("nmi-calls:") and the count invoked
("nmi-cbs:"). rcu_torture_cleanup() disables the counters and then calls
cb_barrier(), which drains every deferred callback, so the two counts must
then match; a mismatch fails the test. This relies on
rcu_barrier()/srcu_barrier() flushing deferred callbacks, as added earlier
in the series.
Set ->nmi_capable on the NMI-safe flavors: rcu, srcu, srcud, and
tasks-tracing (call_srcu() under the hood). Tasks and Tasks Rude are left
alone, as call_rcu_tasks_generic() is not yet NMI-safe.
Enabled by default; the nmi_calls parameter disables it, which helps rule
NMI handling in or out when triaging a failure. Requires
CONFIG_PERF_EVENTS and a hardware PMU: without one nothing is issued from
NMI and the end-of-test check compares zero against zero, so a pass does
not by itself mean the path ran. That is the case under kvm.sh, which
boots qemu with -cpu kvm64 and no vPMU; rcu_torture_nmi_cleanup() says so
on the console.
Signed-off-by: Puranjay Mohan <puranjay@kernel.org>
Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
---
.../admin-guide/kernel-parameters.txt | 7 +
kernel/rcu/rcutorture.c | 150 +++++++++++++++++-
2 files changed, 155 insertions(+), 2 deletions(-)
diff --git a/Documentation/admin-guide/kernel-parameters.txt b/Documentation/admin-guide/kernel-parameters.txt
index 68647ff4bdd2..fd9acc3fd9b7 100644
--- a/Documentation/admin-guide/kernel-parameters.txt
+++ b/Documentation/admin-guide/kernel-parameters.txt
@@ -6173,6 +6173,13 @@ Kernel parameters
stress RCU, they don't participate in the actual
test, hence the "fake".
+ rcutorture.nmi_calls= [KNL]
+ Enable issuing RCU callbacks from an NMI, on the
+ RCU flavors that support it, to exercise the
+ any-context callback path. Requires
+ CONFIG_PERF_EVENTS and a hardware PMU; without
+ both, nothing is issued. Defaults to enabled.
+
rcutorture.nocbs_nthreads= [KNL]
Set number of RCU callback-offload togglers.
Zero (the default) disables toggling.
diff --git a/kernel/rcu/rcutorture.c b/kernel/rcu/rcutorture.c
index 7e08be857f01..4d1be2a49f01 100644
--- a/kernel/rcu/rcutorture.c
+++ b/kernel/rcu/rcutorture.c
@@ -48,6 +48,7 @@
#include <linux/tick.h>
#include <linux/rcupdate_trace.h>
#include <linux/nmi.h>
+#include <linux/perf_event.h>
#include "rcu.h"
@@ -115,6 +116,7 @@ torture_param(int, leakpointer, 0, "Leak pointer dereferences from readers");
torture_param(int, n_barrier_cbs, 0, "# of callbacks/kthreads for barrier testing");
torture_param(int, n_up_down, 32, "# of concurrent up/down hrtimer-based RCU readers");
torture_param(int, nfakewriters, 4, "Number of RCU fake writer threads");
+torture_param(bool, nmi_calls, true, "Exercise ->call() from NMI on nmi_capable flavors");
torture_param(int, nreaders, -1, "Number of RCU reader threads");
torture_param(bool, nwriters, 1, "Number of RCU writer threads (0 or 1)");
torture_param(int, object_debug, 0, "Enable debug-object double call_rcu() testing");
@@ -216,6 +218,8 @@ static long n_rcu_torture_boost_failure;
static long n_rcu_torture_boosts;
static atomic_long_t n_rcu_torture_timers;
static atomic_long_t n_rcu_torture_irqs;
+static atomic_long_t n_rcu_torture_nmi_call;
+static atomic_long_t n_rcu_torture_nmi_cb;
static long n_barrier_attempts;
static long n_barrier_successes; /* did rcu_barrier test succeed? */
static unsigned long n_read_exits;
@@ -433,6 +437,7 @@ struct rcu_torture_ops {
bool (*is_task_rcu_boosted)(void);
long cbflood_max;
int irq_capable;
+ int nmi_capable;
int can_boost;
int extendables;
int slow_gps;
@@ -650,6 +655,7 @@ static struct rcu_torture_ops rcu_ops = {
.debug_objects = 1,
.start_poll_irqsoff = 1,
.rdrs_handle_load = !IS_ENABLED(CONFIG_PREEMPT_RCU) || IS_ENABLED(CONFIG_RCU_BOOST),
+ .nmi_capable = 1,
.name = "rcu"
};
@@ -944,6 +950,7 @@ static struct rcu_torture_ops srcu_ops = {
.debug_objects = 1,
.have_up_down = IS_ENABLED(CONFIG_TINY_SRCU)
? 0 : SRCU_READ_FLAVOR_NORMAL | SRCU_READ_FLAVOR_FAST_UPDOWN,
+ .nmi_capable = 1,
.name = "srcu"
};
@@ -1007,6 +1014,7 @@ static struct rcu_torture_ops srcud_ops = {
.debug_objects = 1,
.have_up_down = IS_ENABLED(CONFIG_TINY_SRCU)
? 0 : SRCU_READ_FLAVOR_NORMAL | SRCU_READ_FLAVOR_FAST_UPDOWN,
+ .nmi_capable = 1,
.name = "srcud"
};
@@ -1271,6 +1279,7 @@ static struct rcu_torture_ops tasks_tracing_ops = {
.cbflood_max = 50000,
.irq_capable = 1,
.slow_gps = 1,
+ .nmi_capable = 1,
.name = "tasks-tracing"
};
@@ -2663,6 +2672,124 @@ static bool rcu_torture_one_read(struct torture_random_state *trsp, long myid)
static DEFINE_TORTURE_RANDOM_PERCPU(rcu_torture_timer_rand);
+/*
+ * Exercise ->call() from NMI context for flavors that set ->nmi_capable. A
+ * per-CPU hardware perf counter overflows into an NMI, and its handler submits
+ * a preallocated callback via ->call(). One callback per CPU is in flight at a
+ * time (guarded by an atomic) to avoid allocating in NMI.
+ */
+#ifdef CONFIG_PERF_EVENTS
+static struct perf_event_attr rcu_torture_nmi_attr = {
+ .type = PERF_TYPE_HARDWARE,
+ .config = PERF_COUNT_HW_CPU_CYCLES,
+ .size = sizeof(struct perf_event_attr),
+ .pinned = 1,
+ .disabled = 1,
+ /*
+ * A fixed period rather than .freq: a frequency-based event bumps
+ * nr_freq_events, which sets TICK_DEP_BIT_PERF_EVENTS and would pin the
+ * tick for the whole run on NO_HZ_FULL kernels.
+ */
+ .sample_period = 20 * 1000 * 1000,
+};
+
+/* One in-flight callback per CPU; ->inuse is released by the callback. */
+struct rcu_torture_nmi_cb {
+ struct rcu_head rh;
+ atomic_t inuse;
+};
+
+static struct perf_event **rcu_torture_nmi_events;
+static int rcu_torture_nmi_hp_state;
+static DEFINE_PER_CPU(struct rcu_torture_nmi_cb, rcu_torture_nmi_cb);
+
+static void rcu_torture_nmi_invoked(struct rcu_head *rhp)
+{
+ struct rcu_torture_nmi_cb *rtncp = container_of(rhp, struct rcu_torture_nmi_cb, rh);
+
+ atomic_long_inc(&n_rcu_torture_nmi_cb);
+ atomic_set(&rtncp->inuse, 0);
+}
+
+static void rcu_torture_nmi_overflow(struct perf_event *event,
+ struct perf_sample_data *data,
+ struct pt_regs *regs)
+{
+ struct rcu_torture_nmi_cb *rtncp = this_cpu_ptr(&rcu_torture_nmi_cb);
+
+ if (!in_nmi())
+ return;
+ if (cur_ops->call && !atomic_xchg(&rtncp->inuse, 1)) {
+ atomic_long_inc(&n_rcu_torture_nmi_call);
+ cur_ops->call(&rtncp->rh, rcu_torture_nmi_invoked);
+ }
+}
+
+static int rcu_torture_nmi_online(unsigned int cpu)
+{
+ struct perf_event *event;
+
+ event = perf_event_create_kernel_counter(&rcu_torture_nmi_attr, cpu, NULL,
+ rcu_torture_nmi_overflow, NULL);
+ if (IS_ERR(event))
+ return 0;
+ rcu_torture_nmi_events[cpu] = event;
+ perf_event_enable(event);
+ return 0;
+}
+
+static int rcu_torture_nmi_offline(unsigned int cpu)
+{
+ struct perf_event *event = rcu_torture_nmi_events[cpu];
+
+ if (event) {
+ rcu_torture_nmi_events[cpu] = NULL;
+ perf_event_disable(event);
+ perf_event_release_kernel(event);
+ }
+ return 0;
+}
+
+/* Drive the counters from hotplug callbacks so coverage survives onoff. */
+static void rcu_torture_nmi_init(void)
+{
+ int ret;
+
+ if (!nmi_calls || !cur_ops->nmi_capable || !cur_ops->call)
+ return;
+ rcu_torture_nmi_events = kcalloc(nr_cpu_ids, sizeof(*rcu_torture_nmi_events),
+ GFP_KERNEL);
+ if (!rcu_torture_nmi_events)
+ return;
+ ret = cpuhp_setup_state(CPUHP_AP_ONLINE_DYN, "rcutorture/nmi:online",
+ rcu_torture_nmi_online, rcu_torture_nmi_offline);
+ if (ret < 0) {
+ kfree(rcu_torture_nmi_events);
+ rcu_torture_nmi_events = NULL;
+ return;
+ }
+ rcu_torture_nmi_hp_state = ret;
+}
+
+static void rcu_torture_nmi_cleanup(void)
+{
+ if (!rcu_torture_nmi_events)
+ return;
+ if (rcu_torture_nmi_hp_state > 0) {
+ cpuhp_remove_state(rcu_torture_nmi_hp_state);
+ rcu_torture_nmi_hp_state = 0;
+ }
+ kfree(rcu_torture_nmi_events);
+ rcu_torture_nmi_events = NULL;
+ if (!atomic_long_read(&n_rcu_torture_nmi_call))
+ pr_alert("%s: nmi_calls set but no ->call() ever issued from NMI, so NMI ->call() went untested (no PMU, or NMIs unavailable here).\n",
+ __func__);
+}
+#else /* #ifdef CONFIG_PERF_EVENTS */
+static void rcu_torture_nmi_init(void) { }
+static void rcu_torture_nmi_cleanup(void) { }
+#endif /* #else #ifdef CONFIG_PERF_EVENTS */
+
/*
* RCU torture reader from timer handler. Dereferences rcu_torture_current,
* incrementing the corresponding element of the pipeline array. The
@@ -3051,6 +3178,9 @@ rcu_torture_stats_print(void)
data_race(n_barrier_attempts),
data_race(n_rcu_torture_barrier_error));
pr_cont("read-exits: %ld ", data_race(n_read_exits)); // Statistic.
+ pr_cont("nmi-calls: %ld nmi-cbs: %ld ",
+ atomic_long_read(&n_rcu_torture_nmi_call),
+ atomic_long_read(&n_rcu_torture_nmi_cb));
pr_cont("nocb-toggles: %ld:%ld ",
atomic_long_read(&n_nocb_offload), atomic_long_read(&n_nocb_deoffload));
pr_cont("gpwraps: %ld\n", n_gpwraps);
@@ -3199,7 +3329,7 @@ rcu_torture_print_module_parms(struct rcu_torture_ops *cur_ops, const char *tag)
"read_exit_delay=%d read_exit_burst=%d "
"reader_flavor=%x "
"nocbs_nthreads=%d nocbs_toggle=%d "
- "test_nmis=%d "
+ "test_nmis=%d nmi_calls=%d "
"preempt_duration=%d preempt_interval=%d n_up_down=%d\n",
torture_type, tag, nrealreaders, nwriters, nrealfakewriters,
stat_interval, verbose, test_no_idle_hz, shuffle_interval,
@@ -3213,7 +3343,7 @@ rcu_torture_print_module_parms(struct rcu_torture_ops *cur_ops, const char *tag)
read_exit_delay, read_exit_burst,
reader_flavor,
nocbs_nthreads, nocbs_toggle,
- test_nmis,
+ test_nmis, nmi_calls,
preempt_duration, preempt_interval, n_up_down);
}
@@ -4287,6 +4417,7 @@ rcu_torture_cleanup(void)
int i;
if (torture_cleanup_begin()) {
+ rcu_torture_nmi_cleanup();
if (cur_ops->cb_barrier != NULL) {
pr_info("%s: Invoking %pS().\n", __func__, cur_ops->cb_barrier);
cur_ops->cb_barrier();
@@ -4329,6 +4460,7 @@ rcu_torture_cleanup(void)
kfree(reader_tasks);
reader_tasks = NULL;
}
+ rcu_torture_nmi_cleanup();
kfree(rcu_torture_reader_mbchk);
rcu_torture_reader_mbchk = NULL;
@@ -4358,6 +4490,19 @@ rcu_torture_cleanup(void)
pr_info("%s: Invoking %pS().\n", __func__, cur_ops->cb_barrier);
cur_ops->cb_barrier();
}
+
+ /*
+ * cb_barrier() above drained every deferred callback, so the count
+ * issued from NMI must equal the count invoked.
+ */
+ if (atomic_long_read(&n_rcu_torture_nmi_call) !=
+ atomic_long_read(&n_rcu_torture_nmi_cb)) {
+ pr_alert("%s: NMI ->call() lost a callback: issued %ld invoked %ld\n",
+ __func__, atomic_long_read(&n_rcu_torture_nmi_call),
+ atomic_long_read(&n_rcu_torture_nmi_cb));
+ atomic_inc(&n_rcu_torture_error);
+ }
+
if (cur_ops->cleanup != NULL)
cur_ops->cleanup();
@@ -4790,6 +4935,7 @@ rcu_torture_init(void)
firsterr = -ENOMEM;
goto unwind;
}
+ rcu_torture_nmi_init();
for (i = 0; i < nrealreaders; i++) {
rcu_torture_reader_mbchk[i].rtc_chkrdr = -1;
firsterr = torture_create_kthread(rcu_torture_reader, (void *)i,
--
2.40.1
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 7/7] selftests/bpf: Add a call_srcu() re-entry reproducer
2026-09-19 0:32 [PATCH 0/7] Allow call_{s,}rcu() from NMI and BPF environments Paul E. McKenney
` (5 preceding siblings ...)
2026-09-19 0:32 ` [PATCH 6/7] rcutorture: Exercise ->call() from NMI context Paul E. McKenney
@ 2026-09-19 0:32 ` Paul E. McKenney
6 siblings, 0 replies; 8+ messages in thread
From: Paul E. McKenney @ 2026-09-19 0:32 UTC (permalink / raw)
To: rcu
Cc: linux-kernel, kernel-team, rostedt, Puranjay Mohan,
Kumar Kartikeya Dwivedi, Paul E . McKenney
From: Puranjay Mohan <puranjay@kernel.org>
Re-enter call_srcu() from a BPF program to exercise its any-context
safety, via call_rcu_tasks_trace(), which is call_srcu() on
rcu_tasks_trace_srcu_struct.
An fentry program on rcu_segcblist_enqueue() fires mid-enqueue: that
function is reached from srcu_gp_start_if_needed() with the srcu_data
->lock held. The program does a task-storage delete, whose only deferred
work is call_rcu_tasks_trace(), re-entering the enqueue on the same CPU.
The handler matches on TID and fires once; pinning the thread removes the
migration window between picking the srcu_data and taking its lock.
Without the fix the nested call re-takes the same sdp lock and
self-deadlocks; with it the nested __call_srcu() sees interrupts disabled
and defers via irq_work, so the delete returns and the test passes.
The test skips where it does not apply: Tiny RCU has no
rcu_segcblist_enqueue() to attach to, and a UP+PREEMPT kernel pairs Tree
RCU with Tiny SRCU, so the attach succeeds but call_srcu() never reaches
the enqueue. Tiny SRCU is told apart by srcu_expedite_current(), which it
stubs out, so on Tree SRCU a zero hit count fails rather than skips and the
reproducer cannot quietly stop reproducing.
Signed-off-by: Puranjay Mohan <puranjay@kernel.org>
Acked-by: Kumar Kartikeya Dwivedi <memxor@gmail.com>
Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
---
.../selftests/bpf/prog_tests/rcu_reentry.c | 93 +++++++++++++++++++
.../testing/selftests/bpf/progs/rcu_reentry.c | 51 ++++++++++
2 files changed, 144 insertions(+)
create mode 100644 tools/testing/selftests/bpf/prog_tests/rcu_reentry.c
create mode 100644 tools/testing/selftests/bpf/progs/rcu_reentry.c
diff --git a/tools/testing/selftests/bpf/prog_tests/rcu_reentry.c b/tools/testing/selftests/bpf/prog_tests/rcu_reentry.c
new file mode 100644
index 000000000000..de23a14b3d40
--- /dev/null
+++ b/tools/testing/selftests/bpf/prog_tests/rcu_reentry.c
@@ -0,0 +1,93 @@
+// SPDX-License-Identifier: GPL-2.0
+/* Exercise re-entry into call_srcu() from BPF; see progs/rcu_reentry.c. */
+#define _GNU_SOURCE
+#include <sched.h>
+#include <test_progs.h>
+#include "task_local_storage_helpers.h"
+#include "trace_helpers.h"
+#include "rcu_reentry.skel.h"
+
+/* Tiny RCU has no rcu_segcblist_enqueue() to attach to. */
+static bool have_attach_target(void)
+{
+ unsigned long long addr;
+
+ return kallsyms_find("rcu_segcblist_enqueue", &addr) == 0;
+}
+
+/* Tiny SRCU stubs out srcu_expedite_current(); Tree SRCU exports it. */
+static bool have_tree_srcu(void)
+{
+ unsigned long long addr;
+
+ return kallsyms_find("srcu_expedite_current", &addr) == 0;
+}
+
+void test_rcu_reentry(void)
+{
+ struct rcu_reentry *skel;
+ int err, pidfd = -1, map_fd;
+ cpu_set_t set, old_set;
+ bool affinity_saved;
+ __u64 val = 1;
+ int cpu;
+
+ if (!have_attach_target()) {
+ test__skip();
+ return;
+ }
+
+ skel = rcu_reentry__open_and_load();
+ if (!ASSERT_OK_PTR(skel, "skel_open_and_load"))
+ return;
+
+ err = rcu_reentry__attach(skel);
+ if (!ASSERT_OK(err, "skel_attach"))
+ goto out;
+
+ /* Keep the re-entry on a single CPU; a cpuset may exclude CPU 0. */
+ affinity_saved = !sched_getaffinity(0, sizeof(old_set), &old_set);
+ cpu = sched_getcpu();
+ if (!ASSERT_GE(cpu, 0, "getcpu"))
+ goto out;
+ CPU_ZERO(&set);
+ CPU_SET(cpu, &set);
+ if (!ASSERT_OK(sched_setaffinity(0, sizeof(set), &set), "setaffinity"))
+ goto out;
+
+ pidfd = sys_pidfd_open(getpid(), 0);
+ if (!ASSERT_GE(pidfd, 0, "pidfd_open"))
+ goto restore;
+ map_fd = bpf_map__fd(skel->maps.task_stg);
+ err = bpf_map_update_elem(map_fd, &pidfd, &val, BPF_NOEXIST);
+ if (!ASSERT_OK(err, "boot_create"))
+ goto restore;
+
+ /* Arm the handler for this thread, then trigger call_rcu_tasks_trace(). */
+ skel->bss->target_pid = syscall(__NR_gettid);
+ err = bpf_map_delete_elem(map_fd, &pidfd);
+ if (!ASSERT_OK(err, "boot_delete"))
+ goto restore;
+
+ /*
+ * Only Tree SRCU reaches rcu_segcblist_enqueue() from call_srcu(); a
+ * UP+PREEMPT kernel pairs Tree RCU with Tiny SRCU, so the attach
+ * succeeds but nothing fires. On Tree SRCU it must fire.
+ */
+ if (!skel->bss->hits) {
+ if (have_tree_srcu())
+ ASSERT_GT(skel->bss->hits, 0, "prog_fired");
+ else
+ test__skip();
+ goto restore;
+ }
+ ASSERT_EQ(skel->bss->get_errs, 0, "nested_storage_get");
+ ASSERT_EQ(skel->bss->del_errs, 0, "nested_storage_delete");
+restore:
+ if (affinity_saved)
+ sched_setaffinity(0, sizeof(old_set), &old_set);
+out:
+ if (pidfd >= 0)
+ close(pidfd);
+ rcu_reentry__destroy(skel);
+}
diff --git a/tools/testing/selftests/bpf/progs/rcu_reentry.c b/tools/testing/selftests/bpf/progs/rcu_reentry.c
new file mode 100644
index 000000000000..47a36f704cf3
--- /dev/null
+++ b/tools/testing/selftests/bpf/progs/rcu_reentry.c
@@ -0,0 +1,51 @@
+// SPDX-License-Identifier: GPL-2.0
+/*
+ * Re-enter call_srcu() from a BPF program. fentry on rcu_segcblist_enqueue()
+ * fires inside call_srcu()'s enqueue (reached from srcu_gp_start_if_needed()
+ * with the srcu_data ->lock held); the handler then calls call_rcu_tasks_trace()
+ * -- itself call_srcu() on rcu_tasks_trace_srcu_struct -- re-entering the same
+ * srcu_data on the same CPU.
+ */
+#include "vmlinux.h"
+#include <bpf/bpf_helpers.h>
+#include <bpf/bpf_tracing.h>
+
+char _license[] SEC("license") = "GPL";
+
+struct {
+ __uint(type, BPF_MAP_TYPE_TASK_STORAGE);
+ __uint(map_flags, BPF_F_NO_PREALLOC);
+ __type(key, int);
+ __type(value, __u64);
+} task_stg SEC(".maps");
+
+int target_pid;
+int hits;
+int get_errs;
+int del_errs;
+int done;
+
+SEC("fentry/rcu_segcblist_enqueue")
+int BPF_PROG(reenter)
+{
+ struct task_struct *cur;
+
+ if (done || !target_pid)
+ return 0;
+
+ cur = bpf_get_current_task_btf();
+ if (cur->pid != target_pid)
+ return 0;
+
+ /* Issue the nested call exactly once, so the test is deterministic. */
+ done = 1;
+ __sync_fetch_and_add(&hits, 1);
+
+ /* Re-enter via a task-storage delete, which calls call_rcu_tasks_trace(). */
+ if (!bpf_task_storage_get(&task_stg, cur, 0, BPF_LOCAL_STORAGE_GET_F_CREATE))
+ __sync_fetch_and_add(&get_errs, 1);
+ else if (bpf_task_storage_delete(&task_stg, cur))
+ __sync_fetch_and_add(&del_errs, 1);
+
+ return 0;
+}
--
2.40.1
^ permalink raw reply [flat|nested] 8+ messages in thread