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.
perf(pool): submit to the calling worker's own queue, and skip a notify
2026-08-08 22:27:55 +02:00 · 2026-08-08 22:26:14 +02:00 · 2026-08-08 22:25:00 +02:00 · 2026-08-08 12:13:11 +02:00 · 2026-08-06 22:53:02 +02:00 · 2026-08-06 21:57:32 +02:00
12 changed files with 1880 additions and 142 deletions
@@ -25,8 +25,11 @@ 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/
 build-tsan/
-build-*/
+build*/
@@ -0,0 +1,413 @@
 # Performance investigation plan: fanout dispatch cost and deep-chain oversubscription
 **Status:** phase 0 implemented; gate not yet cleared
 **Date:** 2026-08-06 (phase 0 landed 2026-08-06)
 **Baseline:** master @ 3b67b7e
 **Machine:** 20 cores, GCC 16.1.1, TBB 2023.1.0, AC power, `performance` governor
 **Data:** 7 full benchmark passes, medians reported below
 ---
 ## 1. What was measured
 Throughput, items/sec, median of 7 passes. `N` is the sample size the harness
 uses for that row; it is what determines whether a row can be trusted at all.
 ### work_us = 10
 | row | KPN it/s | TBB it/s | TBB faster | N | reliable? |
 |---|---|---|---|---|---|
 | chain-1 | 89381 | 91350 | +2.2% | 3000 | solid |
 | chain-2 | 83759 | 87017 | +3.9% | 1000 | solid |
 | chain-4 | 83725 | 85973 | +2.7% | 1000 | solid |
 | chain-8 | 79730 | 81090 | +1.7% | 1000 | solid |
 | **chain-16** | 53484 | 68745 | **+28.5%** | 200 | weak |
 | **chain-32** | 32780 | 45030 | **+37.4%** | 200 | weak |
 | **wide-4** | 84906 | 95137 | **+12.0%** | 3000 | solid |
 | diamond-4 | 84826 | 86772 | +2.3% | 1000 | solid |
 ### work_us = 100
 Everything except chain-16/32 falls within ±3.4%, with KPN often ahead
 (chain-1 −0.6%, chain-4 −1.7%, chain-8 −3.4%, diamond −2.6% — negative means
 KPN faster). chain-16 is +12.5% and chain-32 +18.6%, both at N=50 and
 therefore unusable.
 ### Two deficits, different causes
 1. **Fanout, +12%.** Solidly measured. `wide-4` performs ~5 node dispatches
   per item; the gap works out to a fixed ~250 ns per dispatch, consistent
   with `chain-1`'s ~290 ns over a single dispatch. This is dispatch
   efficiency.
 2. **Deep chains, +28–37%.** The gap is 1.7–3.9% through depth 8, then jumps
   to 28.5% at depth 16 and 37.4% at depth 32. That is a cliff at core count,
   not a linear per-dispatch cost. `Node<>` owns a private `ThreadPool(1)`
   (`include/kpn/node.hpp:21`), so a depth-32 chain spawns 32 OS threads on
   20 cores. TBB bounds its worker count by hardware concurrency regardless of
   graph size.
 ### Scope note
 At 100 µs+ per node KPN is at parity or ahead. The repository's own examples
 (OpenCV cellshade, frame sources, scene-actor extraction) do milliseconds of
 work per node, where a 290 ns dispatch cost is roughly one part in thirty
 thousand. Everything in this document matters only for fine-grained pipelines.
 ---
 ## 2. Phase 0 — the gate that comes first
 **Is there a target workload with sub-30 µs nodes?**
 If no such workload exists or is planned, the correct output of this document
 is section 3 (harness) plus a README correction, and nothing else. Optimising
 for a benchmark regime the project does not operate in is not worth the risk
 described in section 6.
 ---
 ## 3. Prerequisite — make the harness able to answer
 None of the questions below are decidable with the current harness.
 `benchmarks/bench_pipeline.cpp` shrinks the sample count as work per item
 grows, so the rows under investigation run 50–200 items and swing 4–8×
 run to run.
 | id | change | why |
 |---|---|---|
 | M1 | `items_for()` → fixed floor, e.g. `max(2000, …)`, independent of work_us and depth | deep rows are currently unmeasurable |
 | M2 | report items/sec as the primary metric; keep derived overhead as secondary | overhead is `elapsed − work`, a difference of large numbers; it magnifies noise roughly 10× |
 | M3 | K in-process repetitions per config; report median and IQR | one shot per config is the root of the present noise |
 | M4 | discard a warm-up repetition | first-touch page faults, thread spin-up |
 | M5 | extend `pool_sizes[]` to `{1,2,4,8,16,20}` | currently `{1,2,4}` — the configuration the README recommends is never run |
 | M6 | record nproc, governor and AC state in the CSV header | run-to-run attribution |
 **Acceptance:** the same configuration run 7× lands within ±5% on every row.
 Until that holds, no number below should be acted on.
 This touches only the benchmark, not the library.
 ### Status — implemented 2026-08-06
 All of M1–M6 are in `benchmarks/bench_pipeline.cpp`, plus a CLI so the phase-1
 experiments are invocations rather than edits (`--depths`, `--pools`,
 `--work`, `--topos`, `--modes`, `--reps`, `--target-sec`, `--min-items`).
 M1 is not a fixed floor but a time budget with a floor: sample size derives
 from `work_us × stages / units`, the steady-state throughput bound, then
 clamps to `[--min-items, --max-sec]`. A flat 2000-item floor would have made
 `chain-32` on a 1-thread pool at 1000 µs a 64-second row; the ceiling keeps
 such rows short and reports their true `N` so a short row is visible rather
 than silent. The old ladder's error was treating depth as a throughput cost —
 in a pipeline, depth beyond the core count costs throughput, below it only
 latency.
 Also added, ahead of schedule because it is free: `ru_nivcsw` / `ru_nvcsw` per
 item are captured around every timed region, so **A3 is now a matter of
 reading a column** rather than a separate experiment.
 `scripts/bench_repro_check.py` runs the acceptance criterion directly — K
 passes, per-row deviation from the median, non-zero exit if any row exceeds
 tolerance.
 **Gate not yet cleared.** A 3-pass run of `chain-{1,8}` at 10 µs on the
 development laptop (20 cores, **powersave governor, on battery** — the header
 now records this) lands every row within 0.7%, against the 4–8× swings this
 section describes. That is encouraging but is not the acceptance run: it must
 be 7 passes over the full row set on the reference machine.
 **Provisional and not to be acted on:** in that same run `chain-16` private
 was 6% behind TBB, not the 28.5% in the table above. If that survives the
 real acceptance run, the deep-chain deficit is substantially a measurement
 artefact of the N=200 rows and workstream A shrinks accordingly.
 ---
 ## 4. Workstream A — deep chains
 **Hypothesis:** the deficit is thread oversubscription from the private-pool
 model, not dispatch cost.
 ### Investigation
 | id | experiment | falsifies the hypothesis if |
 |---|---|---|
 | A1 | sweep depth 8, 12, 16, 20, 24, 32 at 10 µs, private pools | the cliff is not near nproc |
 | A2 | repeat A1 under `taskset -c 0-7` | the cliff does **not** move to ~depth 8 |
 | A3 | `getrusage(RUSAGE_SELF).ru_nivcsw` per item, depth 8 vs 32 | involuntary context switches do not scale with depth |
 | A4 | chain-16/32 on a shared pool sized 16 and 20, vs private and vs TBB | a correctly sized shared pool does not recover the gap |
 A2 is decisive and costs one run: if the cliff tracks the core count, the
 mechanism is established.
 ### Improvement, conditional on A4
 If a correctly sized shared pool closes the gap, this is not an optimisation
 problem — the mechanism already exists and is simply not the default:
 - **A5** — change `Network`'s default from per-node private pools to a single
  shared pool sized `hardware_concurrency()`. Users should not have to know.
 - **A6** — emit a diagnostic when total node threads exceed
  `hardware_concurrency()`.
 - **A7** — README: state the threshold, with the measured cliff.
 A5 is a change to the default execution model and must clear section 6 in full.
 ### A5 now has a prerequisite (from B1/B2, 2026-08-06)
 The dispatch microbenchmark measured what a shared pool costs per dispatch,
 and it is not free: **466 ns on a private `ThreadPool(1)` against ~1.7 µs on a
 shared pool of 4**, because round-robin submission wakes a sleeping worker on
 every dispatch (see §5). A5 as written would therefore make every graph that
 currently fits inside its core count roughly 3–4× *worse* per dispatch, in
 exchange for fixing graphs that exceed it.
 **A5 must not land before the wake cost does.** The order is B9/B5 first,
 then A5, and A4 must be read with this in mind: if a shared pool "recovers the
 gap" at depth 32, check what it costs at depth 4 in the same run before
 changing any default.
 This partially inverts the prediction in §7: workstream A is not purely a
 default-and-documentation change, because the default it would switch to is
 currently the slower one per dispatch.
 ---
 ## 5. Workstream B — fanout dispatch cost
 **Hypothesis:** a fixed ~250 ns per node dispatch, paid ~5× per item in
 `wide-4`. Unlike workstream A, this genuinely is dispatch efficiency.
 Estimated budget for ~290 ns, per item — **estimates, to be replaced by B3**:
 | cost | est. |
 |---|---|
 | `shared_lock(lifecycle_mx_)` in `submit()` | 20–40 ns |
 | `queues_[target]->mx` lock/unlock | 20–40 ns |
 | `priority_queue` push + pop (heap ops, `std::function` moves) | 50–100 ns |
 | `{ lock_guard lk(cv_mx_); }` + `notify_one()` | 20–40 ns, or µs if a worker actually sleeps |
 | 2–3 × `clock_t::now()` in `fire_once` | 50–75 ns |
 | gate CAS + ~6 stats atomics | 30–60 ns |
 ### Investigation — measure before touching anything
 - **B1** — microbenchmark submit→execute turnaround for a null task on
  `ThreadPool(1)` and `ThreadPool(4)`. Yields ns/dispatch directly, in seconds
  rather than minutes.
 - **B2** — **does a worker actually sleep per item?** Count `cv_.wait` returns,
  or `strace -c -f -e futex`. The entire spin-window hypothesis depends on
  this; if workers are not sleeping, B5 is worthless and drops off the list.
 - **B3** — ablation, one variant per suspected cost, each measured against B1
  rather than guessed at:
 | variant | suspected cost |
 |---|---|
 | stats and clock calls compiled out | 2–3 × `clock_t::now()` plus ~6 atomics per firing |
 | `priority_queue` → FIFO ring | heap operations, `std::function` moves |
 | `shared_lock(lifecycle_mx_)` removed (**measurement only, unsafe**) | `include/kpn/scheduler.hpp:113` |
 | bounded spin before sleeping | `include/kpn/scheduler.hpp:210-227` |
 ### B1/B2 — first results, 2026-08-06
 `benchmarks/bench_dispatch.cpp` answers both without touching the library.
 Sleeping is inferred from `ru_nvcsw`: a thread blocking on a condition
 variable books a voluntary context switch, so voluntary switches per task is
 sleeps per task. Three modes, because "the cost of a dispatch" is three
 numbers: `latency` (idle pool, one task in flight), `batch` (submit flat out,
 drain once), `steady` (the task resubmits its successor, as `fire_once` does).
 Laptop, powersave, battery, 3 reps — **the nanoseconds are provisional; the
 sleep counts are structural and will hold.** `steady`, 10 µs payload:
 | pool threads | ns/dispatch | sleeps/task |
 |---|---|---|
 | 1 | 466 | **0.00** |
 | 2 | 1494 | 0.97 |
 | 4 | 1722 | 1.00 |
 | 8 | 1996 | 1.00 |
 **B1 is answered and the abandon criterion is not met.** A `ThreadPool(1)`
 dispatch is 291 ns for a null task, 466 ns with a payload — against the ~290 ns
 the section-1 budget estimated for `chain-1`. The estimate was good. Dispatch
 cost is not already under 100 ns, so workstream B stays alive.
 **B2 is answered, and the answer is conditional — which the question did not
 anticipate.** It is not "do workers sleep?" but "which pool?":
 - On a private `ThreadPool(1)` — the `Node<>` default — the worker **never**
  sleeps. It resubmits into its own queue and finds the work already there.
 - On any pool of 2 or more, a worker sleeps **exactly once per task**.
 `submit()` round-robins (`next_.fetch_add(1) % thread_count_`,
 `scheduler.hpp:131`), so on a shared pool every task is handed to a *different*
 worker, which is asleep, and every single dispatch pays a futex wake. That is
 the entire 466 ns → 1.7 µs difference.
 Consequently **B5 (bounded spin) is worthless for the default configuration**
 and is the highest-value item for shared pools. It does not drop off the list,
 it moves onto a different one.
 ### B9 — submit-to-self affinity (new, not in the original plan)
 If a `submit()` originating on a pool worker pushed to *that worker's own*
 queue instead of round-robining, the shared pool would inherit the property
 that makes `ThreadPool(1)` fast: the work is already local when the worker
 loops, so no wake. This is roughly what TBB does, and it plausibly subsumes
 most of B5 at lower risk — it changes task placement, not the sleep/wake
 protocol that the August wedge fixes hardened. Work stealing already exists to
 correct the resulting imbalance.
 Measure before believing it: an affinity policy can starve peers, and
 `try_steal` only rebalances when a peer goes idle.
 ### Improvement — only what B3 shows pays
 1. **B4 — compile-time-optional instrumentation.** No concurrency risk; the
   only item here that cannot reintroduce a wedge. Worth doing regardless.
 2. **B5 — bounded spin before sleeping**, mirroring the channel's existing
   `spin_count_` (~4 µs). Note the tension: b9698fa deliberately moved from
   "spin whenever any task runs" to "sleep as soon as nothing is queued" in
   order to fix pathological spinning. A *bounded* window is the middle
   ground; unbounded spin would undo that fix.
 3. **B6 — cheaper queue on the common path.** A private pool holds ≤1–2 tasks;
   `priority_queue<Task>` is heavy for that.
 4. **B7 — batched firing.** `fire_once` processes one token then re-submits;
   looping while inputs stay ready, bounded, amortises the submit, gate CAS
   and wake. The largest algorithmic win, but it changes latency and
   interacts with `compute_priority()`.
 5. **B8 — `lifecycle_mx_` off the hot path.** Last, and possibly never. It is
   load-bearing: it prevents `submit()` racing `stop()`'s `queues_.clear()`,
   a documented segfault reproducible "about 12 runs in 20".
 **Abandon criteria:** if B1 shows dispatch cost already under ~100 ns, or the
 best surviving variant buys under 5%, stop and document the finding.
 ---
 ## 6. Guardrails
 Both workstreams modify the machinery responsible for roughly twenty wedge
 fixes in August 2026, plus the lost wake fixed in 6802328. Every change:
 1. **146/146** ctest, examples included.
 2. **Wedge soak before and after** — `benchmarks/repro_wedge.cpp`, ≥50k
   iterations clean. Reference point: the pre-6802328 code wedged 5/5 inside
   45 s, at iterations 149, 1249, 332, 1740 and 493.
 3. **ThreadSanitizer** on scheduler and pool_node tests for any change to
   either.
 4. **One change at a time**, measured independently. Bundling is how the
   August audit became twenty commits.
 5. **G1 — wire the reproducer in as an opt-in CTest stress target**
   (e.g. `-L soak`) so that performance work cannot silently reintroduce a
   wedge. This should land before either workstream starts.
 ### G1 — implemented 2026-08-06
 `tests/soak_wedge.cpp` (supersedes `benchmarks/repro_wedge.cpp`, which was
 never wired into any build and can be deleted). Always compiled so it cannot
 rot; its CTest cases register only under `-DKPN_ENABLE_SOAK_TESTS=ON`, so the
 default `ctest` count is unchanged.
 ```
 cmake -B build -DKPN_ENABLE_SOAK_TESTS=ON -DKPN_SOAK_ITERS=50000
 cmake --build build --target kpn_soak_wedge
 ctest --test-dir build -L soak
 ```
 Two cases: `soak.wedge.pool` (depth 4, 4 threads — the configuration the
 August wedges were reproduced on) and `soak.wedge.private` (depth 8, one pool
 per node — the model workstream A would change). Both parameterised, so
 A5-style changes can be soaked at the depth that matters.
 A wedge is a hang, and a hang under CTest is an unattributable timeout, so the
 binary carries a watchdog: if an iteration stops making progress for
 `--watchdog-sec` it aborts naming the iteration and the phase (`pushed`,
 `drained`, `nodes stopped`, `pool stopped`). Measured cost: ~13 ms per
 iteration, so the 50k-iteration guardrail is ~11 minutes.
 **Guardrail 1 needs a correction.** The stated reference is 146/146; the
 tests-only configuration used here reports **136/136 passing**, and neither
 `examples/` nor `python/` registers any `add_test`. The true reference count
 must be pinned down before it is used to certify a change.
 ---
 ## 7. Sequencing
 | phase | contents | gate to proceed | state |
 |---|---|---|---|
 | 0 | workload question; M1–M6; G1 | ±5% reproducibility achieved | **tooling done**, acceptance run outstanding |
 | 1 | A1–A4 | A2 confirms the cliff tracks core count | harness supports it; not run |
 | 2 | A5–A7, or documentation only | A4 shows a shared pool recovers the gap | **now gated on B9/B5** |
 | 3 | B1–B3 | B2 answers the sleep question | **B1/B2 answered**; B3 outstanding |
 | 4 | B4, then whichever of B5–B8 survived B3 | each ≥5% and soak-clean | B5 rescoped to shared pools |
 B1/B2 ran early because the microbenchmark cost seconds rather than minutes,
 and the result reordered phases 2 and 4 — the shared-pool default now depends
 on the wake cost being fixed first. Phase 1 is unchanged but its A4 row needs
 a shallow-depth control, per §4.
 ### Reproducing this
 ```
 cmake -B build_bench -DKPN_BUILD_BENCHMARKS=ON -DCMAKE_BUILD_TYPE=Release
 cmake --build build_bench -j
 # Phase 0 acceptance — must pass before any number below is acted on
 python3 scripts/bench_repro_check.py ./build_bench/benchmarks/bench_pipeline \
    --passes 7 --tolerance 5 -- --work=10,100 --reps=5
 # B1/B2
 ./build_bench/benchmarks/bench_dispatch --threads=1,2,4,8,20 --reps=5 \
    | tee dispatch.csv
 # A1/A2 — the depth sweep, and the same under taskset to move the cliff
 ./build_bench/benchmarks/bench_pipeline --work=10 --topos=chain \
    --depths=8,12,16,20,24,32 --modes=priv,tbb --reps=5 | tee a1.csv
 taskset -c 0-7 ./build_bench/benchmarks/bench_pipeline --work=10 \
    --topos=chain --depths=4,6,8,10,12,16,32 --modes=priv,tbb --reps=5 | tee a2.csv
 # A4 — shared pool sized to the machine, against private and TBB.
 # Include a shallow depth: A5's risk is what a shared pool costs when the
 # graph already fits in its cores.
 ./build_bench/benchmarks/bench_pipeline --work=10 --topos=chain \
    --depths=4,16,32 --pools=16,20 --reps=5 | tee a4.csv
 ```
 Check the `# governor=` line in each CSV before trusting it. A3 needs no
 separate run: `ivcsw_per_item` is a column in every row above.
 **Success criteria**
 - chain-32 @10 within 10% of TBB in the recommended configuration
 - wide-4 @10 within 5% of TBB
 - zero wedges across 100k soak iterations
 **Prediction, recorded so it can be proven wrong:** workstream A resolves into
 a default-and-documentation change rather than an optimisation, and workstream
 B yields 5–10% on fanout from B4 and B5, with the remainder not worth the risk.
 **Prediction, revised 2026-08-06 after B1/B2** — the original is already half
 wrong and is left above unedited:
 - Workstream A does *not* resolve into a documentation change, because the
  shared pool it would recommend costs 3–4× more per dispatch than the private
  default. It resolves into B9 first.
 - The largest single win is not B4, B5 or B7 but **B9, submit-to-self
  affinity**: one sleep per dispatch is being paid on every shared pool, and
  eliminating it is worth roughly 1.2 µs per dispatch — far more than the
  5–10% predicted for fanout.
 - Standing: `chain-16`'s 28.5% deficit is a measurement artefact of N=200.
 ---
 ## 8. Related correction
 Independently of the above, the README's TBB comparison overstates its case.
 The claim that KPN++ beats TBB "for every chain and diamond topology at
 100 µs/node" is not supported: at 100 µs only chain-1 and diamond lean KPN,
 while chain-16, chain-32 and wide-4 lean TBB. The tables are also quoted in
 derived overhead, which magnifies small differences — the same rows expressed
 as throughput are mostly within a few percent. Restating them in items/sec
 would be both more accurate and more favourable.
