[PATCH 00/28] hazptr: shared-scan kthread, drain-assist, and stall detector

[PATCH 00/28] hazptr: shared-scan kthread, drain-assist, and stall detector

[Kunwu Chan]

This series adds a shared-scan kthread to the hazard pointer
implementation so that concurrent hazptr_synchronize() callers share
a single Bloom-walk cycle instead of each busy-waiting on remote
slots.  It also adds a drain-assist mechanism for sticky-address
forward progress (the lockdep hash-bucket scenario raised by Lai
Jiangshan [1]) and a waiter stall detector modeled on the RCU CPU
stall detector.

Rebased onto origin/dev at 6e7df7157a91, which includes the
try_acquire fast path and Andrea Parri's hazptrtorture fixes
(shutdown hang, slot-owner race detection, nested acquire).

v3: https://lore.kernel.org/all/20261005171529.1378809-1-kunwu.chan@gmail.com/

1.  Drain-assist and forward progress
====================================

Lai Jiangshan pointed out [1] that the lockdep conversion, which
uses a permanently-reachable hash-bucket address as the
hazptr_synchronize() target, could starve: new lookups for
unrelated keys in the same bucket keep re-acquiring the address,
indefinitely blocking the waiter.

Drain-assist (CONFIG_HAZPTR_DRAIN_ASSIST, default y) addresses
this.  When a waiter has been blocked for 100+ scan cycles, the
kthread snapshots which per-CPU and overflow-list slots hold the
address: per-CPU slots are tracked via a bitmap, and each
overflow-list slot is marked with a drain generation counter
(drain_gen).  The reader cooperates on release by checking its
own slot's drain_gen, clearing it, and decrementing the
outstanding count.  The drain disengages when the bitmap is
empty and the count reaches zero.  A remove-only re-snapshot
recovers releases that happened during the engage walk, before
the drain address was published.

The mechanism is verified by:

  - Drain litmus tests (herd7 + linux-kernel.cfg).
  - HOTADDR: the same address is continuously re-published by
    hammer readers; the waiter completes after the hammer stops.
  - LONGHOLD: a legitimate 8-second hold with the stall timeout
    set to 2 s so the detector reports during the hold; the
    waiter completes after the holder releases.
  - FAIRNESS: two stuck addresses with two waiters; drain
    rotation confirmed (2 distinct addresses engaged, 3
    fairness-cap disengages, both waiters complete after
    release).

The drain-assist provides empirical forward progress, but does
not provide a formal starvation-freedom guarantee under
continuous reader traffic.

2.  Lockdep performance
=======================

The lockdep conversion (commit 20) replaces the system-wide
synchronize_rcu_expedited() in lockdep_unregister_key() with a
per-bucket hazptr_synchronize().  The motivation, following Boqun
Feng's shazptr series, is to eliminate the function-call IPI to
every online CPU per grace period.

Measured on 96 CPUs with PROVE_LOCKING=y and 96 background
hazptr readers: the wq_churn workload (workqueue create/destroy,
each exercising lockdep_unregister_key()) generates about 3
function-call IPIs per operation after subtracting the
scheduling baseline.  The previous system-wide
synchronize_rcu_expedited() path required an IPI to each of
the 96 online CPUs per grace period, so this represents
roughly a 30-fold reduction for this workload.

Writer-side grace-period cost (rcuscale, nwriters=16, nreaders=0;
effective per-GP = measuring-window duration divided by the total
number of grace periods, from the module's summary line; ARM64
KVM; the per-GP figure is an aggregate throughput metric, not
a per-call latency):

               24 CPU      96 CPU      128 CPU     256 CPU
  hazptr        526 us       528 us       526 us       526 us
  RCU          2650 us       928 us      3836 us      4230 us
  SRCU          450 us       639 us      2352 us       706 us

The comparison uses hazptr_synchronize(), synchronize_rcu(),
and synchronize_srcu(), respectively.  Hazptr per-GP cost
remains nearly constant across the tested CPU counts.  RCU and
SRCU grace-period costs show larger variations across the
tested CPU counts in this QEMU ARM64 environment; these
results are from individual runs.

This matches the v3 numbers (493-500 us for hazptr, 865 us for RCU,
599 us for SRCU at 96 CPUs) within run-to-run variance: the shared
scan kthread preserves the aggregate throughput of the v3
per-waiter-scan protocol in this workload.  An individual hazptr_synchronize() call
under this 16-writer load takes about 8 ms because a single scan
cycle serves all concurrent waiters.

Reader overhead (refscale, nreaders=-1, nruns=5, median, ARM64 KVM):

               PROVE_LOCKING=n    PROVE_LOCKING=y
  hazptr       22.2 ns            22.2 ns
  RCU           4.8 ns          147.6 ns
  SRCU         36.4 ns          163.8 ns

Hazptr reader overhead is unaffected by PROVE_LOCKING (22.2 ns
with and without).  Without PROVE_LOCKING, RCU has lower
reader overhead than hazptr.  With PROVE_LOCKING enabled,
while RCU and SRCU readers are heavily
instrumented: RCU degrades 31x (4.8 ns to 147.6 ns) and SRCU
degrades 4.5x (36.4 ns to 163.8 ns).  With PROVE_LOCKING enabled,
the hazptr read path is about 7x faster than RCU's and SRCU's.

These properties — flat per-GP cost across CPU counts, low
IPI, and reader overhead that is largely unaffected by lockdep — make hazptr suited
for short-critical-section readers under a stable, shared
address, such as the lockdep key-hash lookup.

(v3 reported 26.1 ns; the improvement is consistent with the try_acquire
fast path in the base tree.)

3.  Stall detector
==================

CONFIG_HAZPTR_STALL_DETECTOR (default y, 21 s timeout) is modeled
on the RCU CPU stall detector.  After each scan cycle the kthread
checks still-blocked waiters; on timeout it walks all per-CPU and
overflow-list slots and reports each holder's pid, held time,
acquire call site, and task stack under CONFIG_HAZPTR_DEBUG.
Comparing consecutive reports distinguishes a single long-held
reference from repeated publication by different readers.  The
diagnostic message explains that long-lived references are a
hazard-pointer feature, not necessarily a problem.

Suggested by Paul E. McKenney.  Runtime sysfs controls
(/sys/kernel/hazptr/hazptr_stall_timeout_ms and
hazptr_stall_suppress) allow the administrator to adjust the
timeout or suppress reports at runtime.

4.  Changes since RFC v3
====================

- Rebased onto the try_acquire protocol.  Per-CPU
  slots publish their target directly; the shared scan uses a
  single-pass Bloom walk.  Dropped three v3 patches already
  addressed upstream or by the rebase: "scan all per-CPU slots
  before overflow lists" and "elide redundant first drain
  pass" (rebase), and "hazptr: Fix two-phase hazptr_synchronize
  race with detach" (merged upstream from Mathieu Desnoyers
  [3]).

- Added drain-assist, stall detector, lockdep conversion,
  HOTADDR/LONGHOLD/FAIRNESS torture scenarios, hazptr.sh selftest,
  rcuscale hazptr scale type.

- Safety fixes found during verification: Bloom early-return on
  empty slot, wildcard deferral in scans, exact-match fallback
  for Bloom false positives, FORCE_SLOWPATH skip-before-try,
  drain state re-initialized at engage, drain addr read with
  smp_load_acquire, slot type identified via
  hazptr_slot_is_backup().  Bloom filter collision-rate data
  on request from Joel Fernandes: [2].

- Build fixes: select TRACING for the torture test, export
  drain_key and tracepoint for =m linkage.

- The new base includes Andrea Parri's hazptrtorture fixes
  (reader-exit deferred cleanup, nest_acquire, slot-owner
  checks).  The HOTADDR/LONGHOLD/FAIRNESS scenarios are
  integrated with these changes; all configured hazptrtorture scenarios
  pass on the new base with nest_acquire enabled.

- Suggested-by: Boqun Feng (drain-assist direction, lockdep
  conversion, Bloom filter, slowpath debug, stall detector),
  Paul E. McKenney (stall detector, runtime sysfs controls),
  Mathieu Desnoyers (Bloom filter design). 

Testing
=======

ARM64 KVM (all scenarios at 16 vCPUs, 30 s each; the CPU sweep
runs PREEMPT at 8/16/32/64/128/256 vCPUs):

  Scenario     hazptr BUG   verdict
  ------------ -----------  -------
  PREEMPT      0            SUCCESS
  NOPREEMPT    0            SUCCESS
  SLOWPATH     0            SUCCESS
  LOCKDEP      0            SUCCESS
  READERS0     0            SUCCESS
  READERS4     0            SUCCESS
  HOTADDR      0            SUCCESS
  LONGHOLD     0            SUCCESS
  FAIRNESS     0            SUCCESS
  CPU 8-256    0            SUCCESS

The PREEMPT scenario runs with deferral enabled
(defer_modulus=16000) and nest_acquire=2, exercising the
reader-exit deferred cleanup that Andrea Parri's fixes
introduced; shutdown completes in under 100 ms.

HOTADDR drain verification: 450 engage events, 445 complete
disengages, 5 orphaned, 0 fairness-cap (the 5 orphaned
drains were disengaged because the Bloom scan completed the
waiter before the drain shut down).  LONGHOLD: 5 engage, 5 orphaned
disengage (outstanding=0).  FAIRNESS: 4 engage across 2 distinct
addresses, 3 fairness-cap disengages — drain rotation confirmed.

LKMM litmus tests (herd7 + linux-kernel.cfg):

  Litmus                            Result      Property
  --------------------------------  ----------  --------------------
  acquire-before-scan               Never       Publication ordering
  bloom-wildcard-hole               Never       Wildcard deferral
  drain-engage-wildcard             Never       Engage defers on wildcard
  drain-generation-visibility       Never       Generation ordering for release hooks
  drain-resnapshot-slot-race        Sometimes   Remove-only walk; handled by design
  drain-safety                      Sometimes   Engage snapshot; covered by wildcard+re-engage
  wildcard-flip-escape              Never       Legacy wildcard-flip protocol

All 28 commits compile with 0 errors / 0 warnings ([ARCH], [GCC],
four DRAIN x STALL Kconfig combinations, HAZPTR_TORTURE_TEST=m).

Design notes
============

- The drain engages for one address at a time; a fairness cap
  (HAZPTR_DRAIN_MAX_CYCLES) bounds each round and rotates to
  the next stalled waiter.  Rotation is bounded but not
  formally starvation-free: waiters positioned after the drain
  target in the scanning list do not accumulate re-engage
  cycles while another waiter's drain is active, so a
  late-arriving waiter can wait roughly two drain windows
  before its first turn.  The FAIRNESS torture scenario added
  in this series observes the rotation empirically via the
  kernel drain logs; the ARM64 run confirmed 2 distinct engaged
  addresses and 3 fairness-cap disengages.

- Key memory-ordering properties are tested by litmus tests: engage-wildcard,
  generation-visibility, resnapshot-slot-race, and drain-safety
  (see the litmus table above).  Forward progress requires
  readers to release, which holds for the lockdep use case
  (is_dynamic_key() is a short critical section); the stall
  detector reports any waiter blocked beyond the timeout.

[1] https://lore.kernel.org/all/as3aMmPdCHCXW6Ci@U-YXDQC754-96.local/
[2] https://lore.kernel.org/rcu/CAN_Opa-=QsYFgZ-tENoqvpfqMhr=sZToFBs6NfbyTOpNLYQB1g@mail.gmail.com/
[3] https://lore.kernel.org/all/20261011194902.97364-2-mathieu.desnoyers@efficios.com/

Kunwu Chan (28):
  hazptr: Add shared scan kthread
  hazptr: Use Bloom filter for shared scan waiters
  hazptr: Add scoped_guard() support
  hazptr: Add debug option to force the acquire slow path
  Documentation/litmus-tests: Add hazptr wildcard-flip escape test
  Documentation/litmus-tests: Add hazptr acquire-before-scan test
  hazptr: Fix kernel-doc of hazptr_release()
  hazptrtorture: Add slowpath and lockdep scenarios
  hazptrtorture: Add READERS4 and READERS0 torture configs
  hazptrtorture: Add 128- and 256-CPU configs
  selftests/rcutorture: Add hazptr torture test script
  hazptr: Remove per-waiter scan fallback, simplify synchronize
  hazptr: Add waiter-enqueue tracepoint
  hazptr: Add waiter stall detector
  hazptr: Add reader-assisted drain for stalled waiters
  Documentation/litmus-tests: Add hazptr drain engage-wildcard test
  Documentation/litmus-tests: Add hazptr drain resnapshot-slot-race test
  Documentation/litmus-tests: Add hazptr drain safety test
  Documentation/litmus-tests: Add hazptr drain generation-visibility test
  locking/lockdep: Use hazptr to wait for dynamic key lookups
  hazptrtorture: Add HOTADDR and LONGHOLD scenarios
  hazptrtorture: Add drain fairness rotation scenario
  rcuscale: Add hazptr scale type
  hazptr: Add runtime sysfs controls for stall detector
  hazptr: Select TRACING for HAZPTR_TORTURE_TEST to enable tracepoint
  hazptrtorture: Add torture_type=none clean skip for rcuscale coexistence
  hazptr: Add Bloom false-positive exact-match fallback
  Documentation/litmus-tests: Add hazptr bloom wildcard-hole test

 Documentation/litmus-tests/README             |   37 +
 .../hazptr/hazptr-acquire-before-scan.litmus  |   45 +
 .../hazptr/hazptr-bloom-wildcard-hole.litmus  |   49 +
 .../hazptr-drain-engage-wildcard.litmus       |   44 +
 .../hazptr-drain-generation-visibility.litmus |   41 +
 .../hazptr-drain-resnapshot-slot-race.litmus  |   36 +
 .../hazptr/hazptr-drain-safety.litmus         |   43 +
 .../hazptr/hazptr-wildcard-flip-escape.litmus |   46 +
 include/linux/hazptr.h                        |   71 +-
 include/trace/events/hazptr.h                 |   29 +
 kernel/hazptr.c                               | 1120 ++++++++++++++++-
 kernel/locking/lockdep.c                      |   25 +-
 kernel/rcu/Kconfig.debug                      |   42 +
 kernel/rcu/hazptrtorture.c                    |  633 +++++++++-
 kernel/rcu/rcuscale.c                         |   67 +
 .../selftests/rcutorture/bin/hazptr.sh        |  382 ++++++
 .../rcutorture/configs/hazptr/CFLIST          |    9 +
 .../rcutorture/configs/hazptr/CPU128          |   16 +
 .../rcutorture/configs/hazptr/CPU128.boot     |    1 +
 .../rcutorture/configs/hazptr/CPU256          |   16 +
 .../rcutorture/configs/hazptr/CPU256.boot     |    1 +
 .../rcutorture/configs/hazptr/FAIRNESS        |   19 +
 .../rcutorture/configs/hazptr/FAIRNESS.boot   |    1 +
 .../rcutorture/configs/hazptr/HOTADDR         |   18 +
 .../rcutorture/configs/hazptr/HOTADDR.boot    |    1 +
 .../rcutorture/configs/hazptr/LOCKDEP         |   16 +
 .../rcutorture/configs/hazptr/LOCKDEP.boot    |    1 +
 .../rcutorture/configs/hazptr/LONGHOLD        |   18 +
 .../rcutorture/configs/hazptr/LONGHOLD.boot   |    1 +
 .../rcutorture/configs/hazptr/READERS0        |   16 +
 .../rcutorture/configs/hazptr/READERS0.boot   |    2 +
 .../rcutorture/configs/hazptr/READERS4        |   16 +
 .../rcutorture/configs/hazptr/READERS4.boot   |    2 +
 .../rcutorture/configs/hazptr/SLOWPATH        |   16 +
 .../rcutorture/configs/rcuscale/HAZPTR        |   13 +
 .../rcutorture/configs/rcuscale/HAZPTR.boot   |    1 +
 36 files changed, 2854 insertions(+), 40 deletions(-)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-acquire-before-scan.litmus
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-bloom-wildcard-hole.litmus
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-engage-wildcard.litmus
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-generation-visibility.litmus
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-resnapshot-slot-race.litmus
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-safety.litmus
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-wildcard-flip-escape.litmus
 create mode 100644 include/trace/events/hazptr.h
 create mode 100755 tools/testing/selftests/rcutorture/bin/hazptr.sh
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU128
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU128.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU256
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU256.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS0
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS0.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS4
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS4.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/SLOWPATH
 create mode 100644 tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR
 create mode 100644 tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR.boot


base-commit: 6e7df7157a91757355e22a2ca46bafc35fbd6465
-- 
2.43.0

[PATCH 01/28] hazptr: Add shared scan kthread

[Kunwu Chan]

Batch concurrent hazptr_synchronize() callers into a single kthread
scan.  Waiters sleep on completions instead of busy-waiting.

The kthread processes queued waiters sequentially: for each waiter,
it scans per-CPU slots and both overflow-list phases until the
address is released.  This is the same scan the direct fallback
performs, delegated to a kthread.

Waiters are embedded in the caller's stack frame, so no dynamic
allocation is needed in the synchronize path.  Fall back to the
direct scan if the kthread is unavailable.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 kernel/hazptr.c | 149 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 149 insertions(+)

diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index b06164aac3a7..013c24021c43 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -12,6 +12,10 @@
 #include <linux/mutex.h>
 #include <linux/list.h>
 #include <linux/export.h>
+#include <linux/completion.h>
+#include <linux/kthread.h>
+#include <linux/swait.h>
+#include <linux/sched.h>
 
 #define HAZPTR_WILDCARD	((void *) 1UL)
 
@@ -234,6 +238,145 @@ void hazptr_scan_overflow_list_period(void *addr, unsigned int scan_idx)
 	}
 }
 
+/* Helper: scan both overflow-list phases. */
+static void hazptr_scan_overflow_list_all(void *addr)
+{
+	unsigned int scan_idx = READ_ONCE(hazptr_overflow_list_phase);
+	int cpu;
+
+	for_each_possible_cpu(cpu) {
+		struct hazptr_overflow_list_flip *overflow_list_flip = per_cpu_ptr(&percpu_overflow_list_flip, cpu);
+
+		hazptr_synchronize_overflow_list(&overflow_list_flip->array[scan_idx], addr);
+		hazptr_synchronize_overflow_list(&overflow_list_flip->array[flip_list_phase(scan_idx)], addr);
+	}
+}
+
+/*
+ * Shared scan: batch concurrent hazptr_synchronize() callers into a
+ * single kthread-driven scan, avoiding redundant per-CPU and
+ * overflow-list walks per caller.  Waiters sleep on completions
+ * instead of busy-waiting.
+ */
+
+struct hazptr_waiter {
+	struct list_head node;
+	void *addr;
+	struct completion done;
+};
+
+struct hazptr_scan_state {
+	struct task_struct *kthread;
+	struct swait_queue_head wq;
+	bool wakeup;
+	struct mutex lock;
+	struct list_head pending;
+	struct list_head scanning;	/* kthread only */
+};
+static struct hazptr_scan_state hazptr_scan;
+
+/*
+ * Queue @addr for the shared scan.  The waiter lives on the caller's
+ * stack, so no allocation is needed.
+ */
+static void hazptr_synchronize_queued(void *addr)
+{
+	struct hazptr_waiter waiter = {
+		.addr = addr,
+	};
+
+	init_completion(&waiter.done);
+	INIT_LIST_HEAD(&waiter.node);
+
+	mutex_lock(&hazptr_scan.lock);
+	list_add_tail(&waiter.node, &hazptr_scan.pending);
+	if (!READ_ONCE(hazptr_scan.wakeup)) {
+		WRITE_ONCE(hazptr_scan.wakeup, true);
+		swake_up_one(&hazptr_scan.wq);
+	}
+	mutex_unlock(&hazptr_scan.lock);
+
+	wait_for_completion(&waiter.done);
+}
+
+static void hazptr_scan_do_cycle(void)
+{
+	struct hazptr_waiter *w, *n;
+	LIST_HEAD(done);
+
+	mutex_lock(&hazptr_phase_lock);
+	mutex_lock(&hazptr_scan.lock);
+	list_splice_tail_init(&hazptr_scan.pending, &hazptr_scan.scanning);
+	mutex_unlock(&hazptr_scan.lock);
+
+	if (list_empty(&hazptr_scan.scanning)) {
+		mutex_unlock(&hazptr_phase_lock);
+		return;
+	}
+
+	/*
+	 * Process each waiter: scan per-CPU slots and overflow lists
+	 * until the waiter's address is released.  This is the same
+	 * scan the direct fallback performs, delegated to a kthread so
+	 * callers sleep instead of busy-waiting.
+	 */
+	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
+		hazptr_scan_cpu_slots(w->addr);
+		hazptr_scan_overflow_list_all(w->addr);
+		list_move(&w->node, &done);
+	}
+
+	mutex_unlock(&hazptr_phase_lock);
+
+	list_for_each_entry_safe(w, n, &done, node) {
+		list_del_init(&w->node);
+		complete(&w->done);
+	}
+}
+
+static int hazptr_scan_kthread(void *unused)
+{
+	for (;;) {
+		bool idle;
+
+		swait_event_idle_exclusive(hazptr_scan.wq,
+					   READ_ONCE(hazptr_scan.wakeup));
+		hazptr_scan_do_cycle();
+		mutex_lock(&hazptr_scan.lock);
+		idle = list_empty(&hazptr_scan.pending) &&
+		      list_empty(&hazptr_scan.scanning);
+		if (idle)
+			WRITE_ONCE(hazptr_scan.wakeup, false);
+		mutex_unlock(&hazptr_scan.lock);
+		if (idle)
+			continue;
+		schedule_timeout_idle(1);
+	}
+	return 0;
+}
+
+/*
+ * Initialize the scan kthread.  Failure falls back to the
+ * direct scan.
+ */
+static int __init hazptr_scan_init(void)
+{
+	struct task_struct *t;
+
+	init_swait_queue_head(&hazptr_scan.wq);
+	mutex_init(&hazptr_scan.lock);
+	INIT_LIST_HEAD(&hazptr_scan.pending);
+	INIT_LIST_HEAD(&hazptr_scan.scanning);
+
+	t = kthread_run(hazptr_scan_kthread, NULL, "hazptr_scan");
+	if (!IS_ERR(t))
+		smp_store_release(&hazptr_scan.kthread, t);
+	else
+		pr_warn("hazptr: scan kthread failed, using direct scan\n");
+	return 0;
+}
+core_initcall(hazptr_scan_init);
+
 /*
  * hazptr_synchronize: Wait until @addr is released from all slots.
  *
@@ -261,6 +404,12 @@ void hazptr_synchronize(void *addr)
 	/* Memory ordering: Store A before Load B. */
 	smp_mb();
 
+	/* Pairs with smp_store_release in hazptr_scan_init(). */
+	if (smp_load_acquire(&hazptr_scan.kthread)) {
+		hazptr_synchronize_queued(addr);
+		return;
+	}
+
 	guard(mutex)(&hazptr_phase_lock);
 
 	/* Scan per-CPU slots. */
-- 
2.43.0

[PATCH 02/28] hazptr: Use Bloom filter for shared scan waiters

[Kunwu Chan]

Replace the per-waiter sequential scan with a shared Bloom filter
built during a single walk: each observed non-wildcard address
sets three bits.  A waiter whose address hashes to an unset bit
can be completed without another slot walk.

A Bloom filter has no false negatives, so an address absent from
the scan cannot be held by an observed slot; false positives only
delay completion to a later scan cycle.

The filter uses a one-page bitmap with three multiply-shift hash
functions.  It lives in the hazptr_scan_state and is only touched
by the scan kthread.

Suggested-by: Boqun Feng <boqun@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Suggested-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 kernel/hazptr.c | 122 +++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 116 insertions(+), 6 deletions(-)

diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index 013c24021c43..059c356ceaa7 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -16,6 +16,8 @@
 #include <linux/kthread.h>
 #include <linux/swait.h>
 #include <linux/sched.h>
+#include <linux/bitops.h>
+#include <linux/bitmap.h>
 
 #define HAZPTR_WILDCARD	((void *) 1UL)
 
