[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@xxxxxxxxx>
---
 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