@@ -4,6 +4,13 @@ add_executable(bench_pipeline bench_pipeline.cpp)
 target_link_libraries(bench_pipeline PRIVATE kpn)
 target_compile_options(bench_pipeline PRIVATE -O3 -march=native)
 # Dispatch microbenchmark (PERF_PLAN B1/B2): ns per ThreadPool dispatch, and
 # whether a worker actually sleeps per task. No TBB comparison — it measures
 # KPN's own scheduler, not a competitor.
 add_executable(bench_dispatch bench_dispatch.cpp)
 target_link_libraries(bench_dispatch PRIVATE kpn)
 target_compile_options(bench_dispatch PRIVATE -O3 -march=native)
 find_package(TBB QUIET)
 if(TBB_FOUND)
    target_link_libraries(bench_pipeline PRIVATE TBB::tbb)
@@ -0,0 +1,307 @@
 // Dispatch microbenchmark — PERF_PLAN B1 and B2.
 //
 // B1 asks what a single ThreadPool dispatch costs. B2 asks whether a worker
 // actually sleeps per item, because the whole spin-window hypothesis (B5)
 // depends on the answer: if workers are not sleeping, a spin window buys
 // nothing and drops off the list.
 //
 // Both are answered here without touching the library. Sleeping is inferred
 // from ru_nvcsw — a thread blocking on a condition variable books a voluntary
 // context switch — so `vcsw/task` near 1.0 means a sleep per dispatch and near
 // 0 means the worker never went to sleep at all.
 //
 // Three modes, because "the cost of a dispatch" is three different numbers:
 //
 //   latency — one task in flight, pool idle in between. The worker is asleep
 //             at every submission, so this is dispatch cost *including* a
 //             wake. Worst case, and the case a spin window would attack.
 //
 //   batch   — submit K no-op tasks flat out, then drain. The worker is never
 //             idle, so this is the amortised floor: queue and heap operations
 //             with no wake at all. Reports the producer-side submit() cost
 //             separately from end-to-end throughput.
 //
 //   steady  — the task resubmits its successor, one in flight, each doing
 //             --work-us of work. This is what a KPN node actually does:
 //             fire_once processes a token and resubmits. On ThreadPool(1) the
 //             worker resubmits to its own queue; on ThreadPool(4) round-robin
 //             hands the task to a *different* worker, which may be asleep.
 //             That difference is the fanout cost wide-4 pays ~5x per item.
 //
 // Usage: ./bench_dispatch [--threads=1,2,4] [--mode=latency,batch,steady]
 //                         [--tasks=200000] [--work-us=0] [--reps=5] [--warmup=1]
 #include <kpn/kpn.hpp>
 #include "bench_env.hpp"
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
 #include <cstdio>
 #include <cstdlib>
 #include <mutex>
 #include <string>
 #include <vector>
 using namespace kpn;
 using sclock = std::chrono::steady_clock;
 struct Opts {
    std::vector<int> threads {1, 2, 4};
    std::vector<std::string> modes {"latency", "batch", "steady"};
    long tasks   = 200000;
    int  work_us = 0;
    int  reps    = 5;
    int  warmup  = 1;
 };
 static Opts g_opts;
 static void busy_us(int us) {
    if (us <= 0) return;
    auto end = sclock::now() + std::chrono::microseconds(us);
    while (sclock::now() < end);
 }
 struct Sample {
    double ns_per_dispatch  = 0;   // end-to-end, minus the work payload
    double submit_ns        = 0;   // producer side only (batch mode)
    double vcsw_per_task    = 0;   // B2: sleeps per dispatch
    double ivcsw_per_task   = 0;
 };
 // ── latency: one task at a time, worker asleep between submissions ────────────
 static Sample run_latency(int threads, long tasks) {
    ThreadPool pool(threads);
    pool.start();
    std::mutex mx;
    std::condition_variable cv;
    bool done = false;
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    for (long i = 0; i < tasks; ++i) {
        { std::lock_guard lk(mx); done = false; }
        pool.submit([&] {
            busy_us(g_opts.work_us);
            { std::lock_guard lk(mx); done = true; }
            cv.notify_one();
        });
        std::unique_lock lk(mx);
        cv.wait(lk, [&] { return done; });
    }
    auto t1 = sclock::now();
    Sample s;
    long iv = 0, vc = 0;
    ru.finish(iv, vc);
    pool.stop();
    double elapsed_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
    s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0;
    // The requesting thread blocks once per task too, so it books a voluntary
    // switch of its own; halve to attribute per side rather than per process.
    s.vcsw_per_task  = static_cast<double>(vc) / tasks / 2.0;
    s.ivcsw_per_task = static_cast<double>(iv) / tasks;
    return s;
 }
 // ── batch: submit flat out, drain once. No wake in the steady state ───────────
 static Sample run_batch(int threads, long tasks) {
    ThreadPool pool(threads);
    pool.start();
    std::atomic<long> ran{0};
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    for (long i = 0; i < tasks; ++i)
        pool.submit([&] {
            busy_us(g_opts.work_us);
            ran.fetch_add(1, std::memory_order_relaxed);
        });
    auto t_submitted = sclock::now();
    pool.drain();
    auto t1 = sclock::now();
    Sample s;
    long iv = 0, vc = 0;
    ru.finish(iv, vc);
    pool.stop();
    if (ran.load() != tasks)
        std::fprintf(stderr, "WARNING: batch ran %ld of %ld tasks\n",
                     ran.load(), tasks);
    double elapsed_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
    s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0;
    s.submit_ns = std::chrono::duration<double, std::nano>(t_submitted - t0).count() / tasks;
    s.vcsw_per_task  = static_cast<double>(vc) / tasks;
    s.ivcsw_per_task = static_cast<double>(iv) / tasks;
    return s;
 }
 // ── steady: the task resubmits its successor, as fire_once does ──────────────
 static Sample run_steady(int threads, long tasks) {
    ThreadPool pool(threads);
    pool.start();
    std::mutex mx;
    std::condition_variable cv;
    std::atomic<long> count{0};
    bool  finished = false;
    // Recursive submission: hold the chain in a std::function so the task can
    // resubmit itself. Captured by reference; it outlives the drain below.
    //
    // The counter is atomic rather than mutex-guarded so that this loop
    // measures the pool's dispatch path and not a lock of the benchmark's own.
    std::function<void()> step = [&] {
        busy_us(g_opts.work_us);
        long n = count.fetch_add(1, std::memory_order_relaxed) + 1;
        if (n < tasks) {
            pool.submit(step);
        } else {
            { std::lock_guard lk(mx); finished = true; }
            cv.notify_one();
        }
    };
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    pool.submit(step);
    {
        std::unique_lock lk(mx);
        cv.wait(lk, [&] { return finished; });
    }
    auto t1 = sclock::now();
    Sample s;
    long iv = 0, vc = 0;
    ru.finish(iv, vc);
    pool.stop();
    double elapsed_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
    s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0;
    s.vcsw_per_task  = static_cast<double>(vc) / tasks;
    s.ivcsw_per_task = static_cast<double>(iv) / tasks;
    return s;
 }
 // ── driver ────────────────────────────────────────────────────────────────────
 static void run_row(const std::string& mode, int threads, long tasks) {
    auto once = [&] {
        if (mode == "latency") return run_latency(threads, tasks);
        if (mode == "batch")   return run_batch(threads, tasks);
        return run_steady(threads, tasks);
    };
    for (int i = 0; i < g_opts.warmup; ++i) (void)once();
    std::vector<double> ns, sub, vcsw, ivcsw;
    for (int i = 0; i < g_opts.reps; ++i) {
        Sample s = once();
        ns.push_back(s.ns_per_dispatch);
        sub.push_back(s.submit_ns);
        vcsw.push_back(s.vcsw_per_task);
        ivcsw.push_back(s.ivcsw_per_task);
    }
    const double med = bench::percentile(ns, 0.5);
    const double q1  = bench::percentile(ns, 0.25);
    const double q3  = bench::percentile(ns, 0.75);
    const double iqr = med > 0 ? 100.0 * (q3 - q1) / med : 0.0;
    const double sleeps = bench::percentile(vcsw, 0.5);
    std::fprintf(stderr, "%-9s %-8d %-8d %-10ld %-12.0f %-7.1f %-11.0f %-10.2f %-10.2f\n",
                 mode.c_str(), threads, g_opts.work_us, tasks, med, iqr,
                 bench::percentile(sub, 0.5), sleeps,
                 bench::percentile(ivcsw, 0.5));
    // Column names deliberately match bench_pipeline's key columns so that
    // scripts/bench_repro_check.py can gate this benchmark too.
    std::printf("%s,%d,%d,%d,%ld,%d,%.1f,%.2f,%.1f,%.3f,%.3f\n",
                mode.c_str(), threads, g_opts.work_us, threads, tasks,
                g_opts.reps, med, iqr, bench::percentile(sub, 0.5),
                sleeps, bench::percentile(ivcsw, 0.5));
    std::fflush(stdout);
 }
 static std::vector<int> parse_int_list(const char* s) {
    std::vector<int> out;
    const char* p = s;
    while (*p) {
        char* end = nullptr;
        long v = std::strtol(p, &end, 10);
        if (end == p) break;
        out.push_back(static_cast<int>(v));
        p = end;
        while (*p == ',' || *p == ' ') ++p;
    }
    return out;
 }
 static std::vector<std::string> parse_word_list(const std::string& s) {
    std::vector<std::string> out;
    std::size_t pos = 0;
    while (pos <= s.size()) {
        std::size_t c = s.find(',', pos);
        if (c == std::string::npos) c = s.size();
        if (c > pos) out.push_back(s.substr(pos, c - pos));
        pos = c + 1;
    }
    return out;
 }
 static void usage() {
    std::fprintf(stderr,
        "usage: bench_dispatch [options]\n"
        "  --threads=1,2,4                pool sizes\n"
        "  --mode=latency,batch,steady    which measurements to run\n"
        "  --tasks=200000                 dispatches per repetition\n"
        "  --work-us=0                    payload per task\n"
        "  --reps=5 --warmup=1\n");
 }
 int main(int argc, char** argv) {
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        auto eq = a.find('=');
        std::string key = a.substr(0, eq);
        std::string val = eq == std::string::npos ? "" : a.substr(eq + 1);
        if      (key == "--help" || key == "-h") { usage(); return 0; }
        else if (key == "--threads") g_opts.threads = parse_int_list(val.c_str());
        else if (key == "--mode")    g_opts.modes   = parse_word_list(val);
        else if (key == "--tasks")   g_opts.tasks   = std::atol(val.c_str());
        else if (key == "--work-us") g_opts.work_us = std::atoi(val.c_str());
        else if (key == "--reps")    g_opts.reps    = std::atoi(val.c_str());
        else if (key == "--warmup")  g_opts.warmup  = std::atoi(val.c_str());
        else { std::fprintf(stderr, "unknown option: %s\n", a.c_str()); usage(); return 2; }
    }
    if (g_opts.reps < 1) g_opts.reps = 1;
    if (g_opts.warmup < 0) g_opts.warmup = 0;
    char cfg[160];
    std::snprintf(cfg, sizeof cfg, "tasks=%ld work_us=%d reps=%d warmup=%d",
                  g_opts.tasks, g_opts.work_us, g_opts.reps, g_opts.warmup);
    bench::print_environment(cfg);
    std::fprintf(stderr, "\n%-9s %-8s %-8s %-10s %-12s %-7s %-11s %-10s %-10s\n",
                 "mode", "threads", "work_us", "tasks", "ns/dispatch", "iqr%",
                 "submit_ns", "vcsw/task", "ivcsw/task");
    std::fprintf(stderr, "%s\n", std::string(96, '-').c_str());
    std::printf("topology,size,work_us,threads,items,reps,ns_per_dispatch,"
                "iqr_pct,submit_ns,vcsw_per_task,ivcsw_per_task\n");
    // latency is a round trip per task, so it is far slower per dispatch than
    // the other modes; scale it down rather than run for minutes.
    for (const auto& mode : g_opts.modes)
        for (int t : g_opts.threads) {
            long tasks = mode == "latency"
                       ? std::max(2000L, g_opts.tasks / 20)
                       : g_opts.tasks;
            run_row(mode, t, tasks);
        }
 }