@@ -57,6 +59,12 @@ unsigned int flip_list_phase(unsigned int phase)
 	return 1 - phase;
 }
 
+static
+bool is_wildcard(void *addr)
+{
+	return (unsigned long)addr == (unsigned long)HAZPTR_WILDCARD;
+}
+
 static
 struct hazptr_slot_item *hazptr_get_free_percpu_slot_item(void)
 {
@@ -252,6 +260,98 @@ static void hazptr_scan_overflow_list_all(void *addr)
 	}
 }
 
+/* Number of hash functions for the Bloom filter. */
+#define HAZPTR_BLOOM_HASHES 3
+
+/* Number of bits in the Bloom filter bitmap: one page. */
+#define HAZPTR_BLOOM_NBITS (PAGE_SIZE * 8)
+
+struct hazptr_bloom {
+	unsigned long map[PAGE_SIZE / sizeof(unsigned long)];
+};
+
+/*
+ * Multiply @addr by a per-hash odd constant and use the high bits
+ * as the Bloom filter index.
+ */
+static unsigned long hazptr_bloom_hash(void *addr, unsigned int i)
+{
+	static const u64 mult[HAZPTR_BLOOM_HASHES] = {
+		0x9E3779B97F4A7C15ULL,
+		0xC2B2AE3D27D4EB4FULL,
+		0x165667B19E3779F9ULL,
+	};
+	u64 hash = (u64)(unsigned long)addr * mult[i];
+
+	return hash >> (64 - ilog2(HAZPTR_BLOOM_NBITS));
+}
+
+static void hazptr_bloom_reset(struct hazptr_bloom *bloom)
+{
+	bitmap_zero(bloom->map, HAZPTR_BLOOM_NBITS);
+}
+
+static void hazptr_bloom_add(struct hazptr_bloom *bloom, void *addr)
+{
+	unsigned int i;
+
+	for (i = 0; i < HAZPTR_BLOOM_HASHES; i++)
+		__set_bit(hazptr_bloom_hash(addr, i), bloom->map);
+}
+
+static bool hazptr_bloom_contains(const struct hazptr_bloom *bloom, void *addr)
+{
+	unsigned int i;
+
+	for (i = 0; i < HAZPTR_BLOOM_HASHES; i++)
+		if (!test_bit(hazptr_bloom_hash(addr, i), bloom->map))
+			return false;
+	return true;
+}
+
+/*
+ * Walk all slots and record observed non-wildcard addresses into
+ * the Bloom filter.  A single pass suffices on the try_acquire
+ * fast path (readers publish their target directly).
+ */
+static void hazptr_scan_walk(struct hazptr_bloom *bloom)
+{
+	int cpu;
+
+	hazptr_bloom_reset(bloom);
+
+	for_each_possible_cpu(cpu) {
+		struct hazptr_percpu_slots *percpu_slots = per_cpu_ptr(&hazptr_percpu_slots, cpu);
+		struct hazptr_overflow_list_flip *overflow_list_flip = per_cpu_ptr(&percpu_overflow_list_flip, cpu);
+		unsigned int idx;
+
+		for (idx = 0; idx < NR_HAZPTR_PERCPU_SLOTS; idx++) {
+			struct hazptr_slot_item *item = &percpu_slots->items[idx];
+			void *v;
+
+			v = smp_load_acquire(&item->slot.addr);
+			if (v && !is_wildcard(v))
+				hazptr_bloom_add(bloom, v);
+		}
+		for (int i = 0; i < 2; i++) {
+			struct hazptr_overflow_list *list = &overflow_list_flip->array[i];
+			struct hazptr_backup_slot *b;
+			unsigned long flags;
+
+			raw_spin_lock_irqsave(&list->lock, flags);
+			hlist_for_each_entry(b, &list->head, overflow_node) {
+				void *v;
+
+				v = smp_load_acquire(&b->slot.addr);
+				if (v && !is_wildcard(v))
+					hazptr_bloom_add(bloom, v);
+			}
+			raw_spin_unlock_irqrestore(&list->lock, flags);
+		}
+	}
+}
+
+/* Need is_wildcard for the Bloom filter */
 /*
  * Shared scan: batch concurrent hazptr_synchronize() callers into a
  * single kthread-driven scan, avoiding redundant per-CPU and
@@ -272,6 +372,7 @@ struct hazptr_scan_state {
 	struct mutex lock;
 	struct list_head pending;
 	struct list_head scanning;	/* kthread only */
+	struct hazptr_bloom bloom;	/* kthread only */
 };
 static struct hazptr_scan_state hazptr_scan;
 
@@ -315,19 +416,28 @@ static void hazptr_scan_do_cycle(void)
 	}
 
 	/*
-	 * Process each waiter: scan per-CPU slots and overflow lists
-	 * until the waiter's address is released.  This is the same
-	 * scan the direct fallback performs, delegated to a kthread so
-	 * callers sleep instead of busy-waiting.
+	 * Walk all slots and record observed addresses into the
+	 * Bloom filter, then complete waiters whose address is
+	 * not present.  On the try_acquire fast path, readers
+	 * publish their target directly (no wildcard transient),
+	 * so a single pass suffices.
 	 */
+	hazptr_scan_walk(&hazptr_scan.bloom);
+
+	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
+		if (!hazptr_bloom_contains(&hazptr_scan.bloom, w->addr))
+			list_move(&w->node, &done);
+	}
+
+	mutex_unlock(&hazptr_phase_lock);
+
+	/* Waiters still blocked: their address was observed present. */
 	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
 		hazptr_scan_cpu_slots(w->addr);
 		hazptr_scan_overflow_list_all(w->addr);
 		list_move(&w->node, &done);
 	}
 
-	mutex_unlock(&hazptr_phase_lock);
-
 	list_for_each_entry_safe(w, n, &done, node) {
 		list_del_init(&w->node);
 		complete(&w->done);
-- 
2.43.0

[PATCH 03/28] hazptr: Add scoped_guard() support

[Kunwu Chan]

Add a scoped_guard() class for hazptr so that callers can pair
hazptr_acquire() with hazptr_release() on scope exit.

The guard holds a pointer to a caller-provided hazptr_ctx rather
than embedding it.  The ctx must remain at a stable address while
the hazard pointer is held because the slow path can chain its
backup slot into the overflow list by address.

Suggested-by: Boqun Feng <boqun@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 include/linux/hazptr.h | 37 +++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/include/linux/hazptr.h b/include/linux/hazptr.h
index d37792af3493..3372b1cdbee6 100644
--- a/include/linux/hazptr.h
+++ b/include/linux/hazptr.h
@@ -341,6 +341,43 @@ void hazptr_release(struct hazptr_ctx *ctx, void *addr)
 		hazptr_unchain_backup_slot(ctx);
 }
 
+
+/**
+ * struct hazptr_guard - cleanup.h guard for a single hazard pointer
+ *
+ * @ctx: Pointer to the caller-provided hazard-pointer context.
+ * @addr: The address returned by the matching hazptr_acquire().
+ */
+struct hazptr_guard {
+	struct hazptr_ctx *ctx;
+	void *addr;
+};
+
+/**
+ * hazptr_guard_acquire - Acquire a hazard pointer for scoped_guard() use
+ *
+ * @ctx: The caller-provided hazard-pointer context.
+ * @addr_p: Pointer to the pointer that is to be hazard-pointer protected.
+ *
+ * Acquire a hazard pointer using @ctx and return a guard that
+ * cleanup.h releases via hazptr_release() on scope exit.
+ */
+static inline struct hazptr_guard
+hazptr_guard_acquire(struct hazptr_ctx *ctx, void * const *addr_p)
+{
+	struct hazptr_guard g;
+
+	g.ctx = ctx;
+	g.addr = hazptr_acquire(ctx, addr_p);
+	return g;
+}
+
+DEFINE_CLASS(hazptr, struct hazptr_guard,
+	     hazptr_release(_T.ctx, _T.addr),
+	     hazptr_guard_acquire(ctx, addr_p),
+	     struct hazptr_ctx *ctx, void * const *addr_p)
+DEFINE_CLASS_IS_UNCONDITIONAL(hazptr)
+
 void hazptr_init(void);
 
 #endif /* _LINUX_HAZPTR_H */
-- 
2.43.0

[PATCH 04/28] hazptr: Add debug option to force the acquire slow path

[Kunwu Chan]

Add CONFIG_HAZPTR_ACQUIRE_FORCE_SLOWPATH: makes hazptr_acquire()
skip the per-CPU fast path entirely and enter __hazptr_acquire(),
even when a per-CPU slot is available.  This is useful for testing
slow-path locking, especially with lockdep.  It also lets
hazptrtorture exercise the slow path without requiring the per-CPU
slots to be exhausted.

Note that the forced path skips __hazptr_try_acquire() rather than
running it and discarding its result: a successful try_acquire
publishes the address to the per-CPU slot, and abandoning that
publication would leave a slot pointing at a stale ctx, which
hazptr_note_context_switch() then dereferences (use-after-free).

Suggested-by: Boqun Feng <boqun@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 include/linux/hazptr.h   |  2 ++
 kernel/rcu/Kconfig.debug | 10 ++++++++++
 2 files changed, 12 insertions(+)

diff --git a/include/linux/hazptr.h b/include/linux/hazptr.h
index 3372b1cdbee6..299ad8a07156 100644
--- a/include/linux/hazptr.h
+++ b/include/linux/hazptr.h
@@ -271,6 +271,8 @@ void *hazptr_acquire(struct hazptr_ctx *ctx, void * const *addr_p)
 	ctx->acquire_cpu = smp_processor_id();
 	ctx->acquire_caller = _THIS_IP_;
 #endif
+	if (IS_ENABLED(CONFIG_HAZPTR_ACQUIRE_FORCE_SLOWPATH))
+		return __hazptr_acquire(ctx, addr_p);
 	addr = __hazptr_try_acquire(ctx, addr_p, slot);
 	if (unlikely(!addr))
 		return __hazptr_acquire(ctx, addr_p);
diff --git a/kernel/rcu/Kconfig.debug b/kernel/rcu/Kconfig.debug
index 7c3c6017b266..3a2e9d7f58f8 100644
--- a/kernel/rcu/Kconfig.debug
+++ b/kernel/rcu/Kconfig.debug
@@ -265,4 +265,14 @@ config HAZPTR_DEBUG
 
 	  Say Y here if you want to enable those assert, N otherwise.
 
+config HAZPTR_ACQUIRE_FORCE_SLOWPATH
+	bool "Force hazard-pointer acquire into the slow path"
+	depends on DEBUG_KERNEL
+	default n
+	help
+	  Make hazptr_acquire() always enter __hazptr_acquire().
+	  This is useful for testing slow-path locking, especially
+	  with lockdep.  It also lets hazptrtorture exercise the slow
+	  path without filling the per-CPU slots.
+
 endmenu # "RCU Debugging"
-- 
2.43.0

[PATCH 05/28] Documentation/litmus-tests: Add hazptr wildcard-flip escape test

[Kunwu Chan]

Add a three-process litmus test for the in-flight wildcard window
of the single-pass scan: a straggling acquire which published the
old wildcard after the final walk cannot still load the
unpublished object.

Verified with herd7 (linux-kernel.cfg): Never 0 13.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 .../hazptr/hazptr-wildcard-flip-escape.litmus | 46 +++++++++++++++++++
 1 file changed, 46 insertions(+)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-wildcard-flip-escape.litmus

diff --git a/Documentation/litmus-tests/hazptr/hazptr-wildcard-flip-escape.litmus b/Documentation/litmus-tests/hazptr/hazptr-wildcard-flip-escape.litmus
new file mode 100644
index 000000000000..de467bb82d20
--- /dev/null
+++ b/Documentation/litmus-tests/hazptr/hazptr-wildcard-flip-escape.litmus
@@ -0,0 +1,46 @@
+C hazptr-wildcard-flip-escape
+
+(*
+ * Result: Never
+ *
+ * Check that a straggling acquire which published the old
+ * wildcard after the final walk cannot still load the
+ * unpublished object.
+ *)
+
+{
+	int ptr = 3;
+	int hp = 0;
+	int wildcard = 1;
+	int enq = 0;
+}
+
+P0(int *ptr, int *enq)
+{
+	WRITE_ONCE(*ptr, 0);
+	smp_mb();
+	smp_store_release(enq, 1);
+}
+
+P1(int *enq, int *wildcard, int *hp)
+{
+	int r1;
+	int r2;
+
+	r1 = smp_load_acquire(enq);
+	WRITE_ONCE(*wildcard, 2);
+	r2 = READ_ONCE(*hp);
+}
+
+P2(int *wildcard, int *hp, int *ptr)
+{
+	int r0;
+	int r3;
+
+	r0 = READ_ONCE(*wildcard);
+	WRITE_ONCE(*hp, r0);
+	smp_mb();
+	r3 = READ_ONCE(*ptr);
+}
+
+exists (1:r1=1 /\ 1:r2=0 /\ 2:r3=3)
-- 
2.43.0

[PATCH 06/28] Documentation/litmus-tests: Add hazptr acquire-before-scan test

[Kunwu Chan]

Add a two-process litmus test for the hazard-pointer publication
protocol: the reclaimer unpublishes the pointer, executes smp_mb(),
then scans; the reader publishes the hazard pointer, executes
smp_mb(), then loads the pointer.  The smp_mb() pair forbids the
reclaimer from missing the publication while the reader still
observes the pre-unpublish pointer.

This is the publication protocol used by the lockdep
is_dynamic_key()/lockdep_unregister_key() conversion.

Verified with herd7: Never 0 3.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 Documentation/litmus-tests/README             | 13 ++++++
 .../hazptr/hazptr-acquire-before-scan.litmus  | 45 +++++++++++++++++++
 2 files changed, 58 insertions(+)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-acquire-before-scan.litmus

diff --git a/Documentation/litmus-tests/README b/Documentation/litmus-tests/README
index 4d4ec9c6f2cc..c2bf20fa4092 100644
--- a/Documentation/litmus-tests/README
+++ b/Documentation/litmus-tests/README
@@ -97,3 +97,16 @@ SRCU-fastpath-scan-before-anchor.litmus
     period, violating the SRCU grace-period guarantee.  See
     SRCU-fastpath-anchor-before-scan.litmus for the opposite
     ordering.
+
+
+hazptr (/hazptr directory)
+--------------------------
+
+hazptr-acquire-before-scan.litmus
+    Test the resolved-publication ordering between hazard-pointer
+    publication and the scan.
+
+hazptr-wildcard-flip-escape.litmus
+    Test the in-flight wildcard window of the single-pass wildcard-flip
+    scan: a straggling acquire that published the old wildcard after
+    the final walk cannot still load the unpublished object.
diff --git a/Documentation/litmus-tests/hazptr/hazptr-acquire-before-scan.litmus b/Documentation/litmus-tests/hazptr/hazptr-acquire-before-scan.litmus
new file mode 100644
index 000000000000..6dc13830c07c
--- /dev/null
+++ b/Documentation/litmus-tests/hazptr/hazptr-acquire-before-scan.litmus
@@ -0,0 +1,45 @@
+C hazptr-acquire-before-scan
+
+(*
+ * Result: Never
+ *
+ * The reclaimer unpublishes the pointer, executes smp_mb(), then
+ * scans the hazard-pointer slot.  The reader publishes the
+ * protected address, executes smp_mb(), then loads the pointer.
+ *
+ * The smp_mb() pair forbids the reclaimer from missing the
+ * publication while the reader still observes the pre-unpublish
+ * pointer.
+ *
+ * This is the publication protocol used by the lockdep
+ * is_dynamic_key()/lockdep_unregister_key() conversion.
+ *
+ * This models the resolved-publication case.  The in-flight
+ * wildcard window of the wildcard-flip scan is covered by
+ * hazptr-wildcard-flip-escape.litmus.
+ *)
+
+{
+int ptr = 1;
+int hp = 0;
+}
+
+P0(int *ptr, int *hp)
+{
+	int r0;
+
+	WRITE_ONCE(*ptr, 0);	/* Unpublish. */
+	smp_mb();
+	r0 = READ_ONCE(*hp);	/* Scan. */
+}
+
+P1(int *ptr, int *hp)
+{
+	int r0;
+
+	WRITE_ONCE(*hp, 1);	/* Publish. */
+	smp_mb();
+	r0 = READ_ONCE(*ptr);	/* Load pointer. */
+}
+
+exists (0:r0=0 /\ 1:r0=1)
-- 
2.43.0

[PATCH 07/28] hazptr: Fix kernel-doc of hazptr_release()

[Kunwu Chan]

The second parameter of hazptr_release() is named @addr, not
@addr_p.  It is the address passed to the matching
hazptr_acquire().

Fixes: 17969a0e7f7e ("hazptr: Upgrade kernel-doc headers")
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 include/linux/hazptr.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/include/linux/hazptr.h b/include/linux/hazptr.h
index 299ad8a07156..0fb48518525b 100644
--- a/include/linux/hazptr.h
+++ b/include/linux/hazptr.h
@@ -315,7 +315,7 @@ static inline void hazptr_release_debug(struct hazptr_ctx *ctx, void *addr) { }
  * hazptr_release - Release the specified hazard pointer
  *
  * @ctx: The hazard-pointer context that was passed to hazptr_acquire().
- * @addr_p: The pointer that is to be hazard-pointer unprotected.
+ * @addr: The address passed to the matching hazptr_acquire().
  *
  * Release the protected hazard pointer recorded in @ctx.
  *
-- 
2.43.0

[PATCH 08/28] hazptrtorture: Add slowpath and lockdep scenarios

[Kunwu Chan]

Add SLOWPATH and LOCKDEP configs: SLOWPATH enables
CONFIG_HAZPTR_ACQUIRE_FORCE_SLOWPATH for slow-path stress; LOCKDEP
combines PROVE_LOCKING with wq_churn=1 so each alloc_workqueue/
destroy_workqueue pair exercises the lockdep hazptr paths.

Suggested-by: Boqun Feng <boqun@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/rcu/hazptrtorture.c                    | 48 ++++++++++++++++++-
 .../rcutorture/configs/hazptr/LOCKDEP         | 16 +++++++
 .../rcutorture/configs/hazptr/LOCKDEP.boot    |  1 +
 .../rcutorture/configs/hazptr/SLOWPATH        | 16 +++++++
 4 files changed, 79 insertions(+), 2 deletions(-)
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/SLOWPATH

diff --git a/kernel/rcu/hazptrtorture.c b/kernel/rcu/hazptrtorture.c
index c8b7c208775b..fe09280d063c 100644
--- a/kernel/rcu/hazptrtorture.c
+++ b/kernel/rcu/hazptrtorture.c
@@ -23,6 +23,7 @@
 #include <linux/torture.h>
 #include <linux/hazptr.h>
 #include <linux/rcupdate.h>
+#include <linux/workqueue.h>
 
 #include "rcu.h"
 
@@ -51,6 +52,8 @@ torture_param(int, shutdown_secs, 0, "Shutdown time (s), <= zero to disable.");
 torture_param(int, stat_interval, 60, "Number of seconds between stats printk()s");
 torture_param(int, stutter, 5, "Number of seconds to run/halt test");
 torture_param(int, verbose, 1, "Enable verbose debugging printk()s");
+torture_param(bool, wq_churn, false,
+	      "Churn workqueues to exercise lockdep dynamic keys");
 
 static char *torture_type = "hazptr";
 module_param(torture_type, charp, 0444);
@@ -62,6 +65,7 @@ static struct task_struct *preempt_task;
 static struct task_struct **reader_tasks;
 static struct task_struct *do_pending_task;
 static struct task_struct *stats_task;
+static struct task_struct *wq_churn_task;
 
 #define HAZPTR_TORTURE_PIPE_LEN 10
 
@@ -83,6 +87,7 @@ static atomic_t n_hazptr_torture_alloc;
 static atomic_t n_hazptr_torture_alloc_fail;
 static atomic_t n_hazptr_torture_free;
 static atomic_t n_hazptr_torture_error;
+static atomic_t n_hazptr_torture_wq_churn;
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_acquires);
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_releases);
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_acquires_irq);
@@ -668,13 +673,14 @@ hazptr_torture_stats_print(void)
 		atomic_read(&n_hazptr_torture_alloc_fail),
 		atomic_read(&n_hazptr_torture_free));
 	torture_onoff_stats();
-	pr_cont("acq: %lld rel: %lld acqirq: %lld relirq: %lld reldefer: %lld relundefer %lld\n",
+	pr_cont("acq: %lld rel: %lld acqirq: %lld relirq: %lld reldefer: %lld relundefer %lld wqchurn: %d\n",
 		torture_sum_pcpu_atomic_long(&hazptr_torture_acquires),
 		torture_sum_pcpu_atomic_long(&hazptr_torture_releases),
 		torture_sum_pcpu_atomic_long(&hazptr_torture_acquires_irq),
 		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_irq),
 		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_defer),
-		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_undefer));
+		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_undefer),
+		atomic_read(&n_hazptr_torture_wq_churn));
 
 	pr_alert("%s%s ", torture_type, TORTURE_FLAG);
 	if (i > 1) {
@@ -788,6 +794,38 @@ static int hazptr_torture_preempt(void *unused)
 	return 0;
 }
 
