Merge branch 'perf/phase0-harness'

Phase 0 of PERF_PLAN -- the benchmark harness that can support a
conclusion, plus the first two optimisations it justified.

The harness work came first because none of the plan's questions were
decidable with the old one: items_for() shrank the sample as work per
item grew, so exactly the rows under investigation ran 50-200 items and
swung 4-8x between passes. Sample size now derives from a time budget,
rows report median-of-N with IQR, and a run is observable and
restartable rather than opaque for hours.

That immediately paid for itself. chain-16's 28.5% deficit against TBB,
which section 4 built a thread-oversubscription hypothesis on, was an
artefact of the N=200 rows: it measures +3.2% at N=30000. chain-32's
37.4% is 15.1%.

The scheduler change is B9 plus the notify gate: a worker resubmitting
into a shared pool was handing the task to a sleeping peer and paying a
futex wake on every dispatch. Shared pools gain 12-15% on the fanout and
shallow-chain rows; private pools, which are the Node<> default, are
unchanged.

Also settled: the reference test count section 6 was unsure of is 150.

Still open -- the 7-pass gate has not been run against the scheduler
change, and the remaining TBB gap (wide-4 +11.5%, chain-32 +15.1%) is
private-pool-bound and untouched by this work.

.gitignore

@@ -25,6 +25,9 @@ venv/
 .DS_Store
 Thumbs.db
 
+# Benchmark output (scripts/bench_repro_check.py --out-dir)
+bench_runs/
+
 # Claude Code local settings
 .claude/settings.local.json
 include/kpn/ort_cache/

include/kpn/scheduler.hpp

@@ -128,7 +128,28 @@ public:
             rejected_.fetch_add(1, std::memory_order_relaxed);
             return;
         }