@@ -0,0 +1,108 @@
 // Shared benchmark plumbing: machine attribution (PERF_PLAN M6), repetition
 // statistics (M3), and context-switch capture.
 //
 // The attribution is not decoration. A result taken under the powersave
 // governor or on battery is not comparable with one taken on AC under
 // performance, and a stored CSV that does not say which it was cannot be
 // argued about later.
 #pragma once
 #include <algorithm>
 #include <cmath>
 #include <cstdio>
 #include <string>
 #include <thread>
 #include <vector>
 #include <sys/resource.h>
 namespace bench {
 inline int hw_units() {
    unsigned n = std::thread::hardware_concurrency();
    return n ? static_cast<int>(n) : 1;
 }
 inline std::string read_line_of(const char* path) {
    std::FILE* f = std::fopen(path, "r");
    if (!f) return "unknown";
    char buf[128] = {0};
    if (!std::fgets(buf, sizeof buf, f)) { std::fclose(f); return "unknown"; }
    std::fclose(f);
    std::string s(buf);
    while (!s.empty() && (s.back() == '\n' || s.back() == ' ')) s.pop_back();
    return s.empty() ? "unknown" : s;
 }
 inline std::string ac_state() {
    for (const char* p : {"/sys/class/power_supply/AC/online",
                          "/sys/class/power_supply/AC0/online",
                          "/sys/class/power_supply/ACAD/online",
                          "/sys/class/power_supply/ADP1/online"}) {
        std::string v = read_line_of(p);
        if (v != "unknown") return v == "1" ? "ac" : "battery";
    }
    return "unknown";
 }
 // M6 — emitted to both streams: the CSV so a stored result can be attributed,
 // the terminal so a run under the wrong governor is noticed while it happens.
 inline void print_environment(const std::string& config_line) {
    const std::string gov = read_line_of(
        "/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor");
    const std::string ac = ac_state();
    for (std::FILE* out : {stdout, stderr}) {
        std::fprintf(out, "# nproc=%d governor=%s power=%s\n",
                     hw_units(), gov.c_str(), ac.c_str());
        if (!config_line.empty())
            std::fprintf(out, "# %s\n", config_line.c_str());
 #if defined(__GNUC__) && !defined(__clang__)
        std::fprintf(out, "# compiler=gcc-%d.%d.%d\n",
                     __GNUC__, __GNUC_MINOR__, __GNUC_PATCHLEVEL__);
 #elif defined(__clang__)
        std::fprintf(out, "# compiler=clang-%d.%d.%d\n",
                     __clang_major__, __clang_minor__, __clang_patchlevel__);
 #endif
    }
    if (gov != "performance" || ac == "battery")
        std::fprintf(stderr,
            "# WARNING: governor=%s power=%s — results are not comparable with\n"
            "#          a run on AC power under the performance governor.\n",
            gov.c_str(), ac.c_str());
 }
 inline double percentile(std::vector<double> v, double p) {
    if (v.empty()) return 0;
    std::sort(v.begin(), v.end());
    double idx = p * (v.size() - 1);
    auto   lo  = static_cast<std::size_t>(std::floor(idx));
    auto   hi  = static_cast<std::size_t>(std::ceil(idx));
    return v[lo] + (v[hi] - v[lo]) * (idx - lo);
 }
 // Process-wide context-switch counters, sampled around a timed region.
 //
 // ru_nvcsw (voluntary) is the cheap answer to PERF_PLAN B2: a thread that
 // blocks on a condition variable books a voluntary switch, so voluntary
 // switches per dispatch is, near enough, sleeps per dispatch. ru_nivcsw
 // (involuntary) is preemption, which is what oversubscription looks like (A3).
 struct RusageDelta {
    long ivcsw0 = 0, vcsw0 = 0;
    void start() {
        rusage ru{};
        getrusage(RUSAGE_SELF, &ru);
        ivcsw0 = ru.ru_nivcsw;
        vcsw0  = ru.ru_nvcsw;
    }
    void finish(long& nivcsw, long& nvcsw) const {
        rusage ru{};
        getrusage(RUSAGE_SELF, &ru);
        nivcsw = ru.ru_nivcsw - ivcsw0;
        nvcsw  = ru.ru_nvcsw  - vcsw0;
    }
 };
 } // namespace bench