+static void hazptr_torture_wq_churn_fn(struct work_struct *work)
+{
+}
+
+/*
+ * Repeatedly create and destroy workqueues.  Each workqueue creation
+ * registers a lockdep dynamic key and each destruction unregisters it,
+ * which exercises both is_dynamic_key() and lockdep_unregister_key().
+ */
+static int hazptr_torture_wq_churn(void *unused)
+{
+	VERBOSE_TOROUT_STRING("hazptr_torture_wq_churn task started");
+	do {
+		struct workqueue_struct *wq;
+		struct work_struct churn_work;
+
+		wq = alloc_workqueue("hazptr-torture-churn", WQ_PERCPU, 1);
+		if (!wq) {
+			torture_hrtimeout_ms(10, 10, NULL);
+			continue;
+		}
+		INIT_WORK(&churn_work, hazptr_torture_wq_churn_fn);
+		queue_work(wq, &churn_work);
+		flush_work(&churn_work);
+		destroy_workqueue(wq);
+		atomic_inc(&n_hazptr_torture_wq_churn);
+		stutter_wait("hazptr_torture_wq_churn");
+	} while (!torture_must_stop());
+	torture_kthread_stopping("hazptr_torture_wq_churn");
+	return 0;
+}
+
 static void
 hazptr_torture_cleanup(void)
 {
@@ -880,6 +918,7 @@ static int __init hazptr_torture_init(void)
 	atomic_set(&n_hazptr_torture_alloc_fail, 0);
 	atomic_set(&n_hazptr_torture_free, 0);
 	atomic_set(&n_hazptr_torture_error, 0);
+	atomic_set(&n_hazptr_torture_wq_churn, 0);
 	for (i = 0; i < HAZPTR_TORTURE_PIPE_LEN + 1; i++)
 		atomic_set(&hazptr_torture_wcount[i], 0);
 	for_each_possible_cpu(cpu) {
@@ -960,6 +999,11 @@ static int __init hazptr_torture_init(void)
 		if (torture_init_error(firsterr))
 			goto unwind;
 	}
+	if (wq_churn) {
+		firsterr = torture_create_kthread(hazptr_torture_wq_churn, NULL, wq_churn_task);
+		if (torture_init_error(firsterr))
+			goto unwind;
+	}
 
 	torture_init_end();
 	return 0;
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP b/tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP
new file mode 100644
index 000000000000..f0ff29925a54
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP
@@ -0,0 +1,16 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=16
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_DEBUG_LOCK_ALLOC=y
+CONFIG_PROVE_LOCKING=y
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP.boot b/tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP.boot
new file mode 100644
index 000000000000..7b16720214bd
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/LOCKDEP.boot
@@ -0,0 +1 @@
+hazptrtorture.wq_churn=1
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/SLOWPATH b/tools/testing/selftests/rcutorture/configs/hazptr/SLOWPATH
new file mode 100644
index 000000000000..4d13028ca84c
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/SLOWPATH
@@ -0,0 +1,16 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=16
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
+CONFIG_HAZPTR_ACQUIRE_FORCE_SLOWPATH=y
-- 
2.43.0

[PATCH 09/28] hazptrtorture: Add READERS4 and READERS0 torture configs

[Kunwu Chan]

Add READERS4 and READERS0 configs to cover reader-active and
reader-free synchronize paths.

Both use the PREEMPT baseline and hazptr-stack torture type,
with four readers for READERS4 and no readers for READERS0.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 .../selftests/rcutorture/configs/hazptr/READERS0 | 16 ++++++++++++++++
 .../rcutorture/configs/hazptr/READERS0.boot      |  2 ++
 .../selftests/rcutorture/configs/hazptr/READERS4 | 16 ++++++++++++++++
 .../rcutorture/configs/hazptr/READERS4.boot      |  2 ++
 4 files changed, 36 insertions(+)
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS0
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS0.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS4
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/READERS4.boot

diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/READERS0 b/tools/testing/selftests/rcutorture/configs/hazptr/READERS0
new file mode 100644
index 000000000000..c5b12f3bef74
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/READERS0
@@ -0,0 +1,16 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=16
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_DEBUG_LOCK_ALLOC=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/READERS0.boot b/tools/testing/selftests/rcutorture/configs/hazptr/READERS0.boot
new file mode 100644
index 000000000000..f5564c218902
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/READERS0.boot
@@ -0,0 +1,2 @@
+hazptrtorture.nreaders=0
+hazptrtorture.torture_type=hazptr-stack
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/READERS4 b/tools/testing/selftests/rcutorture/configs/hazptr/READERS4
new file mode 100644
index 000000000000..c5b12f3bef74
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/READERS4
@@ -0,0 +1,16 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=16
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_DEBUG_LOCK_ALLOC=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/READERS4.boot b/tools/testing/selftests/rcutorture/configs/hazptr/READERS4.boot
new file mode 100644
index 000000000000..0d0df3f7dca4
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/READERS4.boot
@@ -0,0 +1,2 @@
+hazptrtorture.nreaders=4
+hazptrtorture.torture_type=hazptr-stack
-- 
2.43.0

[PATCH 10/28] hazptrtorture: Add 128- and 256-CPU configs

[Kunwu Chan]

Add CPU128 and CPU256 configs for hazptr-stack torture testing.
These configs exercise the shared-scan path at larger CPU counts.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
Acked-by: Bradley Morgan <brads@mainlining.org>
---
 .../selftests/rcutorture/configs/hazptr/CPU128   | 16 ++++++++++++++++
 .../rcutorture/configs/hazptr/CPU128.boot        |  1 +
 .../selftests/rcutorture/configs/hazptr/CPU256   | 16 ++++++++++++++++
 .../rcutorture/configs/hazptr/CPU256.boot        |  1 +
 4 files changed, 34 insertions(+)
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU128
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU128.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU256
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/CPU256.boot

diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/CPU128 b/tools/testing/selftests/rcutorture/configs/hazptr/CPU128
new file mode 100644
index 000000000000..079b463607e8
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/CPU128
@@ -0,0 +1,16 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=128
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_DEBUG_LOCK_ALLOC=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/CPU128.boot b/tools/testing/selftests/rcutorture/configs/hazptr/CPU128.boot
new file mode 100644
index 000000000000..1d09a6446080
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/CPU128.boot
@@ -0,0 +1 @@
+hazptrtorture.torture_type=hazptr-stack
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/CPU256 b/tools/testing/selftests/rcutorture/configs/hazptr/CPU256
new file mode 100644
index 000000000000..b00a984977f3
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/CPU256
@@ -0,0 +1,16 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=256
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_DEBUG_LOCK_ALLOC=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/CPU256.boot b/tools/testing/selftests/rcutorture/configs/hazptr/CPU256.boot
new file mode 100644
index 000000000000..1d09a6446080
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/CPU256.boot
@@ -0,0 +1 @@
+hazptrtorture.torture_type=hazptr-stack
-- 
2.43.0

[PATCH 11/28] selftests/rcutorture: Add hazptr torture test script

[Kunwu Chan]

Add hazptr.sh: runs the NOPREEMPT/PREEMPT/READERS0/READERS4 basic
scenarios, the LOCKDEP (PROVE_LOCKING + wq_churn) and SLOWPATH
(forced slow path) combinations, the CPU-scaling sweep, and the
HOTADDR/LONGHOLD staged scenarios, checking each console log for
the module verdict (End of test: FAILURE), lockdep splats, and
unexpected stall-detector reports.  Also adds the torture configs
to CFLIST.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 .../selftests/rcutorture/bin/hazptr.sh        | 319 ++++++++++++++++++
 .../rcutorture/configs/hazptr/CFLIST          |   6 +
 2 files changed, 325 insertions(+)
 create mode 100755 tools/testing/selftests/rcutorture/bin/hazptr.sh

diff --git a/tools/testing/selftests/rcutorture/bin/hazptr.sh b/tools/testing/selftests/rcutorture/bin/hazptr.sh
new file mode 100755
index 000000000000..4e30ea51d9fc
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/bin/hazptr.sh
@@ -0,0 +1,319 @@
+#!/bin/bash
+# SPDX-License-Identifier: GPL-2.0+
+#
+# Run hazptr torture tests and report any that fail.
+#
+# Usage: hazptr.sh [ --cpus N ] [ --allcpus ] [ --datestamp string ]
+#
+# Copyright (C) 2026 Kunwu Chan
+#
+# Authors: Kunwu Chan <kunwu.chan@gmail.com>
+
+usage () {
+	echo "Usage: $scriptname optional arguments:"
+	echo "       --cpus N       override CPU count (default: auto-detect)"
+	echo "       --allcpus      test at CPU counts 8,16,32,...,256"
+	echo "       --datestamp string"
+	exit 1
+}
+
+ds=`date +%Y.%m.%d-%H.%M.%S`-hazptr
+scriptname="$0"
+
+RCUTORTURE="`pwd`/tools/testing/selftests/rcutorture"; export RCUTORTURE
+PATH=${RCUTORTURE}/bin:$PATH; export PATH
+. functions.sh
+
+cpus=""		# empty = let kvm.sh auto-detect
+allcpus=0
+
+while test $# -gt 0
+do
+	case "$1" in
+	--datestamp)
+		checkarg --datestamp "(relative pathname)" "$#" "$2" '^[a-zA-Z0-9._/-]*$' '^--'
+		ds=$2
+		shift
+		;;
+	--cpus)
+		checkarg --cpus "(number)" "$#" "$2" '^[0-9]\+$' '^--'
+		cpus="$2"
+		shift
+		;;
+	--allcpus)
+		allcpus=1
+		;;
+	*)
+		echo Unknown argument $1
+		usage
+		;;
+	esac
+	shift
+done
+
+nerrs=0
+
+run_kvm () {
+	local ds_suffix="$1"
+	local configs="$2"
+	local cpu_arg="$3"
+	local kconfig="$4"
+	local duration="${5:-5s}"
+	local resdir="$RCUTORTURE/res/$ds/$ds_suffix"
+
+	mkdir -p "$resdir"
+
+	tools/testing/selftests/rcutorture/bin/kvm.sh \
+		--torture hazptr \
+		--duration "$duration" \
+		--configs "$configs" \
+		${cpu_arg:+--cpus "$cpu_arg"} \
+		${kconfig:+--kconfig "$kconfig"} \
+		--trust-make \
+		--datestamp "$ds/$ds_suffix" > "$resdir/kvm.sh.out" 2>&1
+	return $?
+}
+
+check_lockdep () {
+	local ds_suffix="$1" config_dir="$2" label="$3"
+	local resdir="$RCUTORTURE/res/$ds/$ds_suffix"
+	local console="$resdir/$config_dir/console.log"
+
+	if ! test -f "$console"
+	then
+		echo "Missing console.log for $label ($ds_suffix)" > "$resdir/kvm.sh.err"
+		return 1
+	fi
+	if grep -qE "WARNING: possible (recursive locking|circular locking dependency)" "$console"
+	then
+		echo "Lockdep warning in $label" > "$resdir/kvm.sh.err"
+		return 1
+	fi
+	return 0
+}
+
+# The stall detector must not fire in normal scenarios: a splat here is
+# either a detector false positive (deadline arithmetic, too-aggressive
+# timeout) or a genuine kernel bug making legitimate waits exceed the
+# timeout.  Either way, fail the test.  (HOTADDR is the positive case
+# and is checked separately.)
+check_stall () {
+	local ds_suffix="$1" config_dir="$2" label="$3"
+	local resdir="$RCUTORTURE/res/$ds/$ds_suffix"
+	local console="$resdir/$config_dir/console.log"
+
+	if ! test -f "$console"
+	then
+		echo "Missing console.log for $label ($ds_suffix)" > "$resdir/kvm.sh.err"
+		return 1
+	fi
+	if grep -q "hazptr: waiter stall" "$console"
+	then
+		echo "Unexpected hazptr stall splat in $label" > "$resdir/kvm.sh.err"
+		return 1
+	fi
+	return 0
+}
+
+# Classify the in-kernel torture verdict from the console log:
+#   FAILURE    -- the module explicitly reported failure
+#   INCOMPLETE -- no module verdict reached (e.g. external limit hit)
+#   SUCCESS    -- the module explicitly reported success
+# Only SUCCESS passes.  kvm.sh does not propagate the per-run exit
+# status of its background kvm-test-1-run.sh jobs, so grep the
+# console directly.
+check_verdict () {
+	local ds_suffix="$1" config_dir="$2" label="$3"
+	local resdir="$RCUTORTURE/res/$ds/$ds_suffix"
+	local console="$resdir/$config_dir/console.log"
+
+	if ! test -f "$console"
+	then
+		echo "Missing console.log for $label ($ds_suffix)" > "$resdir/kvm.sh.err"
+		return 1
+	fi
+	if grep -q "End of test: FAILURE" "$console"
+	then
+		echo "Module verdict FAILURE in $label: $(grep -m1 "End of test: FAILURE" "$console" | cut -c1-160)" > "$resdir/kvm.sh.err"
+		return 1
+	fi
+	if ! grep -q "End of test: SUCCESS" "$console"
+	then
+		echo "No module verdict in $label (INCOMPLETE): $(grep -m1 "MAX_LOCKDEP_KEYS\|BUG:\|hung\|Killed" "$console" 2>/dev/null | cut -c1-120)" > "$resdir/kvm.sh.err"
+		return 1
+	fi
+	return 0
+}
+
+# Basic torture: all hazptr configs, CPU count auto-detected
+echo "--- hazptr basic torture${cpus:+ ($cpus CPUs)} ---"
+if run_kvm basic "NOPREEMPT PREEMPT READERS0 READERS4" "$cpus"
+then
+	for cfg in NOPREEMPT PREEMPT READERS0 READERS4
+	do
+		check_verdict basic "$cfg" "basic/$cfg" ||
+			nerrs=$((nerrs+1))
+		check_stall basic "$cfg" "basic/$cfg" ||
+			nerrs=$((nerrs+1))
+	done
+else
+	nerrs=$((nerrs+1))
+fi
+
+# LOCKDEP: verify no lockdep splats with PROVE_LOCKING + wq_churn
+
+echo "--- hazptr lockdep + wq_churn ---"
+if run_kvm lockdep-wq LOCKDEP "$cpus"
+then
+	check_verdict lockdep-wq LOCKDEP "LOCKDEP wq_churn" ||
+		nerrs=$((nerrs+1))
+	check_lockdep lockdep-wq LOCKDEP "LOCKDEP wq_churn" ||
+		nerrs=$((nerrs+1))
+	check_stall lockdep-wq LOCKDEP "LOCKDEP wq_churn" ||
+		nerrs=$((nerrs+1))
+else
+	nerrs=$((nerrs+1))
+fi
+
+# SLOWPATH + PROVE_LOCKING: verify forced-slowpath does not deadlock
+
+echo "--- hazptr SLOWPATH + lockdep ---"
+if run_kvm slowpath-lockdep SLOWPATH "$cpus" \
+	"CONFIG_PROVE_LOCKING=y"
+then
+		check_verdict slowpath-lockdep SLOWPATH \
+			"SLOWPATH+PROVE_LOCKING" ||
+			nerrs=$((nerrs+1))
+		check_lockdep slowpath-lockdep SLOWPATH \
+			"SLOWPATH+PROVE_LOCKING" ||
+			nerrs=$((nerrs+1))
+		check_stall slowpath-lockdep SLOWPATH \
+			"SLOWPATH+PROVE_LOCKING" ||
+			nerrs=$((nerrs+1))
+else
+	nerrs=$((nerrs+1))
+fi
+
+# HOTADDR: Lai-scenario documenting mode.  Starvation episodes and
+# stall-detector splats are EXPECTED here (baseline behavior); the
+# check verifies the scenario ran and the machinery stayed sane.
+
+echo "--- hazptr HOTADDR (Lai-scenario documenting) ---"
+if run_kvm hotaddr HOTADDR "$cpus" "" "30s"
+then
+	resdir="$RCUTORTURE/res/$ds/hotaddr/HOTADDR"
+	console="$resdir/console.log"
+	hotaddr_err=""
+	if ! test -f "$console"
+	then
+		echo "Missing console.log for HOTADDR" > "$RCUTORTURE/res/$ds/hotaddr/kvm.sh.err"
+		hotaddr_err=1
+	elif ! grep -q "End of test: SUCCESS" "$console"
+	then
+		echo "HOTADDR: module verdict not SUCCESS" > "$resdir/kvm.sh.err"
+		hotaddr_err=1
+	elif ! grep -q "hot r" "$console"
+	then
+		echo "HOTADDR: staged rounds not found" > "$resdir/kvm.sh.err"
+		hotaddr_err=1
+	fi
+	# Independent checks: a failure must not be masked by an earlier
+	# elif taking the note branch.
+	if grep -q "positive control timed out" "$console" 2>/dev/null
+	then
+		echo "HOTADDR: positive control failure (machinery fault)" >> "$resdir/kvm.sh.err"
+		hotaddr_err=1
+	fi
+	if [ -z "$hotaddr_err" ]
+	then
+		if ! grep -q "hazptr: waiter stall" "$console"
+		then
+			echo "HOTADDR note: no stall splats observed (record as not-reproduced, investigate interleaving)" > "$resdir/kvm.sh.note"
+		else
+			echo "HOTADDR: starvation reproduced, stall detector confirmed, drain verified"
+		fi
+	else
+		nerrs=$((nerrs+1))
+	fi
+else
+	nerrs=$((nerrs+1))
+fi
+
+# LONGHOLD: legitimate long-lived references.  The stall detector must
+# fire (observation), the holder must release, and the waiter must drain
+# normally afterwards: diagnostics without interruption, module verdict
+# stays SUCCESS.
+
+echo "--- hazptr LONGHOLD (legitimate long-held reference) ---"
+if run_kvm longhold LONGHOLD "$cpus" "" "60s"
+then
+	resdir="$RCUTORTURE/res/$ds/longhold/LONGHOLD"
+	console="$resdir/console.log"
+	longhold_err=""
+	if ! test -f "$console"
+	then
+		echo "Missing console.log for LONGHOLD" > "$RCUTORTURE/res/$ds/longhold/kvm.sh.err"
+		longhold_err=1
+	elif ! grep -q "End of test: SUCCESS" "$console"
+	then
+		echo "LONGHOLD: module verdict not SUCCESS" > "$resdir/kvm.sh.err"
+		longhold_err=1
+	elif ! grep -q "longhold drain=1" "$console"
+	then
+		echo "LONGHOLD: drain after release not observed" > "$resdir/kvm.sh.err"
+		longhold_err=1
+	fi
+	# Independent check: the sync must drain after release; this
+	# must not be masked by the stall-report note branch.
+	if grep -q "longhold: sync did not drain" "$console" 2>/dev/null
+	then
+		echo "LONGHOLD: sync failed to drain after release" >> "$resdir/kvm.sh.err"
+		longhold_err=1
+	fi
+	if [ -z "$longhold_err" ]
+	then
+		if ! grep -q "hazptr: waiter stall" "$console"
+		then
+			echo "LONGHOLD note: stall detector did not fire on a legitimate long hold" > "$resdir/kvm.sh.note"
+		else
+			echo "LONGHOLD: detector fired on legitimate hold, drain verified, no interruption"
+		fi
+	else
+		nerrs=$((nerrs+1))
+	fi
+else
+	nerrs=$((nerrs+1))
+fi
+
+# CPU scaling: optionally test at multiple CPU counts
+
+if test "$allcpus" -ne 0
+then
+	max_cpus=`grep '^processor' /proc/cpuinfo 2>/dev/null | wc -l`
+	for n in 8 16 32 64 128 256
+	do
+		if test "$n" -le "$max_cpus"
+		then
+		echo "--- hazptr CPU=$n ---"
+		if run_kvm "cpu$n" "PREEMPT" "$n" "CONFIG_NR_CPUS=$n"
+		then
+			check_verdict "cpu$n" PREEMPT "cpu$n/PREEMPT" ||
+				nerrs=$((nerrs+1))
+			check_stall "cpu$n" PREEMPT "cpu$n/PREEMPT" ||
+				nerrs=$((nerrs+1))
+		else
+			nerrs=$((nerrs+1))
+		fi
+		fi
+	done
+fi
+
+# Report
+
+if test "$nerrs" -ne 0
+then
+	echo "hazptr.sh: $nerrs test(s) failed"
+	exit 1
+fi
+echo "hazptr.sh: all tests passed"
+exit 0
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST b/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
index 4d62eb4a39f9..90ef09b0e654 100644
--- a/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
@@ -1,2 +1,8 @@
 NOPREEMPT
 PREEMPT
+READERS0
+READERS4
+SLOWPATH
+LOCKDEP
+CPU128
+CPU256
-- 
2.43.0

[PATCH 12/28] hazptr: Remove per-waiter scan fallback, simplify synchronize

[Kunwu Chan]

With try_acquire, per-CPU slots publish their target address
directly (no wildcard transient), so a single Bloom-walk pass
suffices.  Drop the per-waiter exact-match fallback that
busy-waited on remote slots; waiters whose address is in the
Bloom filter wait for the next scan cycle.

Extend hazptr_scan_walk() with an optional @watch parameter for
exact-match lookups (used by the upcoming stall detector).

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/hazptr.c | 104 +++++++++++++++++++++++++++---------------------
 1 file changed, 58 insertions(+), 46 deletions(-)

diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index 059c356ceaa7..4784df78be4e 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -39,10 +39,6 @@ struct hazptr_overflow_list {
  * successively iterates on both lists. Therefore, only list removals
  * can cause the iteration to retry, and the number of removals is
  * limited to the number of list elements.
- *
- * Due to the overflow list raw spin lock, the hazard pointer readers are
- * blocking, starvation-free with bounded waiting, assuming bounded critical
- * sections and no NMI or virtualization-induced holder preemption.
  */
 struct hazptr_overflow_list_flip {
 	struct hazptr_overflow_list array[2];
@@ -153,12 +149,12 @@ void *__hazptr_acquire(struct hazptr_ctx *ctx, void * const *addr_p)
 EXPORT_SYMBOL_GPL(__hazptr_acquire);
 
 /*
- * Perform piecewise iteration on overflow list waiting until "addr" and
- * wildcard are not present. Raw spinlock is released and taken between each
- * list item and busy loop iteration. The overflow list generation is checked
- * each time the lock is taken to validate that the list has not changed before
- * resuming iteration or busy wait. If the generation has changed, retry the
- * entire list traversal.
+ * Perform piecewise iteration on overflow list waiting until "addr" is
+ * not present. Raw spinlock is released and taken between each list
+ * item and busy loop iteration. The overflow list generation is checked
+ * each time the lock is taken to validate that the list has not changed
+ * before resuming iteration or busy wait. If the generation has
+ * changed, retry the entire list traversal.
  */
 static
 void hazptr_synchronize_overflow_list(struct hazptr_overflow_list *overflow_list, void *addr)
@@ -228,8 +224,8 @@ void hazptr_scan_cpu_slots(void *addr)
 	}
 }
 
-static
-void hazptr_scan_overflow_list_period(void *addr, unsigned int scan_idx)
+static void __maybe_unused
+hazptr_scan_overflow_list_period(void *addr, unsigned int scan_idx)
 {
 	int cpu;
 
@@ -310,15 +306,28 @@ static bool hazptr_bloom_contains(const struct hazptr_bloom *bloom, void *addr)
 }
 
 /*
- * Walk all slots and record observed non-wildcard addresses into
- * the Bloom filter.  A single pass suffices on the try_acquire
- * fast path (readers publish their target directly).
+ * Walk all slots and return true if @watch (non-NULL) is present.
+ * If @bloom is non-NULL, record observed non-wildcard addresses in it.
+ * A NULL @watch means pure collection mode: never early-returns on a
+ * match.
+ *
+ * A wildcard in an overflow-list slot means a slow-path acquire is in
+ * flight: that slot may resolve to any address (including @watch or a
+ * Bloom-collected one) after the walk.  To keep the single-pass
+ * completion decision safe, the caller must observe wildcards and
+ * defer completion decisions for one cycle; @wildcard (if non-NULL)
+ * is set to true in that case.  In watch mode, a wildcard also counts
+ * as "present" so exact-match callers do not complete.
  */
-static void hazptr_scan_walk(struct hazptr_bloom *bloom)
+static bool hazptr_scan_walk(void *watch, struct hazptr_bloom *bloom,
+			     bool *wildcard)
 {
 	int cpu;
 
-	hazptr_bloom_reset(bloom);
+	if (bloom)
+		hazptr_bloom_reset(bloom);
+	if (wildcard)
+		*wildcard = false;
 
 	for_each_possible_cpu(cpu) {
 		struct hazptr_percpu_slots *percpu_slots = per_cpu_ptr(&hazptr_percpu_slots, cpu);
@@ -329,8 +338,11 @@ static void hazptr_scan_walk(struct hazptr_bloom *bloom)
 			struct hazptr_slot_item *item = &percpu_slots->items[idx];
 			void *v;
 
+			/* Pairs with smp_store_release in hazptr_release(). */
 			v = smp_load_acquire(&item->slot.addr);
-			if (v && !is_wildcard(v))
+			if (watch && (v == watch || is_wildcard(v)))
+				return true;
+			if (bloom && v && !is_wildcard(v))
 				hazptr_bloom_add(bloom, v);
 		}
 		for (int i = 0; i < 2; i++) {
@@ -342,16 +354,24 @@ static void hazptr_scan_walk(struct hazptr_bloom *bloom)
 			hlist_for_each_entry(b, &list->head, overflow_node) {
 				void *v;
 
+				/* Pairs with smp_store_release in hazptr_release(). */
 				v = smp_load_acquire(&b->slot.addr);
-				if (v && !is_wildcard(v))
+				if (watch && (v == watch || is_wildcard(v))) {
+					raw_spin_unlock_irqrestore(&list->lock, flags);
+					return true;
+				}
+				if (bloom && v && !is_wildcard(v))
 					hazptr_bloom_add(bloom, v);
+				else if (v && is_wildcard(v) && wildcard)
+					*wildcard = true;
 			}
 			raw_spin_unlock_irqrestore(&list->lock, flags);
 		}
 	}
+
+	return false;
 }
 
-/* Need is_wildcard for the Bloom filter */
 /*
  * Shared scan: batch concurrent hazptr_synchronize() callers into a
  * single kthread-driven scan, avoiding redundant per-CPU and
@@ -416,28 +436,32 @@ static void hazptr_scan_do_cycle(void)
 	}
 
 	/*
-	 * Walk all slots and record observed addresses into the
+	 * Scan all slots: record observed addresses into the
 	 * Bloom filter, then complete waiters whose address is
 	 * not present.  On the try_acquire fast path, readers
 	 * publish their target directly (no wildcard transient),
-	 * so a single pass suffices.
+	 * so a single pass suffices for per-CPU slots.
 	 */
