diff options
| author | Paul E. McKenney <paulmck@kernel.org> | 2026-09-18 16:16:34 -0700 |
|---|---|---|
| committer | Paul E. McKenney <paulmck@kernel.org> | 2026-09-18 16:16:34 -0700 |
| commit | aebf6c4777d534ccd089b3ccab98d71ea3167a0d (patch) | |
| tree | bdc90b4188953643d37f76bbc9d767f8891baea6 | |
| parent | 64b7dbc3e2226f1259577b32d5e7675c2839d93d (diff) | |
| parent | c3f9b7bf999c2d7f5f0fdb696ec80895895e37ba (diff) | |
| download | linux-next-aebf6c4777d534ccd089b3ccab98d71ea3167a0d.tar.gz linux-next-aebf6c4777d534ccd089b3ccab98d71ea3167a0d.zip | |
Merge branches 'misc.2026.09.18a' and 'torture.2026.09.18a' into HEAD
misc.2026.09.18a: Miscellaneous fixes.
torture.2026.09.18a: Make call_{s,}rcu() invocable from any environment,
add atomic SRCU, additional SRCU updates, and torture-test updates.
| -rw-r--r-- | Documentation/admin-guide/kernel-parameters.txt | 7 | ||||
| -rw-r--r-- | include/linux/srcu.h | 83 | ||||
| -rw-r--r-- | include/linux/srcutiny.h | 18 | ||||
| -rw-r--r-- | include/linux/srcutree.h | 19 | ||||
| -rw-r--r-- | kernel/rcu/Kconfig | 6 | ||||
| -rw-r--r-- | kernel/rcu/rcu.h | 14 | ||||
| -rw-r--r-- | kernel/rcu/rcutorture.c | 208 | ||||
| -rw-r--r-- | kernel/rcu/srcutiny.c | 182 | ||||
| -rw-r--r-- | kernel/rcu/srcutree.c | 462 | ||||
| -rw-r--r-- | kernel/rcu/tiny.c | 127 | ||||
| -rw-r--r-- | kernel/rcu/tree.c | 133 | ||||
| -rw-r--r-- | kernel/rcu/tree.h | 6 | ||||
| -rw-r--r-- | tools/testing/selftests/bpf/prog_tests/rcu_reentry.c | 93 | ||||
| -rw-r--r-- | tools/testing/selftests/bpf/progs/rcu_reentry.c | 51 | ||||
| -rwxr-xr-x | tools/testing/selftests/rcutorture/bin/kvm-remote.sh | 22 | ||||
| -rwxr-xr-x | tools/testing/selftests/rcutorture/bin/torture.sh | 24 |
16 files changed, 1342 insertions, 113 deletions
diff --git a/Documentation/admin-guide/kernel-parameters.txt b/Documentation/admin-guide/kernel-parameters.txt index 6cc6d45b59d6..38af3a20b57e 100644 --- a/Documentation/admin-guide/kernel-parameters.txt +++ b/Documentation/admin-guide/kernel-parameters.txt @@ -6176,6 +6176,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/include/linux/srcu.h b/include/linux/srcu.h index 7d9bc06df98d..1a8a465a5650 100644 --- a/include/linux/srcu.h +++ b/include/linux/srcu.h @@ -36,6 +36,8 @@ static inline int __init_srcu_struct(struct srcu_struct *ssp, const char *name, int __init_srcu_struct_fast(struct srcu_struct *ssp, const char *name, struct lock_class_key *key); int __init_srcu_struct_fast_updown(struct srcu_struct *ssp, const char *name, struct lock_class_key *key); +int __init_srcu_struct_atomic(struct srcu_struct *ssp, const char *name, + struct lock_class_key *key); #endif // #ifndef CONFIG_TINY_SRCU #define init_srcu_struct_fast(ssp) \ @@ -52,6 +54,13 @@ int __init_srcu_struct_fast_updown(struct srcu_struct *ssp, const char *name, __init_srcu_struct_fast_updown((ssp), #ssp, &__srcu_key); \ }) +#define init_srcu_struct_atomic(ssp) \ +({ \ + static struct lock_class_key __srcu_key; \ + \ + __init_srcu_struct_atomic((ssp), #ssp, &__srcu_key); \ +}) + #define __SRCU_DEP_MAP_INIT(srcu_name) .dep_map = { .name = #srcu_name }, #else /* #ifdef CONFIG_DEBUG_LOCK_ALLOC */ @@ -64,6 +73,7 @@ static inline int __init_srcu_struct(struct srcu_struct *ssp, const char *name, #ifndef CONFIG_TINY_SRCU int init_srcu_struct_fast(struct srcu_struct *ssp); int init_srcu_struct_fast_updown(struct srcu_struct *ssp); +int init_srcu_struct_atomic(struct srcu_struct *ssp); #endif // #ifndef CONFIG_TINY_SRCU #define __SRCU_DEP_MAP_INIT(srcu_name) @@ -77,14 +87,18 @@ int init_srcu_struct_fast_updown(struct srcu_struct *ssp); }) /* Values for SRCU Tree srcu_data ->srcu_reader_flavor, but also used by rcutorture. */ -#define SRCU_READ_FLAVOR_NORMAL 0x1 // srcu_read_lock(). -#define SRCU_READ_FLAVOR_NMI 0x2 // srcu_read_lock_nmisafe(). -// 0x4 // SRCU-lite is no longer with us. -#define SRCU_READ_FLAVOR_FAST 0x4 // srcu_read_lock_fast(), also NMI-safe. -#define SRCU_READ_FLAVOR_FAST_UPDOWN 0x8 // srcu_read_lock_fast_updown(). +#define SRCU_READ_FLAVOR_NORMAL 0x01 // srcu_read_lock(). +#define SRCU_READ_FLAVOR_NMI 0x02 // srcu_read_lock_nmisafe(). +// 0x04 // SRCU-lite is no longer with us. +#define SRCU_READ_FLAVOR_FAST 0x04 // srcu_read_lock_fast(), also NMI-safe. +#define SRCU_READ_FLAVOR_FAST_UPDOWN 0x08 // srcu_read_lock_fast_updown(). +#define SRCU_READ_FLAVOR_ATOMIC 0x10 // srcu_read_lock_atomic(). #define SRCU_READ_FLAVOR_ALL (SRCU_READ_FLAVOR_NORMAL | SRCU_READ_FLAVOR_NMI | \ - SRCU_READ_FLAVOR_FAST | SRCU_READ_FLAVOR_FAST_UPDOWN) + SRCU_READ_FLAVOR_FAST | SRCU_READ_FLAVOR_FAST_UPDOWN | \ + SRCU_READ_FLAVOR_ATOMIC) // All of the above. +#define SRCU_READ_FLAVOR_PREDEF (SRCU_READ_FLAVOR_FAST | SRCU_READ_FLAVOR_ATOMIC) + // Flavors special DEFINE_SRCU() flavors. #define SRCU_READ_FLAVOR_SLOWGP (SRCU_READ_FLAVOR_FAST | SRCU_READ_FLAVOR_FAST_UPDOWN) // Flavors requiring synchronize_rcu() // instead of smp_mb(). @@ -102,6 +116,7 @@ void call_srcu(struct srcu_struct *ssp, struct rcu_head *head, void (*func)(struct rcu_head *head)); void cleanup_srcu_struct(struct srcu_struct *ssp); void synchronize_srcu(struct srcu_struct *ssp); +void synchronize_srcu_atomic(struct srcu_struct *ssp); #define SRCU_GET_STATE_COMPLETED 0x1 @@ -307,6 +322,45 @@ static inline int srcu_read_lock(struct srcu_struct *ssp) } /** + * srcu_read_lock_atomic - register a new reader promising an atomic section + * @ssp: srcu_struct in which to register the new reader. + * + * As srcu_read_lock(), but the caller promises that the read-side + * critical section never sleeps and never blocks on anything which + * may itself depend on memory allocation to make progress. Preemption + * is disabled for the duration, which both enforces that promise (any + * sleepable call in the section will splat on every configuration) + * and bounds the section so that the update side may spin rather + * than sleep when waiting for readers: see synchronize_srcu_atomic(). + * + * The lock and matching srcu_read_unlock_atomic() must be invoked on + * the same CPU, from the same context; passing the return value to + * another task is not permitted for this flavor. + */ +static inline int srcu_read_lock_atomic(struct srcu_struct *ssp) + __acquires_shared(ssp) +{ + int retval; + + preempt_disable(); + /* + * Arm might_sleep() to catch even a *potentially* sleeping call + * in the section, not just an actual schedule: the atomic-domain + * promise must hold on every path, contended or not. In hardirq, + * softirq, or NMI the annotation would land on the interrupted + * task, and can also result in data races against that task's + * own non_block_start()/non_block_end() invocations; it is also + * redundant there, so skip it. + */ + if (in_task()) + non_block_start(); + srcu_check_read_flavor(ssp, SRCU_READ_FLAVOR_ATOMIC); + retval = __srcu_read_lock(ssp); + srcu_lock_acquire(&ssp->dep_map); + return retval; +} + +/** * srcu_read_lock_fast - register a new reader for an SRCU-protected structure. * @ssp: srcu_struct in which to register the new reader. * @@ -499,6 +553,23 @@ static inline void srcu_read_unlock(struct srcu_struct *ssp, int idx) } /** + * srcu_read_unlock_atomic - unregister an atomic-section reader + * @ssp: srcu_struct from which to unregister the old reader. + * @idx: return value from corresponding srcu_read_lock_atomic(). + */ +static inline void srcu_read_unlock_atomic(struct srcu_struct *ssp, int idx) + __releases_shared(ssp) +{ + WARN_ON_ONCE(idx & ~0x1); + srcu_check_read_flavor(ssp, SRCU_READ_FLAVOR_ATOMIC); + srcu_lock_release(&ssp->dep_map); + __srcu_read_unlock(ssp, idx); + if (in_task()) + non_block_end(); + preempt_enable(); +} + +/** * srcu_read_unlock_fast - unregister a old reader from an SRCU-protected structure. * @ssp: srcu_struct in which to unregister the old reader. * @scp: return value from corresponding srcu_read_lock_fast(). diff --git a/include/linux/srcutiny.h b/include/linux/srcutiny.h index fbcf13bc12d1..47a368f945e3 100644 --- a/include/linux/srcutiny.h +++ b/include/linux/srcutiny.h @@ -12,12 +12,14 @@ #define _LINUX_SRCU_TINY_H #include <linux/irq_work_types.h> +#include <linux/llist.h> #include <linux/swait.h> struct srcu_struct { short srcu_lock_nesting[2]; /* srcu_read_lock() nesting depth. */ u8 srcu_gp_running; /* GP workqueue running? */ u8 srcu_gp_waiting; /* GP waiting for readers? */ + u8 srcu_atomic_gp_flag; /* Serialize atomic GP work.*/ unsigned long srcu_idx; /* Current reader array element in bit 0x2. */ unsigned long srcu_idx_max; /* Furthest future srcu_idx request. */ struct swait_queue_head srcu_wq; @@ -26,6 +28,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 +37,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 +45,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) \ } @@ -57,15 +65,20 @@ void srcu_tiny_irq_work(struct irq_work *irq_work); #define DEFINE_SRCU_FAST_UPDOWN(name) DEFINE_SRCU(name) #define DEFINE_STATIC_SRCU_FAST_UPDOWN(name) \ static struct srcu_struct name = __SRCU_STRUCT_INIT(name, name, name, name) +#define DEFINE_SRCU_ATOMIC(name) DEFINE_SRCU(name) +#define DEFINE_STATIC_SRCU_ATOMIC(name) \ + static struct srcu_struct name = __SRCU_STRUCT_INIT(name, name, name, name) // Dummy structure for srcu_notifier_head. struct srcu_usage { }; #define __SRCU_USAGE_INIT(name) { } #define __init_srcu_struct_fast __init_srcu_struct #define __init_srcu_struct_fast_updown __init_srcu_struct +#define __init_srcu_struct_atomic __init_srcu_struct #ifndef CONFIG_DEBUG_LOCK_ALLOC #define init_srcu_struct_fast init_srcu_struct #define init_srcu_struct_fast_updown init_srcu_struct +#define init_srcu_struct_atomic init_srcu_struct #endif // #ifndef CONFIG_DEBUG_LOCK_ALLOC void synchronize_srcu(struct srcu_struct *ssp); @@ -131,10 +144,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/include/linux/srcutree.h b/include/linux/srcutree.h index 75e54e4f963f..93b76e543894 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(). */ @@ -76,6 +80,7 @@ struct srcu_usage { struct mutex srcu_cb_mutex; /* Serialize CB preparation. */ raw_spinlock_t __private lock; /* Protect counters and size state. */ struct mutex srcu_gp_mutex; /* Serialize GP work. */ + atomic_t srcu_atomic_gp_flag; /* Serialize atomic GP work. */ unsigned long srcu_gp_seq; /* Grace-period seq #. */ unsigned long srcu_gp_seq_needed; /* Latest gp_seq needed. */ unsigned long srcu_gp_seq_needed_exp; /* Furthest future exp GP. */ @@ -209,6 +214,12 @@ struct srcu_struct { * instead of smp_mb(), and given that the first (for example) * srcu_read_lock_fast() might race with the first synchronize_srcu(), * this different must be specified at initialization time. + * + * If you use any of the DEFINE_SRCU() functions within a module, the + * module-entry code will invoke init_srcu_struct() and the module-exit + * code will invoke cleanup_srcu_struct(). This means that if your module + * passes the resulting srcu_struct structure to call_srcu(), you will + * need to also pass this structure to srcu_barrier() prior to module exit. */ #ifdef MODULE # define __DEFINE_SRCU(name, fast, is_static) \ @@ -225,14 +236,16 @@ struct srcu_struct { is_static struct srcu_struct name = \ __SRCU_STRUCT_INIT(name, name##_srcu_usage, name##_srcu_data, fast) #endif -#define DEFINE_SRCU(name) __DEFINE_SRCU(name, 0, /* not static */) +#define DEFINE_SRCU(name) __DEFINE_SRCU(name, 0, /* !static */) #define DEFINE_STATIC_SRCU(name) __DEFINE_SRCU(name, 0, static) -#define DEFINE_SRCU_FAST(name) __DEFINE_SRCU(name, SRCU_READ_FLAVOR_FAST, /* not static */) +#define DEFINE_SRCU_FAST(name) __DEFINE_SRCU(name, SRCU_READ_FLAVOR_FAST, /* !static */) #define DEFINE_STATIC_SRCU_FAST(name) __DEFINE_SRCU(name, SRCU_READ_FLAVOR_FAST, static) #define DEFINE_SRCU_FAST_UPDOWN(name) __DEFINE_SRCU(name, SRCU_READ_FLAVOR_FAST_UPDOWN, \ - /* not static */) + /* !static */) #define DEFINE_STATIC_SRCU_FAST_UPDOWN(name) \ __DEFINE_SRCU(name, SRCU_READ_FLAVOR_FAST_UPDOWN, static) +#define DEFINE_SRCU_ATOMIC(name) __DEFINE_SRCU(name, SRCU_READ_FLAVOR_ATOMIC, /* !static */) +#define DEFINE_STATIC_SRCU_ATOMIC(name) __DEFINE_SRCU(name, SRCU_READ_FLAVOR_ATOMIC, static) int __srcu_read_lock(struct srcu_struct *ssp) __acquires_shared(ssp); void synchronize_srcu_expedited(struct srcu_struct *ssp); 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..d60444bf3a02 100644 --- a/kernel/rcu/rcu.h +++ b/kernel/rcu/rcu.h @@ -572,6 +572,20 @@ 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; +} + +/* 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/rcutorture.c b/kernel/rcu/rcutorture.c index 794937e13e7c..51ed35f5aab1 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; @@ -440,6 +445,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 +654,8 @@ 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), + .nmi_capable = 1, .name = "rcu" }; @@ -702,10 +710,21 @@ static struct rcu_torture_ops rcu_busted_ops = { DEFINE_STATIC_SRCU(srcu_ctl); DEFINE_STATIC_SRCU_FAST(srcu_ctlf); DEFINE_STATIC_SRCU_FAST_UPDOWN(srcu_ctlfud); +DEFINE_STATIC_SRCU_ATOMIC(srcu_ctla); static struct srcu_struct srcu_ctld; static struct srcu_struct *srcu_ctlp = &srcu_ctl; static struct rcu_torture_ops srcud_ops; +// Restrict APIs permitted for atomic SRCU. +static void srcu_torture_init_forbidden_apis(void) +{ + cur_ops->call = NULL; + cur_ops->cb_barrier = NULL; + cur_ops->deferred_free = NULL; + cur_ops->exp_current = NULL; + cur_ops->start_gp_poll = NULL; +} + static void srcu_torture_init(void) { rcu_sync_torture_init(); @@ -721,6 +740,11 @@ static void srcu_torture_init(void) srcu_ctlp = &srcu_ctlfud; VERBOSE_TOROUT_STRING("srcu_torture_init fast-up/down SRCU"); } + if (reader_flavor & SRCU_READ_FLAVOR_ATOMIC) { + srcu_ctlp = &srcu_ctla; + VERBOSE_TOROUT_STRING("srcu_torture_init atomic SRCU"); + srcu_torture_init_forbidden_apis(); + } } static void srcu_get_gp_data(int *flags, unsigned long *gp_seq) @@ -758,6 +782,11 @@ static int srcu_torture_read_lock(void) WARN_ON_ONCE(idx & ~0x1); ret += idx << 3; } + if (reader_flavor & SRCU_READ_FLAVOR_ATOMIC) { + idx = srcu_read_lock_atomic(srcu_ctlp); + WARN_ON_ONCE(idx & ~0x1); + ret += idx << 4; + } return ret; } @@ -777,7 +806,8 @@ srcu_read_delay(struct torture_random_state *rrsp, struct rt_read_seg *rtrsp) delay = torture_random(rrsp) % (nrealreaders * 2 * longdelay * uspertick); - if (!delay && !in_atomic() && !rcu_preempt_depth() && !irqs_disabled()) { + if (!delay && !in_atomic() && !rcu_preempt_depth() && !irqs_disabled() && + !(reader_flavor & SRCU_READ_FLAVOR_ATOMIC)) { schedule_timeout_interruptible(longdelay); rtrsp->rt_delay_jiffies = longdelay; } else { @@ -788,6 +818,8 @@ srcu_read_delay(struct torture_random_state *rrsp, struct rt_read_seg *rtrsp) static void srcu_torture_read_unlock(int idx) { WARN_ON_ONCE((reader_flavor && (idx & ~reader_flavor)) || (!reader_flavor && (idx & ~0x1))); + if (reader_flavor & SRCU_READ_FLAVOR_ATOMIC) + srcu_read_unlock_atomic(srcu_ctlp, (idx & 0x10) >> 4); if (reader_flavor & SRCU_READ_FLAVOR_FAST_UPDOWN) srcu_read_unlock_fast_updown(srcu_ctlp, __srcu_ctr_to_ptr(srcu_ctlp, (idx & 0x8) >> 3)); @@ -867,7 +899,10 @@ static void srcu_torture_deferred_free(struct rcu_torture *rp) static void srcu_torture_synchronize(void) { - synchronize_srcu(srcu_ctlp); + if (reader_flavor & SRCU_READ_FLAVOR_ATOMIC) + synchronize_srcu_atomic(srcu_ctlp); + else + synchronize_srcu(srcu_ctlp); } static unsigned long srcu_torture_get_gp_state(void) @@ -903,7 +938,10 @@ static void srcu_torture_stats(void) static void srcu_torture_synchronize_expedited(void) { - synchronize_srcu_expedited(srcu_ctlp); + if (reader_flavor & SRCU_READ_FLAVOR_ATOMIC) + synchronize_srcu_atomic(srcu_ctlp); + else + synchronize_srcu_expedited(srcu_ctlp); } static void srcu_torture_expedite_current(void) @@ -942,6 +980,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" }; @@ -960,6 +999,10 @@ static void srcud_torture_init(void) } else if (reader_flavor & SRCU_READ_FLAVOR_FAST_UPDOWN) { WARN_ON(init_srcu_struct_fast_updown(&srcu_ctld)); VERBOSE_TOROUT_STRING("srcud_torture_init fast-up/down SRCU"); + } else if (reader_flavor & SRCU_READ_FLAVOR_ATOMIC) { + WARN_ON(init_srcu_struct_atomic(&srcu_ctld)); + VERBOSE_TOROUT_STRING("srcud_torture_init atomic SRCU"); + srcu_torture_init_forbidden_apis(); } else { WARN_ON(init_srcu_struct(&srcu_ctld)); } @@ -1005,6 +1048,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" }; @@ -1269,6 +1313,7 @@ static struct rcu_torture_ops tasks_tracing_ops = { .cbflood_max = 50000, .irq_capable = 1, .slow_gps = 1, + .nmi_capable = 1, .name = "tasks-tracing" }; @@ -1738,7 +1783,7 @@ rcu_torture_writer(void *arg) pr_alert("%s" TORTURE_FLAG " Waited %lu jiffies for boot to complete.\n", torture_type, jiffies - j); - if (IS_ENABLED(CONFIG_RCU_LAZY)) + if (IS_ENABLED(CONFIG_RCU_LAZY) && cur_ops->call) INIT_WORK_ONSTACK(&lazy_work, rcu_torture_writer_work); do { @@ -1933,7 +1978,7 @@ rcu_torture_writer(void *arg) !rcu_gp_is_normal(); } rcu_torture_writer_state = RTWS_STUTTER; - if (IS_ENABLED(CONFIG_RCU_LAZY)) + if (IS_ENABLED(CONFIG_RCU_LAZY) && cur_ops->call) queue_work(system_percpu_wq, &lazy_work); stutter_waited = stutter_wait("rcu_torture_writer"); if (stutter_waited && @@ -1966,7 +2011,7 @@ rcu_torture_writer(void *arg) " Dynamic grace-period expediting was disabled.\n", torture_type); - if (IS_ENABLED(CONFIG_RCU_LAZY)) { + if (IS_ENABLED(CONFIG_RCU_LAZY) && cur_ops->call) { cancel_work_sync(&lazy_work); destroy_work_on_stack(&lazy_work); } @@ -2534,8 +2579,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; } @@ -2660,6 +2707,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 * counter in the element should never be greater than 1, otherwise, the @@ -3047,6 +3212,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); @@ -3195,7 +3363,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, @@ -3209,7 +3377,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); } @@ -3856,7 +4024,7 @@ static int __init rcu_torture_fwd_prog_init(void) } if (fwd_progress_holdoff <= 0) fwd_progress_holdoff = 1; - if (fwd_progress_div <= 0) + if (fwd_progress_div < 2) fwd_progress_div = 4; rfp = kzalloc_objs(*rfp, fwd_progress); fwd_prog_tasks = kzalloc_objs(*fwd_prog_tasks, fwd_progress); @@ -4283,6 +4451,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(); @@ -4322,9 +4491,12 @@ rcu_torture_cleanup(void) for (i = 0; i < nrealreaders; i++) torture_stop_kthread(rcu_torture_reader, reader_tasks[i]); + if (irqreader && cur_ops->irq_capable) + kick_all_cpus_sync(); kfree(reader_tasks); reader_tasks = NULL; } + rcu_torture_nmi_cleanup(); kfree(rcu_torture_reader_mbchk); rcu_torture_reader_mbchk = NULL; @@ -4354,6 +4526,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(); @@ -4786,6 +4971,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, diff --git a/kernel/rcu/srcutiny.c b/kernel/rcu/srcutiny.c index 558ba8d316db..99f8bfd98b04 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; @@ -38,11 +41,14 @@ static int init_srcu_struct_fields(struct srcu_struct *ssp) ssp->srcu_cb_tail = &ssp->srcu_cb_head; ssp->srcu_gp_running = false; ssp->srcu_gp_waiting = false; + ssp->srcu_atomic_gp_flag = 0; ssp->srcu_idx = 0; ssp->srcu_idx_max = 0; 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 +92,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); @@ -144,6 +160,8 @@ void srcu_drive_gp(struct work_struct *wp) WRITE_ONCE(ssp->srcu_idx, ssp->srcu_idx + 1); WRITE_ONCE(ssp->srcu_gp_waiting, true); /* srcu_read_unlock() wakes! */ preempt_enable(); + if (IS_ENABLED(CONFIG_PREEMPTION)) + synchronize_rcu(); // Needed for RCU Tasks Trace to imply RCU grace period do { // Deadlock issues prevent __srcu_read_unlock() from // doing an unconditional wakeup, so polling is required. @@ -213,11 +231,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 +249,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); /* @@ -261,6 +341,95 @@ void synchronize_srcu(struct srcu_struct *ssp) EXPORT_SYMBOL_GPL(synchronize_srcu); /* + * synchronize_srcu_atomic - spinning grace period for atomic-reader domains + * @ssp: srcu_struct with which to synchronize. + * + * On !SMP this cannot spin: a reader observed mid-section is preempted + * or interrupted-out, and can only finish if we yield the CPU. But it + * is also never needed: an atomic-flavor reader (preemption disabled) + * cannot be observed mid-section from process context on the sole CPU. + * So a reader observed here has broken the atomic-domain promise, and + * the only correct wait for it is a real grace period. + * + * (Actual kernel-doc header is in Tree SRCU.) + */ +void synchronize_srcu_atomic(struct srcu_struct *ssp) +{ + int idx; + bool ret; + unsigned long srcu_state = get_state_synchronize_srcu(ssp); + + srcu_lock_sync(&ssp->dep_map); + + if (IS_ENABLED(CONFIG_PREEMPTION)) + synchronize_rcu(); // Needed for RCU Tasks Trace to imply RCU grace period. + // And in Tiny RCU, it is near zero cost and doesn't block. + + // Usually, there will be no readers. + preempt_disable(); // Guard against lazy preemption and some other grace period. + ret = !READ_ONCE(ssp->srcu_lock_nesting[0]) && !READ_ONCE(ssp->srcu_lock_nesting[1]); + if (ret) { + WRITE_ONCE(ssp->srcu_idx_max, ssp->srcu_idx + 2); + WRITE_ONCE(ssp->srcu_idx, ssp->srcu_idx + 2); + preempt_enable(); + return; + } + + // Because readers disable preemption, we should never get here. + // However, a splat and some spinning is usually preferable to + // memory corruption due to a too-short grace period. There is + // the possibility that this will hang if the preempted reader is + // not looked upon favorably by the scheduler, but this is still + // preferable to memory corruption. + WARN_ON_ONCE(1); + + // Wait to drive a grace period or for someone else to do it + // for us while we are lazily preempted. + while (ssp->srcu_atomic_gp_flag) { + if (poll_state_synchronize_srcu(ssp, srcu_state)) { + preempt_enable(); + return; + } + preempt_enable(); + cpu_relax(); + cond_resched_tasks_rcu_qs(); + preempt_disable(); + } + ssp->srcu_atomic_gp_flag = 1; + preempt_enable(); + + // We get here if a reader has been lazily preempted. + // First, wait for old readers, which are quite unlikely. + WRITE_ONCE(ssp->srcu_idx_max, get_state_synchronize_srcu(ssp)); + idx = !(((READ_ONCE(ssp->srcu_idx) + 1) & 0x2) >> 1); + while (READ_ONCE(ssp->srcu_lock_nesting[idx])) { + cond_resched_tasks_rcu_qs(); + cpu_relax(); + } + + // Next, flip the index and wait for the other group of readers. + WRITE_ONCE(ssp->srcu_idx, ssp->srcu_idx + 1); + idx = !idx; + while (READ_ONCE(ssp->srcu_lock_nesting[idx])) { + cond_resched_tasks_rcu_qs(); + cpu_relax(); + } + + // Finally, flip the index again for poll_state_synchronize_srcu(). + WRITE_ONCE(ssp->srcu_idx, ssp->srcu_idx + 1); + WARN_ON_ONCE(!poll_state_synchronize_srcu(ssp, srcu_state)); +} +EXPORT_SYMBOL_GPL(synchronize_srcu_atomic); + +/* 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 */ unsigned long get_state_synchronize_srcu(struct srcu_struct *ssp) @@ -280,6 +449,11 @@ EXPORT_SYMBOL_GPL(get_state_synchronize_srcu); * The difference between this and get_state_synchronize_srcu() is that * this function ensures that the poll_state_synchronize_srcu() will * eventually return the value true. + * + * This function cannot be used with atomic SRCU, which only has + * atomic grace periods. Doing so will silently corrupt internal + * SRCU state. Tree SRCU has appropriate checking with splats, + * so please test with CONFIG_SMP=y as well as CONFIG_SMP=n. */ unsigned long start_poll_synchronize_srcu(struct srcu_struct *ssp) { diff --git a/kernel/rcu/srcutree.c b/kernel/rcu/srcutree.c index 01f19bb17dd6..402718d08588 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; } } @@ -200,8 +238,10 @@ static bool init_srcu_struct_nodes(struct srcu_struct *ssp, gfp_t gfp_flags) * Initialize non-compile-time initialized fields, including the * associated srcu_node and srcu_data structures. The is_static parameter * tells us that ->sda has already been wired up to srcu_data. + * The is_atomic parameter tells us that there is no reason to + * ever transition to big. */ -static int init_srcu_struct_fields(struct srcu_struct *ssp, bool is_static) +static int init_srcu_struct_fields(struct srcu_struct *ssp, bool is_static, bool is_atomic) { if (!is_static) ssp->srcu_sup = kzalloc_obj(*ssp->srcu_sup); @@ -213,12 +253,18 @@ static int init_srcu_struct_fields(struct srcu_struct *ssp, bool is_static) ssp->srcu_sup->node = NULL; mutex_init(&ssp->srcu_sup->srcu_cb_mutex); mutex_init(&ssp->srcu_sup->srcu_gp_mutex); + atomic_set(&ssp->srcu_sup->srcu_atomic_gp_flag, 0); ssp->srcu_sup->srcu_gp_seq = SRCU_GP_SEQ_INITIAL_VAL; ssp->srcu_sup->srcu_barrier_seq = 0; mutex_init(&ssp->srcu_sup->srcu_barrier_mutex); atomic_set(&ssp->srcu_sup->srcu_barrier_cpu_cnt, 0); INIT_DELAYED_WORK(&ssp->srcu_sup->work, process_srcu); - init_irq_work(&ssp->srcu_sup->irq_work, srcu_irq_work); + /* + * trace events started on the command line require SRCU before + * the irq_work kthread starts. Since all it does is a simple + * wakeup, having it as a hard irq, even on PREEMPT_RT is fine. + */ + ssp->srcu_sup->irq_work = IRQ_WORK_INIT_HARD(srcu_irq_work); ssp->srcu_sup->sda_is_static = is_static; if (!is_static) { ssp->sda = alloc_percpu(struct srcu_data); @@ -229,7 +275,8 @@ static int init_srcu_struct_fields(struct srcu_struct *ssp, bool is_static) init_srcu_struct_data(ssp); ssp->srcu_sup->srcu_gp_seq_needed_exp = SRCU_GP_SEQ_INITIAL_VAL; ssp->srcu_sup->srcu_last_gp_end = ktime_get_mono_fast_ns(); - if (READ_ONCE(ssp->srcu_sup->srcu_size_state) == SRCU_SIZE_SMALL && SRCU_SIZING_IS_INIT()) { + if (!is_atomic && + READ_ONCE(ssp->srcu_sup->srcu_size_state) == SRCU_SIZE_SMALL && SRCU_SIZING_IS_INIT()) { if (!preemptible()) WRITE_ONCE(ssp->srcu_sup->srcu_size_state, SRCU_SIZE_ALLOC); else if (init_srcu_struct_nodes(ssp, GFP_KERNEL)) @@ -258,26 +305,27 @@ err_free_sup: #ifdef CONFIG_DEBUG_LOCK_ALLOC static int -__init_srcu_struct_common(struct srcu_struct *ssp, const char *name, struct lock_class_key *key) +__init_srcu_struct_common(struct srcu_struct *ssp, const char *name, + struct lock_class_key *key, bool is_atomic) { /* Don't re-initialize a lock while it is held. */ debug_check_no_locks_freed((void *)ssp, sizeof(*ssp)); lockdep_init_map(&ssp->dep_map, name, key, 0); - return init_srcu_struct_fields(ssp, false); + return init_srcu_struct_fields(ssp, false, is_atomic); } int init_srcu_struct_lockdep(struct srcu_struct *ssp, const char *name, struct lock_class_key *key) { ssp->srcu_reader_flavor = 0; - return __init_srcu_struct_common(ssp, name, key); + return __init_srcu_struct_common(ssp, name, key, false); } EXPORT_SYMBOL_GPL(init_srcu_struct_lockdep); int __init_srcu_struct_fast(struct srcu_struct *ssp, const char *name, struct lock_class_key *key) { ssp->srcu_reader_flavor = SRCU_READ_FLAVOR_FAST; - return __init_srcu_struct_common(ssp, name, key); + return __init_srcu_struct_common(ssp, name, key, false); } EXPORT_SYMBOL_GPL(__init_srcu_struct_fast); @@ -285,10 +333,17 @@ int __init_srcu_struct_fast_updown(struct srcu_struct *ssp, const char *name, struct lock_class_key *key) { ssp->srcu_reader_flavor = SRCU_READ_FLAVOR_FAST_UPDOWN; - return __init_srcu_struct_common(ssp, name, key); + return __init_srcu_struct_common(ssp, name, key, false); } EXPORT_SYMBOL_GPL(__init_srcu_struct_fast_updown); +int __init_srcu_struct_atomic(struct srcu_struct *ssp, const char *name, struct lock_class_key *key) +{ + ssp->srcu_reader_flavor = SRCU_READ_FLAVOR_ATOMIC; + return __init_srcu_struct_common(ssp, name, key, true); +} +EXPORT_SYMBOL_GPL(__init_srcu_struct_atomic); + #else /* #ifdef CONFIG_DEBUG_LOCK_ALLOC */ /** @@ -305,7 +360,7 @@ EXPORT_SYMBOL_GPL(__init_srcu_struct_fast_updown); int init_srcu_struct_generic(struct srcu_struct *ssp) { ssp->srcu_reader_flavor = 0; - return init_srcu_struct_fields(ssp, false); + return init_srcu_struct_fields(ssp, false, false); } EXPORT_SYMBOL_GPL(init_srcu_struct_generic); @@ -322,7 +377,7 @@ EXPORT_SYMBOL_GPL(init_srcu_struct_generic); int init_srcu_struct_fast(struct srcu_struct *ssp) { ssp->srcu_reader_flavor = SRCU_READ_FLAVOR_FAST; - return init_srcu_struct_fields(ssp, false); + return init_srcu_struct_fields(ssp, false, false); } EXPORT_SYMBOL_GPL(init_srcu_struct_fast); @@ -340,10 +395,30 @@ EXPORT_SYMBOL_GPL(init_srcu_struct_fast); int init_srcu_struct_fast_updown(struct srcu_struct *ssp) { ssp->srcu_reader_flavor = SRCU_READ_FLAVOR_FAST_UPDOWN; - return init_srcu_struct_fields(ssp, false); + return init_srcu_struct_fields(ssp, false, false); } EXPORT_SYMBOL_GPL(init_srcu_struct_fast_updown); +/** + * init_srcu_struct_atomic - initialize an atomic sleep-RCU structure + * @ssp: structure to initialize. + * + * Use this in place of DEFINE_SRCU_ATOMIC() and DEFINE_STATIC_SRCU_ATOMIC() + * for non-static srcu_struct structures that are to be passed to + * srcu_read_lock_atomic() and friends. It is necessary to invoke this on a + * given srcu_struct before passing that srcu_struct to any other function. + * Each srcu_struct represents a separate domain of SRCU protection. + * + * And yes, we really are defining a sleepable RCU implementation that + * cannot sleep. Strange universe we live in, isn't it? + */ +int init_srcu_struct_atomic(struct srcu_struct *ssp) +{ + ssp->srcu_reader_flavor = SRCU_READ_FLAVOR_ATOMIC; + return init_srcu_struct_fields(ssp, false, true); +} +EXPORT_SYMBOL_GPL(init_srcu_struct_atomic); + #endif /* #else #ifdef CONFIG_DEBUG_LOCK_ALLOC */ /* @@ -432,9 +507,11 @@ static void raw_spin_lock_irqsave_ssp_contention(struct srcu_struct *ssp, unsign * done with compile-time initialization, so this check is added * to each update-side SRCU primitive. Use ssp->lock, which -is- * compile-time initialized, to resolve races involving multiple - * CPUs trying to garner first-use privileges. + * CPUs trying to garner first-use privileges. The is_atomic + * parameter tells us that there will never be a reason to + * transition to big. */ -static void check_init_srcu_struct(struct srcu_struct *ssp) +static void check_init_srcu_struct(struct srcu_struct *ssp, bool is_atomic) { unsigned long flags; @@ -446,7 +523,7 @@ static void check_init_srcu_struct(struct srcu_struct *ssp) raw_spin_unlock_irqrestore_rcu_node(ssp->srcu_sup, flags); return; } - init_srcu_struct_fields(ssp, true); + init_srcu_struct_fields(ssp, true, is_atomic); raw_spin_unlock_irqrestore_rcu_node(ssp->srcu_sup, flags); } @@ -688,6 +765,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 +780,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) { @@ -704,7 +788,7 @@ void cleanup_srcu_struct(struct srcu_struct *ssp) // Call srcu_barrier() before this cleanup_srcu_struct() // to avoid triggering this WARN_ON(). if (WARN_ON(timer_delete_sync(&sdp->delay_work) && - rcu_segcblist_n_cbs(&sdp->srcu_cblist)) && + !rcu_segcblist_empty(&sdp->srcu_cblist)) && rcu_cpu_beenfullyonline(sdp->cpu)) queue_work_on(sdp->cpu, rcu_gp_wq, &sdp->work); flush_work(&sdp->work); @@ -752,7 +836,10 @@ void __srcu_check_read_flavor(struct srcu_struct *ssp, int read_flavor) WARN_ON_ONCE(ssp->srcu_reader_flavor && read_flavor != ssp->srcu_reader_flavor); WARN_ON_ONCE(old_read_flavor && ssp->srcu_reader_flavor && old_read_flavor != ssp->srcu_reader_flavor); - WARN_ON_ONCE(read_flavor == SRCU_READ_FLAVOR_FAST && !ssp->srcu_reader_flavor); + WARN_ON_ONCE(!!(read_flavor & SRCU_READ_FLAVOR_PREDEF) && + read_flavor != ssp->srcu_reader_flavor); + WARN_ON_ONCE(!!(ssp->srcu_reader_flavor & SRCU_READ_FLAVOR_PREDEF) && + read_flavor != ssp->srcu_reader_flavor); if (!old_read_flavor) { old_read_flavor = cmpxchg(&sdp->srcu_reader_flavor, 0, read_flavor); if (!old_read_flavor) @@ -892,7 +979,7 @@ static void srcu_schedule_cbs_snp(struct srcu_struct *ssp, struct srcu_node *snp * are initiating callback invocation. This allows the ->srcu_have_cbs[] * array to have a finite number of elements. */ -static void srcu_gp_end(struct srcu_struct *ssp) +static void srcu_gp_end(struct srcu_struct *ssp, bool is_atomic) { unsigned long cbdelay = 1; bool cbs; @@ -908,7 +995,8 @@ static void srcu_gp_end(struct srcu_struct *ssp) struct srcu_usage *sup = ssp->srcu_sup; /* Prevent more than one additional grace period. */ - mutex_lock(&sup->srcu_cb_mutex); + if (!is_atomic) + mutex_lock(&sup->srcu_cb_mutex); /* End the current grace period. */ raw_spin_lock_irq_rcu_node(sup); @@ -923,15 +1011,16 @@ static void srcu_gp_end(struct srcu_struct *ssp) if (ULONG_CMP_LT(sup->srcu_gp_seq_needed_exp, gpseq)) WRITE_ONCE(sup->srcu_gp_seq_needed_exp, gpseq); raw_spin_unlock_irq_rcu_node(sup); - mutex_unlock(&sup->srcu_gp_mutex); + if (!is_atomic) + mutex_unlock(&sup->srcu_gp_mutex); /* A new grace period can start at this point. But only one. */ /* Initiate callback invocation as needed. */ ss_state = smp_load_acquire(&sup->srcu_size_state); - if (ss_state < SRCU_SIZE_WAIT_BARRIER) { + if (!is_atomic && ss_state < SRCU_SIZE_WAIT_BARRIER) { srcu_schedule_cbs_sdp(per_cpu_ptr(ssp->sda, get_boot_cpu_id()), cbdelay); - } else { + } else if (!is_atomic) { idx = rcu_seq_ctr(gpseq) % ARRAY_SIZE(snp->srcu_have_cbs); srcu_for_each_node_breadth_first(ssp, snp) { raw_spin_lock_irq_rcu_node(snp); @@ -968,13 +1057,15 @@ static void srcu_gp_end(struct srcu_struct *ssp) } /* Callback initiation done, allow grace periods after next. */ - mutex_unlock(&sup->srcu_cb_mutex); + if (!is_atomic) + mutex_unlock(&sup->srcu_cb_mutex); /* Start a new grace period if needed. */ raw_spin_lock_irq_rcu_node(sup); gpseq = rcu_seq_current(&sup->srcu_gp_seq); if (!rcu_seq_state(gpseq) && ULONG_CMP_LT(gpseq, sup->srcu_gp_seq_needed)) { + WARN_ON_ONCE(ssp->srcu_reader_flavor & SRCU_READ_FLAVOR_ATOMIC); srcu_gp_start(ssp); raw_spin_unlock_irq_rcu_node(sup); srcu_reschedule(ssp, 0); @@ -982,8 +1073,10 @@ static void srcu_gp_end(struct srcu_struct *ssp) raw_spin_unlock_irq_rcu_node(sup); } - /* Transition to big if needed. */ - if (ss_state != SRCU_SIZE_SMALL && ss_state != SRCU_SIZE_BIG) { + /* Transition to big if needed, but never for atomic SRCU. */ + if (ssp->srcu_reader_flavor == SRCU_READ_FLAVOR_ATOMIC && ss_state != SRCU_SIZE_SMALL) { + WARN_ON_ONCE(1); + } else if (ss_state != SRCU_SIZE_SMALL && ss_state != SRCU_SIZE_BIG) { if (ss_state == SRCU_SIZE_ALLOC) init_srcu_struct_nodes(ssp, GFP_KERNEL); else @@ -1098,6 +1191,7 @@ static void srcu_funnel_gp_start(struct srcu_struct *ssp, struct srcu_data *sdp, /* If grace period not already in progress, start it. */ if (!WARN_ON_ONCE(rcu_seq_done(&sup->srcu_gp_seq, s)) && rcu_seq_state(sup->srcu_gp_seq) == SRCU_STATE_IDLE) { + WARN_ON_ONCE(ssp->srcu_reader_flavor & SRCU_READ_FLAVOR_ATOMIC); srcu_gp_start(ssp); // And how can that list_add() in the "else" clause @@ -1234,7 +1328,7 @@ static bool srcu_should_expedite(struct srcu_struct *ssp) unsigned long t; unsigned long tlast; - check_init_srcu_struct(ssp); + check_init_srcu_struct(ssp, false); /* If _lite() readers, don't do unsolicited expediting. */ if (this_cpu_read(ssp->sda->srcu_reader_flavor) & SRCU_READ_FLAVOR_SLOWGP) return false; @@ -1293,7 +1387,7 @@ static unsigned long srcu_gp_start_if_needed(struct srcu_struct *ssp, struct srcu_node *sdp_mynode; int ss_state; - check_init_srcu_struct(ssp); + check_init_srcu_struct(ssp, false); /* * While starting a new grace period, make sure we are in an * SRCU read-side critical section so that the grace-period @@ -1411,8 +1505,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 +1518,110 @@ 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 (WARN_ON_ONCE(ssp->srcu_reader_flavor == SRCU_READ_FLAVOR_ATOMIC)) + return; // Leak the callback rather than corrupt SRCU state. + 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 @@ -1469,8 +1667,13 @@ static void __synchronize_srcu(struct srcu_struct *ssp, bool do_norm) if (rcu_scheduler_active == RCU_SCHEDULER_INACTIVE) return; + if (WARN_ON_ONCE(ssp->srcu_reader_flavor == SRCU_READ_FLAVOR_ATOMIC)) { + // This works, and exposes a possible bug. + synchronize_srcu_atomic(ssp); + return; + } might_sleep(); - check_init_srcu_struct(ssp); + check_init_srcu_struct(ssp, false); init_completion(&rcu.completion); init_rcu_head_on_stack(&rcu.head); __call_srcu(ssp, &rcu.head, wakeme_after_rcu, do_norm); @@ -1589,10 +1792,21 @@ EXPORT_SYMBOL_GPL(get_state_synchronize_srcu); * period has elapsed in the meantime. Unlike get_state_synchronize_srcu(), * this function also ensures that any needed SRCU grace period will be * started. This convenience does come at a cost in terms of CPU overhead. + * + * This function cannot be used with atomic SRCU, which only has + * atomic grace periods. Give a warning if someone tries, and return + * the same cookie that would have been returned, but refrain from + * messing up state by starting a grace period. If someone somewhere + * somehow invokes synchronize_srcu_atomic(), passing this cookie to + * poll_state_synchronize_srcu() will return true. If no one ever invokes + * synchronize_srcu_atomic(), too bad. */ unsigned long start_poll_synchronize_srcu(struct srcu_struct *ssp) { - return srcu_gp_start_if_needed(ssp, NULL, true); + if (WARN_ON_ONCE(ssp->srcu_reader_flavor == SRCU_READ_FLAVOR_ATOMIC)) + return get_state_synchronize_srcu(ssp); + else + return srcu_gp_start_if_needed(ssp, NULL, true); } EXPORT_SYMBOL_GPL(start_poll_synchronize_srcu); @@ -1678,9 +1892,21 @@ 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, false); + // Atomic SRCU has no callbacks, so there is nothing to wait on. + if (WARN_ON_ONCE(ssp->srcu_reader_flavor == SRCU_READ_FLAVOR_ATOMIC)) + return; + + /* + * 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(); - check_init_srcu_struct(ssp); + 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. */ @@ -1743,6 +1969,9 @@ static void srcu_expedite_current_cb(struct rcu_head *rhp) * no current grace period, one might be created. If the current grace * period is currently sleeping, that sleep will complete before expediting * will take effect. + * + * This function must not be invoked on srcu_struct structures that are + * used with srcu_read_lock_atomic() and synchronize_srcu_atomic(). */ void srcu_expedite_current(struct srcu_struct *ssp) { @@ -1750,6 +1979,9 @@ void srcu_expedite_current(struct srcu_struct *ssp) bool needcb = false; struct srcu_data *sdp; + // Atomic SRCU has no callbacks, so there is nothing to expedite. + if (WARN_ON_ONCE(ssp->srcu_reader_flavor == SRCU_READ_FLAVOR_ATOMIC)) + return; migrate_disable(); sdp = this_cpu_ptr(ssp->sda); raw_spin_lock_irqsave_sdp_contention(sdp, &flags); @@ -1785,13 +2017,16 @@ EXPORT_SYMBOL_GPL(srcu_batches_completed); /* * Core SRCU state machine. Push state bits of ->srcu_gp_seq * to SRCU_STATE_SCAN2, and invoke srcu_gp_end() when scan has - * completed in that state. + * completed in that state. Set is_atomic to indicate that + * the caller is excluding other calls so that ->srcu_gp_mutex + * is not needed, and to indicate that blocking is forbidden. */ -static void srcu_advance_state(struct srcu_struct *ssp) +static void srcu_advance_state(struct srcu_struct *ssp, bool is_atomic) { int idx; - mutex_lock(&ssp->srcu_sup->srcu_gp_mutex); + if (!is_atomic) + mutex_lock(&ssp->srcu_sup->srcu_gp_mutex); /* * Because readers might be delayed for an extended period after @@ -1809,15 +2044,19 @@ static void srcu_advance_state(struct srcu_struct *ssp) if (ULONG_CMP_GE(ssp->srcu_sup->srcu_gp_seq, ssp->srcu_sup->srcu_gp_seq_needed)) { WARN_ON_ONCE(rcu_seq_state(ssp->srcu_sup->srcu_gp_seq)); raw_spin_unlock_irq_rcu_node(ssp->srcu_sup); - mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); + if (!is_atomic) + mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); return; } idx = rcu_seq_state(READ_ONCE(ssp->srcu_sup->srcu_gp_seq)); - if (idx == SRCU_STATE_IDLE) + if (idx == SRCU_STATE_IDLE) { + WARN_ON_ONCE(ssp->srcu_reader_flavor & SRCU_READ_FLAVOR_ATOMIC); srcu_gp_start(ssp); + } raw_spin_unlock_irq_rcu_node(ssp->srcu_sup); if (idx != SRCU_STATE_IDLE) { - mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); + if (!is_atomic) + mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); return; /* Someone else started the grace period. */ } } @@ -1825,7 +2064,8 @@ static void srcu_advance_state(struct srcu_struct *ssp) if (rcu_seq_state(READ_ONCE(ssp->srcu_sup->srcu_gp_seq)) == SRCU_STATE_SCAN1) { idx = !(ssp->srcu_ctrp - &ssp->sda->srcu_ctrs[0]); if (!try_check_zero(ssp, idx, 1)) { - mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); + if (!is_atomic) + mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); return; /* readers present, retry later. */ } srcu_flip(ssp); @@ -1843,14 +2083,137 @@ static void srcu_advance_state(struct srcu_struct *ssp) */ idx = !(ssp->srcu_ctrp - &ssp->sda->srcu_ctrs[0]); if (!try_check_zero(ssp, idx, 2)) { - mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); + if (!is_atomic) + mutex_unlock(&ssp->srcu_sup->srcu_gp_mutex); return; /* readers present, retry later. */ } ssp->srcu_sup->srcu_n_exp_nodelay = 0; - srcu_gp_end(ssp); /* Releases ->srcu_gp_mutex. */ + srcu_gp_end(ssp, is_atomic); /* Releases ->srcu_gp_mutex. */ } } +/** + * synchronize_srcu_atomic - spin for prior SRCU read-side critical-section completion + * @ssp: srcu_struct with which to synchronize. + * + * Similar to synchronize_srcu(), but spins rather than blocking. + * Use only with srcu_read_lock_atomic() and srcu_read_unlock_atomic(), + * which are forbidden from voluntarily context switching. If + * synchronize_srcu_atomic() is invoked from a more restrictive context + * (for example, interrupts disabled) for a given srcu_struct structure, + * then for that structure, all calls to both srcu_read_lock_atomic() + * and srcu_read_unlock_atomic() must be invoked from that same context, + * or one that is even more strict. + * + * If synchronize_srcu_atomic() is invoked on a given srcu_struct + * structure, then none of call_srcu(), synchronize_srcu(), + * synchronize_srcu_expedited(), start_poll_synchronize_srcu(), + * srcu_barrier(), or srcu_expedite_current() may be invoked on that + * same structure. + * + * Because synchronize_srcu_atomic() is even more expedited than is + * synchronize_srcu_expedited(), there is no expedited counterpart to + * this function. + */ +void synchronize_srcu_atomic(struct srcu_struct *ssp) +{ + unsigned long srcu_state; + struct srcu_usage *sup = ssp->srcu_sup; + unsigned long rdm0, rdm1; + unsigned long unlocks0, unlocks1; + + // Initialize. Either init_srcu_struct() was invoked or + // DEFINE_SRCU() or similar was used. Therefore, no allocation + // will be done here. + check_init_srcu_struct(ssp, true); + srcu_check_read_flavor(ssp, SRCU_READ_FLAVOR_ATOMIC); + + // Perhaps others will do our work for us. + srcu_state = get_state_synchronize_srcu(ssp); + preempt_disable(); + while (atomic_read(&sup->srcu_atomic_gp_flag) || + atomic_xchg(&sup->srcu_atomic_gp_flag, 1)) { + preempt_enable(); + if (poll_state_synchronize_srcu(ssp, srcu_state)) + return; + cpu_relax(); + preempt_disable(); + } + + // One last check for others doing our work for us under the lock. + raw_spin_lock_irq_rcu_node(sup); + if (poll_state_synchronize_srcu(ssp, srcu_state)) { + raw_spin_unlock_irq_rcu_node(sup); + atomic_set(&sup->srcu_atomic_gp_flag, 0); + preempt_enable(); + return; + } + + // OK, we really have to do it ourselves. Start the grace period. + non_block_start(); // We must not voluntarily block! + smp_store_release(&sup->srcu_gp_seq_needed, srcu_state); // See srcu_funnel_gp_start(). + ASSERT_EXCLUSIVE_WRITER(ssp->srcu_sup->srcu_gp_seq); + srcu_gp_start(ssp); + raw_spin_unlock_irq_rcu_node(sup); + + // + // Fastpath: If there are no readers at all, neither grace-period + // scan need wait, so both can be satisfied at once without doing + // the index flip. The counter-sum proof is the same as that of + // srcu_readers_active_idx_check(), but spanning both indices. + // Atomic SRCU guarantees that all readers are of + // SRCU_READ_FLAVOR_ATOMIC, so the SLOWGP check never triggers and + // the ->srcu_reader_flavor masks returned by + // srcu_readers_unlock_idx() are unused. + // + // This proof must follow the grace-period anchor written by the + // srcu_gp_start() above, never precede it. With the anchor first, + // a reader whose lock increment is missed by the sums below cannot + // have incremented its lock counter before the anchor, and therefore + // cannot be a pre-existing reader of this grace period. Placing the + // proof before the anchor would let this grace period miss a + // pre-existing reader and return without waiting for it. + // + // The smp_mb() pairs with the smp_mb() in __srcu_read_lock() + // (store-buffering pattern), which guarantees that a lock is always + // counted if the corresponding unlock is counted, the same + // memory-ordering guarantee as is provided by + // srcu_readers_active_idx_check(). + // + unlocks0 = srcu_readers_unlock_idx(ssp, 0, &rdm0); + unlocks1 = srcu_readers_unlock_idx(ssp, 1, &rdm1); + smp_mb(); /* A */ + if (srcu_readers_lock_idx(ssp, 0, false, unlocks0) && + srcu_readers_lock_idx(ssp, 1, false, unlocks1)) { + // No readers, so end this grace period manually, skipping + // the index flip. Advancing the sequence number via + // rcu_seq_start() in srcu_gp_start() above and rcu_seq_end() + // below keeps get_state_synchronize_srcu() and + // poll_state_synchronize_srcu() working, all under ->lock + // and ->srcu_atomic_gp_flag, which excludes concurrent + // sequence-number updates. + raw_spin_lock_irq_rcu_node(sup); + rcu_seq_end(&sup->srcu_gp_seq); + raw_spin_unlock_irq_rcu_node(sup); + WARN_ON_ONCE(!poll_state_synchronize_srcu(ssp, srcu_state)); + atomic_set_release(&sup->srcu_atomic_gp_flag, 0); + preempt_enable(); + non_block_end(); + return; + } + + // Wait for it to complete, helping it along. + while (!poll_state_synchronize_srcu(ssp, srcu_state)) { + cpu_relax(); + srcu_advance_state(ssp, true); + } + ASSERT_EXCLUSIVE_WRITER(sup->srcu_atomic_gp_flag); + atomic_set_release(&sup->srcu_atomic_gp_flag, 0); + preempt_enable(); + non_block_end(); +} +EXPORT_SYMBOL_GPL(synchronize_srcu_atomic); + /* * Invoke a limited number of SRCU callbacks that have passed through * their grace period. If there are more to do, SRCU will reschedule @@ -1930,6 +2293,7 @@ static void srcu_reschedule(struct srcu_struct *ssp, unsigned long delay) } } else if (!rcu_seq_state(ssp->srcu_sup->srcu_gp_seq)) { /* Outstanding request and no GP. Start one. */ + WARN_ON_ONCE(ssp->srcu_reader_flavor & SRCU_READ_FLAVOR_ATOMIC); srcu_gp_start(ssp); } raw_spin_unlock_irq_rcu_node(ssp->srcu_sup); @@ -1951,7 +2315,7 @@ static void process_srcu(struct work_struct *work) sup = container_of(work, struct srcu_usage, work.work); ssp = sup->srcu_ssp; - srcu_advance_state(ssp); + srcu_advance_state(ssp, false); raw_spin_lock_irq_rcu_node(ssp->srcu_sup); curdelay = srcu_get_delay(ssp); raw_spin_unlock_irq_rcu_node(ssp->srcu_sup); @@ -2058,8 +2422,10 @@ void srcu_torture_stats_print(struct srcu_struct *ssp, char *tt, char *tf) } pr_cont(" T(%ld,%ld)\n", s0, s1); } - if (SRCU_SIZING_IS_TORTURE()) - srcu_transition_to_big(ssp); + if (SRCU_SIZING_IS_TORTURE()) { + if (!WARN_ON_ONCE(ssp->srcu_reader_flavor == SRCU_READ_FLAVOR_ATOMIC)) + srcu_transition_to_big(ssp); + } } EXPORT_SYMBOL_GPL(srcu_torture_stats_print); @@ -2135,6 +2501,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/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); diff --git a/kernel/rcu/tree.c b/kernel/rcu/tree.c index f60252390d5d..aea3aef7397e 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> @@ -3147,21 +3148,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. @@ -3177,7 +3176,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); @@ -3205,6 +3203,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); @@ -3895,8 +3990,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. */ @@ -4240,6 +4339,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); } @@ -4537,6 +4639,15 @@ 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); + /* Likewise for the outgoing CPU's deferred call_srcu() callbacks. */ + srcu_offline_drain(cpu); + 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. */ 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; +} diff --git a/tools/testing/selftests/rcutorture/bin/kvm-remote.sh b/tools/testing/selftests/rcutorture/bin/kvm-remote.sh index 48a8052d5dae..a8397f86e49d 100755 --- a/tools/testing/selftests/rcutorture/bin/kvm-remote.sh +++ b/tools/testing/selftests/rcutorture/bin/kvm-remote.sh @@ -6,6 +6,11 @@ # Usage: kvm-remote.sh "systems" [ <kvm.sh args> ] # kvm-remote.sh "systems" /path/to/old/run [ <kvm-again.sh args> ] # +# The caller may set the KVM_REMOTE_SSH environment in order to specify +# an alternative ssh command, which is necessary in some environments +# for authentication purposes. This alternative ssn command must support +# ssh's usual arguments. +# # Copyright (C) 2021 Facebook, Inc. # # Authors: Paul E. McKenney <paulmck@kernel.org> @@ -13,6 +18,11 @@ scriptname=$0 args="$*" +if test -z "${KVM_REMOTE_SSH}" +then + KVM_REMOTE_SSH=ssh; export KVM_REMOTE_SSH +fi + if ! test -d tools/testing/selftests/rcutorture/bin then echo $scriptname must be run from top-level directory of kernel source tree. @@ -137,7 +147,7 @@ chmod +x $T/bin/kvm-remote-*.sh # Check first to avoid the need for cleanup for system-name typos for i in $systems do - ssh -o BatchMode=yes $i getconf _NPROCESSORS_ONLN > $T/ssh.stdout 2> $T/ssh.stderr + ${KVM_REMOTE_SSH} -o BatchMode=yes $i getconf _NPROCESSORS_ONLN > $T/ssh.stdout 2> $T/ssh.stderr ret=$? if test "$ret" -ne 0 then @@ -158,14 +168,14 @@ echo Build-products tarball: `du -h $T/binres.tgz` | tee -a "$oldrun/remote-log" for i in $systems do echo Downloading tarball to $i `date` | tee -a "$oldrun/remote-log" - cat $T/binres.tgz | ssh -o BatchMode=yes $i "cd /tmp; tar -xzf -" + cat $T/binres.tgz | ${KVM_REMOTE_SSH} -o BatchMode=yes $i "cd /tmp; tar -xzf -" ret=$? tries=0 while test "$ret" -ne 0 do echo Unable to download $T/binres.tgz to system $i, waiting and then retrying. $tries prior retries. | tee -a "$oldrun/remote-log" sleep 60 - cat $T/binres.tgz | ssh -o BatchMode=yes $i "cd /tmp; tar -xzf -" + cat $T/binres.tgz | ${KVM_REMOTE_SSH} -o BatchMode=yes $i "cd /tmp; tar -xzf -" ret=$? if test "$ret" -ne 0 then @@ -191,7 +201,7 @@ checkremotefile () { while : do - ssh -o BatchMode=yes $1 "test -f \"$2\"" + ${KVM_REMOTE_SSH} -o BatchMode=yes $1 "test -f \"$2\"" ret=$? if test "$ret" -eq 255 then @@ -239,7 +249,7 @@ startbatches () { then continue # System still running last test, skip. fi - ssh -o BatchMode=yes "$i" "cd \"$resdir/$ds\"; touch remote.run; PATH=\"$T/bin:$PATH\" nohup kvm-remote-$curbatch.sh > kvm-remote-$curbatch.sh.out 2>&1 &" 1>&2 + ${KVM_REMOTE_SSH} -o BatchMode=yes "$i" "cd \"$resdir/$ds\"; touch remote.run; PATH=\"$T/bin:$PATH\" nohup kvm-remote-$curbatch.sh > kvm-remote-$curbatch.sh.out 2>&1 &" 1>&2 ret=$? if test "$ret" -ne 0 then @@ -281,7 +291,7 @@ do if test "$ret" -eq 1 then echo " ---" Collecting results from $i `date` | tee -a "$oldrun/remote-log" - ( cd "$oldrun"; ssh -o BatchMode=yes $i "cd $rundir; tar -czf - kvm-remote-*.sh.out */console.log */kvm-test-1-run*.sh.out */qemu[_-]pid */qemu-retval */qemu-affinity; rm -rf $T > /dev/null 2>&1" | tar -xzf - ) + ( cd "$oldrun"; ${KVM_REMOTE_SSH} -o BatchMode=yes $i "cd $rundir; tar -czf - kvm-remote-*.sh.out */console.log */kvm-test-1-run*.sh.out */qemu[_-]pid */qemu-retval */qemu-affinity; rm -rf $T > /dev/null 2>&1" | tar -xzf - ) break; fi if test "$ret" -eq 255 diff --git a/tools/testing/selftests/rcutorture/bin/torture.sh b/tools/testing/selftests/rcutorture/bin/torture.sh index f0083891ee81..e7b578ec152f 100755 --- a/tools/testing/selftests/rcutorture/bin/torture.sh +++ b/tools/testing/selftests/rcutorture/bin/torture.sh @@ -68,6 +68,7 @@ do_clocksourcewd="${ifnotaarch64}" do_rt=yes do_rcutasksflavors="${ifnotaarch64}" # FIXME: Back to "yes" when SMP=n auto-avoided do_srcu_lockdep=yes +do_atomic_srcu=yes do_rcu_rust=no # doyesno - Helper function for yes/no arguments @@ -103,6 +104,7 @@ usage () { echo " --do-rcu-rust / --do-no-rcu-rust / --no-rcu-rust" echo " --do-scftorture / --do-no-scftorture / --no-scftorture" echo " --do-srcu-lockdep / --do-no-srcu-lockdep / --no-srcu-lockdep" + echo " --do-atomic-srcu / --do-no-atomic-srcu / --no-atomic-srcu" echo " --duration [ <minutes> | <hours>h | <days>d ]" echo " --guest-cpu-limit N" echo " --kcsan-kmake-arg kernel-make-arguments" @@ -148,6 +150,7 @@ do do_kcsan=yes do_clocksourcewd="${ifnotaarch64}" do_srcu_lockdep=yes + do_atomic_srcu=yes ;; --do-allmodconfig|--do-no-allmodconfig|--no-allmodconfig) do_allmodconfig=`doyesno "$1" --do-allmodconfig` @@ -183,6 +186,7 @@ do do_kcsan=no do_clocksourcewd=no do_srcu_lockdep=no + do_atomic_srcu=no ;; --do-normal|--do-norm|--do-no-normal|--do-no-norm|--no-normal|--no-norm) do_normal=`doyesno "$1" --do-normal` @@ -212,6 +216,9 @@ do --do-srcu-lockdep|--do-no-srcu-lockdep|--no-srcu-lockdep) do_srcu_lockdep=`doyesno "$1" --do-srcu-lockdep` ;; + --do-atomic-srcu|--do-no-atomic-srcu|--no-atomic-srcu) + do_atomic_srcu=`doyesno "$1" --do-atomic-srcu` + ;; --duration) checkarg --duration "(minutes)" $# "$2" '^[0-9][0-9]*\(m\|h\|d\|\)$' '^error' mult=1 @@ -497,6 +504,23 @@ then torture_set "rcutorture" tools/testing/selftests/rcutorture/bin/kvm.sh --allcpus --duration "$duration_rcutorture" --configs "$configs_rcutorture" --trust-make fi +# Test atomic SRCU across Tree SRCU (SRCU-N and SRCU-P) and Tiny SRCU +# (SRCU-T). The reader flavor selects srcu_read_lock_atomic() and +# synchronize_srcu_atomic(). Tiny SRCU requires SMP=n, which aarch64 +# does not support. +if test "$do_atomic_srcu" = "yes" +then + torture_bootargs="rcutorture.reader_flavor=0x10" + configs_atomic_srcu="SRCU-N SRCU-P" + if test "$ifnotaarch64" = yes + then + configs_atomic_srcu="$configs_atomic_srcu SRCU-T" + fi + torture_set "atomic-srcu" tools/testing/selftests/rcutorture/bin/kvm.sh \ + --allcpus --duration "$duration_rcutorture" \ + --configs "$configs_atomic_srcu" --trust-make +fi + if test "$do_locktorture" = "yes" then torture_bootargs="torture.disable_onoff_at_boot" |
