[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@xxxxxxxxxx>
Signed-off-by: Kunwu Chan <kunwu.chan@xxxxxxxxx>
---
 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