-	hazptr_scan_walk(&hazptr_scan.bloom);
+	{
+		bool wildcard;
 
-	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
-		if (!hazptr_bloom_contains(&hazptr_scan.bloom, w->addr))
-			list_move(&w->node, &done);
+		hazptr_scan_walk(NULL, &hazptr_scan.bloom, &wildcard);
+		/*
+		 * A wildcard in an overflow-list slot is a slow-path
+		 * acquire in flight: that slot can resolve to any
+		 * address after the walk, so defer completion
+		 * decisions to the next cycle.
+		 */
+		if (!wildcard)
+			/* Complete waiters whose address is not in the filter. */
+			list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
+				if (!hazptr_bloom_contains(&hazptr_scan.bloom, w->addr))
+					list_move(&w->node, &done);
+			}
 	}
 
 	mutex_unlock(&hazptr_phase_lock);
 
-	/* Waiters still blocked: their address was observed present. */
-	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
-		hazptr_scan_cpu_slots(w->addr);
-		hazptr_scan_overflow_list_all(w->addr);
-		list_move(&w->node, &done);
-	}
-
 	list_for_each_entry_safe(w, n, &done, node) {
 		list_del_init(&w->node);
 		complete(&w->done);
@@ -496,8 +520,6 @@ core_initcall(hazptr_scan_init);
  */
 void hazptr_synchronize(void *addr)
 {
-	unsigned int scan_list_phase;
-
 	/*
 	 * Busy-wait should only be done from preemptible context.
 	 */
@@ -521,18 +543,8 @@ void hazptr_synchronize(void *addr)
 	}
 
 	guard(mutex)(&hazptr_phase_lock);
-
-	/* Scan per-CPU slots. */
 	hazptr_scan_cpu_slots(addr);
-
-	/*
-	 * Scan overflow lists *after* scanning per-CPU slots. See
-	 * hazptr_promote_to_backup_slot() for scan ordering requirement.
-	 */
-	scan_list_phase = flip_list_phase(hazptr_overflow_list_phase);
-	hazptr_scan_overflow_list_period(addr, scan_list_phase);
-	WRITE_ONCE(hazptr_overflow_list_phase, scan_list_phase);	/* Flip the current list phase. */
-	hazptr_scan_overflow_list_period(addr, flip_list_phase(scan_list_phase));
+	hazptr_scan_overflow_list_all(addr);
 }
 EXPORT_SYMBOL_GPL(hazptr_synchronize);
 
-- 
2.43.0

[PATCH 13/28] hazptr: Add waiter-enqueue tracepoint

[Kunwu Chan]

Production tracepoint fired when hazptr_synchronize() enqueues a
waiter onto the scan kthread's pending list.  Downstream users
include hazptrtorture's staged hot-address scenario, which uses
it to confirm enqueue before running its controls.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 include/trace/events/hazptr.h | 29 +++++++++++++++++++++++++++++
 kernel/hazptr.c               |  5 +++++
 2 files changed, 34 insertions(+)
 create mode 100644 include/trace/events/hazptr.h

diff --git a/include/trace/events/hazptr.h b/include/trace/events/hazptr.h
new file mode 100644
index 000000000000..1eed4c9fd682
--- /dev/null
+++ b/include/trace/events/hazptr.h
@@ -0,0 +1,29 @@
+/* SPDX-License-Identifier: GPL-2.0+ */
+#undef TRACE_SYSTEM
+#define TRACE_SYSTEM hazptr
+
+#if !defined(_TRACE_HAZPTR_H) || defined(TRACE_HEADER_MULTI_READ)
+#define _TRACE_HAZPTR_H
+
+#include <linux/tracepoint.h>
+
+TRACE_EVENT(hazptr_waiter_enqueue,
+
+	TP_PROTO(void *addr),
+
+	TP_ARGS(addr),
+
+	TP_STRUCT__entry(
+		__field(void *, addr)
+	),
+
+	TP_fast_assign(
+		__entry->addr = addr;
+	),
+
+	TP_printk("addr=%p", __entry->addr)
+);
+
+#endif /* _TRACE_HAZPTR_H */
+
+#include <trace/define_trace.h>
diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index 4784df78be4e..16f6e2acaf87 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -19,6 +19,10 @@
 #include <linux/bitops.h>
 #include <linux/bitmap.h>
 
+#define CREATE_TRACE_POINTS
+#include <trace/events/hazptr.h>
+EXPORT_TRACEPOINT_SYMBOL_GPL(hazptr_waiter_enqueue);
+
 #define HAZPTR_WILDCARD	((void *) 1UL)
 
 static DEFINE_MUTEX(hazptr_phase_lock);	/* Protect the list phase flip. */
@@ -411,6 +415,7 @@ static void hazptr_synchronize_queued(void *addr)
 
 	mutex_lock(&hazptr_scan.lock);
 	list_add_tail(&waiter.node, &hazptr_scan.pending);
+	trace_hazptr_waiter_enqueue(addr);
 	if (!READ_ONCE(hazptr_scan.wakeup)) {
 		WRITE_ONCE(hazptr_scan.wakeup, true);
 		swake_up_one(&hazptr_scan.wq);
-- 
2.43.0

[PATCH 14/28] hazptr: Add waiter stall detector

[Kunwu Chan]

RCU-CPU-stall-style diagnostic for hazptr_synchronize() waiters
that remain blocked beyond a configurable timeout
(CONFIG_HAZPTR_STALL_DETECTOR, default y; timeout 21s,
CONFIG_HAZPTR_STALL_TIMEOUT).  Purely observational: never
changes synchronization semantics or completion conditions.

The scan kthread checks still-blocked waiters after each cycle.
On timeout, an exact-match walk collects holder blame: per-CPU
slot (cpu, idx), overflow-list entry, and under
CONFIG_HAZPTR_DEBUG the holder's pid, held time, acquire caller,
and task stack (sched_show_task).  Per-holder held times are
evidence, not classification.

/sys/kernel/hazptr/hazptr_stall_count exposes the cumulative
event count.  Adds acquire_jiffies to hazptr_ctx under
CONFIG_HAZPTR_DEBUG so held time can be reported.

Suggested-by: Paul E. McKenney <paulmck@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 include/linux/hazptr.h   |  13 ++
 kernel/hazptr.c          | 263 ++++++++++++++++++++++++++++++++++++++-
 kernel/rcu/Kconfig.debug |  16 +++
 3 files changed, 288 insertions(+), 4 deletions(-)

diff --git a/include/linux/hazptr.h b/include/linux/hazptr.h
index 0fb48518525b..ebf804315a26 100644
--- a/include/linux/hazptr.h
+++ b/include/linux/hazptr.h
@@ -24,6 +24,7 @@
 #include <linux/percpu.h>
 #include <linux/types.h>
 #include <linux/cleanup.h>
+#include <linux/jiffies.h>
 #include <linux/sched.h>
 #include <linux/ptr_eq.h>
 
@@ -55,6 +56,7 @@ struct hazptr_ctx {
 	bool detach_task, detach_cpu;	/* Whether the ctx has been detached from task/cpu. */
 	int acquire_pid, acquire_cpu;	/* Note the task and cpu number at acquire. */
 	unsigned long acquire_caller;	/* Acquire instruction pointer. */
+	unsigned long acquire_jiffies;	/* Acquire-attempt time, for stall diagnostics. */
 #endif
 };
 
@@ -65,6 +67,11 @@ struct hazptr_slot_ctx {
 struct hazptr_slot_item {
 	struct hazptr_slot slot;
 	struct hazptr_slot_ctx ctx;
+#ifdef CONFIG_HAZPTR_DEBUG
+	int acquire_pid;
+	unsigned long acquire_caller;
+	unsigned long acquire_jiffies;
+#endif
 };
 
 struct hazptr_percpu_slots {
@@ -270,6 +277,7 @@ void *hazptr_acquire(struct hazptr_ctx *ctx, void * const *addr_p)
 	ctx->acquire_pid = current->pid;
 	ctx->acquire_cpu = smp_processor_id();
 	ctx->acquire_caller = _THIS_IP_;
+	ctx->acquire_jiffies = jiffies;
 #endif
 	if (IS_ENABLED(CONFIG_HAZPTR_ACQUIRE_FORCE_SLOWPATH))
 		return __hazptr_acquire(ctx, addr_p);
@@ -278,6 +286,11 @@ void *hazptr_acquire(struct hazptr_ctx *ctx, void * const *addr_p)
 		return __hazptr_acquire(ctx, addr_p);
 	slot_item->ctx.ctx = ctx;
 	ctx->slot = slot;
+#ifdef CONFIG_HAZPTR_DEBUG
+	slot_item->acquire_pid = ctx->acquire_pid;
+	slot_item->acquire_caller = ctx->acquire_caller;
+	slot_item->acquire_jiffies = ctx->acquire_jiffies;
+#endif
 	return addr;
 }
 
diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index 16f6e2acaf87..8de194c07eea 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -16,6 +16,8 @@
 #include <linux/kthread.h>
 #include <linux/swait.h>
 #include <linux/sched.h>
+#include <linux/sched/debug.h>
+#include <linux/nmi.h>
 #include <linux/bitops.h>
 #include <linux/bitmap.h>
 
@@ -107,6 +109,11 @@ void *__hazptr_acquire(struct hazptr_ctx *ctx, void * const *addr_p)
 			 */
 			item->ctx.ctx = ctx;
 			ctx->slot = slot;
+#ifdef CONFIG_HAZPTR_DEBUG
+			item->acquire_pid = ctx->acquire_pid;
+			item->acquire_caller = ctx->acquire_caller;
+			item->acquire_jiffies = ctx->acquire_jiffies;
+#endif
 			return addr;
 		}
 	}
@@ -387,6 +394,14 @@ struct hazptr_waiter {
 	struct list_head node;
 	void *addr;
 	struct completion done;
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+	unsigned long enqueue_jiffies;
+	unsigned long last_report;
+	unsigned int cycles;
+	unsigned int reports;
+	int pid;
+	unsigned long caller;
+#endif
 };
 
 struct hazptr_scan_state {
@@ -400,11 +415,14 @@ struct hazptr_scan_state {
 };
 static struct hazptr_scan_state hazptr_scan;
 
+/* Sysfs counters, cf. RCU's rcu_stall_count. */
+static atomic_t hazptr_stall_count = ATOMIC_INIT(0);
+
 /*
- * Queue @addr for the shared scan.  The waiter lives on the caller's
- * stack, so no allocation is needed.
+ * Queue @addr for the shared scan.  The waiter lives on the
+ * caller's stack, so no allocation is needed.
  */
-static void hazptr_synchronize_queued(void *addr)
+static void hazptr_synchronize_queued(void *addr, unsigned long caller)
 {
 	struct hazptr_waiter waiter = {
 		.addr = addr,
@@ -412,7 +430,16 @@ static void hazptr_synchronize_queued(void *addr)
 
 	init_completion(&waiter.done);
 	INIT_LIST_HEAD(&waiter.node);
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+	waiter.enqueue_jiffies = jiffies;
+	waiter.last_report = jiffies;
+	waiter.cycles = 0;
+	waiter.reports = 0;
+	waiter.pid = task_pid_nr(current);
+	waiter.caller = caller;
+#endif
 
+	/* Enqueue and wake the scan kthread. */
 	mutex_lock(&hazptr_scan.lock);
 	list_add_tail(&waiter.node, &hazptr_scan.pending);
 	trace_hazptr_waiter_enqueue(addr);
@@ -422,9 +449,14 @@ static void hazptr_synchronize_queued(void *addr)
 	}
 	mutex_unlock(&hazptr_scan.lock);
 
+	/* Sleep until the scan kthread completes this waiter. */
 	wait_for_completion(&waiter.done);
 }
 
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+static void hazptr_waiter_stall_check(struct hazptr_waiter *w);
+#endif
+
 static void hazptr_scan_do_cycle(void)
 {
 	struct hazptr_waiter *w, *n;
@@ -471,8 +503,201 @@ static void hazptr_scan_do_cycle(void)
 		list_del_init(&w->node);
 		complete(&w->done);
 	}
+
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+	/* Stall diagnostics for waiters still blocked. */
+	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node)
+		hazptr_waiter_stall_check(w);
+#endif
+}
+
+
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+
+/*
+ * Holder blame collected by the stall walk.  Under
+ * CONFIG_HAZPTR_DEBUG, the per-CPU slot's ctx pointer and the
+ * overflow-list node's embedding ctx provide pid and acquire
+ * caller.  These reads are racy diagnostics: the holder may be
+ * releasing concurrently, so blame may be stale -- acceptable for
+ * a debug splat, never used for correctness.
+ */
+struct hazptr_stall_holder {
+	int cpu;		/* -1: overflow-list entry */
+	int idx;
+	int pid;
+	unsigned long caller;
+	unsigned long held;	/* jiffies held at observation (DEBUG) */
+};
+
+/*
+ * Exact-match walk for stall diagnostics: count slots holding
+ * @addr and fill up to @max holders with blame.  Unlike the
+ * wildcard drain, this is safe outside hazptr_phase_lock: it
+ * matches plain slot values and takes the overflow-list locks.
+ * Expensive: only called for stalled waiters, never per cycle.
+ *
+ * The result is a concurrent observation, not an atomic snapshot:
+ * holders may release, re-acquire, or migrate to the overflow list
+ * while the walk runs.  The per-holder fields (pid, held) are
+ * racy reads best-effort diagnostics.
+ */
+static int hazptr_stall_walk(void *addr,
+			     struct hazptr_stall_holder *holders, int max)
+{
+	int cpu, n = 0;
+
+	for_each_possible_cpu(cpu) {
+		struct hazptr_percpu_slots *percpu_slots = per_cpu_ptr(&hazptr_percpu_slots, cpu);
+		struct hazptr_overflow_list_flip *overflow_list_flip = per_cpu_ptr(&percpu_overflow_list_flip, cpu);
+		unsigned int idx;
+
+		for (idx = 0; idx < NR_HAZPTR_PERCPU_SLOTS; idx++) {
+			struct hazptr_slot_item *item = &percpu_slots->items[idx];
+			struct hazptr_stall_holder *h;
+
+			if (smp_load_acquire(&item->slot.addr) != addr)
+				continue;
+			if (n < max) {
+				h = &holders[n];
+				h->cpu = cpu;
+				h->idx = idx;
+				h->pid = -1;
+				h->caller = 0;
+				h->held = 0;
+#ifdef CONFIG_HAZPTR_DEBUG
+				/*
+				 * Read diagnostics from the stable per-CPU
+				 * slot_item rather than the ctx pointer,
+				 * which may belong to a reader that has
+				 * already exited its scope.
+				 */
+				/*
+				 * Best-effort: fields are written after
+				 * the slot address is published, so they
+				 * may briefly describe the previous
+				 * holder.  held is measured from the
+				 * acquire attempt start, not from slot
+				 * publication.
+				 */
+				h->pid = READ_ONCE(item->acquire_pid);
+				h->caller = READ_ONCE(item->acquire_caller);
+				h->held = jiffies - READ_ONCE(item->acquire_jiffies);
+#endif
+			}
+			n++;
+		}
+		for (int i = 0; i < 2; i++) {
+			struct hazptr_overflow_list *list = &overflow_list_flip->array[i];
+			struct hazptr_backup_slot *b;
+			unsigned long flags;
+
+			raw_spin_lock_irqsave(&list->lock, flags);
+			hlist_for_each_entry(b, &list->head, overflow_node) {
+				struct hazptr_stall_holder *h;
+
+				if (smp_load_acquire(&b->slot.addr) != addr)
+					continue;
+				if (n < max) {
+					h = &holders[n];
+					h->cpu = -1;
+					h->idx = i;
+					h->pid = -1;
+					h->caller = 0;
+					h->held = 0;
+#ifdef CONFIG_HAZPTR_DEBUG
+					{
+						struct hazptr_ctx *ctx = container_of(b, struct hazptr_ctx, backup_slot);
+
+						h->pid = READ_ONCE(ctx->acquire_pid);
+						h->caller = READ_ONCE(ctx->acquire_caller);
+						h->held = jiffies - READ_ONCE(ctx->acquire_jiffies);
+					}
+#endif
+				}
+				n++;
+			}
+			raw_spin_unlock_irqrestore(&list->lock, flags);
+		}
+	}
+	return n;
+}
+
+/*
+ * Collect and print holder blame for a stalled waiter.
+ */
+static void hazptr_waiter_stall_report(struct hazptr_waiter *w)
+{
+	struct hazptr_stall_holder holders[16];
+	unsigned long waited = jiffies - w->enqueue_jiffies;
+	int n, i;
+
+	pr_warn("hazptr: waiter stall: pid %d blocked %ums in hazptr_synchronize(%p) caller %pS, %u scan cycles, report #%u\n",
+		w->pid, jiffies_to_msecs(waited), w->addr,
+		(void *)w->caller, w->cycles, w->reports);
+
+	n = hazptr_stall_walk(w->addr, holders, 16);
+	if (!n) {
+		pr_warn("hazptr: stall: no matching holder found during diagnostic walk\n");
+		return;
+	}
+	pr_warn("hazptr: stall: %d slot(s) hold this address\n", n);
+	pr_warn("hazptr: stall: long-lived references are a hazard-pointer feature, so a long wait is not necessarily a problem.  Evidence for interpretation: across reports, per-holder held times that reset mean that slot was re-acquired; growing held times on the same slot mean that acquisition persists (long-held reference, or a holder stuck in its critical section -- check the task).  Compare the full holder set across reports to help interpret re-publication vs. a single long-held reference.\n");
+	for (i = 0; i < min(n, 16); i++) {
+		if (holders[i].cpu >= 0)
+			pr_warn("hazptr: stall:   cpu=%d slot=%d",
+				holders[i].cpu, holders[i].idx);
+		else
+			pr_warn("hazptr: stall:   overflow-list%d",
+				holders[i].idx);
+#ifdef CONFIG_HAZPTR_DEBUG
+		if (holders[i].pid >= 0) {
+			struct task_struct *t;
+
+			pr_cont(" pid=%d held=%ums caller=%pS",
+				holders[i].pid,
+				jiffies_to_msecs(holders[i].held),
+				(void *)holders[i].caller);
+			rcu_read_lock();
+			t = find_task_by_vpid(holders[i].pid);
+			if (t) {
+				touch_nmi_watchdog();
+				sched_show_task(t);
+			}
+			rcu_read_unlock();
+		}
+#endif
+		pr_cont("\n");
+	}
+	if (n > 16)
+		pr_warn("hazptr: stall:   ... and %d more\n", n - 16);
 }
 
+/*
+ * Stall check for one still-blocked waiter, called from the scan
+ * kthread after each cycle.  Purely diagnostic: never changes
+ * completion conditions.  First report at the timeout, re-reports
+ * at twice the timeout while the wait continues.
+ */
+static void hazptr_waiter_stall_check(struct hazptr_waiter *w)
+{
+	unsigned long now = jiffies;
+	unsigned long timeout = msecs_to_jiffies(CONFIG_HAZPTR_STALL_TIMEOUT);
+	unsigned long deadline;
+
+	w->cycles++;
+	deadline = w->reports ? w->last_report + 2 * timeout
+			      : w->enqueue_jiffies + timeout;
+	if (time_before(now, deadline))
+		return;
+	w->reports++;
+	hazptr_waiter_stall_report(w);
+	atomic_inc(&hazptr_stall_count);
+	w->last_report = jiffies;
+}
+
+#endif /* CONFIG_HAZPTR_STALL_DETECTOR */
+
 static int hazptr_scan_kthread(void *unused)
 {
 	for (;;) {
@@ -543,7 +768,7 @@ void hazptr_synchronize(void *addr)
 
 	/* Pairs with smp_store_release in hazptr_scan_init(). */
 	if (smp_load_acquire(&hazptr_scan.kthread)) {
-		hazptr_synchronize_queued(addr);
+		hazptr_synchronize_queued(addr, _RET_IP_);
 		return;
 	}
 
@@ -592,3 +817,33 @@ void __init hazptr_init(void)
 		}
 	}
 }
+#ifdef CONFIG_SYSFS
+
+static ssize_t hazptr_stall_count_show(struct kobject *kobj,
+					struct kobj_attribute *attr,
+					char *page)
+{
+	return sysfs_emit(page, "%d\n", atomic_read(&hazptr_stall_count));
+}
+
+static struct kobj_attribute hazptr_stall_count_attr =
+	__ATTR_RO(hazptr_stall_count);
+
+static struct attribute *hazptr_sysfs_attrs[] = {
+	&hazptr_stall_count_attr.attr,
+	NULL,
+};
+
+static struct attribute_group hazptr_sysfs_group = {
+	.name = "hazptr",
+	.attrs = hazptr_sysfs_attrs,
+};
+
+static __init int hazptr_sysfs_init(void)
+{
+	return sysfs_create_group(kernel_kobj, &hazptr_sysfs_group);
+}
+
+late_initcall(hazptr_sysfs_init);
+
+#endif /* CONFIG_SYSFS */
diff --git a/kernel/rcu/Kconfig.debug b/kernel/rcu/Kconfig.debug
index 3a2e9d7f58f8..b64ae06f3298 100644
--- a/kernel/rcu/Kconfig.debug
+++ b/kernel/rcu/Kconfig.debug
@@ -275,4 +275,20 @@ config HAZPTR_ACQUIRE_FORCE_SLOWPATH
 	  with lockdep.  It also lets hazptrtorture exercise the slow
 	  path without filling the per-CPU slots.
 
+config HAZPTR_STALL_DETECTOR
+	bool "Hazard-pointer waiter stall detector"
+	default y
+	help
+	  Print diagnostics when a hazptr_synchronize() waiter remains
+	  blocked for longer than the stall timeout, like RCU's CPU
+	  stall detector.  Purely observational.
+
+config HAZPTR_STALL_TIMEOUT
+	int "Hazard-pointer waiter stall timeout (ms)"
+	range 1 3600000
+	default 21000
+	depends on HAZPTR_STALL_DETECTOR
+	help
+	  Waiter stall timeout in milliseconds, defaulting to RCU's 21s.
+
 endmenu # "RCU Debugging"
-- 
2.43.0

[PATCH 15/28] hazptr: Add reader-assisted drain for stalled waiters

[Kunwu Chan]

When a waiter remains blocked on a sticky address for 100+ scan
cycles, engage a drain: snapshot which per-CPU and overflow-list
slots hold the address, then wait for those holders to release.

Per-CPU slots are tracked via a bitmap mapping (cpu, idx) to the
stable per-CPU slot_item.  Overflow-list slots (struct
hazptr_backup_slot) lack stable identity, so each one is marked
with the current drain generation (drain_gen) at engage; the
reader cooperates on release by checking its own slot's drain_gen,
clearing it, and decrementing backup_outstanding.

The unchain path and the engage both hold the overflow-list lock
during their marker operations, serializing the reader's marker
consumption with the engage's marker writes and preventing the
stranded-marker race.

A cmpxchg() ensures only one of the release hook, the resnapshot
recovery path, or the unchain path consumes each marker.

The outstanding count is maintained incrementally (atomic_inc
during the engage walk, atomic_dec on consumption) so a
concurrent unchain cannot be overwritten by a final atomic_set.

A fairness cap (HAZPTR_DRAIN_MAX_CYCLES) bounds consecutive drain
cycles for a single address: on expiry the drain disengages and
the waiter's re-engage counter resets.

Each disengage logs its reason (complete, fairness-cap, or
orphaned) for test observability.

Wildcards never appear in per-CPU slots: the try_acquire fast
path publishes the actual address directly; only the slow path's
backup slot goes through a wildcard transient.

Suggested-by: Boqun Feng <boqun@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 include/linux/hazptr.h                        |  17 +
 kernel/hazptr.c                               | 468 +++++++++++++++++-
 kernel/rcu/Kconfig.debug                      |  12 +
 .../selftests/rcutorture/bin/hazptr.sh        |  22 +-
 4 files changed, 513 insertions(+), 6 deletions(-)

diff --git a/include/linux/hazptr.h b/include/linux/hazptr.h
index ebf804315a26..018e9309e3ea 100644
--- a/include/linux/hazptr.h
+++ b/include/linux/hazptr.h
@@ -39,12 +39,22 @@ struct hazptr_slot {
 };
 
 struct hazptr_overflow_list;