-        std::size_t target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_;
+        // B9 — submit-to-self affinity. Round-robin hands every task to a
+        // *different* worker, and on a pool of two or more that worker is
+        // asleep, so each dispatch pays a futex wake: measured 182 ns/dispatch
+        // on a 1-thread pool against 3197 ns on 20 threads, with voluntary
+        // context switches per task rising 0.00 -> 1.19 in step.
+        //
+        // A submission originating on one of *our own* workers goes to that
+        // worker's queue instead. It is about to return to worker_loop and
+        // try_pop its own queue, so the work is already there and nothing
+        // sleeps — the property that makes ThreadPool(1) fast, extended to
+        // any pool size. Imbalance is corrected by the existing try_steal.
+        //
+        // The pool identity check is load-bearing: a worker of pool A
+        // submitting into pool B must not use A's index, which may exceed B's
+        // thread_count_ or alias an unrelated queue. Nested networks do
+        // exactly this.
+        std::size_t target;
+        if (tls_pool == this && tls_worker < thread_count_) {
+            target = tls_worker;
+        } else {
+            target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_;
+        }
         {
             std::lock_guard lock(queues_[target]->mx);
             queues_[target]->pq.push(
@@ -142,9 +163,24 @@ public:
         // will observe total_ > 0) or already blocked in wait() (and will be
         // woken). Without this, notify_one() can slip into the gap between the
         // worker's predicate check and its wait(), and be lost — a deadlock.
+        //
+        // Skipped entirely when no worker is parked. waiters_ is incremented
+        // *before* wait() releases cv_mx_ and decremented after it returns,
+        // both under that mutex, so a worker on its way to sleep is already
+        // counted here. Reading zero therefore means no worker can be in
+        // wait(), and there is nothing a notify could reach — as opposed to
+        // reading zero because we raced one, which the mutex prevents.
+        //
+        // This is the hot path for an already-busy pool: with B9 the work is
+        // in the local queue and the submitting worker will find it itself,
+        // so the lock round-trip and notify were pure overhead. Measured 1.00
+        // voluntary context switches per task before this, on a pool where
+        // only one task is ever in flight.
+        if (waiters_.load(std::memory_order_seq_cst) != 0) {
             { std::lock_guard<std::mutex> lk(cv_mx_); }
             cv_.notify_one();
         }
+    }
 
     std::size_t thread_count() const { return thread_count_; }
 
@@ -193,6 +229,17 @@ private:
 
     std::optional<std::function<void()>> try_steal(std::size_t thief) {
         // Find the most-loaded peer without blocking — racy peek is fine.
+        //
+        // The threshold is >0: a peer holding a single task is a valid victim.
+        //
+        // Raising it to >1 — to stop a thief winning the race for a task its
+        // owner just submitted to itself (B9) — deadlocks. `latency` mode
+        // hangs at 12 and 20 threads: an external submit() round-robins one
+        // task onto an idle worker's queue, and if that worker is parked, no
+        // peer will take it because a queue of one is no longer stealable.
+        // Nothing else is coming to wake it, so the pool sits forever.
+        // Measured before reverting: it also made steady-state *worse*,
+        // 2229 -> 4546 ns at 12 threads.
         std::size_t victim = thief, best = 0;
         for (std::size_t i = 0; i < queues_.size(); ++i) {
             if (i == thief) continue;
@@ -221,15 +268,41 @@ private:
     }
 
     void worker_loop(std::size_t id) {
+        // Identify this thread as one of our workers, for B9's affinity check
+        // in submit(). Restored on exit rather than merely cleared: a pool
+        // whose worker runs a task that itself starts and stops a nested pool
+        // would otherwise come back with its identity erased.
+        ThreadPool* const prev_pool   = tls_pool;
+        const std::size_t prev_worker = tls_worker;
+        tls_pool   = this;
+        tls_worker = id;
+        struct Restore {
+            ThreadPool* p; std::size_t w;
+            ~Restore() { tls_pool = p; tls_worker = w; }
+        } restore{prev_pool, prev_worker};
+
         while (true) {
             if (auto fn = try_pop(*queues_[id]))   { execute(*fn); continue; }
             if (auto fn = try_steal(id))            { execute(*fn); continue; }
 
+            // B5 (bounded spin before parking) was tried here and removed: it
+            // does not pay. Swept at 50/200/1000 rounds on a 12-thread pool,
+            // steady state went 2123 / 2230 / 2574 ns against 2229 ns without
+            // it, and voluntary context switches per task stayed at ~0.97
+            // throughout. The spin cannot catch what it is aimed at, because
+            // a peer is woken the moment queued_ becomes non-zero — which
+            // happens before this worker reaches the spin at all.
             std::unique_lock lock(cv_mx_);
+            // Counted under cv_mx_ and before the predicate is evaluated, so
+            // that a submit() which reads waiters_ == 0 can be certain this
+            // worker is not about to block: to get here we already hold the
+            // mutex that submit() must take to notify.
+            waiters_.fetch_add(1, std::memory_order_seq_cst);
             cv_.wait(lock, [this] {
                 return stopped_.load(std::memory_order_seq_cst)
                     || queued_.load(std::memory_order_relaxed) > 0;
             });
+            waiters_.fetch_sub(1, std::memory_order_seq_cst);
             // Exit on queued_, not total_: waiting for total_ to reach zero
             // meant waiting for someone else's task to finish, which this
             // worker cannot help with and would spin through until it did.
@@ -239,6 +312,13 @@ private:
         }
     }
 
+    /// Which pool, and which of its workers, the calling thread is — or
+    /// nullptr on any thread that is not a pool worker. Read by submit() to
+    /// decide whether a local push is safe (B9). inline so the header stays
+    /// header-only.
+    static inline thread_local ThreadPool* tls_pool   = nullptr;
+    static inline thread_local std::size_t tls_worker = 0;
+
     const std::size_t                          thread_count_;
     std::vector<std::unique_ptr<WorkerQueue>>  queues_;
     std::vector<std::thread>                   workers_;
@@ -267,6 +347,9 @@ private:
     std::atomic<size_t>   queued_{0};     // waiting to run
     std::atomic<size_t>   active_{0};     // executing only (for snapshot)
     std::atomic<size_t>   next_{0};       // round-robin submit cursor
+    /// Workers currently inside cv_.wait(), maintained under cv_mx_. Lets
+    /// submit() skip the lock round-trip and notify when nobody is parked.
+    std::atomic<size_t>   waiters_{0};
     std::atomic<uint64_t> seq_{0};        // tie-break for equal-priority tasks
     std::atomic<uint64_t> submitted_{0};
     /// Submissions refused because the pool was already stopped. Not an error —

scripts/bench_repro_check.py

@@ -9,12 +9,20 @@ between KPN and TBB is worth acting on.
 Exits non-zero if any row exceeds the tolerance, so it can gate a session of
 performance work rather than merely inform one.
 
+A full sweep is hours, so the run is observable and restartable rather than
+opaque: rows stream to --out-dir as each pass produces them, and a progress
+bar tracks rows within the pass. Killing the run keeps everything already
+written; --resume picks up from the completed passes on disk.
+
 Usage:
     scripts/bench_repro_check.py ./build_bench/benchmarks/bench_pipeline \\
         --passes 7 --tolerance 5 -- --work=10 --topos=chain,wide --reps=5
 """
 
 import argparse
+import datetime
+import os
+import pathlib
 import statistics
 import subprocess
 import sys
@@ -27,11 +35,47 @@ KEY_COLS = ("topology", "size", "work_us", "threads")
 METRIC_COLS = ("items_per_sec", "ns_per_dispatch")
 
 
-def parse_csv(text, metric=None):
+def _progress(total, desc):
+    """A tqdm bar if tqdm is installed, else a minimal stderr fallback.
+
+    The fallback exists because this script gates a benchmark run; refusing to
+    start over a missing progress dependency would be the wrong trade.
+    """
+    try:
+        from tqdm import tqdm
+    except ImportError:
+        class Fallback:
+            def __init__(self):
+                self.n = 0
+
+            def update(self, k=1):
+                self.n += k
+                end = "\n" if (total and self.n >= total) else "\r"
+                print(f"  {desc}: {self.n}/{total or '?'} rows",
+                      file=sys.stderr, end=end, flush=True)
+
+            def close(self):
+                pass
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *exc):
+                self.close()
+
+        return Fallback()
+
+    return tqdm(total=total, desc=desc, unit="row", leave=False,
+                bar_format="  {desc}: {n_fmt}/{total_fmt} rows "
+                           "|{bar}| {elapsed}<{remaining}",
+                file=sys.stderr)
+
+
+def parse_csv_lines(lines, metric=None):
     """Return ({(topology, size, work_us, threads): value}, metric_name)."""
     rows = {}
     header = None
-    for line in text.splitlines():
+    for line in lines:
         line = line.strip()
         if not line or line.startswith("#"):
             continue
@@ -59,36 +103,64 @@ def parse_csv(text, metric=None):
     return rows, metric
 
 
-def main():
-    ap = argparse.ArgumentParser()
-    ap.add_argument("binary", help="path to bench_pipeline")
-    ap.add_argument("--passes", type=int, default=7)
-    ap.add_argument("--tolerance", type=float, default=5.0,
-                    help="max allowed deviation from the median, percent")
-    ap.add_argument("--metric", default=None, choices=METRIC_COLS,
-                    help="column to check (default: whichever the CSV carries)")
+def parse_csv(text, metric=None):
+    return parse_csv_lines(text.splitlines(), metric)
 
-    # Everything after a standalone `--` goes to bench_pipeline verbatim.
-    # argparse.REMAINDER would swallow this script's own flags instead.
-    argv = sys.argv[1:]
-    extra = []
-    if "--" in argv:
-        cut = argv.index("--")
-        argv, extra = argv[:cut], argv[cut + 1:]
-    args = ap.parse_args(argv)
 
-    passes = []
-    metric = args.metric
-    for i in range(args.passes):
-        print(f"pass {i + 1}/{args.passes} ...", file=sys.stderr, flush=True)
-        proc = subprocess.run([args.binary] + extra,
-                              capture_output=True, text=True)
+def count_rows(binary, extra):
+    """Enumerate the sweep cheaply, so the progress bar has a real total.
+
+    Asks the binary itself rather than reimplementing the sweep in Python,
+    which would silently drift from the C++ defaults. Returns None if the
+    probe fails -- an unknown total degrades the bar, it does not stop the run.
+    """
+    probe = [binary] + extra + ["--reps=0", "--warmup=0"]
+    try:
+        proc = subprocess.run(probe, capture_output=True, text=True,
+                              timeout=600)
+    except (subprocess.SubprocessError, OSError):
+        return None
     if proc.returncode != 0:
-            print(proc.stderr, file=sys.stderr)
-            sys.exit(f"{args.binary} failed with {proc.returncode}")
-        rows, metric = parse_csv(proc.stdout, metric)
-        passes.append(rows)
+        return None
+    rows, _ = parse_csv(proc.stdout)
+    return len(rows) or None
 
+
+def run_pass(binary, extra, total, desc, sink, metric):
+    """Run one pass, streaming rows to `sink` and the bar as they arrive.
+
+    capture_output would withhold every row until the pass ended, which for a
+    multi-hour sweep means no way to tell a slow run from a wedged one.
+    """
+    lines = []
+    bar = _progress(total, desc)
+    proc = subprocess.Popen([binary] + extra, stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE, text=True, bufsize=1)
+    try:
+        for line in proc.stdout:
+            lines.append(line)
+            if sink:
+                sink.write(line)
+                sink.flush()          # a killed run keeps its rows
+            stripped = line.strip()
+            if (stripped and not stripped.startswith("#")
+                    and "," in stripped
+                    and not stripped.startswith("topology,")):
+                bar.update(1)
+    finally:
+        bar.close()
+        proc.stdout.close()
+        stderr = proc.stderr.read()
+        proc.stderr.close()
+        rc = proc.wait()
+
+    if rc != 0:
+        print(stderr, file=sys.stderr)
+        sys.exit(f"{binary} failed with {rc}")
+    return parse_csv_lines(lines, metric)
+
+
+def report(passes, metric, tolerance, npasses):
     keys = set(passes[0])
     for p in passes[1:]:
         keys &= set(p)
@@ -103,20 +175,84 @@ def main():
         values = [p[key] for p in passes]
         med = statistics.median(values)
         worst = max(abs(v - med) / med * 100 for v in values) if med else 0.0
-        ok = worst <= args.tolerance
+        ok = worst <= tolerance
         failures += not ok
         label = "{}-{} w={} s={}".format(*key)
         print(f"{label:<34} {med:>20.1f} {worst:>9.1f}%  {'ok' if ok else 'NOISY'}")
 
     print("-" * 72)
     if failures:
-        print(f"{failures}/{len(keys)} rows exceed +/-{args.tolerance:g}% — "
+        print(f"{failures}/{len(keys)} rows exceed +/-{tolerance:g}% — "
               f"the Phase-0 gate is not met.")
         return 1
-    print(f"all {len(keys)} rows within +/-{args.tolerance:g}% "
-          f"over {args.passes} passes — Phase-0 gate met.")
+    print(f"all {len(keys)} rows within +/-{tolerance:g}% "
+          f"over {npasses} passes — Phase-0 gate met.")
     return 0
 
 
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("binary", help="path to bench_pipeline")
+    ap.add_argument("--passes", type=int, default=7)
+    ap.add_argument("--tolerance", type=float, default=5.0,
+                    help="max allowed deviation from the median, percent")
+    ap.add_argument("--metric", default=None, choices=METRIC_COLS,
+                    help="column to check (default: whichever the CSV carries)")
+    ap.add_argument("--out-dir", default=None,
+                    help="write pass-NN.csv as rows arrive "
+                         "(default: bench_runs/<timestamp>)")
+    ap.add_argument("--resume", action="store_true",
+                    help="reuse complete pass-NN.csv files in --out-dir")
+
+    # Everything after a standalone `--` goes to bench_pipeline verbatim.
+    # argparse.REMAINDER would swallow this script's own flags instead.
+    argv = sys.argv[1:]
+    extra = []
+    if "--" in argv:
+        cut = argv.index("--")
+        argv, extra = argv[:cut], argv[cut + 1:]
+    args = ap.parse_args(argv)
+
+    out_dir = args.out_dir
+    if out_dir is None:
+        if args.resume:
+            sys.exit("--resume needs an explicit --out-dir")
+        stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
+        out_dir = os.path.join("bench_runs", stamp)
+    out = pathlib.Path(out_dir)
+    out.mkdir(parents=True, exist_ok=True)
+
+    total = count_rows(args.binary, extra)
+    print(f"writing to {out}/", file=sys.stderr)
+    if total:
+        print(f"{total} rows per pass, {args.passes} passes", file=sys.stderr)
+
+    passes = []
+    metric = args.metric
+    for i in range(args.passes):
+        path = out / f"pass-{i + 1:02d}.csv"
+
+        if args.resume and path.exists():
+            rows, metric = parse_csv(path.read_text(), metric)
+            # A partial file from a killed run must not be silently averaged
+            # in as if it were a whole pass.
+            if total and len(rows) < total:
+                print(f"pass {i + 1}/{args.passes}: {path.name} has "
+                      f"{len(rows)}/{total} rows — rerunning", file=sys.stderr)
+            else:
+                print(f"pass {i + 1}/{args.passes}: reusing {path.name} "
+                      f"({len(rows)} rows)", file=sys.stderr)
+                passes.append(rows)
+                continue
+
+        print(f"pass {i + 1}/{args.passes} ...", file=sys.stderr, flush=True)
+        with open(path, "w") as sink:
+            rows, metric = run_pass(args.binary, extra, total,
+                                    f"pass {i + 1}/{args.passes}", sink, metric)
+        passes.append(rows)
+
+    return report(passes, metric, args.tolerance, args.passes)
+
+
 if __name__ == "__main__":
     sys.exit(main())