@@ -9,24 +9,38 @@
 //   private — each node owns a private ThreadPool(1)  [Node<>]
 //   pool    — all nodes share one ThreadPool(T)        [PoolNode<> + shared pool]
 //
-// Usage: ./bench_pipeline | tee results.csv
+// Each row is run --reps times (plus discarded warm-up runs); the reported
 // figure is the median items/sec, with the inter-quartile spread as a
 // reliability indicator. A row whose iqr_pct is above a few percent is not
 // measuring what it claims to measure.
 //
 // Usage: ./bench_pipeline [options] | tee results.csv
 //        ./bench_pipeline --help
 #include <kpn/kpn.hpp>
 #include "bench_env.hpp"
 #ifdef KPN_BENCH_TBB
 #include <oneapi/tbb/flow_graph.h>
 namespace tbb_flow = oneapi::tbb::flow;
 #endif
 #include <algorithm>
 #include <array>
 #include <atomic>
 #include <chrono>
 #include <cmath>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <memory>
 #include <string>
 #include <thread>
 #include <vector>
 #include <sys/resource.h>
 using namespace kpn;
 using namespace std::chrono_literals;
 using sclock = std::chrono::steady_clock;
@@ -57,31 +71,63 @@ static void push_retry(Channel<int>& ch, int val) {
    }
 }