+struct hazptr_ctx;
+
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+DECLARE_STATIC_KEY_FALSE(hazptr_drain_key);
+void hazptr_drain_release(void *addr, struct hazptr_slot *slot,
+			   struct hazptr_ctx *ctx);
+#endif
 
 struct hazptr_backup_slot {
 	struct hlist_node overflow_node;
 	struct hazptr_slot slot;
 	/* Overflow list where the backup slot is added. */
 	struct hazptr_overflow_list *overflow_list;
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+	unsigned long drain_gen;
+#endif
 };
 
 struct hazptr_ctx {
@@ -72,6 +82,9 @@ struct hazptr_slot_item {
 	unsigned long acquire_caller;
 	unsigned long acquire_jiffies;
 #endif
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+	unsigned long drain_gen;
+#endif
 };
 
 struct hazptr_percpu_slots {
@@ -352,6 +365,10 @@ void hazptr_release(struct hazptr_ctx *ctx, void *addr)
 	hazptr_release_debug(ctx, addr);
 	slot = ctx->slot;
 	smp_store_release(&slot->addr, NULL);
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+	if (static_branch_unlikely(&hazptr_drain_key))
+		hazptr_drain_release(addr, slot, ctx);
+#endif
 	if (unlikely(hazptr_slot_is_backup(ctx, slot)))
 		hazptr_unchain_backup_slot(ctx);
 }
diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index 8de194c07eea..7d4273a44183 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -323,6 +323,9 @@ static bool hazptr_bloom_contains(const struct hazptr_bloom *bloom, void *addr)
  * match.
  *
  * A wildcard in an overflow-list slot means a slow-path acquire is in
+ * flight.  Wildcards never appear in per-CPU slots: the try_acquire
+ * fast path publishes the actual address directly, and only the
+ * slow path's backup slot goes through a wildcard transient.
  * flight: that slot may resolve to any address (including @watch or a
  * Bloom-collected one) after the walk.  To keep the single-pass
  * completion decision safe, the caller must observe wildcards and
@@ -402,6 +405,9 @@ struct hazptr_waiter {
 	int pid;
 	unsigned long caller;
 #endif
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+	unsigned int drain_cycles;
+#endif
 };
 
 struct hazptr_scan_state {
@@ -417,6 +423,346 @@ static struct hazptr_scan_state hazptr_scan;
 
 /* Sysfs counters, cf. RCU's rcu_stall_count. */
 static atomic_t hazptr_stall_count = ATOMIC_INIT(0);
+static atomic_t hazptr_drain_count = ATOMIC_INIT(0);
+
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+
+/*
+ * Drain-assist: internal forward-progress mechanism for sticky addresses.
+ *
+ * When a waiter is blocked for too many scan cycles, the kthread
+ * snapshots which per-CPU slots hold the waited-for address and
+ * tracks their release via a per-slot bitmap.  A remove-only
+ * re-snapshot (periodic re-walk that only clears bits, never sets
+ * new ones) recovers lost notifications.
+ *
+ * Completion requires: bitmap empty (all per-CPU snapshot holders
+ * released) AND all marked backup slots released (tracked via a
+ * per-slot drain_gen marker and a backup_outstanding count).
+ * Per-CPU slots are bitmap-tracked; backup slots are
+ * generation-marked.
+ *
+ * The release path carries a static_branch-guarded check:
+ * near-zero overhead when no drain is active.
+ */
+
+#define HAZPTR_DRAIN_BITS	(NR_CPUS * NR_HAZPTR_PERCPU_SLOTS)
+
+/*
+ * Backup-slot tracking: overflow-list holders lack stable
+ * (cpu, idx) identity.  Instead of storing dangling pointers,
+ * each backup slot carries a drain_gen marker set by engage
+ * and cleared by the release path (a synchronous hook called from
+ * hazptr_release() while the reader's ctx is still valid).
+ * The completion check waits for backup_outstanding to reach
+ * zero, plus the per-CPU bitmap to become empty.
+ */
+
+struct hazptr_drain_state {
+	void *addr;		/* NULL = inactive; written by kthread, read by release hook */
+	DECLARE_BITMAP(map, HAZPTR_DRAIN_BITS);
+	unsigned long generation;
+	atomic_t backup_outstanding;
+	unsigned int cycle_count;	/* cycles since engage, for fairness */
+};
+
+/*
+ * Cap on consecutive drain cycles for a single address.  When reached,
+ * the drain disengages and the current waiter's re-engage counter is
+ * reset, giving other stalled waiters a chance to engage.  This
+ * prevents one continuously-re-published address from starving
+ * waiters on other addresses.
+ */
+#define HAZPTR_DRAIN_MAX_CYCLES	1000
+
+static struct hazptr_drain_state hazptr_drain;
+DEFINE_STATIC_KEY_FALSE(hazptr_drain_key);
+EXPORT_SYMBOL_GPL(hazptr_drain_key);
+
+/*
+ * Engage: record the address, snapshot which per-CPU slots hold it,
+ * and enable the release-path check.  Defers (returns false) if any
+ * slot (per-CPU or overflow-list) holds a wildcard value: an
+ * in-flight acquire could resolve to the target address after the
+ * snapshot, missing the bitmap.
+ */
+/*
+ * Clear backup-slot markers for the given generation, under the
+ * overflow-list locks.  Used by the wildcard-defer and orphaned-
+ * disengage paths, where the drain is being abandoned before or
+ * after formal publication.  Does not decrement backup_outstanding:
+ * the defer path runs before the count is established, and the
+ * orphan path runs after the drain is no longer consulted.
+ */
+static void hazptr_drain_clear_markers(unsigned long gen)
+{
+	int cpu;
+
+	for_each_possible_cpu(cpu) {
+		struct hazptr_overflow_list_flip *overflow_list_flip =
+			per_cpu_ptr(&percpu_overflow_list_flip, cpu);
+
+		for (int i = 0; i < 2; i++) {
+			struct hazptr_overflow_list *list =
+				&overflow_list_flip->array[i];
+			struct hazptr_backup_slot *b;
+			unsigned long flags;
+
+			raw_spin_lock_irqsave(&list->lock, flags);
+			hlist_for_each_entry(b, &list->head, overflow_node) {
+				if (READ_ONCE(b->drain_gen) == gen)
+					WRITE_ONCE(b->drain_gen, 0);
+			}
+			raw_spin_unlock_irqrestore(&list->lock, flags);
+		}
+	}
+}
+static bool hazptr_drain_engage(void *addr)
+{
+	int cpu;
+	unsigned int idx;
+	bool wildcard_present = false;
+
+	bitmap_zero(hazptr_drain.map, HAZPTR_DRAIN_BITS);
+	hazptr_drain.generation++;
+	hazptr_drain.cycle_count = 0;
+	atomic_set(&hazptr_drain.backup_outstanding, 0);
+
+	for_each_possible_cpu(cpu) {
+		struct hazptr_percpu_slots *percpu_slots = per_cpu_ptr(&hazptr_percpu_slots, cpu);
+		struct hazptr_overflow_list_flip *overflow_list_flip = per_cpu_ptr(&percpu_overflow_list_flip, cpu);
+
+		for (idx = 0; idx < NR_HAZPTR_PERCPU_SLOTS; idx++) {
+			struct hazptr_slot_item *item = &percpu_slots->items[idx];
+			void *v = smp_load_acquire(&item->slot.addr);
+
+			if (is_wildcard(v))
+				wildcard_present = true;
+			if (v == addr)
+				set_bit(cpu * NR_HAZPTR_PERCPU_SLOTS + idx,
+					hazptr_drain.map);
+		}
+		for (int i = 0; i < 2; i++) {
+			struct hazptr_overflow_list *list = &overflow_list_flip->array[i];
+			struct hazptr_backup_slot *b;
+			unsigned long flags;
+
+			raw_spin_lock_irqsave(&list->lock, flags);
+			hlist_for_each_entry(b, &list->head, overflow_node) {
+				void *v = smp_load_acquire(&b->slot.addr);
+
+				if (is_wildcard(v)) {
+					wildcard_present = true;
+					continue;
+				}
+				if (v != addr)
+					continue;
+
+				/*
+				 * Mark this backup slot with the current
+				 * drain generation and increment the
+				 * outstanding count immediately.  The
+				 * count is maintained incrementally so
+				 * that a concurrent unchain (which
+				 * consumes the marker and decrements)
+				 * cannot be overwritten by a final
+				 * atomic_set.
+				 */
+				WRITE_ONCE(b->drain_gen,
+					   hazptr_drain.generation);
+				atomic_inc(&hazptr_drain.backup_outstanding);
+			}
+			raw_spin_unlock_irqrestore(&list->lock, flags);
+		}
+	}
+
+	/*
+	 * Defer if any in-flight acquire holds a wildcard.
+	 */
+	if (wildcard_present) {
+		/*
+		 * Clear any markers set before the wildcard was found:
+		 * they belong to an abandoned round.  The count was
+		 * already incremented for these markers; reset it
+		 * because clear_markers does not decrement.
+		 */
+		hazptr_drain_clear_markers(hazptr_drain.generation);
+		bitmap_zero(hazptr_drain.map, HAZPTR_DRAIN_BITS);
+		atomic_set(&hazptr_drain.backup_outstanding, 0);
+		return false;
+	}
+
+	WRITE_ONCE(hazptr_drain.addr, addr);
+	smp_mb(); /* addr publish before static_branch enable */
+
+	static_branch_inc(&hazptr_drain_key);
+	pr_info("hazptr: drain engaged for %p (gen %lu, outstanding %d)\n",
+		addr, hazptr_drain.generation,
+		atomic_read(&hazptr_drain.backup_outstanding));
+	return true;
+}
+
+/*
+ * Release notification: the release path calls this when it clears
+ * a slot that was holding the drain address.
+ *
+ * Context: preemption disabled (hazptr_release always runs under
+ * guard(preempt)).  cpu = this CPU; idx = slot index (0..3).
+ */
+static void hazptr_drain_notify(int cpu, int idx)
+{
+	int bit = cpu * NR_HAZPTR_PERCPU_SLOTS + idx;
+
+	if (bit >= HAZPTR_DRAIN_BITS)
+		return;
+	if (test_and_clear_bit(bit, hazptr_drain.map) &&
+	    bitmap_empty(hazptr_drain.map, HAZPTR_DRAIN_BITS))
+		swake_up_one(&hazptr_scan.wq);
+}
+
+/*
+ * Re-snapshot: remove-only re-walk.  Clear bits for slots that no
+ * longer hold the drain address; never set new bits.  Also recovers
+ * backup-slot releases that occurred during the engage walk (before
+ * the drain address was published): if a backup slot's drain_gen
+ * marker still matches the current generation but the slot no longer
+ * holds the target address, the reader has already released it;
+ * clear the marker and decrement backup_outstanding.
+ * Called periodically while draining.
+ */
+static void hazptr_drain_resnapshot(void)
+{
+	int cpu;
+	unsigned int idx;
+
+	for_each_possible_cpu(cpu) {
+		struct hazptr_percpu_slots *percpu_slots = per_cpu_ptr(&hazptr_percpu_slots, cpu);
+
+		for (idx = 0; idx < NR_HAZPTR_PERCPU_SLOTS; idx++) {
+			int bit = cpu * NR_HAZPTR_PERCPU_SLOTS + idx;
+			struct hazptr_slot_item *item = &percpu_slots->items[idx];
+
+			if (test_bit(bit, hazptr_drain.map) &&
+			    smp_load_acquire(&item->slot.addr) != hazptr_drain.addr)
+				clear_bit(bit, hazptr_drain.map);
+		}
+		for (int i = 0; i < 2; i++) {
+			struct hazptr_overflow_list *list =
+				&per_cpu_ptr(&percpu_overflow_list_flip, cpu)->array[i];
+			struct hazptr_backup_slot *b;
+			unsigned long flags;
+
+			raw_spin_lock_irqsave(&list->lock, flags);
+			hlist_for_each_entry(b, &list->head, overflow_node) {
+				if (READ_ONCE(b->drain_gen) !=
+				    hazptr_drain.generation)
+					continue;
+				/*
+				 * Still holds the drain address: the reader
+				 * has not released yet, so the marker is
+				 * legitimately active.  Skip.
+				 */
+				if (smp_load_acquire(&b->slot.addr) ==
+				    hazptr_drain.addr)
+					continue;
+				/*
+				 * The slot no longer holds the address:
+				 * the reader released during the engage
+				 * walk, before the drain address was
+				 * published.  The release hook could not
+				 * have seen the address, so recover here.
+				 * Atomically claim the marker so that the
+				 * release hook and this recovery path
+				 * cannot both decrement for the same slot.
+				 */
+				if (cmpxchg(&b->drain_gen,
+					    hazptr_drain.generation, 0) ==
+				    hazptr_drain.generation)
+					atomic_dec(&hazptr_drain.backup_outstanding);
+			}
+			raw_spin_unlock_irqrestore(&list->lock, flags);
+		}
+	}
+}
+
+
+/*
+ * Disengage: clear the address, disable the release hook, and
+ * clear any backup-slot markers still marked for this generation.
+ */
+static void hazptr_drain_disengage(const char *reason)
+{
+	unsigned long gen = hazptr_drain.generation;
+
+	WRITE_ONCE(hazptr_drain.addr, NULL);
+	smp_mb(); /* addr-NULL visible before static_branch off */
+	static_branch_dec(&hazptr_drain_key);
+	pr_info("hazptr: drain disengaged (%s, gen %lu, outstanding %d)\n",
+		reason, gen, atomic_read(&hazptr_drain.backup_outstanding));
+	/*
+	 * Clear markers for the abandoned generation.  The static key
+	 * is off, so no new release hook can enter; an in-flight hook
+	 * that already passed the addr check may still consume a
+	 * marker and decrement backup_outstanding, leaving the count
+	 * stale.  The count is only meaningful within a generation's
+	 * lifetime and is reset on the next engage, so no adjustment
+	 * is needed here.
+	 */
+	hazptr_drain_clear_markers(gen);
+}
+
+/*
+ * hazptr_drain_release - release-path hook under static_branch.
+ *
+ * Called from hazptr_release() *after* smp_store_release(&slot->addr, NULL)
+ * has committed.  For per-CPU slots, clears the bitmap bit.  For
+ * backup (overflow-list) slots, cooperates via the slot's drain_gen
+ * marker: the engage set it, and here we clear it and decrement
+ * backup_outstanding.
+ *
+ * The slot type is identified via hazptr_slot_is_backup(ctx, slot).
+ */
+void hazptr_drain_release(void *addr, struct hazptr_slot *slot,
+			   struct hazptr_ctx *ctx)
+{
+	struct hazptr_percpu_slots *base;
+	struct hazptr_slot_item *item;
+	int cpu;
+
+	if (smp_load_acquire(&hazptr_drain.addr) != addr)
+		return;
+
+	cpu = smp_processor_id();
+
+	/* Backup slot: cooperate via drain_gen marker. */
+	if (hazptr_slot_is_backup(ctx, slot)) {
+		/*
+		 * Atomically claim the marker: only one of the release
+		 * hook or the resnapshot recovery path may consume it,
+		 * so only the cmpxchg() winner decrements.
+		 */
+		if (cmpxchg(&ctx->backup_slot.drain_gen,
+			    hazptr_drain.generation, 0) ==
+		    hazptr_drain.generation)
+			atomic_dec(&hazptr_drain.backup_outstanding);
+		return;
+	}
+
+	/*
+	 * Per-CPU slot: derive the bit index from the slot's position.
+	 * Safe because hazptr_release() runs under guard(preempt) and
+	 * the per-CPU slot was acquired on the same CPU, so the slot
+	 * item is in this CPU's items[] array.
+	 */
+	base = per_cpu_ptr(&hazptr_percpu_slots, cpu);
+	item = container_of(slot, struct hazptr_slot_item, slot);
+	if (item >= base->items &&
+	    item < base->items + NR_HAZPTR_PERCPU_SLOTS)
+		hazptr_drain_notify(cpu, item - base->items);
+}
+EXPORT_SYMBOL_GPL(hazptr_drain_release);
+
+#endif /* CONFIG_HAZPTR_DRAIN_ASSIST */
 
 /*
  * Queue @addr for the shared scan.  The waiter lives on the
@@ -438,6 +784,9 @@ static void hazptr_synchronize_queued(void *addr, unsigned long caller)
 	waiter.pid = task_pid_nr(current);
 	waiter.caller = caller;
 #endif
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+	waiter.drain_cycles = 0;
+#endif
 
 	/* Enqueue and wake the scan kthread. */
 	mutex_lock(&hazptr_scan.lock);
@@ -509,6 +858,90 @@ static void hazptr_scan_do_cycle(void)
 	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node)
 		hazptr_waiter_stall_check(w);
 #endif
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+	/*
+	 * Drain-assist: for still-blocked waiters, advance the cycle
+	 * counter and engage or manage a sticky-address drain.
+	 *
+	 * When a waiter has been blocked for 100+ scan cycles, the
+	 * kthread snapshots which slots hold the address (engage)
+	 * and tracks their release via a per-slot bitmap.
+	 *
+	 * When all tracked holders have released (the per-CPU bitmap
+	 * is empty and no backup slots remain marked for the current
+	 * generation), disengage the drain.  The next
+	 * scan cycle's Bloom walk then verifies the address is
+	 * truly absent before completing the waiter, avoiding a race
+	 * with post-engage readers.
+	 */
+	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
+		w->drain_cycles++;
+
+		/* Engage when a waiter has been stuck long enough. */
+		if (!hazptr_drain.addr && w->drain_cycles > 100) {
+			hazptr_drain_engage(w->addr);
+			if (hazptr_drain.addr == w->addr)
+				atomic_inc(&hazptr_drain_count);
+		}
+
+		/* If draining for this address, check progress. */
+		if (hazptr_drain.addr == w->addr) {
+			/* Periodic remove-only re-snapshot. */
+			if (!(w->drain_cycles & 63))
+				hazptr_drain_resnapshot();
+
+			{
+				/*
+				 * All tracked holders have released.
+				 * Disengage and let the next cycle's
+				 * Bloom walk verify the address is
+				 * truly absent before completing the
+				 * waiter.
+				 */
+				if (bitmap_empty(hazptr_drain.map,
+						 HAZPTR_DRAIN_BITS) &&
+				    !atomic_read(&hazptr_drain.backup_outstanding)) {
+					hazptr_drain_disengage("complete");
+					hazptr_drain.cycle_count = 0;
+				}
+			}
+
+			/*
+			 * Fairness: cap consecutive drain cycles for a
+			 * single address.  On expiry, disengage and reset
+			 * this waiter's re-engage counter so the next
+			 * stalled waiter (if any) gets the next turn
+			 * instead of the same address immediately
+			 * re-engaging.
+			 */
+			if (++hazptr_drain.cycle_count >
+			    HAZPTR_DRAIN_MAX_CYCLES) {
+				hazptr_drain_disengage("fairness-cap");
+				hazptr_drain.cycle_count = 0;
+				w->drain_cycles = 0;
+			}
+			break;	/* one drain target at a time */
+		}
+	}
+
+	/*
+	 * Disengage orphaned drain: if the drain address no longer
+	 * has a matching waiter in ->scanning (the Bloom path may
+	 * have completed it this cycle), clean up.
+	 */
+	if (hazptr_drain.addr) {
+		bool orphaned = true;
+
+		list_for_each_entry(w, &hazptr_scan.scanning, node) {
+			if (w->addr == hazptr_drain.addr) {
+				orphaned = false;
+				break;
+			}
+		}
+		if (orphaned)
+			hazptr_drain_disengage("orphaned");
+	}
+#endif
 }
 
 
@@ -684,7 +1117,6 @@ static void hazptr_waiter_stall_check(struct hazptr_waiter *w)
 	unsigned long now = jiffies;
 	unsigned long timeout = msecs_to_jiffies(CONFIG_HAZPTR_STALL_TIMEOUT);
 	unsigned long deadline;
-
 	w->cycles++;
 	deadline = w->reports ? w->last_report + 2 * timeout
 			      : w->enqueue_jiffies + timeout;
@@ -799,6 +1231,30 @@ void hazptr_unchain_backup_slot(struct hazptr_ctx *ctx)
 	struct hazptr_overflow_list *overflow_list = ctx->backup_slot.overflow_list;
 
 	guard(raw_spinlock_irqsave)(&overflow_list->lock);
+#ifdef CONFIG_HAZPTR_DRAIN_ASSIST
+	unsigned long gen;
+
+	/*
+	 * Consume any pending drain marker under the list lock,
+	 * before removing the slot from the overflow list.  The
+	 * engage sets markers while holding this same lock, so
+	 * checking under the lock guarantees that if the engage
+	 * marks this slot, we see the marker before the slot is
+	 * unchained.  Without the lock, the unchain could check
+	 * drain_gen before the engage sets it, then block on the
+	 * lock while the engage sets the marker and increments
+	 * the count, and finally remove the slot without ever
+	 * consuming the orphaned marker.
+	 *
+	 * The cmpxchg ensures only one of the release hook, the
+	 * resnapshot, or this path decrements backup_outstanding.
+	 */
+	gen = READ_ONCE(ctx->backup_slot.drain_gen);
+	if (gen) {
+		if (cmpxchg(&ctx->backup_slot.drain_gen, gen, 0) == gen)
+			atomic_dec(&hazptr_drain.backup_outstanding);
+	}
+#endif
 	overflow_list->gen++;
 	hlist_del(&ctx->backup_slot.overflow_node);
 }
@@ -825,12 +1281,22 @@ static ssize_t hazptr_stall_count_show(struct kobject *kobj,
 {
 	return sysfs_emit(page, "%d\n", atomic_read(&hazptr_stall_count));
 }
+static ssize_t hazptr_drain_count_show(struct kobject *kobj,
+					struct kobj_attribute *attr,
+					char *page)
+{
+	return sysfs_emit(page, "%d\n", atomic_read(&hazptr_drain_count));
+}
+
+static struct kobj_attribute hazptr_drain_count_attr =
+	__ATTR_RO(hazptr_drain_count);
 
 static struct kobj_attribute hazptr_stall_count_attr =
 	__ATTR_RO(hazptr_stall_count);
 
 static struct attribute *hazptr_sysfs_attrs[] = {
 	&hazptr_stall_count_attr.attr,
+	&hazptr_drain_count_attr.attr,
 	NULL,
 };
 
diff --git a/kernel/rcu/Kconfig.debug b/kernel/rcu/Kconfig.debug
index b64ae06f3298..b500e96ea1c1 100644
--- a/kernel/rcu/Kconfig.debug
+++ b/kernel/rcu/Kconfig.debug
@@ -291,4 +291,16 @@ config HAZPTR_STALL_TIMEOUT
 	help
 	  Waiter stall timeout in milliseconds, defaulting to RCU's 21s.
 
+config HAZPTR_DRAIN_ASSIST
+	bool "Hazard-pointer sticky-address drain assist"
+	default y
+	help
+	  Internal forward-progress mechanism for hazptr_synchronize()
+	  waiters whose target address is continuously re-published
+	  by new readers.  When a waiter has been blocked for many
+	  scan cycles, the scan kthread snapshots which slots hold
+	  the address and tracks their release via per-slot tracking.
+
+	  If unsure, say Y.
+
 endmenu # "RCU Debugging"
diff --git a/tools/testing/selftests/rcutorture/bin/hazptr.sh b/tools/testing/selftests/rcutorture/bin/hazptr.sh
index 4e30ea51d9fc..d12735ddfefc 100755
--- a/tools/testing/selftests/rcutorture/bin/hazptr.sh
+++ b/tools/testing/selftests/rcutorture/bin/hazptr.sh
@@ -229,8 +229,19 @@ then
 		if ! grep -q "hazptr: waiter stall" "$console"
 		then
 			echo "HOTADDR note: no stall splats observed (record as not-reproduced, investigate interleaving)" > "$resdir/kvm.sh.note"
+		elif ! grep -q "drain engaged" "$console"
+		then
+			echo "HOTADDR note: drain-assist did not engage (Bloom completion may have handled it)" > "$resdir/kvm.sh.note"
+		elif ! grep -q "drain disengaged (complete" "$console"
+		then
+			if grep -q "drain disengaged (fairness-cap" "$console"
+			then
+				echo "HOTADDR: drain hit fairness cap without completing" > "$resdir/kvm.sh.err"
+			else
+				echo "HOTADDR note: drain engaged but did not disengage normally" > "$resdir/kvm.sh.note"
+			fi
 		else
-			echo "HOTADDR: starvation reproduced, stall detector confirmed, drain verified"
+			echo "HOTADDR: starvation reproduced, stall detector confirmed, drain engaged"
 		fi
 	else
 		nerrs=$((nerrs+1))
@@ -239,10 +250,11 @@ else
 	nerrs=$((nerrs+1))
 fi
 
-# LONGHOLD: legitimate long-lived references.  The stall detector must
-# fire (observation), the holder must release, and the waiter must drain
-# normally afterwards: diagnostics without interruption, module verdict
-# stays SUCCESS.
+# LONGHOLD: legitimate long-lived references.  The holder must release,
+# and the waiter must drain normally afterwards.  The stall detector
+# may or may not fire (the 8-second hold may not exceed the 21-second
+# timeout depending on when the waiter started relative to the hold);
+# this is observation data only, not a correctness requirement.
 
 echo "--- hazptr LONGHOLD (legitimate long-held reference) ---"
 if run_kvm longhold LONGHOLD "$cpus" "" "60s"
-- 
2.43.0

[PATCH 16/28] Documentation/litmus-tests: Add hazptr drain engage-wildcard test

[Kunwu Chan]

Model the drain engage's wildcard defer: if the engage reads a
slot holding a wildcard (an in-flight slow-path acquire), the
engage must not complete for the target address.  The smp_mb()
pairing between the reader's wildcard publication and the
engage's slot read forbids the completion interleaving
(herd7: Never).

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 Documentation/litmus-tests/README             |  5 +++
 .../hazptr-drain-engage-wildcard.litmus       | 44 +++++++++++++++++++
 2 files changed, 49 insertions(+)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-engage-wildcard.litmus

diff --git a/Documentation/litmus-tests/README b/Documentation/litmus-tests/README
index c2bf20fa4092..daab126e45a2 100644
--- a/Documentation/litmus-tests/README
+++ b/Documentation/litmus-tests/README
@@ -110,3 +110,8 @@ hazptr-wildcard-flip-escape.litmus
     Test the in-flight wildcard window of the single-pass wildcard-flip
     scan: a straggling acquire that published the old wildcard after
     the final walk cannot still load the unpublished object.
+
+hazptr-drain-engage-wildcard.litmus
+    Test the drain engage's wildcard defer: an engage that reads a
+    slot holding a wildcard (an in-flight slow-path acquire) must
+    not complete for the target address.
diff --git a/Documentation/litmus-tests/hazptr/hazptr-drain-engage-wildcard.litmus b/Documentation/litmus-tests/hazptr/hazptr-drain-engage-wildcard.litmus
new file mode 100644
index 000000000000..0cee89893dad
--- /dev/null
+++ b/Documentation/litmus-tests/hazptr/hazptr-drain-engage-wildcard.litmus
@@ -0,0 +1,44 @@
+C hazptr-drain-engage-wildcard
+
+(*
+ * Result: Never
+ *
+ * Models the engage's wildcard defer: if the slot holds a
+ * wildcard (transient value 1) at the time the engage reads
+ * it, the engage must not complete for the target address.
+ *
+ * P0 (reclaimer): unpublishes head, then reads the slot.
+ * P1 (reader): publishes wildcard (1), then replaces with
+ *              the target address (2).
+ *
+ * The smp_mb() pair ensures the engage cannot observe the
+ * target (2) while the reader still sees the pre-unpublish
+ * head.  This verifies the publication ordering that the
+ * engage relies on to defer when it sees a wildcard.
+ *)
+
+{
+	head = 1;
+	slot = 0;
+}
+
+P0(int *head, int *slot)
+{
+	int r0;
+
+	WRITE_ONCE(*head, 0);
+	smp_mb();
+	r0 = READ_ONCE(*slot);
+}
+
+P1(int *slot, int *head)
+{
+	int r1;
+
+	WRITE_ONCE(*slot, 1);
+	smp_mb();
+	WRITE_ONCE(*slot, 2);
+	r1 = READ_ONCE(*head);
+}
+
+exists (0:r0=0 /\ 1:r1=1)
-- 
2.43.0

[PATCH 17/28] Documentation/litmus-tests: Add hazptr drain resnapshot-slot-race test

[Kunwu Chan]

Model the marker consumption race between the resnapshot
recovery path and a reader release: only one path may consume
the drain_gen marker (the cmpxchg in the implementation).  The
Sometimes result confirms the race window exists
(herd7: Sometimes).

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 Documentation/litmus-tests/README             |  4 +++
 .../hazptr-drain-resnapshot-slot-race.litmus  | 36 +++++++++++++++++++
 2 files changed, 40 insertions(+)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-resnapshot-slot-race.litmus

diff --git a/Documentation/litmus-tests/README b/Documentation/litmus-tests/README
index daab126e45a2..02c26d6bea50 100644
--- a/Documentation/litmus-tests/README
+++ b/Documentation/litmus-tests/README
@@ -115,3 +115,7 @@ hazptr-drain-engage-wildcard.litmus
     Test the drain engage's wildcard defer: an engage that reads a
     slot holding a wildcard (an in-flight slow-path acquire) must
     not complete for the target address.
+
+hazptr-drain-resnapshot-slot-race.litmus
+    Test the marker consumption race between the resnapshot recovery
+    path and a reader release: only one path may consume the marker.
diff --git a/Documentation/litmus-tests/hazptr/hazptr-drain-resnapshot-slot-race.litmus b/Documentation/litmus-tests/hazptr/hazptr-drain-resnapshot-slot-race.litmus
new file mode 100644
index 000000000000..699cbf0e3d7a
--- /dev/null
+++ b/Documentation/litmus-tests/hazptr/hazptr-drain-resnapshot-slot-race.litmus
@@ -0,0 +1,36 @@
+C hazptr-drain-resnapshot-slot-race
+
+(*
+ * Result: Sometimes
+ *
+ * Models the marker consumption race between the resnapshot
+ * recovery path (P0) and a reader release (P1).  The shared
+ * slot_gen represents drain_gen; only one path may consume
+ * the marker (cmpxchg in the implementation).
+ *
+ * The Sometimes result confirms the race window exists; the
+ * cmpxchg ensures only one of the release hook, resnapshot,
+ * or unchain path decrements backup_outstanding.
+ *)
+
+{
+	slot_addr = 3;
+	slot_gen = 1;
+}
+
+P0(int *slot_addr, int *slot_gen)
+{
+	int r0;
+
+	r0 = READ_ONCE(*slot_addr);
+	if (r0 != 3)
+		WRITE_ONCE(*slot_gen, 0);
+}
+
+P1(int *slot_addr, int *slot_gen)
+{
+	WRITE_ONCE(*slot_addr, 0);
+	smp_store_release(slot_gen, 0);
+}
+
+exists (0:r0=3 /\ slot_gen=0)
-- 
2.43.0

[PATCH 18/28] Documentation/litmus-tests: Add hazptr drain safety test

[Kunwu Chan]

Model the drain engage snapshot race: the engage may read a
slot as empty or wildcard while an in-flight slow-path acquire
still sees the pre-unpublish head.  The Sometimes result
confirms the window exists; the engage's wildcard check (see
the engage-wildcard test) closes the gap by deferring
(herd7: Sometimes).

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 Documentation/litmus-tests/README             |  4 ++
 .../hazptr/hazptr-drain-safety.litmus         | 43 +++++++++++++++++++
 2 files changed, 47 insertions(+)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-safety.litmus

diff --git a/Documentation/litmus-tests/README b/Documentation/litmus-tests/README
index 02c26d6bea50..604c2c0025d9 100644
--- a/Documentation/litmus-tests/README
+++ b/Documentation/litmus-tests/README
@@ -119,3 +119,7 @@ hazptr-drain-engage-wildcard.litmus
 hazptr-drain-resnapshot-slot-race.litmus
     Test the marker consumption race between the resnapshot recovery
     path and a reader release: only one path may consume the marker.
+
+hazptr-drain-safety.litmus
+    Test the engage snapshot race: the engage can miss an in-flight
+    slow-path acquire, so completion must defer on wildcards.
diff --git a/Documentation/litmus-tests/hazptr/hazptr-drain-safety.litmus b/Documentation/litmus-tests/hazptr/hazptr-drain-safety.litmus
new file mode 100644
index 000000000000..a0ec1a063642
--- /dev/null
+++ b/Documentation/litmus-tests/hazptr/hazptr-drain-safety.litmus
@@ -0,0 +1,43 @@
+C hazptr-drain-safety
+
+(*
+ * Result: Sometimes
+ *
+ * Models the drain engage snapshot race: the engage (P0)
+ * unpublishes the object (head=0) and reads the slot, while
+ * a slow-path reader (P1) publishes a wildcard (1), then
+ * replaces it with the target address (2).
+ *
+ * The engage may read 0 (empty), 1 (wildcard), or 2 (target).
+ * Reading 0 or 1 while the reader sees head=1 means the
+ * snapshot missed an in-flight acquire — the Sometimes result
+ * confirms this window exists.  The engage's wildcard check
+ * (hazptr-drain-engage-wildcard) closes the gap by deferring
+ * when it reads the wildcard value.
+ *)
+
+{
+	head = 1;
+	slot = 0;
+}
+
+P0(int *head, int *slot)
+{
+	int r0;
+
+	WRITE_ONCE(*head, 0);
+	smp_mb();
+	r0 = READ_ONCE(*slot);
+}
+
+P1(int *slot, int *head)
+{
+	int r1;
+
+	WRITE_ONCE(*slot, 1);
+	smp_mb();
+	WRITE_ONCE(*slot, 2);
+	r1 = READ_ONCE(*head);
+}
+
+exists ((0:r0=0 \/ 0:r0=1) /\ 1:r1=1)
-- 
2.43.0

[PATCH 19/28] Documentation/litmus-tests: Add hazptr drain generation-visibility test

[Kunwu Chan]

Prove that the drain engage's smp_mb() publication ordering is
correct: a release hook that observes the new drain address also
observes the new generation counter (herd7: Never).

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 Documentation/litmus-tests/README             |  4 ++
 .../hazptr-drain-generation-visibility.litmus | 41 +++++++++++++++++++
 2 files changed, 45 insertions(+)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-drain-generation-visibility.litmus

diff --git a/Documentation/litmus-tests/README b/Documentation/litmus-tests/README
index 604c2c0025d9..79b0634ddb13 100644
--- a/Documentation/litmus-tests/README
+++ b/Documentation/litmus-tests/README
@@ -123,3 +123,7 @@ hazptr-drain-resnapshot-slot-race.litmus
 hazptr-drain-safety.litmus
     Test the engage snapshot race: the engage can miss an in-flight
     slow-path acquire, so completion must defer on wildcards.
+
+hazptr-drain-generation-visibility.litmus
+    Test the drain generation ordering: a release hook that observes
+    the new drain address also observes the new generation.
diff --git a/Documentation/litmus-tests/hazptr/hazptr-drain-generation-visibility.litmus b/Documentation/litmus-tests/hazptr/hazptr-drain-generation-visibility.litmus
new file mode 100644
index 000000000000..d5c1ca247d71
--- /dev/null
+++ b/Documentation/litmus-tests/hazptr/hazptr-drain-generation-visibility.litmus
@@ -0,0 +1,41 @@
+C hazptr-drain-generation-visibility
+
+(*
+ * Result: Never
+ *
+ * Drain generation visibility: the engage publishes the
+ * generation counter before the drain address (separated by
+ * smp_mb()).  The release hook reads the address with
+ * smp_load_acquire().  If it observes the new address, the
+ * subsequent generation read must also observe the new value,
+ * so the cmpxchg compares against the correct round.
+ *
+ * If this were not the case, a release could read a stale
+ * generation (from the previous drain round) while the address
+ * already belongs to the new round, causing the cmpxchg to
+ * succeed against the wrong round's marker and falsely
+ * decrement the new round's backup_outstanding.
+ *)
+
+{
+	gen = 5;
+	addr = 0;
+}
+
+P0(int *gen, int *addr)
+{
+	WRITE_ONCE(*gen, 6);
+	smp_mb();
+	WRITE_ONCE(*addr, 1);
+}
+
+P1(int *addr, int *gen)
+{
+	int r0;
+	int r1;
+
+	r0 = smp_load_acquire(addr);
+	r1 = READ_ONCE(*gen);
+}
+
+exists (1:r0=1 /\ 1:r1=5)
\ No newline at end of file
-- 
2.43.0

[PATCH 20/28] locking/lockdep: Use hazptr to wait for dynamic key lookups

[Kunwu Chan]

Replace the system-wide synchronize_rcu_expedited() in
lockdep_unregister_key() with a hazptr_synchronize() scoped to
the per-key hash bucket.

Suggested-by: Boqun Feng <boqun@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/locking/lockdep.c | 25 +++++++++++++++----------
 1 file changed, 15 insertions(+), 10 deletions(-)

diff --git a/kernel/locking/lockdep.c b/kernel/locking/lockdep.c
index c3dc84a7cef2..256f23a9041d 100644
--- a/kernel/locking/lockdep.c
+++ b/kernel/locking/lockdep.c
@@ -58,6 +58,7 @@
 #include <linux/context_tracking.h>
 #include <linux/console.h>
 #include <linux/kasan.h>
+#include <linux/hazptr.h>
 
 #include <asm/sections.h>
 
@@ -1308,14 +1309,19 @@ static bool is_dynamic_key(const struct lock_class_key *key)
 
 	hash_head = keyhashentry(key);
 
-	rcu_read_lock();
-	hlist_for_each_entry_rcu(k, hash_head, hash_entry) {
-		if (k == key) {
-			found = true;
-			break;
+	{
+		struct hazptr_ctx ctx;
+		void *bucket = hash_head;
+
+		scoped_guard(hazptr, &ctx, &bucket) {
+			hlist_for_each_entry_rcu(k, hash_head, hash_entry, 1) {
+				if (k == key) {
+					found = true;
+					break;
+				}
+			}
 		}
 	}
-	rcu_read_unlock();
 
 	return found;
 }
@@ -6717,11 +6723,10 @@ void lockdep_unregister_key(struct lock_class_key *key)
 	 *
 	 * Some operations like __qdisc_destroy() will call this in a debug
 	 * kernel, and the network traffic is disabled while waiting, hence
-	 * the delay of the wait matters in debugging cases. Currently use a
-	 * synchronize_rcu_expedited() to speed up the wait at the cost of
-	 * system IPIs. TODO: Replace RCU with hazptr for this.
+	 * the delay of the wait matters in debugging cases. Replace the
+	 * expedited RCU wait with hazptr_synchronize().
 	 */
-	synchronize_rcu_expedited();
+	hazptr_synchronize(keyhashentry(key));
 }
 EXPORT_SYMBOL_GPL(lockdep_unregister_key);
 
-- 
2.43.0

[PATCH 21/28] hazptrtorture: Add HOTADDR and LONGHOLD scenarios

[Kunwu Chan]

HOTADDR: stages the sticky-address forward-progress scenario.
LONGHOLD: validates legitimate long-held references.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/rcu/hazptrtorture.c                    | 384 +++++++++++++++++-
 .../rcutorture/configs/hazptr/CFLIST          |   2 +
 .../rcutorture/configs/hazptr/HOTADDR         |  18 +
 .../rcutorture/configs/hazptr/HOTADDR.boot    |   1 +
 .../rcutorture/configs/hazptr/LONGHOLD        |  18 +
 .../rcutorture/configs/hazptr/LONGHOLD.boot   |   1 +
 6 files changed, 419 insertions(+), 5 deletions(-)
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR.boot
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD.boot

diff --git a/kernel/rcu/hazptrtorture.c b/kernel/rcu/hazptrtorture.c
index fe09280d063c..5dbe04af7515 100644
--- a/kernel/rcu/hazptrtorture.c
+++ b/kernel/rcu/hazptrtorture.c
@@ -24,6 +24,7 @@
 #include <linux/hazptr.h>
 #include <linux/rcupdate.h>
 #include <linux/workqueue.h>
+#include <trace/events/hazptr.h>
 
 #include "rcu.h"
 
@@ -54,6 +55,15 @@ torture_param(int, stutter, 5, "Number of seconds to run/halt test");
 torture_param(int, verbose, 1, "Enable verbose debugging printk()s");
 torture_param(bool, wq_churn, false,
 	      "Churn workqueues to exercise lockdep dynamic keys");
+torture_param(int, nhotreaders, 0,
+	      "Hot-addr hammer readers, 0 disables Lai-scenario");
+torture_param(int, hot_hold_us, 20000,
+	      "Hammer reader hold (us)");
+torture_param(int, hot_hammer_ms, 3000, "Hammer window (ms)");
+torture_param(int, hot_longhold_ms, 0,
+	      "Long-hold round duration (ms), 0 disables; R0 holds past the stall timeout, no hammer");
+torture_param(bool, starve_expected, false,
+	      "Starvation episodes are expected (documenting mode)");
 
 static char *torture_type = "hazptr";
 module_param(torture_type, charp, 0444);
@@ -66,6 +76,22 @@ static struct task_struct **reader_tasks;
 static struct task_struct *do_pending_task;
 static struct task_struct *stats_task;
 static struct task_struct *wq_churn_task;
+static struct task_struct *hot_stage_task;
+static struct task_struct *hot_r0_task;
+static struct task_struct **hot_reader_tasks;
+static int hot_target_storage;
+static void *hot_addr = &hot_target_storage;
+static atomic_t n_hot_syncs;
+static atomic_t n_hot_starves;
+static atomic_t n_hot_in_seq;
+static atomic_t hot_live_syncs;		/* transient sync kthreads alive */
+static atomic_t hot_hammer_holding;	/* hammer readers currently holding */
+static atomic_t hot_r0_gate;		/* 1: R0 may acquire for this round */
+static DECLARE_COMPLETION(hot_enqueued);
+static DECLARE_COMPLETION(hot_sync_done);
+static DECLARE_COMPLETION(r0_held);
+static DECLARE_COMPLETION(r0_release);
+static bool hot_hammer_active;
 
 #define HAZPTR_TORTURE_PIPE_LEN 10
 
@@ -87,13 +113,13 @@ static atomic_t n_hazptr_torture_alloc;
 static atomic_t n_hazptr_torture_alloc_fail;
 static atomic_t n_hazptr_torture_free;
 static atomic_t n_hazptr_torture_error;
-static atomic_t n_hazptr_torture_wq_churn;
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_acquires);
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_releases);
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_acquires_irq);
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_releases_irq);
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_releases_defer);
 static DEFINE_PER_CPU(atomic_long_t, hazptr_torture_releases_undefer);
+static atomic_t n_hazptr_torture_wq_churn;
 static struct list_head hazptr_torture_removed;
 
 // State for a deferred (AKA pending) hazard pointer
@@ -681,6 +707,10 @@ hazptr_torture_stats_print(void)
 		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_defer),
 		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_undefer),
 		atomic_read(&n_hazptr_torture_wq_churn));