-// ── result ────────────────────────────────────────────────────────────────────
+// ── configuration (M1, M3, M4, M5) ────────────────────────────────────────────
-struct Result {
+struct Config {
-    const char* topology;
+    std::vector<int> work_amts {10, 100, 1000};
-    int         size;
+    std::vector<int> pool_sizes{1, 2, 4, 8, 16, 20};   // M5
-    int         work_us;
+    std::vector<int> depths    {1, 2, 4, 8, 16, 32};
-    int         threads;     // 0 = private (1 thread per node), N = shared pool size
+    std::vector<int> widths    {1, 2, 3, 4};
-    double      items_per_sec;
+    int    reps        = 5;      // M3: measured repetitions per row
-    double      overhead_us;
+    int    warmup      = 1;      // M4: discarded repetitions per row
    double target_sec  = 0.30;   // aimed-for duration of one repetition
    long   min_items   = 2000;   // M1: floor, independent of work_us and depth
    double max_sec     = 3.0;    // ceiling; only bites where min_items cannot fit
    bool   do_chain = true, do_wide = true, do_diamond = true;
    bool   do_priv  = true, do_pool = true, do_tbb = true;
 };
 static Config g_cfg;
 // M1 — sample size from a time budget with a hard floor, rather than a
 // hand-tuned ladder that collapsed to 50–200 items on exactly the rows under
 // investigation.
 //
 // `stages` is the number of node firings per item; `units` the number of
 // threads able to run them concurrently. Steady-state throughput of the
 // pipeline is bounded by work_us * stages / units, so that is the per-item
 // cost the sample size is derived from. Depth beyond `units` costs throughput;
 // depth below it costs only latency, which does not scale the run.
 static long pick_items(int work_us, int stages, int units) {
    units = std::max(1, std::min(units, bench::hw_units()));
    const double per_item_us =
        std::max(1.0, static_cast<double>(work_us)) *
        std::max(1.0, static_cast<double>(stages) / units);
    long want = static_cast<long>(g_cfg.target_sec * 1e6 / per_item_us);
    long cap  = static_cast<long>(g_cfg.max_sec    * 1e6 / per_item_us);
    want = std::max(want, g_cfg.min_items);
    // The floor wins unless honouring it would blow the time ceiling by more
    // than the ceiling allows; such rows are reported with their true N so the
    // reader can see they are short.
    if (want > cap) want = std::max(cap, 200L);
    return want;
 }
 // ── one measured repetition ───────────────────────────────────────────────────
 struct Sample {
    double items_per_sec = 0;
    double overhead_us   = 0;
    long   nivcsw        = 0;   // involuntary context switches during the run
    long   nvcsw         = 0;   // voluntary   context switches during the run
 };
 // ── chain ─────────────────────────────────────────────────────────────────────
-static int items_for(int work_us, int depth = 1) {
+static Sample bench_chain(int depth, int work_us, long N) {
-    int effective = std::max(1, work_us) * std::max(1, depth);
+    const std::size_t CAP = static_cast<std::size_t>(N);
    if (effective <= 1)     return 5000;
    if (effective <= 10)    return 3000;
    if (effective <= 100)   return 1000;
    if (effective <= 1000)  return 200;
    return 50;
 }
 static Result bench_chain(int depth, int work_us) {
    const int N   = items_for(work_us, depth);
    const int CAP = N;
    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= depth; ++i)
@@ -98,17 +144,20 @@ static Result bench_chain(int depth, int work_us) {
    std::atomic<sclock::time_point> t1;
    std::thread reader([&] {
-        for (int i = 0; i < N; ++i) chs.back()->pop();
+        for (long i = 0; i < N; ++i) chs.back()->pop();
        t1.store(sclock::now(), std::memory_order_release);
    });
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (int i = 0; i < N; ++i) push_retry(*chs[0], i);
+        for (long i = 0; i < N; ++i) push_retry(*chs[0], static_cast<int>(i));
    });
    pusher.join();
    reader.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    for (auto& n : nodes) n->stop();
    double elapsed = std::chrono::duration<double>(
@@ -116,13 +165,13 @@ static Result bench_chain(int depth, int work_us) {
    // Subtract theoretical pipeline fill cost (depth-1)*W so that overhead
    // reflects only framework latency, not the expected pipeline startup time.
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
-    double wus = (elapsed * 1e6 - pipeline_us) / N;
+    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
-    return {"chain", depth, work_us, 0, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
-static Result bench_chain_pool(int depth, int work_us, int pool_threads) {
+static Sample bench_chain_pool(int depth, int work_us, int pool_threads, long N) {
-    const int N   = items_for(work_us, depth);
+    const std::size_t CAP = static_cast<std::size_t>(N);
    const int CAP = N;
    auto pool = std::make_shared<ThreadPool>(pool_threads);
@@ -142,33 +191,36 @@ static Result bench_chain_pool(int depth, int work_us, int pool_threads) {
    std::atomic<sclock::time_point> t1;
    std::thread reader([&] {
-        for (int i = 0; i < N; ++i) chs.back()->pop();
+        for (long i = 0; i < N; ++i) chs.back()->pop();
        t1.store(sclock::now(), std::memory_order_release);
    });
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (int i = 0; i < N; ++i) push_retry(*chs[0], i);
+        for (long i = 0; i < N; ++i) push_retry(*chs[0], static_cast<int>(i));
    });
    pusher.join();
    reader.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    for (auto& n : nodes) n->stop();
    pool->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
-    double wus = (elapsed * 1e6 - pipeline_us) / N;
+    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
-    return {"chain", depth, work_us, pool_threads, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
 // ── wide (fanout<W>) ──────────────────────────────────────────────────────────
 template<std::size_t W>
-static Result bench_wide(int work_us) {
+static Sample bench_wide(int work_us, long N) {
-    const int N   = items_for(work_us);
+    const std::size_t CAP = static_cast<std::size_t>(N);
    const int CAP = N;
    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, W>>(CAP);
@@ -197,33 +249,36 @@ static Result bench_wide(int work_us) {
    for (std::size_t w = 0; w < W; ++w) {
        readers[w] = std::thread([&, w] {
-            for (int i = 0; i < N; ++i) sink_chs[w]->pop();
+            for (long i = 0; i < N; ++i) sink_chs[w]->pop();
            if (readers_done.fetch_add(1, std::memory_order_acq_rel) + 1
                    == static_cast<int>(W))
                t1.store(sclock::now(), std::memory_order_release);
        });
    }
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
+        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });
    pusher.join();
    for (auto& r : readers) r.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    for (auto& n : nodes) n->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    return {"wide", static_cast<int>(W), work_us, 0, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
 template<std::size_t W>
-static Result bench_wide_pool(int work_us, int pool_threads) {
+static Sample bench_wide_pool(int work_us, int pool_threads, long N) {
-    const int N   = items_for(work_us);
+    const std::size_t CAP = static_cast<std::size_t>(N);
    const int CAP = N;
    auto pool   = std::make_shared<ThreadPool>(pool_threads);
    auto src_ch = std::make_shared<Channel<int>>(CAP);
@@ -254,35 +309,38 @@ static Result bench_wide_pool(int work_us, int pool_threads) {
    for (std::size_t w = 0; w < W; ++w) {
        readers[w] = std::thread([&, w] {
-            for (int i = 0; i < N; ++i) sink_chs[w]->pop();
+            for (long i = 0; i < N; ++i) sink_chs[w]->pop();
            if (readers_done.fetch_add(1, std::memory_order_acq_rel) + 1
                    == static_cast<int>(W))
                t1.store(sclock::now(), std::memory_order_release);
        });
    }
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
+        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });
    pusher.join();
    for (auto& r : readers) r.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    for (auto& n : nodes) n->stop();
    pool->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    return {"wide", static_cast<int>(W), work_us, pool_threads, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
 // ── diamond ───────────────────────────────────────────────────────────────────
-static Result bench_diamond(int work_us) {
+static Sample bench_diamond(int work_us, long N) {
-    const int N   = items_for(work_us, 2);
+    const std::size_t CAP = static_cast<std::size_t>(N);
    const int CAP = N;
    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, 2>>(CAP);
@@ -312,7 +370,7 @@ static Result bench_diamond(int work_us) {
    std::atomic<int> done{0};
    auto make_reader = [&](Channel<int>& ch) {
        return std::thread([&] {
-            for (int i = 0; i < N; ++i) ch.pop();
+            for (long i = 0; i < N; ++i) ch.pop();
            if (done.fetch_add(1, std::memory_order_acq_rel) + 1 == 2)
                t1.store(sclock::now(), std::memory_order_release);
        });
@@ -320,23 +378,26 @@ static Result bench_diamond(int work_us) {
    auto rL = make_reader(*snkL);
    auto rR = make_reader(*snkR);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
+        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });
    pusher.join(); rL.join(); rR.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop(); nL->stop(); nR->stop(); nL2->stop(); nR2->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    return {"diamond", 4, work_us, 0, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
-static Result bench_diamond_pool(int work_us, int pool_threads) {
+static Sample bench_diamond_pool(int work_us, int pool_threads, long N) {
-    const int N   = items_for(work_us, 2);
+    const std::size_t CAP = static_cast<std::size_t>(N);
    const int CAP = N;
    auto pool   = std::make_shared<ThreadPool>(pool_threads);
    auto src_ch = std::make_shared<Channel<int>>(CAP);
@@ -369,7 +430,7 @@ static Result bench_diamond_pool(int work_us, int pool_threads) {
    std::atomic<int> done{0};
    auto make_reader = [&](Channel<int>& ch) {
        return std::thread([&] {
-            for (int i = 0; i < N; ++i) ch.pop();
+            for (long i = 0; i < N; ++i) ch.pop();
            if (done.fetch_add(1, std::memory_order_acq_rel) + 1 == 2)
                t1.store(sclock::now(), std::memory_order_release);
        });
@@ -377,28 +438,30 @@ static Result bench_diamond_pool(int work_us, int pool_threads) {
    auto rL = make_reader(*snkL);
    auto rR = make_reader(*snkR);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
+        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });
    pusher.join(); rL.join(); rR.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    nL->stop(); nR->stop(); nL2->stop(); nR2->stop();
    pool->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    return {"diamond", 4, work_us, pool_threads, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
 // ── TBB flow graph ────────────────────────────────────────────────────────────
 #ifdef KPN_BENCH_TBB
-static Result bench_chain_tbb(int depth, int work_us) {
+static Sample bench_chain_tbb(int depth, int work_us, long N) {
    const int N = items_for(work_us, depth);
    tbb_flow::graph g;
    using FN = tbb_flow::function_node<int, int>;
    std::vector<std::unique_ptr<FN>> nodes;
@@ -409,21 +472,23 @@ static Result bench_chain_tbb(int depth, int work_us) {
    for (int i = 0; i + 1 < depth; ++i)
        tbb_flow::make_edge(*nodes[i], *nodes[i + 1]);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
-    for (int i = 0; i < N; ++i) nodes[0]->try_put(i);
+    for (long i = 0; i < N; ++i) nodes[0]->try_put(static_cast<int>(i));
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
-    double elapsed    = std::chrono::duration<double>(t1 - t0).count();
+    double elapsed     = std::chrono::duration<double>(t1 - t0).count();
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
-    double wus        = (elapsed * 1e6 - pipeline_us) / N;
+    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
-    return {"chain_tbb", depth, work_us, -1, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
 template<std::size_t W>
-static Result bench_wide_tbb(int work_us) {
+static Sample bench_wide_tbb(int work_us, long N) {
    const int N = items_for(work_us);
    tbb_flow::graph g;
    tbb_flow::broadcast_node<int> fan(g);
    using FN = tbb_flow::function_node<int, int>;
@@ -434,19 +499,21 @@ static Result bench_wide_tbb(int work_us) {
        tbb_flow::make_edge(fan, *n);
    }
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
-    for (int i = 0; i < N; ++i) fan.try_put(i);
+    for (long i = 0; i < N; ++i) fan.try_put(static_cast<int>(i));
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    double elapsed = std::chrono::duration<double>(t1 - t0).count();
-    double wus     = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    return {"wide_tbb", static_cast<int>(W), work_us, -1, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
-static Result bench_diamond_tbb(int work_us) {
+static Sample bench_diamond_tbb(int work_us, long N) {
    const int N = items_for(work_us, 2);
    tbb_flow::graph g;
    tbb_flow::broadcast_node<int> fan(g);
    using FN = tbb_flow::function_node<int, int>;
@@ -456,71 +523,242 @@ static Result bench_diamond_tbb(int work_us) {
    tbb_flow::make_edge(fan, nL);  tbb_flow::make_edge(fan, nR);
    tbb_flow::make_edge(nL, nL2);  tbb_flow::make_edge(nR, nR2);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
-    for (int i = 0; i < N; ++i) fan.try_put(i);
+    for (long i = 0; i < N; ++i) fan.try_put(static_cast<int>(i));
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    double elapsed = std::chrono::duration<double>(t1 - t0).count();
-    double wus     = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    return {"diamond_tbb", 4, work_us, -1, N / elapsed, wus};
+    s.items_per_sec = N / elapsed;
    return s;
 }
 #endif // KPN_BENCH_TBB
 // ── repetition driver (M2, M3, M4) ────────────────────────────────────────────
 using bench::percentile;
 // A row: median of `reps` repetitions, after `warmup` discarded ones.
 // M2 — items/sec is the primary figure; derived overhead is secondary,
 // because it is a difference of large numbers and magnifies noise ~10×.
 template<class Fn>
 static void run_row(const char* topology, int size, int work_us, int sched,
                    long N, Fn&& one_rep) {
    for (int i = 0; i < g_cfg.warmup; ++i) (void)one_rep();   // M4
    std::vector<double> ips, ovh;
    long ivcsw = 0, vcsw = 0;
    for (int i = 0; i < g_cfg.reps; ++i) {
        Sample s = one_rep();
        ips.push_back(s.items_per_sec);
        ovh.push_back(s.overhead_us);
        ivcsw += s.nivcsw;
        vcsw  += s.nvcsw;
    }
    const double med  = percentile(ips, 0.5);
    const double q1   = percentile(ips, 0.25);
    const double q3   = percentile(ips, 0.75);
    const double iqr  = med > 0 ? 100.0 * (q3 - q1) / med : 0.0;
    const double lo   = *std::min_element(ips.begin(), ips.end());
    const double hi   = *std::max_element(ips.begin(), ips.end());
    const double spread = med > 0 ? 100.0 * (hi - lo) / med : 0.0;
    const double ivcsw_per_item = static_cast<double>(ivcsw) / (double(N) * g_cfg.reps);
    const double vcsw_per_item  = static_cast<double>(vcsw)  / (double(N) * g_cfg.reps);
    const std::string s = sched  < 0 ? "tbb"
                        : sched == 0 ? "priv"
                                     : std::to_string(sched);
    std::fprintf(stderr, "%-10s %-5d %-8d %-6s %-8ld %-12.0f %-7.1f %-7.1f %-9.1f %-8.2f %-8.2f\n",
                 topology, size, work_us, s.c_str(), N,
                 med, iqr, spread, percentile(ovh, 0.5), ivcsw_per_item, vcsw_per_item);
    std::printf("%s,%d,%d,%s,%ld,%d,%.0f,%.0f,%.0f,%.2f,%.2f,%.2f,%.3f,%.3f\n",
                topology, size, work_us, s.c_str(), N, g_cfg.reps,
                med, lo, hi, iqr, spread, percentile(ovh, 0.5),
                ivcsw_per_item, vcsw_per_item);
    std::fflush(stdout);
 }
 // ── argument parsing ──────────────────────────────────────────────────────────
 static std::vector<int> parse_int_list(const char* s) {
    std::vector<int> out;
    const char* p = s;
    while (*p) {
        char* end = nullptr;
        long v = std::strtol(p, &end, 10);
        if (end == p) break;
        out.push_back(static_cast<int>(v));
        p = end;
        while (*p == ',' || *p == ' ') ++p;
    }
    return out;
 }
 static bool has_word(const std::string& csv, const char* word) {
    return csv.find(word) != std::string::npos;
 }
 static void usage() {
    std::fprintf(stderr,
        "usage: bench_pipeline [options]\n"
        "  --work=10,100,1000     per-node busy-work, microseconds\n"
        "  --depths=1,2,4,8,16,32 chain depths\n"
        "  --widths=1,2,3,4       fanout widths\n"
        "  --pools=1,2,4,8,16,20  shared-pool thread counts\n"
        "  --topos=chain,wide,diamond\n"
        "  --modes=priv,pool,tbb\n"
        "  --reps=5               measured repetitions per row\n"
        "  --warmup=1             discarded repetitions per row\n"
        "  --target-sec=0.30      aimed-for duration of one repetition\n"
        "  --min-items=2000       sample-size floor\n"
        "  --max-sec=3.0          per-repetition ceiling (overrides the floor)\n");
 }
 static bool parse_args(int argc, char** argv) {
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        auto eq = a.find('=');
        std::string key = a.substr(0, eq);
        std::string val = eq == std::string::npos ? "" : a.substr(eq + 1);
        if (key == "--help" || key == "-h")   { usage(); std::exit(0); }
        else if (key == "--work")       g_cfg.work_amts  = parse_int_list(val.c_str());
        else if (key == "--depths")     g_cfg.depths     = parse_int_list(val.c_str());
        else if (key == "--widths")     g_cfg.widths     = parse_int_list(val.c_str());
        else if (key == "--pools")      g_cfg.pool_sizes = parse_int_list(val.c_str());
        else if (key == "--reps")       g_cfg.reps       = std::atoi(val.c_str());
        else if (key == "--warmup")     g_cfg.warmup     = std::atoi(val.c_str());
        else if (key == "--target-sec") g_cfg.target_sec = std::atof(val.c_str());
        else if (key == "--min-items")  g_cfg.min_items  = std::atol(val.c_str());
        else if (key == "--max-sec")    g_cfg.max_sec    = std::atof(val.c_str());
        else if (key == "--topos") {
            g_cfg.do_chain   = has_word(val, "chain");
            g_cfg.do_wide    = has_word(val, "wide");
            g_cfg.do_diamond = has_word(val, "diamond");
        }
        else if (key == "--modes") {
            g_cfg.do_priv = has_word(val, "priv");
            g_cfg.do_pool = has_word(val, "pool");
            g_cfg.do_tbb  = has_word(val, "tbb");
        }
        else { std::fprintf(stderr, "unknown option: %s\n", a.c_str()); usage(); return false; }
    }
    if (g_cfg.reps < 1) g_cfg.reps = 1;
    if (g_cfg.warmup < 0) g_cfg.warmup = 0;
    return true;
 }
 // `wide` is templated on W, so dispatch the runtime width through a switch.
 template<class F>
 static void with_width(int w, F&& f) {
    switch (w) {
        case 1: f(std::integral_constant<std::size_t, 1>{}); break;
        case 2: f(std::integral_constant<std::size_t, 2>{}); break;
        case 3: f(std::integral_constant<std::size_t, 3>{}); break;
        case 4: f(std::integral_constant<std::size_t, 4>{}); break;
        default:
            std::fprintf(stderr, "width %d not instantiated (1..4 only)\n", w);
    }
 }
 // ── main ──────────────────────────────────────────────────────────────────────
-int main() {
+int main(int argc, char** argv) {
-    const int work_amts[]  = {10, 100, 1000};
+    // A rejected option must fail loudly: a harness driver that silently got
-    const int pool_sizes[] = {1, 2, 4};
+    // no CSV back is worse than one that stops.
    if (!parse_args(argc, argv)) return 2;
-    std::fprintf(stderr, "%-12s %-8s %-10s %-8s %-18s %-20s\n",
+    char cfg[192];
-                 "topology", "size", "work_us", "threads", "items/sec", "overhead_us/item");
+    std::snprintf(cfg, sizeof cfg,
-    std::fprintf(stderr, "%s\n", std::string(78, '-').c_str());
+                  "reps=%d warmup=%d target_sec=%.2f min_items=%ld max_sec=%.1f",
-    std::printf("topology,size,work_us,threads,items_per_sec,overhead_us_per_item\n");
+                  g_cfg.reps, g_cfg.warmup, g_cfg.target_sec,
                  g_cfg.min_items, g_cfg.max_sec);
    bench::print_environment(cfg);
-    auto emit = [](const Result& r) {
+    std::fprintf(stderr, "\n%-10s %-5s %-8s %-6s %-8s %-12s %-7s %-7s %-9s %-8s %-8s\n",
-        std::string sched = r.threads  < 0 ? "tbb"
+                 "topology", "size", "work_us", "sched", "items", "items/sec",
-                          : r.threads == 0 ? "priv"
+                 "iqr%", "range%", "ovh_us", "ivcsw/it", "vcsw/it");
-                          : std::to_string(r.threads);
+    std::fprintf(stderr, "%s\n", std::string(104, '-').c_str());
-        std::fprintf(stderr, "%-12s %-8d %-10d %-8s %-18.0f %-20.1f\n",
+    std::printf("topology,size,work_us,threads,items,reps,items_per_sec,"
-                     r.topology, r.size, r.work_us, sched.c_str(),
+                "items_per_sec_min,items_per_sec_max,iqr_pct,range_pct,"
-                     r.items_per_sec, r.overhead_us);
+                "overhead_us_per_item,ivcsw_per_item,vcsw_per_item\n");
        std::printf("%s,%d,%d,%s,%.0f,%.2f\n",
                    r.topology, r.size, r.work_us, sched.c_str(),
                    r.items_per_sec, r.overhead_us);
        std::fflush(stdout);
    };
-    for (int w : work_amts) {
+    for (int w : g_cfg.work_amts) {
        g_work_us.store(w, std::memory_order_relaxed);
        std::fprintf(stderr, "\n── work_us=%-4d private pools ───────────────────────────────────────\n", w);
-        for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain(d, w));
+        if (g_cfg.do_priv) {
-        emit(bench_wide<1>(w));
+            std::fprintf(stderr, "\n── work_us=%-4d private pools ──────────────────────\n", w);
-        emit(bench_wide<2>(w));
+            if (g_cfg.do_chain)
-        emit(bench_wide<3>(w));
+                for (int d : g_cfg.depths) {
-        emit(bench_wide<4>(w));
+                    long N = pick_items(w, d, d);
-        emit(bench_diamond(w));
+                    run_row("chain", d, w, 0, N, [&] { return bench_chain(d, w, N); });
                }
            if (g_cfg.do_wide)
                for (int wd : g_cfg.widths)
                    with_width(wd, [&](auto W) {
                        long N = pick_items(w, W.value, W.value);
                        run_row("wide", static_cast<int>(W.value), w, 0, N,
                                [&] { return bench_wide<W.value>(w, N); });
                    });
            if (g_cfg.do_diamond) {
                long N = pick_items(w, 4, 4);
                run_row("diamond", 4, w, 0, N, [&] { return bench_diamond(w, N); });
            }
        }
-        for (int pt : pool_sizes) {
+        if (g_cfg.do_pool) {
-            std::fprintf(stderr, "\n── work_us=%-4d shared pool (%d thread%s) ─────────────────────────────\n",
+            for (int pt : g_cfg.pool_sizes) {
-                         w, pt, pt == 1 ? "" : "s");
+                std::fprintf(stderr, "\n── work_us=%-4d shared pool (%d thread%s) ───────────\n",
-            for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain_pool(d, w, pt));
+                             w, pt, pt == 1 ? "" : "s");
-            emit(bench_wide_pool<1>(w, pt));
+                if (g_cfg.do_chain)
-            emit(bench_wide_pool<2>(w, pt));
+                    for (int d : g_cfg.depths) {
-            emit(bench_wide_pool<3>(w, pt));
+                        long N = pick_items(w, d, pt);
-            emit(bench_wide_pool<4>(w, pt));
+                        run_row("chain", d, w, pt, N,
-            emit(bench_diamond_pool(w, pt));
+                                [&] { return bench_chain_pool(d, w, pt, N); });
                    }
                if (g_cfg.do_wide)
                    for (int wd : g_cfg.widths)
                        with_width(wd, [&](auto W) {
                            long N = pick_items(w, W.value, pt);
                            run_row("wide", static_cast<int>(W.value), w, pt, N,
                                    [&] { return bench_wide_pool<W.value>(w, pt, N); });
                        });
                if (g_cfg.do_diamond) {
                    long N = pick_items(w, 4, pt);
                    run_row("diamond", 4, w, pt, N,
                            [&] { return bench_diamond_pool(w, pt, N); });
                }
            }
        }
 #ifdef KPN_BENCH_TBB
-        std::fprintf(stderr, "\n── work_us=%-4d TBB flow graph ──────────────────────────────────────\n", w);
+        if (g_cfg.do_tbb) {
-        for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain_tbb(d, w));
+            std::fprintf(stderr, "\n── work_us=%-4d TBB flow graph ─────────────────────\n", w);
-        emit(bench_wide_tbb<1>(w));
+            if (g_cfg.do_chain)
-        emit(bench_wide_tbb<2>(w));
+                for (int d : g_cfg.depths) {
-        emit(bench_wide_tbb<3>(w));
+                    long N = pick_items(w, d, d);
-        emit(bench_wide_tbb<4>(w));
+                    run_row("chain_tbb", d, w, -1, N,
-        emit(bench_diamond_tbb(w));
+                            [&] { return bench_chain_tbb(d, w, N); });
                }
            if (g_cfg.do_wide)
                for (int wd : g_cfg.widths)
                    with_width(wd, [&](auto W) {
                        long N = pick_items(w, W.value, W.value);
                        run_row("wide_tbb", static_cast<int>(W.value), w, -1, N,
                                [&] { return bench_wide_tbb<W.value>(w, N); });
                    });
            if (g_cfg.do_diamond) {
                long N = pick_items(w, 4, 4);
                run_row("diamond_tbb", 4, w, -1, N,
                        [&] { return bench_diamond_tbb(w, N); });
            }
        }
 #endif
    }
 }
@@ -136,7 +136,6 @@ public:
            throw ChannelOverflowError(capacity_);
        }
        const bool was_empty = (t == h);
        buf_[t & ring_mask_] = make_storage(std::move(value));
        tail_.store(t + 1, std::memory_order_release);
        stats_.record_push(t - h + 1, data_bytes);
@@ -144,7 +143,8 @@ public:
        wake_.fetch_add(1, std::memory_order_release);
        wake_.notify_one();
-        if (was_empty && push_callback_)
+        // Level-triggered, not edge-triggered — see set_push_callback.
        if (push_callback_)
            push_callback_();
    }
@@ -187,13 +187,13 @@ public:
        if (t - h >= capacity_) return PushResult::Full;
        const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
        const bool was_empty = (t == h);
        buf_[t & ring_mask_] = make_storage(std::move(value));
        tail_.store(t + 1, std::memory_order_release);
        stats_.record_push(t - h + 1, data_bytes);
        wake_.fetch_add(1, std::memory_order_release);
        wake_.notify_one();
-        if (was_empty && push_callback_) push_callback_();
+        // Level-triggered, not edge-triggered — see set_push_callback.
        if (push_callback_) push_callback_();
        return PushResult::Taken;
    }
@@ -212,13 +212,13 @@ public:
            const std::size_t h = head_.load(std::memory_order_acquire);
            if (t - h < capacity_) {          // space available → normal push
                const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
                const bool was_empty = (t == h);
                buf_[t & ring_mask_] = make_storage(std::move(value));
                tail_.store(t + 1, std::memory_order_release);
                stats_.record_push(t - h + 1, data_bytes);
                wake_.fetch_add(1, std::memory_order_release);
                wake_.notify_one();
-                if (was_empty && push_callback_) push_callback_();
+                // Level-triggered, not edge-triggered — see set_push_callback.
                if (push_callback_) push_callback_();
                return true;
            }
            // full: yield briefly and retry (consumer will drain)
@@ -400,7 +400,41 @@ public:
        wake_.notify_all();
    }
-    // Register a callback fired when the queue transitions empty→non-empty.
+    // Register a callback fired after every successful push.
    //
    // It fires on every push, not on the empty→non-empty transition, and that
    // is a correctness requirement rather than a simplification.
    //
    // The edge version tested `was_empty = (t == h)` using an `h` sampled
    // *before* the item was published. A PoolNode consumer decides whether to
    // run again from the level (count_ready → approx_size), so the two sides
    // could each read the other as stale and both stand down:
    //
    //   producer (push)                    consumer (PoolNode firing)
    //   ------------------------           ----------------------------
    //   samples t=782, h=781
    //     -> was_empty = false, no wake
    //                                      pops idx 781, head_ = 782
    //                                      count_ready(): head_==tail_==782
    //                                      -> not ready, gate released to Idle
    //   tail_.store(783)
    //
    // The item is in the ring, the node is idle, and no wake is outstanding.
    // Worse, the failure is absorbing: every later push now sees a non-empty
    // ring, so `was_empty` is false forever and the callback never fires again.
    // The node sleeps while its backlog grows and its consumer waits on it.
    //
    // Re-reading head_ after the tail_ store does not fix it. That is the
    // store-buffer pattern, and under acquire/release both sides may legally
    // read stale; forbidding it needs seq_cst on the producer's tail_ store and
    // head_ load *and* on the consumer's head_ store and tail_ load — a fence
    // on both hot paths. Firing unconditionally is correct by construction:
    // the callback runs after the publishing store, so a consumer that observes
    // the level at all observes the item.
    //
    // The redundant wakes are cheap. on_input_ready re-checks the level, and
    // SubmitGate::claim() collapses a wake arriving during a firing into the
    // firing already in flight, so the cost is one CAS, not one extra run.
    void set_push_callback(std::function<void()> cb) {
        push_callback_ = std::move(cb);
    }
@@ -53,6 +53,19 @@ public:
    }
    void start() override {
        // Under the lifecycle lock for the same reason stop() is: submit()
        // reads queues_ and this rebuilds it. A network starts its nodes one at
        // a time, and a node already started fires into the next one's channel,
        // whose push callback submits — so a submission can genuinely land
        // while another pool is still inside start(). ThreadSanitizer reports
        // it as a read at submit() against this write, and the consequence is
        // worse than a torn read: push_back can reallocate the vector under a
        // reader that has already indexed it.
        //
        // Queues are all constructed before any worker is spawned, which is
        // what keeps worker_loop's own queues_[id] out of this — it never takes
        // the lock, so holding it across the spawn cannot deadlock.
        std::unique_lock lk(lifecycle_mx_);
        stopped_.store(false, std::memory_order_relaxed);
        queues_.clear();
        for (std::size_t i = 0; i < thread_count_; ++i)
@@ -115,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(
@@ -129,8 +163,23 @@ 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.
-        { std::lock_guard<std::mutex> lk(cv_mx_); }
+        //
-        cv_.notify_one();
+        // 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_; }
@@ -180,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;
@@ -208,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.
@@ -226,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_;
@@ -254,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 —
@@ -0,0 +1,258 @@
 #!/usr/bin/env python3
 """Check the PERF_PLAN Phase-0 acceptance criterion.
 Runs bench_pipeline several times and reports, per row, how far the passes
 spread around their median. The plan's gate is: the same configuration run 7x
 lands within +/-5% on every row. Until that holds, no measured difference
 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
 KEY_COLS = ("topology", "size", "work_us", "threads")
 # bench_pipeline reports throughput, bench_dispatch reports per-dispatch cost.
 # Either is a valid thing to demand reproducibility of; deviation from the
 # median is symmetric, so it does not matter which direction is "better".
 METRIC_COLS = ("items_per_sec", "ns_per_dispatch")
 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 lines:
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        fields = line.split(",")
        if header is None:
            if fields[0] != "topology":
                continue
            header = fields
            if metric is None:
                for cand in METRIC_COLS:
                    if cand in header:
                        metric = cand
                        break
                else:
                    sys.exit(f"no metric column found in header: {header}")
            elif metric not in header:
                sys.exit(f"metric {metric!r} not in header: {header}")
            continue
        rec = dict(zip(header, fields))
        try:
            key = tuple(rec[c] for c in KEY_COLS)
            rows[key] = float(rec[metric])
        except (KeyError, ValueError):
            continue
    return rows, metric
 def parse_csv(text, metric=None):
    return parse_csv_lines(text.splitlines(), metric)
 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:
        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)
    if not keys:
        sys.exit("no rows common to every pass")
    print(f"\n{'row':<34} {'median ' + metric:>20} {'worst dev':>10}  verdict")
    print("-" * 72)
    failures = 0
    for key in sorted(keys):
        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 <= 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 +/-{tolerance:g}% — "
              f"the Phase-0 gate is not met.")
        return 1
    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())
@@ -52,12 +52,31 @@ target_link_libraries(kpn_tests PRIVATE
 add_executable(kpn_tests_stress test_channel_stress.cpp)
 target_link_libraries(kpn_tests_stress PRIVATE kpn Catch2::Catch2WithMain)
 # ── Wedge soak (PERF_PLAN G1) ─────────────────────────────────────────────────
 # Long-running end-to-end loop over the configurations that historically wedged.
 # Always built, so it cannot rot, but its CTest cases are registered only under
 # -DKPN_ENABLE_SOAK_TESTS=ON: they run for minutes and would otherwise dominate
 # every `ctest` invocation. Performance work runs it before and after a change:
 #
 #   cmake -B build -DKPN_ENABLE_SOAK_TESTS=ON -DKPN_SOAK_ITERS=50000
 #   cmake --build build --target kpn_soak_wedge
 #   ctest --test-dir build -L soak
 #
 # The binary self-diagnoses: an iteration that stops making progress trips a
 # watchdog that aborts naming the iteration and phase, rather than hanging.
 add_executable(kpn_soak_wedge soak_wedge.cpp)
 target_link_libraries(kpn_soak_wedge PRIVATE kpn)
 target_compile_options(kpn_soak_wedge PRIVATE -O2)
 option(KPN_ENABLE_SOAK_TESTS "Register the wedge soak cases with CTest" OFF)
 set(KPN_SOAK_ITERS 5000 CACHE STRING "Iterations per wedge soak case")
 # ── Sanitizer flags ───────────────────────────────────────────────────────────
 # kpn_sanitizer_flags() is defined in the top-level CMakeLists and is a no-op
 # unless -DKPN_SANITIZER=... is set. Sanitizer must be on both compile and link.
 kpn_sanitizer_flags(_kpn_san)
 if(_kpn_san)
-    foreach(_t kpn_tests kpn_tests_stress)
+    foreach(_t kpn_tests kpn_tests_stress kpn_soak_wedge)
        target_compile_options(${_t} PRIVATE ${_kpn_san})
        target_link_options(${_t}    PRIVATE ${_kpn_san})
    endforeach()
@@ -77,3 +96,16 @@ catch_discover_tests(kpn_tests DISCOVERY_MODE PRE_TEST)
 # Register the stress suite under its own label so CI can run / time it
 # separately from the fast unit tests.
 catch_discover_tests(kpn_tests_stress DISCOVERY_MODE PRE_TEST PROPERTIES LABELS "stress")
 if(KPN_ENABLE_SOAK_TESTS)
    # pool: the configuration the August wedges were reproduced on.
    add_test(NAME soak.wedge.pool
             COMMAND kpn_soak_wedge --mode=pool --depth=4 --threads=4
                     --items=1000 --work-us=10 --iters=${KPN_SOAK_ITERS})
    # private: one pool per node — the model workstream A would change.
    add_test(NAME soak.wedge.private
             COMMAND kpn_soak_wedge --mode=priv --depth=8
                     --items=1000 --work-us=10 --iters=${KPN_SOAK_ITERS})
    set_tests_properties(soak.wedge.pool soak.wedge.private PROPERTIES
                         LABELS "soak" TIMEOUT 3600)
 endif()
@@ -0,0 +1,236 @@
 // Wedge soak test (PERF_PLAN G1).
 //
 // Runs a pipeline configuration end-to-end in a loop and fails if any single
 // iteration stops making progress. Its purpose is to keep performance work
 // from silently reintroducing one of the wedges fixed in August 2026 — the
 // scheduler and channel wake paths are where both perf workstreams operate.
 //
 // Originally the minimal reproducer for the shared-pool chain wedge at
 // (chain, depth=4, work_us=10, pool_threads=4); the pre-6802328 code wedged
 // 5/5 within 45 s, at iterations 149, 1249, 332, 1740 and 493.
 //
 // A wedge is a hang, so a plain loop would hang CTest until its timeout with
 // no indication of where. The watchdog turns that into a failure naming the
 // iteration and the phase it stalled in.
 //
 // Usage: ./kpn_soak_wedge [options]
 //   --iters=5000        iterations to run
 //   --mode=pool|priv    shared ThreadPool(--threads), or one private pool/node
 //   --depth=4           chain depth
 //   --threads=4         shared pool size (--mode=pool only)
 //   --items=1000        items pushed per iteration
 //   --work-us=10        busy-work per node
 //   --watchdog-sec=30   per-iteration progress deadline
 #include <kpn/kpn.hpp>
 #include <atomic>
 #include <chrono>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <memory>
 #include <string>
 #include <thread>
 #include <vector>
 #if defined(__linux__)
 #include <sys/prctl.h>
 #endif
 using namespace kpn;
 using sclock = std::chrono::steady_clock;
 static std::atomic<int> g_work_us{10};
 static int chain_fn(int x) {
    int us = g_work_us.load(std::memory_order_relaxed);
    if (us > 0) {
        auto end = sclock::now() + std::chrono::microseconds(us);
        while (sclock::now() < end);
    }
    return x;
 }
 using ChainNode     = Node<chain_fn, in<>, out<>>;
 using PoolChainNode = PoolNode<chain_fn, in<>, out<>>;
 static void push_retry(Channel<int>& ch, int val) {
    while (true) {
        try   { ch.push(val); return; }
        catch (const ChannelOverflowError&) { std::this_thread::yield(); }
        catch (const ChannelClosedError&)   { return; }
    }
 }
 // ── watchdog ──────────────────────────────────────────────────────────────────
 //
 // The worker bumps g_progress at every phase boundary. The watchdog aborts if
 // it stops moving, so a wedge is reported as a failure at a known iteration
 // rather than as an unattributable CTest timeout.
 static std::atomic<unsigned long> g_progress{0};
 static std::atomic<int>           g_iter{0};
 static std::atomic<const char*>   g_phase{"init"};
 static std::atomic<bool>          g_done{false};
 static void mark(const char* phase) {
    g_phase.store(phase, std::memory_order_relaxed);
    g_progress.fetch_add(1, std::memory_order_release);
 }
 static void watchdog(double deadline_sec) {
    unsigned long last = g_progress.load(std::memory_order_acquire);
    auto last_move = sclock::now();
    while (!g_done.load(std::memory_order_acquire)) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        unsigned long now = g_progress.load(std::memory_order_acquire);
        if (now != last) { last = now; last_move = sclock::now(); continue; }
        double stalled = std::chrono::duration<double>(sclock::now() - last_move).count();
        if (stalled > deadline_sec) {
            std::fprintf(stderr,
                "\nWEDGE: no progress for %.0fs at iteration %d, phase '%s'\n",
                stalled, g_iter.load(std::memory_order_relaxed),
                g_phase.load(std::memory_order_relaxed));
            std::fflush(stderr);
            std::abort();   // core dump / stack trace at the point of the wedge
        }
    }
 }
 // ── one iteration ─────────────────────────────────────────────────────────────
 struct Opts {
    int    iters        = 5000;
    int    depth        = 4;
    int    threads      = 4;
    int    items        = 1000;
    int    work_us      = 10;
    bool   shared_pool  = true;
    double watchdog_sec = 30.0;
 };
 static void one_round_pool(const Opts& o) {
    const std::size_t CAP = static_cast<std::size_t>(o.items);
    auto pool = std::make_shared<ThreadPool>(o.threads);
    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= o.depth; ++i)
        chs.push_back(std::make_shared<Channel<int>>(CAP));
    std::vector<std::unique_ptr<PoolChainNode>> nodes;
    for (int i = 0; i < o.depth; ++i) {
        nodes.push_back(std::make_unique<PoolChainNode>(pool, CAP));
        nodes.back()->set_input_channel<0>(chs[i]);
        nodes.back()->set_output_channel<0>(chs[i + 1].get());
    }
    pool->start();
    for (auto& n : nodes) n->start();
    mark("started");
    std::thread reader([&] {
        for (int i = 0; i < o.items; ++i) chs.back()->pop();
    });
    std::thread pusher([&] {
        for (int i = 0; i < o.items; ++i) push_retry(*chs[0], i);
    });
    pusher.join();  mark("pushed");
    reader.join();  mark("drained");
    for (auto& n : nodes) n->stop();
    mark("nodes stopped");
    pool->stop();
    mark("pool stopped");
 }
 static void one_round_private(const Opts& o) {
    const std::size_t CAP = static_cast<std::size_t>(o.items);
    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= o.depth; ++i)
        chs.push_back(std::make_shared<Channel<int>>(CAP));
    std::vector<std::unique_ptr<ChainNode>> nodes;
    for (int i = 0; i < o.depth; ++i) {
        nodes.push_back(std::make_unique<ChainNode>(CAP));
        nodes.back()->set_input_channel<0>(chs[i]);
        nodes.back()->set_output_channel<0>(chs[i + 1].get());
    }
    for (auto& n : nodes) n->start();
    mark("started");
    std::thread reader([&] {
        for (int i = 0; i < o.items; ++i) chs.back()->pop();
    });
    std::thread pusher([&] {
        for (int i = 0; i < o.items; ++i) push_retry(*chs[0], i);
    });
    pusher.join();  mark("pushed");
    reader.join();  mark("drained");
    for (auto& n : nodes) n->stop();
    mark("nodes stopped");
 }
 // ── main ──────────────────────────────────────────────────────────────────────
 static void usage() {
    std::fprintf(stderr,
        "usage: kpn_soak_wedge [--iters=N] [--mode=pool|priv] [--depth=D]\n"
        "                      [--threads=T] [--items=N] [--work-us=U]\n"
        "                      [--watchdog-sec=S]\n");
 }
 int main(int argc, char** argv) {
 #if defined(__linux__)
    // Allow gdb to attach under ptrace_scope=1 when a wedge is caught.
    prctl(PR_SET_PTRACER, PR_SET_PTRACER_ANY, 0, 0, 0);
 #endif
    Opts o;
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        auto eq = a.find('=');
        std::string key = a.substr(0, eq);
        std::string val = eq == std::string::npos ? "" : a.substr(eq + 1);
        if      (key == "--iters")        o.iters        = std::atoi(val.c_str());
        else if (key == "--depth")        o.depth        = std::atoi(val.c_str());
        else if (key == "--threads")      o.threads      = std::atoi(val.c_str());
        else if (key == "--items")        o.items        = std::atoi(val.c_str());
        else if (key == "--work-us")      o.work_us      = std::atoi(val.c_str());
        else if (key == "--watchdog-sec") o.watchdog_sec = std::atof(val.c_str());
        else if (key == "--mode")         o.shared_pool  = (val != "priv");
        else { usage(); return 2; }
    }
    g_work_us.store(o.work_us, std::memory_order_relaxed);
    std::fprintf(stderr,
        "soak: mode=%s depth=%d threads=%d items=%d work_us=%d iters=%d watchdog=%.0fs\n",
        o.shared_pool ? "pool" : "priv", o.depth,
        o.shared_pool ? o.threads : o.depth, o.items, o.work_us,
        o.iters, o.watchdog_sec);
    std::thread wd(watchdog, o.watchdog_sec);
    const auto t0 = sclock::now();
    for (int i = 0; i < o.iters; ++i) {
        g_iter.store(i, std::memory_order_relaxed);
        if (o.shared_pool) one_round_pool(o);
        else               one_round_private(o);
        if ((i + 1) % 100 == 0) {
            std::fprintf(stderr, "\r  %d/%d", i + 1, o.iters);
            std::fflush(stderr);
        }
    }
    g_done.store(true, std::memory_order_release);
    wd.join();
    double secs = std::chrono::duration<double>(sclock::now() - t0).count();
    std::fprintf(stderr, "\ncompleted %d iterations in %.1fs with no wedge\n",
                 o.iters, secs);
    return 0;
 }
@@ -153,11 +153,18 @@ TEST_CASE("SPSC: producer racing a disable() never throws and never hangs",
    }
 }
-TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
+TEST_CASE("SPSC: push_callback fires for every push, never missed",
          "[channel][stress]") {
-    // The empty->non-empty callback ([channel.hpp] was_empty branch) is read by
+    // Regression: this callback is the *only* thing that wakes a PoolNode, and
-    // the consumer-side notification path. Run it under contention to make sure
+    // it used to fire only on the empty->non-empty edge, computed from a head_
-    // the was_empty detection isn't torn by a concurrent pop().
+    // sampled before the item was published. A concurrent pop() could drain the
    // ring to empty in that window, so neither side saw the other: the item sat
    // in the ring with the consumer idle, and because the trigger was an edge it
    // never recovered. See set_push_callback in channel.hpp.
    //
    // The old version of this test asserted only `1 <= callbacks <= N`, which a
    // *missed* callback satisfies — it named the hazard and could not detect it.
    // One callback per successful push is the contract, so assert exactly that.
    Channel<int> ch(/*capacity=*/4, /*spin_count=*/4);
    std::atomic<int> callbacks{0};
    ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); });
@@ -175,10 +182,9 @@ TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
    for (int i = 0; i < N; ++i) (void)ch.pop();
    producer.join();
-    // At least one transition, at most one per item; mainly we assert the run
+    // Exactly one callback per successful push. Fewer means a wake was dropped,
-    // completed without TSan flagging a race on push_callback_/was_empty.
+    // which is the bug; more would mean a spurious wake was manufactured.
-    REQUIRE(callbacks.load() >= 1);
+    REQUIRE(callbacks.load() == N);
    REQUIRE(callbacks.load() <= N);
 }
 // Ordering contract of the out-of-band sentinel under contention.
Author	SHA1	Message	Date
dtourolle	5375ab41b2	Merge branch 'perf/phase0-harness' 🚦 CI / changes (push) Successful in 7s Details 🚦 CI / docker (push) Skipped Details 🚦 CI / docs (push) Skipped Details 🧪 Test / test (push) Successful in 8m18s Details 🚦 CI / test (push) Successful in 8m18s Details 🧵 ThreadSanitizer / tsan (push) Successful in 3m33s Details 🚦 CI / tsan (push) Successful in 3m34s Details 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.	2026-08-08 22:27:55 +02:00
dtourolleandClaude Opus 5	b500570c47	perf(pool): submit to the calling worker's own queue, and skip a notify nobody is waiting for B9 from PERF_PLAN, plus the notify gate. Two changes to submit(), both aimed at the same cost: on any pool of two or more threads, every single dispatch paid a futex wake. submit() round-robins, so a worker resubmitting -- which is what fire_once does on every token -- handed the task to a different worker, and that worker was asleep. bench_dispatch measured it as 182 ns/dispatch on ThreadPool(1) against 3197 ns on 20 threads, with voluntary context switches per task rising 0.00 -> 1.19 in step: the 1-thread pool is fast precisely because it resubmits into its own queue and finds the work already there. A submission originating on one of the pool's own workers now goes to that worker's queue, extending that property to any pool size. try_steal still corrects the imbalance. The identity check is against `this`, not merely "am I a pool worker": a worker of pool A submitting into pool B must not use A's index, which may exceed B's thread_count_. Nested networks do exactly this. tls_pool is a non-owning identity tag, only ever compared, never dereferenced -- its lifetime is strictly nested inside the pool's, since stop() joins every worker before clearing queues_. The notify gate skips the cv_mx_ round-trip and notify_one() when waiters_ is zero. waiters_ is maintained under cv_mx_ and incremented before the predicate is evaluated, so reading zero in submit() means no worker can be in wait() -- as opposed to reading zero because we raced one, which the mutex prevents. stop()'s notify_all() is deliberately left ungated. Shared pools, work_us=10, items/sec: chain-1 59512 -> 66662 (+12.0%), wide-4 53698 -> 61705 (+14.9%), chain-8 +6.8%, chain-32 +3.7%. Private pools -- the Node<> default -- are unchanged at -1.3% to +3.6%, inside the gate's tolerance. Two things tried and removed, recorded in comments so they are not retried: Raising the steal threshold to >1, to stop a thief winning the race for a self-submitted task, DEADLOCKS. An external submit() round-robins a single task onto an idle worker's queue; if that worker is parked, no peer will take it, because a queue of one is no longer stealable. latency mode hangs at 12 and 20 threads. It was also 2x slower in steady state, 2229 -> 4546 ns. B5, bounded spin before parking, does not pay: swept at 50/200/1000 rounds, 2123 / 2230 / 2574 ns against 2229 ns without it, with vcsw/task flat at ~0.97. The spin cannot catch what it targets, because a peer is woken the moment queued_ becomes non-zero -- before this worker reaches the spin at all. Guardrails: 150/150 ctest including soak and examples, and ThreadSanitizer clean over the full suite (128 cases, 302 assertions). That also pins the reference count PERF_PLAN section 6 flagged as uncertain: it is 150, not 146 or 136. Caveat: the throughput figures above were taken on a build that also carried the since-removed spin experiment. The scheduler logic is identical to this tree and correctness was re-verified on it, but the numbers are one build stale and predate the 7-pass gate, which has not been run on this change. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-08 22:26:14 +02:00
dtourolleandClaude Opus 5	73828bcffe	perf(bench): make a multi-hour acceptance run observable and restartable The Phase-0 gate is 7 passes over the full row set, which is long enough that capture_output=True was the wrong default: no rows existed anywhere until a pass ended, so a slow run and a wedged one looked identical, and killing either threw away everything measured. Rows now stream to --out-dir/pass-NN.csv with a flush per line, so an interrupted run keeps what it had. --resume reuses complete pass files and reruns incomplete ones -- checked against the row count rather than merely existing, because a partial file silently averaged in as a whole pass would corrupt the verdict rather than fail it. Progress is a tqdm bar over rows, not passes; a pass counter would sit at 1/7 for twenty minutes and report nothing useful. The row total comes from probing the binary with --reps=0 (0.8s) rather than reimplementing the sweep in Python, which would drift from the C++ defaults. There is a plain-stderr fallback when tqdm is absent -- refusing to start a benchmark over a missing progress dependency is the wrong trade. bench_runs/ is gitignored: the new default output path would otherwise land in git status. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-08 22:25:00 +02:00
dtourolleandClaude Opus 5	a3f61fcb3c	perf: make the benchmark able to answer the question, then ask it PERF_PLAN phase 0, plus B1/B2 which turned out to cost seconds rather than the minutes budgeted for them. No library code is touched. The harness could not support the conclusions drawn from it. items_for() shrank the sample as work per item grew, so exactly the rows under investigation -- chain-16 and chain-32 -- ran 50 to 200 items and swung 4-8x between passes. Sample size now derives from a time budget with a floor, using work_us * stages / units as the per-item cost. The old ladder's error was treating depth as a throughput cost: past the core count it is, below it depth costs only latency. Rows now report median of N repetitions after a discarded warm-up, with IQR and range, so an unreliable row says so instead of being averaged into a table. The CSV header records nproc, governor and AC state, which immediately caught this laptop running on battery under powersave. A3 needed no experiment in the end: ru_nivcsw and ru_nvcsw are captured around every timed region and reported per item, so involuntary switches against depth is a column rather than a run. bench_dispatch answers B1 and B2 without instrumenting the scheduler. Sleeping is inferred from ru_nvcsw, since a thread blocking on a condition variable books a voluntary context switch. B1: a ThreadPool(1) dispatch is 291 ns null, 466 ns with a payload, against the ~290 ns the plan estimated -- so the abandon criterion is not met and workstream B stays alive. B2's answer is not the one the question expected. It is not whether workers sleep but which pool: on a private ThreadPool(1) the worker never sleeps, because it resubmits into its own queue and finds the work already there; on any pool of two or more it sleeps exactly once per task, because submit() round-robins to a different worker, which is asleep. That is the whole 466 ns to 1.7 us difference, and it inverts half the plan. B5 (bounded spin) buys nothing in the default configuration, and A5 must not make a shared pool the default until the wake cost is fixed, or every graph that already fits its cores gets 3-4x worse per dispatch. G1 lands as tests/soak_wedge.cpp, superseding benchmarks/repro_wedge.cpp, which was never wired into any build. Always compiled so it cannot rot; its CTest cases register only under -DKPN_ENABLE_SOAK_TESTS=ON, so the default test count is unchanged. A wedge is a hang, and a hang under CTest is an unattributable timeout, so it carries a watchdog that aborts naming the iteration and phase. Phase 0's gate is not yet cleared: the acceptance run belongs on the reference machine, not here. A 3-pass check lands every row within 0.7% against the 4-8x swings described above, which is encouraging and is not the same thing. Provisional, recorded so it can be checked: chain-16 came out 6% behind TBB rather than 28.5%. If that survives a proper run, the deep-chain deficit is substantially an artefact of the N=200 rows. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-08 12:13:11 +02:00
dtourolle	771b9f8593	fix: ThreadPool::start must take the lifecycle lock too 🚦 CI / changes (push) Successful in 4s Details 🚦 CI / docker (push) Has been skipped Details 🚦 CI / test (push) Successful in 5m0s Details 🚦 CI / tsan (push) Successful in 3m39s Details 🚦 CI / docs (push) Has been skipped Details `abbb2d4` guarded submit() against stop() with a shared/exclusive lock, because stop() ends with queues_.clear() and submit() indexes queues_. It missed the other writer: start() rebuilds the same vector and took no lock at all. A submission can genuinely land while a pool is inside start(). A network starts its nodes one at a time, and a node already started fires into the next one's channel, whose push callback submits. ThreadSanitizer reports it as the read at scheduler.hpp:114 against the write at :59, and the consequence is worse than a torn read: push_back can reallocate the vector under a reader that has already indexed it. Surfaced by `6802328`. Firing push_callback_ unconditionally is correct — the argument in that commit holds — and it makes callbacks frequent enough during startup to hit this window. It went from unobserved to 4 races in one run of the unit suite. Holding the lock across the thread spawn is safe: all queues are constructed before any worker starts, and worker_loop never takes the lifecycle lock, so there is nothing for it to deadlock against. Verified with -DKPN_SANITIZER=thread: 4 races in one run of five before, 0 across five unit runs and two stress runs after. 148/148.	2026-08-06 22:53:02 +02:00
dtourolle	3b67b7e1e9	Merge origin/master into master 🚦 CI / changes (push) Successful in 4s Details 🚦 CI / docker (push) Has been skipped Details 🚦 CI / test (push) Successful in 5m1s Details 🚦 CI / tsan (push) Successful in 3m38s Details 🚦 CI / docs (push) Has been skipped Details	2026-08-06 21:57:32 +02:00
dtourolle	433c3b3859	Merge branch 'fix/channel-push-callback-lost-wake' A push must wake its consumer even when the ring looked non-empty. The edge-triggered push_callback_ was computed from a head_ sampled before the item was published, so a concurrent pop could leave a PoolNode idle holding work — permanently, since the edge never fired again. Validated: 136/136 tests pass; the retargeted stress test fails at 1325/10000 against the old code. ~85k reproducer iterations and 10 consecutive full benchmark passes with no wedge, against 5/5 wedges before the fix.	2026-08-06 21:57:15 +02:00
dtourolleandClaude Opus 5	6802328e97	fix: a push must wake its consumer, even when the ring looked non-empty push(), try_push() and push_blocking() fired push_callback_ only on the empty->non-empty edge, and computed that edge from a head_ sampled before the item was published. A PoolNode consumer decides whether to run again from the level (count_ready -> approx_size), so a pop landing in that window left both sides standing down: producer (push) consumer (PoolNode firing) ------------------------ ---------------------------- samples t=782, h=781 -> was_empty = false, no wake pops idx 781, head_ = 782 count_ready(): head_==tail_==782 -> not ready, gate released to Idle tail_.store(783) The item is in the ring, the node is idle, and no wake is outstanding. The failure is absorbing: every later push then sees a non-empty ring, so the edge never fires again and the node sleeps while its backlog grows. Observed as a hang in bench_pipeline at (chain, depth=4, work_us=10, shared pool): all pool workers asleep in worker_loop, the reader blocked in pop(), and 218 items stranded in one channel with head_ stopped at exactly the index where the edge was dropped. Re-reading head_ after the tail_ store does not fix this. That is the store-buffer pattern, and under acquire/release both sides may legally read stale; forbidding it needs seq_cst on the producer's tail_ store and head_ load and on the consumer's head_ store and tail_ load, a fence on both hot paths. Firing unconditionally is correct by construction: the callback runs after the publishing store, so a consumer that observes the level at all observes the item. The redundant wakes are cheap -- on_input_ready re-checks the level and SubmitGate::claim() collapses a wake arriving mid-firing into the firing already in flight. The stress test named this exact hazard and could not detect it: it asserted only 1 <= callbacks <= N, which a missed callback satisfies. It now requires one callback per successful push, and fails at 1325/10000 against the old code. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-06 20:50:34 +02:00