+	pr_cont(" hotsync: %d hotstarve: %d hotseq: %d",
+		atomic_read(&n_hot_syncs),
+		atomic_read(&n_hot_starves),
+		atomic_read(&n_hot_in_seq));
 
 	pr_alert("%s%s ", torture_type, TORTURE_FLAG);
 	if (i > 1) {
@@ -752,7 +782,8 @@ hazptr_torture_print_module_parms(struct hazptr_torture_ops *cur_ops, const char
 		 "nest_acquire=%d "
 		 "onoff_interval=%d onoff_holdoff=%d "
 		 "preempt_duration=%d preempt_interval=%d "
-		 "reader_sleep_us=%d "
+		 "reader_sleep_us=%d wq_churn=%d "
+		 "nhotreaders=%d hot_hold_us=%d hot_hammer_ms=%d "
 		 "shuffle_interval=%d shutdown_secs=%d stat_interval=%d stutter=%d "
 		 "verbose=%d\n",
 		 torture_type, tag, nrealreaders, nwriters,
@@ -760,7 +791,8 @@ hazptr_torture_print_module_parms(struct hazptr_torture_ops *cur_ops, const char
 		 nest_acquire,
 		 onoff_interval, onoff_holdoff,
 		 preempt_duration, preempt_interval,
-		 reader_sleep_us,
+		 reader_sleep_us, wq_churn,
+		 nhotreaders, hot_hold_us, hot_hammer_ms,
 		 shuffle_interval, shutdown_secs, stat_interval, stutter,
 		 verbose);
 }
@@ -801,10 +833,14 @@ static void hazptr_torture_wq_churn_fn(struct work_struct *work)
 /*
  * Repeatedly create and destroy workqueues.  Each workqueue creation
  * registers a lockdep dynamic key and each destruction unregisters it,
- * which exercises both is_dynamic_key() and lockdep_unregister_key().
+ * which exercises both is_dynamic_key() and lockdep_unregister_key(),
+ * the two lockdep paths that use hazptr.  Most useful in kernels with
+ * CONFIG_PROVE_LOCKING=y and CONFIG_HAZPTR_ACQUIRE_FORCE_SLOWPATH=y.
  */
 static int hazptr_torture_wq_churn(void *unused)
 {
+	DEFINE_TORTURE_RANDOM(rand);
+
 	VERBOSE_TOROUT_STRING("hazptr_torture_wq_churn task started");
 	do {
 		struct workqueue_struct *wq;
@@ -812,7 +848,7 @@ static int hazptr_torture_wq_churn(void *unused)
 
 		wq = alloc_workqueue("hazptr-torture-churn", WQ_PERCPU, 1);
 		if (!wq) {
-			torture_hrtimeout_ms(10, 10, NULL);
+			torture_hrtimeout_ms(10, 10, &rand);
 			continue;
 		}
 		INIT_WORK(&churn_work, hazptr_torture_wq_churn_fn);
@@ -826,6 +862,271 @@ static int hazptr_torture_wq_churn(void *unused)
 	return 0;
 }
 
+/*
+ * Hot-addr staged scenario --- Lai's forward-progress test.
+ *
+ * Three persistent kthreads plus a per-round transient sync kthread:
+ *   hot_r0_task      old reader: acquires hot_addr, signals, waits
+ *                    for release command, then releases.
+ *   hot_reader_tasks hammer readers: tight acquire/release loops on
+ *                    hot_addr, long hold, near-zero gap, no stutter,
+ *                    kthread_bind'd to distinct CPUs.
+ *   hot_stage_task   orchestrator: drives the staged sequence round
+ *                    by round, recording starvation episodes.
+ *   hot_do_sync      transient per-round: hazptr_synchronize(hot_addr).
+ *
+ * A tracepoint on waiter-enqueue confirms the waiter is on
+ * hazptr_scan.pending before the orchestrator verifies the negative
+ * control (sync must NOT have completed while R0 holds).
+ */
+
+static void hot_probe_enqueue(void *__data, void *addr)
+{
+	if (addr == hot_addr)
+		complete(&hot_enqueued);
+}
+
+/* Per-round transient sync kthread.  hot_live_syncs lets the
+ * orchestrator and cleanup wait for exit, so no completion event
+ * from one round can leak into the next.
+ */
+static int hazptr_torture_hot_do_sync(void *unused)
+{
+	hazptr_synchronize(hot_addr);
+	complete(&hot_sync_done);
+	atomic_dec(&hot_live_syncs);
+	return 0;
+}
+
+/* R0: acquires hot_addr, signals orchestrator, waits for release cmd.
+ * The release wait is timeout-bounded so kthread_stop() at cleanup can
+ * always join; the hazard pointer is always released before exit.
+ * The one-shot gate ensures R0 does not re-acquire immediately after
+ * releasing: re-acquisition is reserved for a new round, so a draining
+ * waiter is never blocked by R0 itself.
+ */
+static int hazptr_torture_hot_r0(void *unused)
+{
+	struct hazptr_ctx ctx;
+	void *p;
+
+	VERBOSE_TOROUT_STRING("hot_r0 task started");
+	do {
+		while (!atomic_read(&hot_r0_gate) && !torture_must_stop())
+			schedule_timeout_interruptible(1);
+		if (torture_must_stop())
+			break;
+		atomic_set(&hot_r0_gate, 0);
+		p = hot_addr;
+		(void)hazptr_acquire(&ctx, &p);
+		complete(&r0_held);
+		while (!wait_for_completion_timeout(&r0_release, HZ)) {
+			if (torture_must_stop())
+				break;
+		}
+		hazptr_release(&ctx, p);
+	} while (!torture_must_stop());
+	torture_kthread_stopping("hot_r0");
+	return 0;
+}
+
+/* Hammer reader: parks while not hammering (hot_hammer_active false),
+ * so it persists across rounds.  hot_hammer_holding counts live holds
+ * and lets the orchestrator hand off R0 only once a hammer reader
+ * actually holds the address -- no coverage gap at the seam.
+ */
+static int hazptr_torture_hot_reader(void *arg)
+{
+	int cpu = (long)arg;
+	struct hazptr_ctx ctx;
+	void *p;
+	DEFINE_TORTURE_RANDOM(rand);
+
+	set_cpus_allowed_ptr(current, cpumask_of(cpu));
+	for (;;) {
+		while (!READ_ONCE(hot_hammer_active) && !torture_must_stop())
+			schedule_timeout_interruptible(1);
+		if (torture_must_stop())
+			break;
+		p = hot_addr;
+		p = hazptr_acquire(&ctx, &p);
+		atomic_inc(&hot_hammer_holding);
+		torture_hrtimeout_us(hot_hold_us, 100, &rand);
+		hazptr_release(&ctx, p);
+		atomic_dec(&hot_hammer_holding);
+		torture_hrtimeout_us(1, 1, &rand);
+	}
+	torture_kthread_stopping("hot_reader");
+	return 0;
+}
+
+/*
+ * Orchestrator: one round per loop iteration.  All waits are
+ * timeout-bounded so cleanup can always join this task.
+ *
+ * Phase   Action
+ *  boot   Wait for R0's initial hold.
+ *   P2    Launch transient sync kthread; wait for tracepoint
+ *         confirmation (hot_enqueued).
+ *   P3    Negative control (non-consuming completion_done()): sync
+ *         must NOT be done while R0 still holds.
+ *   P4    Handoff: start hammer, wait until a hammer reader holds
+ *         (hot_hammer_holding > 0), then release R0 -- no gap.
+ *   P5    Hammer window: poll sync for hot_hammer_ms.
+ *         -> completes in window: n_hot_in_seq++.
+ *         -> still pending:       n_hot_starves++ (episode; counts
+ *            as test error only when !starve_expected).
+ *   P6    Positive control: stop hammer, sync must drain within 5 s
+ *         (skipped when P5 already observed completion).
+ *  bottom  Drain this round's sync kthread (hot_live_syncs), wait for
+ *          R0's next hold; jitter; loop unless must_stop.
+ */
+static int hazptr_torture_hot_stage(void *unused)
+{
+	unsigned long round = 0;
+	DEFINE_TORTURE_RANDOM(rand);
+
+	VERBOSE_TOROUT_STRING("hot_stage task started");
+
+	/* If the system is still booting, let it finish (same as the
+	 * classic writer): a busy boot would make staged rounds record
+	 * false starvation episodes and false stall splats. */
+	while (!torture_must_stop() && !rcu_inkernel_boot_has_ended())
+		schedule_timeout_interruptible(HZ);
+
+	do {
+		u64 t_sync_start;
+		int done_in_window = 0, positive_done;
+		struct task_struct *sync_task;
+
+		reinit_completion(&r0_held);
+		reinit_completion(&r0_release);
+		reinit_completion(&hot_sync_done);
+		reinit_completion(&hot_enqueued);
+		WRITE_ONCE(hot_hammer_active, false);
+
+		/* Gate R0 into a fresh hold and wait for it. */
+		atomic_set(&hot_r0_gate, 1);
+		while (!wait_for_completion_timeout(&r0_held, HZ)) {
+			if (torture_must_stop())
+				goto out;
+		}
+
+		/* P2: launch sync; confirm waiter enqueued. */
+		t_sync_start = ktime_get_ns();
+		atomic_inc(&hot_live_syncs);
+		sync_task = kthread_run(hazptr_torture_hot_do_sync, NULL,
+					"hot_do_sync");
+		if (IS_ERR(sync_task)) {
+			atomic_dec(&hot_live_syncs);
+			pr_alert("%shot: sync kthread alloc failed r%lu\n",
+				 TORTURE_FLAG, round);
+			goto skip_round;
+		}
+		if (!wait_for_completion_timeout(&hot_enqueued, HZ)) {
+			pr_alert("%shot: enqueue not confirmed r%lu\n",
+				 TORTURE_FLAG, round);
+			goto skip_round;
+		}
+
+		/* P3 NEGATIVE CONTROL: sync must still be pending. */
+		if (completion_done(&hot_sync_done)) {
+			pr_alert("%shot: completed before R0 released r%lu\n",
+				 TORTURE_FLAG, round);
+			atomic_inc(&n_hazptr_torture_error);
+		}
+
+		/*
+		 * LONGHOLD variant: no hammer.  R0 legitimately holds
+		 * past the stall timeout; the detector should report
+		 * (observation, not interruption), and after release the
+		 * sync must drain normally.  Validates that long-lived
+		 * references -- an hazptr feature -- do not break the
+		 * machinery and are not treated as errors.
+		 */
+		if (hot_longhold_ms) {
+			torture_hrtimeout_ms(hot_longhold_ms, 100, &rand);
+			complete(&r0_release);
+			done_in_window = wait_for_completion_timeout(
+				&hot_sync_done, 5 * HZ);
+			if (!done_in_window) {
+				pr_alert("%shot: longhold: sync did not drain after release r%lu\n",
+					 TORTURE_FLAG, round);
+				atomic_inc(&n_hazptr_torture_error);
+			}
+			atomic_inc(done_in_window ? &n_hot_in_seq
+						  : &n_hot_starves);
+			positive_done = done_in_window;
+			{
+				u64 dt = ktime_get_ns() - t_sync_start;
+
+				atomic_inc(&n_hot_syncs);
+				pr_alert("%shot r%lu: longhold drain=%d dt_ns=%llu\n",
+					 TORTURE_FLAG, round,
+					 !!done_in_window, dt);
+			}
+			goto skip_round;
+		}
+
+		/* P4: handoff -- hammer holds before R0 releases. */
+		WRITE_ONCE(hot_hammer_active, true);
+		while (!atomic_read(&hot_hammer_holding) &&
+		       !torture_must_stop())
+			schedule_timeout_interruptible(1);
+		complete(&r0_release);
+
+		/* P5 HAMMER WINDOW. */
+		done_in_window = wait_for_completion_timeout(
+			&hot_sync_done, msecs_to_jiffies(hot_hammer_ms));
+		if (done_in_window) {
+			atomic_inc(&n_hot_in_seq);
+		} else {
+			atomic_inc(&n_hot_starves);
+			if (!starve_expected)
+				atomic_inc(&n_hazptr_torture_error);
+		}
+
+		/* P6 POSITIVE CONTROL. */
+		WRITE_ONCE(hot_hammer_active, false);
+		if (done_in_window) {
+			positive_done = 1;
+		} else {
+			positive_done = wait_for_completion_timeout(
+				&hot_sync_done, 5 * HZ);
+			if (!positive_done) {
+				pr_alert("%shot: positive control timed out r%lu (machinery fault?)\n",
+					 TORTURE_FLAG, round);
+				atomic_inc(&n_hazptr_torture_error);
+			}
+		}
+
+		{
+			u64 dt = ktime_get_ns() - t_sync_start;
+
+			atomic_inc(&n_hot_syncs);
+			pr_alert("%shot r%lu: starve=%d in_seq=%d pos=%d dt_ns=%llu\n",
+				 TORTURE_FLAG, round, !done_in_window,
+				 !!done_in_window, positive_done, dt);
+		}
+
+	skip_round:
+		/* Release R0 (harmless if already released) so an
+		 * orphan sync from a skipped round can drain. */
+		complete(&r0_release);
+		/* Drain this round's sync kthread: no completion event
+		 * may cross the round boundary. */
+		while (atomic_read(&hot_live_syncs) && !torture_must_stop())
+			schedule_timeout_interruptible(1);
+		/* R0 parks on the gate after its release; the next round
+		 * opens the gate and waits for its fresh hold above. */
+		round++;
+		torture_hrtimeout_ms(500, 200, &rand);
+	} while (!torture_must_stop());
+out:
+	torture_kthread_stopping("hot_stage");
+	return 0;
+}
+
 static void
 hazptr_torture_cleanup(void)
 {
@@ -840,6 +1141,34 @@ hazptr_torture_cleanup(void)
 
 	torture_stop_kthread(hazptr_torture_do_pending, do_pending_task);
 	torture_stop_kthread(hazptr_torture_preempt, preempt_task);
+	torture_stop_kthread(hazptr_torture_wq_churn, wq_churn_task);
+
+	/* Hot scenario: stop hammer readers first so pending syncs drain. */
+	if (nhotreaders > 0 || hot_longhold_ms > 0) {
+		int wait_ms = 5000;
+
+		WRITE_ONCE(hot_hammer_active, false);
+		if (hot_reader_tasks) {
+			for (i = 0; i < nhotreaders; i++)
+				torture_stop_kthread(hazptr_torture_hot_reader,
+						     hot_reader_tasks[i]);
+			kfree(hot_reader_tasks);
+			hot_reader_tasks = NULL;
+		}
+		torture_stop_kthread(hazptr_torture_hot_r0, hot_r0_task);
+		torture_stop_kthread(hazptr_torture_hot_stage, hot_stage_task);
+		/* Wait for any transient sync kthreads to exit so no
+		 * module code is executing at unload. */
+		while (atomic_read(&hot_live_syncs) && wait_ms > 0) {
+			schedule_timeout_uninterruptible(msecs_to_jiffies(50));
+			wait_ms -= 50;
+		}
+		if (atomic_read(&hot_live_syncs))
+			pr_alert("%shot: %d sync kthread(s) still alive at exit\n",
+				 TORTURE_FLAG, atomic_read(&hot_live_syncs));
+		unregister_trace_hazptr_waiter_enqueue(hot_probe_enqueue, NULL);
+	}
+
 	torture_stop_kthread(hazptr_torture_writer, writer_task);
 
 	if (reader_tasks) {
@@ -873,6 +1202,8 @@ static int __init hazptr_torture_init(void)
 	int firsterr = 0;
 	static struct hazptr_torture_ops *torture_ops[] = { &hazptr_ops, &hazptr_stack_ops, };
 
+	if (!strcmp(torture_type, "none"))
+		return 0;
 	if (!torture_init_begin(torture_type, verbose))
 		return -EBUSY;
 
@@ -919,6 +1250,10 @@ static int __init hazptr_torture_init(void)
 	atomic_set(&n_hazptr_torture_free, 0);
 	atomic_set(&n_hazptr_torture_error, 0);
 	atomic_set(&n_hazptr_torture_wq_churn, 0);
+	atomic_set(&n_hot_syncs, 0);
+	atomic_set(&n_hot_starves, 0);
+	atomic_set(&n_hot_in_seq, 0);
+	atomic_set(&hot_r0_gate, 0);
 	for (i = 0; i < HAZPTR_TORTURE_PIPE_LEN + 1; i++)
 		atomic_set(&hazptr_torture_wcount[i], 0);
 	for_each_possible_cpu(cpu) {
@@ -1004,6 +1339,45 @@ static int __init hazptr_torture_init(void)
 		if (torture_init_error(firsterr))
 			goto unwind;
 	}
+	if (nhotreaders > 0 || hot_longhold_ms > 0) {
+		int cpu = 0, j;
+
+		firsterr = register_trace_hazptr_waiter_enqueue(hot_probe_enqueue, NULL);
+		if (firsterr) {
+			pr_alert("%shot: tracepoint registration failed: %d\n",
+				 TORTURE_FLAG, firsterr);
+			goto unwind;
+		}
+
+		if (nhotreaders > 0) {
+			hot_reader_tasks = kcalloc(nhotreaders,
+						  sizeof(hot_reader_tasks[0]),
+						  GFP_KERNEL);
+			if (!hot_reader_tasks) {
+				firsterr = -ENOMEM;
+				goto unwind;
+			}
+			for (j = 0; j < nhotreaders; j++) {
+				cpu = cpumask_next(cpu % num_online_cpus(),
+						   cpu_online_mask);
+				if (cpu >= nr_cpu_ids)
+					cpu = cpumask_first(cpu_online_mask);
+				firsterr = torture_create_kthread(
+					hazptr_torture_hot_reader,
+					(void *)(long)cpu, hot_reader_tasks[j]);
+				if (torture_init_error(firsterr))
+					goto unwind;
+			}
+		}
+		firsterr = torture_create_kthread(hazptr_torture_hot_r0, NULL,
+						  hot_r0_task);
+		if (torture_init_error(firsterr))
+			goto unwind;
+		firsterr = torture_create_kthread(hazptr_torture_hot_stage,
+						  NULL, hot_stage_task);
+		if (torture_init_error(firsterr))
+			goto unwind;
+	}
 
 	torture_init_end();
 	return 0;
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST b/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
index 90ef09b0e654..b3f6137ce8ab 100644
--- a/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
@@ -6,3 +6,5 @@ SLOWPATH
 LOCKDEP
 CPU128
 CPU256
+HOTADDR
+LONGHOLD
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR b/tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR
new file mode 100644
index 000000000000..6f4d6746b63c
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR
@@ -0,0 +1,18 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=16
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_LOCK_ALLOC=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
+CONFIG_HAZPTR_STALL_DETECTOR=y
+CONFIG_HAZPTR_STALL_TIMEOUT=1000
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR.boot b/tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR.boot
new file mode 100644
index 000000000000..878b588e2e39
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/HOTADDR.boot
@@ -0,0 +1 @@
+hazptrtorture.nreaders=0 hazptrtorture.nwriters=0 hazptrtorture.nhotreaders=4 hazptrtorture.hot_hold_us=20000 hazptrtorture.starve_expected=1 hazptrtorture.shutdown_secs=20
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD b/tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD
new file mode 100644
index 000000000000..fe293638bbe2
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD
@@ -0,0 +1,18 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=16
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_LOCK_ALLOC=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
+CONFIG_HAZPTR_STALL_DETECTOR=y
+CONFIG_HAZPTR_STALL_TIMEOUT=2000
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD.boot b/tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD.boot
new file mode 100644
index 000000000000..cbe30b6a09cc
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/LONGHOLD.boot
@@ -0,0 +1 @@
+hazptrtorture.nreaders=0 hazptrtorture.nwriters=0 hazptrtorture.hot_longhold_ms=8000 hazptrtorture.shutdown_secs=40
-- 
2.43.0

[PATCH 22/28] hazptrtorture: Add drain fairness rotation scenario

[Kunwu Chan]

Two stuck addresses, each held by a dedicated holder for the
whole observation window (hot_fair_ms), with two sync waiters
launched back to back.  The first waiter's drain engages and,
because its holder never releases, must hit the fairness cap;
the second waiter's counter kept growing, so it gets the next
drain turn.  Rotation evidence comes from the kernel drain logs:
distinct addresses in "drain engaged" lines plus at least one
"drain disengaged (fairness-cap" line.

The module verdict only requires forward progress: after both
holders release, both syncs must drain within 30 s.  Rotation
observation is reported via console-log checks in hazptr.sh,
graded as notes (not errors) when the window is too short for
the configured HZ.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/rcu/hazptrtorture.c                    | 211 +++++++++++++++++-
 .../selftests/rcutorture/bin/hazptr.sh        |  51 +++++
 .../rcutorture/configs/hazptr/CFLIST          |   1 +
 .../rcutorture/configs/hazptr/FAIRNESS        |  19 ++
 .../rcutorture/configs/hazptr/FAIRNESS.boot   |   1 +
 5 files changed, 281 insertions(+), 2 deletions(-)
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS
 create mode 100644 tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS.boot

diff --git a/kernel/rcu/hazptrtorture.c b/kernel/rcu/hazptrtorture.c
index 5dbe04af7515..787862a4d4eb 100644
--- a/kernel/rcu/hazptrtorture.c
+++ b/kernel/rcu/hazptrtorture.c
@@ -64,6 +64,8 @@ torture_param(int, hot_longhold_ms, 0,
 	      "Long-hold round duration (ms), 0 disables; R0 holds past the stall timeout, no hammer");
 torture_param(bool, starve_expected, false,
 	      "Starvation episodes are expected (documenting mode)");
+torture_param(int, hot_fair_ms, 0,
+	      "Drain fairness rotation window (ms), 0 disables");
 
 static char *torture_type = "hazptr";
 module_param(torture_type, charp, 0444);
@@ -93,6 +95,24 @@ static DECLARE_COMPLETION(r0_held);
 static DECLARE_COMPLETION(r0_release);
 static bool hot_hammer_active;
 
+/* Fairness rotation scenario: two stuck addresses whose holders
+ * never release during the observation window.  Two sync waiters
+ * are launched; the drain fairness cap must rotate the drain
+ * between the two addresses instead of monopolizing one.
+ */
+static int fair_target_a;
+static int fair_target_b;
+static void *const fair_addr[2] = { &fair_target_a, &fair_target_b };
+static struct task_struct *fair_holder_tasks[2];
+static struct task_struct *fair_stage_task;
+static struct completion fair_held[2];
+static struct completion fair_release[2];
+static struct completion fair_released[2];
+static struct completion fair_sync_done[2];
+static atomic_t fair_live_syncs;
+static atomic_t n_fair_rounds;
+static atomic_t n_fair_timeout;
+
 #define HAZPTR_TORTURE_PIPE_LEN 10
 
 // Update-side data structure used to check RCU readers.
@@ -707,10 +727,13 @@ hazptr_torture_stats_print(void)
 		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_defer),
 		torture_sum_pcpu_atomic_long(&hazptr_torture_releases_undefer),
 		atomic_read(&n_hazptr_torture_wq_churn));
-	pr_cont(" hotsync: %d hotstarve: %d hotseq: %d",
+	pr_cont(" hotsync: %d hotstarve: %d hotseq: %d"
+		" fairrounds: %d fairtimeout: %d",
 		atomic_read(&n_hot_syncs),
 		atomic_read(&n_hot_starves),
-		atomic_read(&n_hot_in_seq));
+		atomic_read(&n_hot_in_seq),
+		atomic_read(&n_fair_rounds),
+		atomic_read(&n_fair_timeout));
 
 	pr_alert("%s%s ", torture_type, TORTURE_FLAG);
 	if (i > 1) {
@@ -1127,6 +1150,152 @@ static int hazptr_torture_hot_stage(void *unused)
 	return 0;
 }
 
+/*
+ * Fairness rotation scenario --- two stuck addresses, two
+ * perpetual holders, two sync waiters.
+ *
+ *   fair_holder_tasks[2]  each acquires its own address and holds
+ *                         it until commanded to release; the hold
+ *                         spans the whole observation window.
+ *   fair_stage_task       orchestrator: waits for both holds,
+ *                         launches both sync waiters (address A
+ *                         first so it is at the head of the
+ *                         scanning list), waits out the window,
+ *                         then releases both holders and requires
+ *                         both syncs to drain.
+ *
+ * The window must be long enough for the drain to engage (~100
+ * scan cycles) and hit the fairness cap (HAZPTR_DRAIN_MAX_CYCLES,
+ * default 1000) on address A, after which address B's waiter --
+ * whose re-engage counter kept growing -- gets the next drain
+ * turn.  Rotation evidence is read from the kernel drain logs
+ * ("drain engaged" for both addresses, at least one "drain
+ * disengaged (fairness-cap").
+ *
+ * The module verdict (n_hazptr_torture_error) does not depend on
+ * rotation: it fails only if a sync fails to drain after both
+ * holders released (a forward-progress regression).
+ */
+static int hazptr_torture_fair_holder(void *arg)
+{
+	long i = (long)arg;
+	struct hazptr_ctx ctx;
+	void *p;
+
+	VERBOSE_TOROUT_STRING("fair_holder task started");
+	p = fair_addr[i];
+	(void)hazptr_acquire(&ctx, &p);
+	complete(&fair_held[i]);
+	while (!wait_for_completion_timeout(&fair_release[i], HZ)) {
+		if (torture_must_stop())
+			break;
+	}
+	hazptr_release(&ctx, p);
+	complete(&fair_released[i]);
+	torture_kthread_stopping("fair_holder");
+	return 0;
+}
+
+/* Transient per-round sync kthread.  fair_live_syncs is
+ * incremented by the stage before kthread_run() and decremented
+ * here at exit, so the count covers the full lifetime including
+ * the window between thread creation and first execution.
+ */
+static int hazptr_torture_fair_do_sync(void *arg)
+{
+	long i = (long)arg;
+
+	hazptr_synchronize(fair_addr[i]);
+	complete(&fair_sync_done[i]);
+	atomic_dec(&fair_live_syncs);
+	return 0;
+}
+
+static int hazptr_torture_fair_stage(void *unused)
+{
+	int i;
+
+	VERBOSE_TOROUT_STRING("fair_stage task started");
+
+	while (!torture_must_stop() && !rcu_inkernel_boot_has_ended())
+		schedule_timeout_interruptible(HZ);
+	{
+		bool sync_launched[2] = { false, false };
+
+	/* Wait for both holders to acquire their addresses. */
+	for (i = 0; i < 2; i++) {
+		while (!wait_for_completion_timeout(&fair_held[i], HZ)) {
+			if (torture_must_stop())
+				goto out;
+		}
+	}
+
+	/* Launch both sync waiters; A first so it leads the list. */
+	for (i = 0; i < 2; i++) {
+		struct task_struct *t;
+
+		atomic_inc(&fair_live_syncs);
+		t = kthread_run(hazptr_torture_fair_do_sync, (void *)(long)i,
+				"fair_do_sync");
+		if (IS_ERR(t)) {
+			atomic_dec(&fair_live_syncs);
+			pr_alert("%sfair: sync kthread alloc failed (%d)\n",
+				 TORTURE_FLAG, i);
+			atomic_inc(&n_hazptr_torture_error);
+			goto release;
+		}
+		sync_launched[i] = true;
+	}
+
+	/* Observation window: drains rotate via the fairness cap. */
+	{
+		unsigned long end = jiffies + msecs_to_jiffies(hot_fair_ms);
+
+		while (time_before(jiffies, end) && !torture_must_stop())
+			schedule_timeout_interruptible(HZ);
+	}
+
+	/* Negative observation: holders still hold, so the syncs
+	 * must not have completed during the window.
+	 */
+	for (i = 0; i < 2; i++) {
+		if (completion_done(&fair_sync_done[i])) {
+			pr_alert("%sfair: sync %d completed while holder holds\n",
+				 TORTURE_FLAG, i);
+			atomic_inc(&n_hazptr_torture_error);
+		}
+	}
+
+release:
+	/* Release both holders; the syncs must now drain. */
+	for (i = 0; i < 2; i++)
+		complete(&fair_release[i]);
+	for (i = 0; i < 2; i++) {
+		unsigned long deadline = jiffies + 30 * HZ;
+
+		if (!sync_launched[i])
+			continue;
+		while (!wait_for_completion_timeout(&fair_sync_done[i], HZ)) {
+			if (torture_must_stop() ||
+			    time_after(jiffies, deadline)) {
+				pr_alert("%sfair: sync %d did not drain after release\n",
+					 TORTURE_FLAG, i);
+				atomic_inc(&n_hazptr_torture_error);
+				atomic_inc(&n_fair_timeout);
+				break;
+			}
+		}
+	}
+	atomic_inc(&n_fair_rounds);
+	}
+out:
+	/* Unblock holders even on skip paths. */
+	for (i = 0; i < 2; i++)
+		complete(&fair_release[i]);
+	torture_kthread_stopping("fair_stage");
+	return 0;
+}
+
 static void
 hazptr_torture_cleanup(void)
 {
@@ -1169,6 +1338,24 @@ hazptr_torture_cleanup(void)
 		unregister_trace_hazptr_waiter_enqueue(hot_probe_enqueue, NULL);
 	}
 
+	if (hot_fair_ms > 0) {
+		int wait_ms = 5000;
+
+		for (i = 0; i < 2; i++)
+			complete(&fair_release[i]);
+		for (i = 0; i < 2; i++)
+			torture_stop_kthread(hazptr_torture_fair_holder,
+					     fair_holder_tasks[i]);
+		torture_stop_kthread(hazptr_torture_fair_stage, fair_stage_task);
+		while (atomic_read(&fair_live_syncs) && wait_ms > 0) {
+			schedule_timeout_uninterruptible(msecs_to_jiffies(50));
+			wait_ms -= 50;
+		}
+		if (atomic_read(&fair_live_syncs))
+			pr_alert("%sfair: %d sync kthread(s) still alive at exit\n",
+				 TORTURE_FLAG, atomic_read(&fair_live_syncs));
+	}
+
 	torture_stop_kthread(hazptr_torture_writer, writer_task);
 
 	if (reader_tasks) {
@@ -1378,6 +1565,26 @@ static int __init hazptr_torture_init(void)
 		if (torture_init_error(firsterr))
 			goto unwind;
 	}
+	if (hot_fair_ms > 0) {
+		atomic_set(&fair_live_syncs, 0);
+		atomic_set(&n_fair_rounds, 0);
+		atomic_set(&n_fair_timeout, 0);
+		for (i = 0; i < 2; i++) {
+			init_completion(&fair_held[i]);
+			init_completion(&fair_release[i]);
+			init_completion(&fair_released[i]);
+			init_completion(&fair_sync_done[i]);
+			firsterr = torture_create_kthread(
+				hazptr_torture_fair_holder,
+				(void *)(long)i, fair_holder_tasks[i]);
+			if (torture_init_error(firsterr))
+				goto unwind;
+		}
+		firsterr = torture_create_kthread(hazptr_torture_fair_stage,
+						  NULL, fair_stage_task);
+		if (torture_init_error(firsterr))
+			goto unwind;
+	}
 
 	torture_init_end();
 	return 0;
diff --git a/tools/testing/selftests/rcutorture/bin/hazptr.sh b/tools/testing/selftests/rcutorture/bin/hazptr.sh
index d12735ddfefc..8109cb76bf04 100755
--- a/tools/testing/selftests/rcutorture/bin/hazptr.sh
+++ b/tools/testing/selftests/rcutorture/bin/hazptr.sh
@@ -297,6 +297,57 @@ else
 	nerrs=$((nerrs+1))
 fi
 
+# FAIRNESS: drain rotation between two stuck addresses.  The
+# module verdict covers forward progress (syncs drain after
+# release); rotation itself is observed from the kernel drain
+# logs.
+
+echo "--- hazptr FAIRNESS (drain rotation) ---"
+if run_kvm fairness FAIRNESS "$cpus" "" "50s"
+then
+	resdir="$RCUTORTURE/res/$ds/fairness/FAIRNESS"
+	console="$resdir/console.log"
+	fair_err=""
+	if ! test -f "$console"
+	then
+		echo "Missing console.log for FAIRNESS" > "$RCUTORTURE/res/$ds/fairness/kvm.sh.err"
+		fair_err=1
+	elif ! grep -q "End of test: SUCCESS" "$console"
+	then
+		echo "FAIRNESS: module verdict not SUCCESS" > "$resdir/kvm.sh.err"
+		fair_err=1
+	elif ! grep -q "fair_stage task started" "$console"
+	then
+		echo "FAIRNESS: scenario did not run" > "$resdir/kvm.sh.err"
+		fair_err=1
+	elif grep -q "fair: sync .* did not drain" "$console"
+	then
+		echo "FAIRNESS: sync failed to drain after release" > "$resdir/kvm.sh.err"
+		fair_err=1
+	fi
+	if [ -z "$fair_err" ]
+	then
+		engages=$(grep -o "drain engaged for [0-9a-f]*" "$console" | sort -u | wc -l)
+		caps=$(grep -c "drain disengaged (fairness-cap" "$console")
+		if test "$engages" -ge 2 -a "$caps" -ge 1
+		then
+			echo "FAIRNESS: drain rotated between $engages addresses, $caps fairness-cap disengage(s)"
+		elif test "$caps" -ge 1
+		then
+			echo "FAIRNESS note: fairness-cap hit but only $engages address engaged (window may be short)" > "$resdir/kvm.sh.note"
+		else
+			echo "FAIRNESS note: no fairness-cap observed (window too short for HZ; consider larger hot_fair_ms)" > "$resdir/kvm.sh.note"
+		fi
+	fi
+	if [ -n "$fair_err" ]
+	then
+		nerrs=$((nerrs+1))
+	fi
+else
+	nerrs=$((nerrs+1))
+fi
+echo
+
 # CPU scaling: optionally test at multiple CPU counts
 
 if test "$allcpus" -ne 0
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST b/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
index b3f6137ce8ab..03ce4940ab4a 100644
--- a/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/CFLIST
@@ -8,3 +8,4 @@ CPU128
 CPU256
 HOTADDR
 LONGHOLD
+FAIRNESS
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS b/tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS
new file mode 100644
index 000000000000..b81b57307036
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS
@@ -0,0 +1,19 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=16
+CONFIG_PREEMPT_NONE=n
+CONFIG_PREEMPT_VOLUNTARY=n
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_NO_HZ_FULL=n
+CONFIG_HOTPLUG_CPU=y
+CONFIG_SUSPEND=n
+CONFIG_HIBERNATION=n
+CONFIG_PROVE_LOCKING=n
+CONFIG_DEBUG_LOCK_ALLOC=n
+CONFIG_DEBUG_OBJECTS_RCU_HEAD=n
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_DEBUG=y
+CONFIG_HAZPTR_STALL_DETECTOR=y
+CONFIG_HAZPTR_STALL_TIMEOUT=60000
+CONFIG_HAZPTR_DRAIN_ASSIST=y
diff --git a/tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS.boot b/tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS.boot
new file mode 100644
index 000000000000..ef54cbfc5b50
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/hazptr/FAIRNESS.boot
@@ -0,0 +1 @@
+hazptrtorture.nreaders=0 hazptrtorture.nwriters=0 hazptrtorture.hot_fair_ms=30000 hazptrtorture.shutdown_secs=45
-- 
2.43.0

[PATCH 23/28] rcuscale: Add hazptr scale type

[Kunwu Chan]

Add a hazptr scale type for per-writer hazptr_synchronize()
cost and per-reader hazptr_acquire()/hazptr_release() overhead.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/rcu/rcuscale.c | 67 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 67 insertions(+)

diff --git a/kernel/rcu/rcuscale.c b/kernel/rcu/rcuscale.c
index 5659eaebc5c0..6e3e3f296f85 100644
--- a/kernel/rcu/rcuscale.c
+++ b/kernel/rcu/rcuscale.c
@@ -40,6 +40,7 @@
 #include <linux/vmalloc.h>
 #include <linux/rcupdate_trace.h>
 #include <linux/sched/debug.h>
+#include <linux/hazptr.h>
 
 #include "rcu.h"
 
@@ -460,6 +461,71 @@ static struct rcu_scale_ops tasks_tracing_ops = {
 
 #endif // #else // #ifdef CONFIG_TASKS_TRACE_RCU
 
+#if IS_ENABLED(CONFIG_HAZPTR_TORTURE_TEST)
+
+#define HAZPTR_SCALE_NR_OBJS 8
+static int hazptr_scale_obj[HAZPTR_SCALE_NR_OBJS];
+static atomic_t hazptr_scale_sync_idx = ATOMIC_INIT(0);
+
+struct hazptr_scale_state {
+	struct hazptr_ctx ctx;
+	void *addr;
+};
+static DEFINE_PER_CPU(struct hazptr_scale_state, hazptr_scale_state);
+
+static int hazptr_scale_read_lock(void)
+{
+	struct hazptr_scale_state *state = this_cpu_ptr(&hazptr_scale_state);
+	int idx = smp_processor_id() % HAZPTR_SCALE_NR_OBJS;
+	void *ptr = &hazptr_scale_obj[idx];
+
+	state->addr = hazptr_acquire(&state->ctx, &ptr);
+	return 0;
+}
+
+static void hazptr_scale_read_unlock(int idx)
+{
+	struct hazptr_scale_state *state = this_cpu_ptr(&hazptr_scale_state);
+
+	hazptr_release(&state->ctx, state->addr);
+}
+
+static unsigned long hazptr_scale_completed(void)
+{
+	return 0;
+}
+
+static void hazptr_scale_sync(void)
+{
+	int idx = atomic_fetch_inc(&hazptr_scale_sync_idx) % HAZPTR_SCALE_NR_OBJS;
+
+	hazptr_synchronize(&hazptr_scale_obj[idx]);
+}
+
+static void hazptr_scale_sync_exp(void)
+{
+	int idx = atomic_fetch_inc(&hazptr_scale_sync_idx) % HAZPTR_SCALE_NR_OBJS;
+
+	hazptr_synchronize(&hazptr_scale_obj[idx]);
+}
+
+static struct rcu_scale_ops hazptr_scale_ops = {
+	.ptype		= 0,
+	.readlock	= hazptr_scale_read_lock,
+	.readunlock	= hazptr_scale_read_unlock,
+	.get_gp_seq	= hazptr_scale_completed,
+	.gp_diff	= NULL,
+	.exp_completed	= hazptr_scale_completed,
+	.sync		= hazptr_scale_sync,
+	.exp_sync	= hazptr_scale_sync_exp,
+	.name		= "hazptr",
+};
+
+#define HAZPTR_SCALE_OPS &hazptr_scale_ops,
+#else
+#define HAZPTR_SCALE_OPS
+#endif
+
 static unsigned long rcuscale_seq_diff(unsigned long new, unsigned long old)
 {
 	if (!cur_ops->gp_diff)
@@ -1140,6 +1206,7 @@ rcu_scale_init(void)
 	static struct rcu_scale_ops *scale_ops[] = {
 		&rcu_ops, &srcu_ops, &srcua_ops, &srcud_ops,
 		TASKS_OPS TASKS_RUDE_OPS TASKS_TRACING_OPS
+		HAZPTR_SCALE_OPS
 	};
 
 	if (!torture_init_begin(scale_type, verbose))
-- 
2.43.0

[PATCH 24/28] hazptr: Add runtime sysfs controls for stall detector

[Kunwu Chan]

Add writable sysfs attributes that let the administrator adjust
stall-detector behaviour at runtime.

Suggested-by: Paul E. McKenney <paulmck@kernel.org>
Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/hazptr.c | 62 ++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 61 insertions(+), 1 deletion(-)

diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index 7d4273a44183..66954251c332 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -423,6 +423,10 @@ static struct hazptr_scan_state hazptr_scan;
 
 /* Sysfs counters, cf. RCU's rcu_stall_count. */
 static atomic_t hazptr_stall_count = ATOMIC_INIT(0);
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+static unsigned long hazptr_stall_timeout_ms = CONFIG_HAZPTR_STALL_TIMEOUT;
+static int hazptr_stall_suppress;
+#endif
 static atomic_t hazptr_drain_count = ATOMIC_INIT(0);
 
 #ifdef CONFIG_HAZPTR_DRAIN_ASSIST
@@ -1115,8 +1119,12 @@ static void hazptr_waiter_stall_report(struct hazptr_waiter *w)
 static void hazptr_waiter_stall_check(struct hazptr_waiter *w)
 {
 	unsigned long now = jiffies;
-	unsigned long timeout = msecs_to_jiffies(CONFIG_HAZPTR_STALL_TIMEOUT);
+	unsigned long timeout_ms = READ_ONCE(hazptr_stall_timeout_ms);
+	unsigned long timeout;
 	unsigned long deadline;
+	if (READ_ONCE(hazptr_stall_suppress) || !timeout_ms)
+		return;
+	timeout = msecs_to_jiffies(timeout_ms);
 	w->cycles++;
 	deadline = w->reports ? w->last_report + 2 * timeout
 			      : w->enqueue_jiffies + timeout;
@@ -1275,6 +1283,46 @@ void __init hazptr_init(void)
 }
 #ifdef CONFIG_SYSFS
 
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+static ssize_t hazptr_stall_timeout_show(struct kobject *kobj,
+					  struct kobj_attribute *attr,
+					  char *page)
+{
+	return sysfs_emit(page, "%lu\n", READ_ONCE(hazptr_stall_timeout_ms));
+}
+static ssize_t hazptr_stall_timeout_store(struct kobject *kobj,
+					   struct kobj_attribute *attr,
+					   const char *buf, size_t count)
+{
+	unsigned long val;
+	int ret;
+	ret = kstrtoul(buf, 0, &val);
+	if (ret)
+		return ret;
+	if (val > 3600000)
+		return -EINVAL;
+	WRITE_ONCE(hazptr_stall_timeout_ms, val);
+	return count;
+}
+static ssize_t hazptr_stall_suppress_show(struct kobject *kobj,
+					   struct kobj_attribute *attr,
+					   char *page)
+{
+	return sysfs_emit(page, "%d\n", READ_ONCE(hazptr_stall_suppress));
+}
+static ssize_t hazptr_stall_suppress_store(struct kobject *kobj,
+					    struct kobj_attribute *attr,
+					    const char *buf, size_t count)
+{
+	int val, ret;
+	ret = kstrtoint(buf, 0, &val);
+	if (ret)
+		return ret;
+	WRITE_ONCE(hazptr_stall_suppress, !!val);
+	return count;
+}
+#endif /* CONFIG_HAZPTR_STALL_DETECTOR */
+
 static ssize_t hazptr_stall_count_show(struct kobject *kobj,
 					struct kobj_attribute *attr,
 					char *page)
@@ -1291,12 +1339,24 @@ static ssize_t hazptr_drain_count_show(struct kobject *kobj,
 static struct kobj_attribute hazptr_drain_count_attr =
 	__ATTR_RO(hazptr_drain_count);
 
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+static struct kobj_attribute hazptr_stall_timeout_attr =
+	__ATTR(hazptr_stall_timeout_ms, 0644,
+	       hazptr_stall_timeout_show, hazptr_stall_timeout_store);
+static struct kobj_attribute hazptr_stall_suppress_attr =
+	__ATTR(hazptr_stall_suppress, 0644,
+	       hazptr_stall_suppress_show, hazptr_stall_suppress_store);
+#endif
 static struct kobj_attribute hazptr_stall_count_attr =
 	__ATTR_RO(hazptr_stall_count);
 
 static struct attribute *hazptr_sysfs_attrs[] = {
 	&hazptr_stall_count_attr.attr,
 	&hazptr_drain_count_attr.attr,
+#ifdef CONFIG_HAZPTR_STALL_DETECTOR
+	&hazptr_stall_timeout_attr.attr,
+	&hazptr_stall_suppress_attr.attr,
+#endif
 	NULL,
 };
 
-- 
2.43.0

[PATCH 25/28] hazptr: Select TRACING for HAZPTR_TORTURE_TEST to enable tracepoint

[Kunwu Chan]

The hazptr torture test registers the hazptr_waiter_enqueue
tracepoint for its hot-address scenario; select TRACING so
that the register_trace_*() API is available.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/rcu/Kconfig.debug | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/kernel/rcu/Kconfig.debug b/kernel/rcu/Kconfig.debug
index b500e96ea1c1..8fc6434a05d1 100644
--- a/kernel/rcu/Kconfig.debug
+++ b/kernel/rcu/Kconfig.debug
@@ -27,6 +27,7 @@ config RCU_SCALE_TEST
 	tristate "performance tests for RCU"
 	depends on DEBUG_KERNEL
 	select TORTURE_TEST
+	select TRACING
 	default n
 	help
 	  This option provides a kernel module that runs performance
@@ -42,6 +43,7 @@ config RCU_TORTURE_TEST
 	tristate "torture tests for RCU"
 	depends on DEBUG_KERNEL
 	select TORTURE_TEST
+	select TRACING
 	default n
 	help
 	  This option provides a kernel module that runs torture tests
@@ -102,6 +104,7 @@ config RCU_REF_SCALE_TEST
 	tristate "Scalability tests for read-side synchronization (RCU and others)"
 	depends on DEBUG_KERNEL
 	select TORTURE_TEST
+	select TRACING
 	default n
 	help
 	  This option provides a kernel module that runs performance tests
@@ -117,6 +120,7 @@ config HAZPTR_TORTURE_TEST
 	tristate "Torture tests for hazard pointers"
 	depends on DEBUG_KERNEL
 	select TORTURE_TEST
+	select TRACING
 	default n
 	help
 	  This option provides in-kernel hazard-pointer stress tests.
-- 
2.43.0

[PATCH 26/28] hazptrtorture: Add torture_type=none clean skip for rcuscale coexistence

[Kunwu Chan]

When both HAZPTR_TORTURE_TEST and RCU_SCALE_TEST are built into
the same kernel, hazptrtorture calls torture_init_begin() first
and rcuscale fails with -EBUSY.  torture_type=none makes
hazptrtorture return early without registering, allowing the
rcuscale hazptr scale type to run in the same boot.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 .../selftests/rcutorture/configs/rcuscale/HAZPTR    | 13 +++++++++++++
 .../rcutorture/configs/rcuscale/HAZPTR.boot         |  1 +
 2 files changed, 14 insertions(+)
 create mode 100644 tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR
 create mode 100644 tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR.boot

diff --git a/tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR b/tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR
new file mode 100644
index 000000000000..c5e9cf0184ae
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR
@@ -0,0 +1,13 @@
+CONFIG_SMP=y
+CONFIG_NR_CPUS=128
+CONFIG_PREEMPT=y
+CONFIG_HZ_PERIODIC=n
+CONFIG_NO_HZ_IDLE=y
+CONFIG_HOTPLUG_CPU=y
+CONFIG_DEBUG_KERNEL=y
+CONFIG_HAZPTR_TORTURE_TEST=y
+CONFIG_HAZPTR_DEBUG=y
+CONFIG_HAZPTR_STALL_DETECTOR=y
+CONFIG_HAZPTR_STALL_TIMEOUT=1000
+CONFIG_HAZPTR_DRAIN_ASSIST=y
+CONFIG_RCU_SCALE_TEST=y
diff --git a/tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR.boot b/tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR.boot
new file mode 100644
index 000000000000..e9cca4cf453b
--- /dev/null
+++ b/tools/testing/selftests/rcutorture/configs/rcuscale/HAZPTR.boot
@@ -0,0 +1 @@
+hazptrtorture.torture_type=none
-- 
2.43.0

[PATCH 27/28] hazptr: Add Bloom false-positive exact-match fallback

[Kunwu Chan]

After 200+ scan cycles without progress, do an exact-match
per-address walk to determine whether the address is truly
present in any per-CPU or overflow-list slot.  If absent,
complete the waiter immediately.  A wildcard counts as present.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 kernel/hazptr.c | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/kernel/hazptr.c b/kernel/hazptr.c
index 66954251c332..13ae9eb26a8b 100644
--- a/kernel/hazptr.c
+++ b/kernel/hazptr.c
@@ -407,6 +407,7 @@ struct hazptr_waiter {
 #endif
 #ifdef CONFIG_HAZPTR_DRAIN_ASSIST
 	unsigned int drain_cycles;
+	unsigned int bloom_cycles;	/* cycles since last exact match */
 #endif
 };
 
@@ -852,6 +853,18 @@ static void hazptr_scan_do_cycle(void)
 
 	mutex_unlock(&hazptr_phase_lock);
 
+	/* Bloom-false-positive fallback */
+	list_for_each_entry_safe(w, n, &hazptr_scan.scanning, node) {
+		w->bloom_cycles++;
+		if (w->bloom_cycles > 200) {
+			if (!hazptr_scan_walk(w->addr, NULL, NULL)) {
+				list_move(&w->node, &done);
+			} else {
+				w->bloom_cycles = 0;
+			}
+		}
+	}
+
 	list_for_each_entry_safe(w, n, &done, node) {
 		list_del_init(&w->node);
 		complete(&w->done);
-- 
2.43.0

[PATCH 28/28] Documentation/litmus-tests: Add hazptr bloom wildcard-hole test

[Kunwu Chan]

Test the wildcard-deferral of the shared-scan Bloom walk: a
slow-path acquire that published a wildcard to an overflow-list
slot cannot be missed by the kthread's post-wildcard slot read.
Verified with herd7: Never 0 6.

Signed-off-by: Kunwu Chan <kunwu.chan@gmail.com>
---
 Documentation/litmus-tests/README             |  7 +++
 .../hazptr/hazptr-bloom-wildcard-hole.litmus  | 49 +++++++++++++++++++
 2 files changed, 56 insertions(+)
 create mode 100644 Documentation/litmus-tests/hazptr/hazptr-bloom-wildcard-hole.litmus

diff --git a/Documentation/litmus-tests/README b/Documentation/litmus-tests/README
index 79b0634ddb13..561319404126 100644
--- a/Documentation/litmus-tests/README
+++ b/Documentation/litmus-tests/README
@@ -111,6 +111,7 @@ hazptr-wildcard-flip-escape.litmus
     scan: a straggling acquire that published the old wildcard after
     the final walk cannot still load the unpublished object.
 
+
 hazptr-drain-engage-wildcard.litmus
     Test the drain engage's wildcard defer: an engage that reads a
     slot holding a wildcard (an in-flight slow-path acquire) must
@@ -127,3 +128,9 @@ hazptr-drain-safety.litmus
 hazptr-drain-generation-visibility.litmus
     Test the drain generation ordering: a release hook that observes
     the new drain address also observes the new generation.
+
+hazptr-bloom-wildcard-hole.litmus
+    Test the wildcard-deferral of the shared-scan Bloom walk: a
+    slow-path acquire that published a wildcard to an overflow-list
+    slot cannot be missed by the kthread's post-wildcard slot read
+    (completion is deferred until the wildcard resolves).
diff --git a/Documentation/litmus-tests/hazptr/hazptr-bloom-wildcard-hole.litmus b/Documentation/litmus-tests/hazptr/hazptr-bloom-wildcard-hole.litmus
new file mode 100644
index 000000000000..28a2bc1206c7
--- /dev/null
+++ b/Documentation/litmus-tests/hazptr/hazptr-bloom-wildcard-hole.litmus
@@ -0,0 +1,49 @@
+C hazptr-bloom-wildcard-hole
+
+(*
+ * Result: Never
+ *
+ * Wildcard handling in the shared-scan Bloom walk: a slow-path
+ * acquire publishes HAZPTR_WILDCARD (Store B) to an overflow-list
+ * slot before loading the target pointer (Load A) and publishing the
+ * real address (Store C).  The scan kthread observes the wildcard
+ * (r2=2, its walk loop keeps waiting) and defers completion
+ * decisions; the model captures the final post-wildcard slot read
+ * (r0) that gates completion.
+ *
+ * P0 models the kthread's post-wildcard read: it must not complete
+ * the waiter based on a slot value observed before the reader's
+ * Load A.  P1 is the slow-path reader.
+ *
+ * Question: can the kthread's post-wildcard slot read (r0) miss the
+ * reader's Store C while the reader's Load A still saw the old
+ * pointer (1:r1=1)?  The smp_mb() between Store B and Load A, paired
+ * with the kthread waiting out the wildcard before its final
+ * load-acquire, forbids it: Never.
+ *)
+
+{
+	slot = 2;
+	head = 1;
+}
+
+P0(int *slot, int *head)			/* scan kthread */
+{
+	int r0;
+	int r2;
+
+	r2 = smp_load_acquire(slot);		/* walk sees wildcard... */
+	r0 = smp_load_acquire(slot);		/* ...waits, reads again */
+}
+
+P1(int *slot, int *head)			/* slow-path reader */
+{
+	int r1;
+
+	WRITE_ONCE(*slot, 2);			/* Store B: wildcard */
+	smp_mb();
+	r1 = READ_ONCE(*head);			/* Load A */
+	WRITE_ONCE(*slot, 1);			/* Store C: real address */
+}
+
+exists (0:r0=0 /\ 0:r2=2 /\ 1:r1=1)
-- 
2.43.0