From a3f61fcb3cf104f2c9c2caf853e1399795d215e6 Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Sat, 8 Aug 2026 12:13:11 +0200 Subject: [PATCH] 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 --- .gitignore | 2 +- PERF_PLAN.md | 413 +++++++++++++++++++++++++++++ benchmarks/CMakeLists.txt | 7 + benchmarks/bench_dispatch.cpp | 307 ++++++++++++++++++++++ benchmarks/bench_env.hpp | 108 ++++++++ benchmarks/bench_pipeline.cpp | 482 +++++++++++++++++++++++++--------- scripts/bench_repro_check.py | 122 +++++++++ tests/CMakeLists.txt | 34 ++- tests/soak_wedge.cpp | 236 +++++++++++++++++ 9 files changed, 1587 insertions(+), 124 deletions(-) create mode 100644 PERF_PLAN.md create mode 100644 benchmarks/bench_dispatch.cpp create mode 100644 benchmarks/bench_env.hpp create mode 100755 scripts/bench_repro_check.py create mode 100644 tests/soak_wedge.cpp diff --git a/.gitignore b/.gitignore index d3fb3e0..2d1b7e4 100644 --- a/.gitignore +++ b/.gitignore @@ -29,4 +29,4 @@ Thumbs.db .claude/settings.local.json include/kpn/ort_cache/ build-tsan/ -build-*/ +build*/ diff --git a/PERF_PLAN.md b/PERF_PLAN.md new file mode 100644 index 0000000..28560c4 --- /dev/null +++ b/PERF_PLAN.md @@ -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` 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. diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt index 97ef6f4..8a104fa 100644 --- a/benchmarks/CMakeLists.txt +++ b/benchmarks/CMakeLists.txt @@ -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) diff --git a/benchmarks/bench_dispatch.cpp b/benchmarks/bench_dispatch.cpp new file mode 100644 index 0000000..c159634 --- /dev/null +++ b/benchmarks/bench_dispatch.cpp @@ -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 + +#include "bench_env.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace kpn; +using sclock = std::chrono::steady_clock; + +struct Opts { + std::vector threads {1, 2, 4}; + std::vector 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(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(vc) / tasks / 2.0; + s.ivcsw_per_task = static_cast(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 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(t1 - t0).count(); + s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0; + s.submit_ns = std::chrono::duration(t_submitted - t0).count() / tasks; + s.vcsw_per_task = static_cast(vc) / tasks; + s.ivcsw_per_task = static_cast(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 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 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(t1 - t0).count(); + s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0; + s.vcsw_per_task = static_cast(vc) / tasks; + s.ivcsw_per_task = static_cast(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 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 parse_int_list(const char* s) { + std::vector 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(v)); + p = end; + while (*p == ',' || *p == ' ') ++p; + } + return out; +} + +static std::vector parse_word_list(const std::string& s) { + std::vector 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); + } +} diff --git a/benchmarks/bench_env.hpp b/benchmarks/bench_env.hpp new file mode 100644 index 0000000..92ccc4a --- /dev/null +++ b/benchmarks/bench_env.hpp @@ -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 +#include +#include +#include +#include +#include + +#include + +namespace bench { + +inline int hw_units() { + unsigned n = std::thread::hardware_concurrency(); + return n ? static_cast(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 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::floor(idx)); + auto hi = static_cast(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 diff --git a/benchmarks/bench_pipeline.cpp b/benchmarks/bench_pipeline.cpp index c0e3b09..dc4ed3a 100644 --- a/benchmarks/bench_pipeline.cpp +++ b/benchmarks/bench_pipeline.cpp @@ -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 +#include "bench_env.hpp" + #ifdef KPN_BENCH_TBB #include namespace tbb_flow = oneapi::tbb::flow; #endif +#include #include #include #include +#include #include +#include +#include #include #include #include #include +#include + using namespace kpn; using namespace std::chrono_literals; using sclock = std::chrono::steady_clock; @@ -57,31 +71,63 @@ static void push_retry(Channel& ch, int val) { } } -// ── result ──────────────────────────────────────────────────────────────────── +// ── configuration (M1, M3, M4, M5) ──────────────────────────────────────────── -struct Result { - const char* topology; - int size; - int work_us; - int threads; // 0 = private (1 thread per node), N = shared pool size - double items_per_sec; - double overhead_us; +struct Config { + std::vector work_amts {10, 100, 1000}; + std::vector pool_sizes{1, 2, 4, 8, 16, 20}; // M5 + std::vector depths {1, 2, 4, 8, 16, 32}; + std::vector widths {1, 2, 3, 4}; + int reps = 5; // M3: measured repetitions per row + 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(work_us)) * + std::max(1.0, static_cast(stages) / units); + + long want = static_cast(g_cfg.target_sec * 1e6 / per_item_us); + long cap = static_cast(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) { - int effective = std::max(1, work_us) * std::max(1, depth); - 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; +static Sample bench_chain(int depth, int work_us, long N) { + const std::size_t CAP = static_cast(N); std::vector>> chs; for (int i = 0; i <= depth; ++i) @@ -98,17 +144,20 @@ static Result bench_chain(int depth, int work_us) { std::atomic 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(i)); }); pusher.join(); reader.join(); + Sample s; + ru.finish(s.nivcsw, s.nvcsw); for (auto& n : nodes) n->stop(); double elapsed = std::chrono::duration( @@ -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(work_us) * (N + depth - 1); - double wus = (elapsed * 1e6 - pipeline_us) / N; - return {"chain", depth, work_us, 0, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6 - pipeline_us) / N; + s.items_per_sec = N / elapsed; + return s; } -static Result bench_chain_pool(int depth, int work_us, int pool_threads) { - const int N = items_for(work_us, depth); - const int CAP = N; +static Sample bench_chain_pool(int depth, int work_us, int pool_threads, long N) { + const std::size_t CAP = static_cast(N); auto pool = std::make_shared(pool_threads); @@ -142,33 +191,36 @@ static Result bench_chain_pool(int depth, int work_us, int pool_threads) { std::atomic 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(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( t1.load(std::memory_order_acquire) - t0).count(); double pipeline_us = static_cast(work_us) * (N + depth - 1); - double wus = (elapsed * 1e6 - pipeline_us) / N; - return {"chain", depth, work_us, pool_threads, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6 - pipeline_us) / N; + s.items_per_sec = N / elapsed; + return s; } // ── wide (fanout) ────────────────────────────────────────────────────────── template -static Result bench_wide(int work_us) { - const int N = items_for(work_us); - const int CAP = N; +static Sample bench_wide(int work_us, long N) { + const std::size_t CAP = static_cast(N); auto src_ch = std::make_shared>(CAP); auto fan = std::make_unique>(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(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(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( t1.load(std::memory_order_acquire) - t0).count(); - double wus = (elapsed * 1e6) / N - static_cast(work_us); - return {"wide", static_cast(W), work_us, 0, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6) / N - static_cast(work_us); + s.items_per_sec = N / elapsed; + return s; } template -static Result bench_wide_pool(int work_us, int pool_threads) { - const int N = items_for(work_us); - const int CAP = N; +static Sample bench_wide_pool(int work_us, int pool_threads, long N) { + const std::size_t CAP = static_cast(N); auto pool = std::make_shared(pool_threads); auto src_ch = std::make_shared>(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(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(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( t1.load(std::memory_order_acquire) - t0).count(); - double wus = (elapsed * 1e6) / N - static_cast(work_us); - return {"wide", static_cast(W), work_us, pool_threads, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6) / N - static_cast(work_us); + s.items_per_sec = N / elapsed; + return s; } // ── diamond ─────────────────────────────────────────────────────────────────── -static Result bench_diamond(int work_us) { - const int N = items_for(work_us, 2); - const int CAP = N; +static Sample bench_diamond(int work_us, long N) { + const std::size_t CAP = static_cast(N); auto src_ch = std::make_shared>(CAP); auto fan = std::make_unique>(CAP); @@ -312,7 +370,7 @@ static Result bench_diamond(int work_us) { std::atomic done{0}; auto make_reader = [&](Channel& 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(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( t1.load(std::memory_order_acquire) - t0).count(); - double wus = (elapsed * 1e6) / N - static_cast(work_us); - return {"diamond", 4, work_us, 0, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6) / N - static_cast(work_us); + s.items_per_sec = N / elapsed; + return s; } -static Result bench_diamond_pool(int work_us, int pool_threads) { - const int N = items_for(work_us, 2); - const int CAP = N; +static Sample bench_diamond_pool(int work_us, int pool_threads, long N) { + const std::size_t CAP = static_cast(N); auto pool = std::make_shared(pool_threads); auto src_ch = std::make_shared>(CAP); @@ -369,7 +430,7 @@ static Result bench_diamond_pool(int work_us, int pool_threads) { std::atomic done{0}; auto make_reader = [&](Channel& 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(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( t1.load(std::memory_order_acquire) - t0).count(); - double wus = (elapsed * 1e6) / N - static_cast(work_us); - return {"diamond", 4, work_us, pool_threads, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6) / N - static_cast(work_us); + 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) { - const int N = items_for(work_us, depth); - +static Sample bench_chain_tbb(int depth, int work_us, long N) { tbb_flow::graph g; using FN = tbb_flow::function_node; std::vector> 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(i)); g.wait_for_all(); auto t1 = sclock::now(); + Sample s; + ru.finish(s.nivcsw, s.nvcsw); - double elapsed = std::chrono::duration(t1 - t0).count(); + double elapsed = std::chrono::duration(t1 - t0).count(); double pipeline_us = static_cast(work_us) * (N + depth - 1); - double wus = (elapsed * 1e6 - pipeline_us) / N; - return {"chain_tbb", depth, work_us, -1, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6 - pipeline_us) / N; + s.items_per_sec = N / elapsed; + return s; } template -static Result bench_wide_tbb(int work_us) { - const int N = items_for(work_us); - +static Sample bench_wide_tbb(int work_us, long N) { tbb_flow::graph g; tbb_flow::broadcast_node fan(g); using FN = tbb_flow::function_node; @@ -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(i)); g.wait_for_all(); auto t1 = sclock::now(); + Sample s; + ru.finish(s.nivcsw, s.nvcsw); double elapsed = std::chrono::duration(t1 - t0).count(); - double wus = (elapsed * 1e6) / N - static_cast(work_us); - return {"wide_tbb", static_cast(W), work_us, -1, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6) / N - static_cast(work_us); + s.items_per_sec = N / elapsed; + return s; } -static Result bench_diamond_tbb(int work_us) { - const int N = items_for(work_us, 2); - +static Sample bench_diamond_tbb(int work_us, long N) { tbb_flow::graph g; tbb_flow::broadcast_node fan(g); using FN = tbb_flow::function_node; @@ -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(i)); g.wait_for_all(); auto t1 = sclock::now(); + Sample s; + ru.finish(s.nivcsw, s.nvcsw); double elapsed = std::chrono::duration(t1 - t0).count(); - double wus = (elapsed * 1e6) / N - static_cast(work_us); - return {"diamond_tbb", 4, work_us, -1, N / elapsed, wus}; + s.overhead_us = (elapsed * 1e6) / N - static_cast(work_us); + 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 +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 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(ivcsw) / (double(N) * g_cfg.reps); + const double vcsw_per_item = static_cast(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 parse_int_list(const char* s) { + std::vector 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(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 +static void with_width(int w, F&& f) { + switch (w) { + case 1: f(std::integral_constant{}); break; + case 2: f(std::integral_constant{}); break; + case 3: f(std::integral_constant{}); break; + case 4: f(std::integral_constant{}); break; + default: + std::fprintf(stderr, "width %d not instantiated (1..4 only)\n", w); + } +} + // ── main ────────────────────────────────────────────────────────────────────── -int main() { - const int work_amts[] = {10, 100, 1000}; - const int pool_sizes[] = {1, 2, 4}; +int main(int argc, char** argv) { + // A rejected option must fail loudly: a harness driver that silently got + // 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", - "topology", "size", "work_us", "threads", "items/sec", "overhead_us/item"); - std::fprintf(stderr, "%s\n", std::string(78, '-').c_str()); - std::printf("topology,size,work_us,threads,items_per_sec,overhead_us_per_item\n"); + char cfg[192]; + std::snprintf(cfg, sizeof cfg, + "reps=%d warmup=%d target_sec=%.2f min_items=%ld max_sec=%.1f", + 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::string sched = r.threads < 0 ? "tbb" - : r.threads == 0 ? "priv" - : std::to_string(r.threads); - std::fprintf(stderr, "%-12s %-8d %-10d %-8s %-18.0f %-20.1f\n", - r.topology, r.size, r.work_us, sched.c_str(), - r.items_per_sec, r.overhead_us); - 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); - }; + std::fprintf(stderr, "\n%-10s %-5s %-8s %-6s %-8s %-12s %-7s %-7s %-9s %-8s %-8s\n", + "topology", "size", "work_us", "sched", "items", "items/sec", + "iqr%", "range%", "ovh_us", "ivcsw/it", "vcsw/it"); + std::fprintf(stderr, "%s\n", std::string(104, '-').c_str()); + std::printf("topology,size,work_us,threads,items,reps,items_per_sec," + "items_per_sec_min,items_per_sec_max,iqr_pct,range_pct," + "overhead_us_per_item,ivcsw_per_item,vcsw_per_item\n"); - 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)); - emit(bench_wide<1>(w)); - emit(bench_wide<2>(w)); - emit(bench_wide<3>(w)); - emit(bench_wide<4>(w)); - emit(bench_diamond(w)); + if (g_cfg.do_priv) { + std::fprintf(stderr, "\n── work_us=%-4d private pools ──────────────────────\n", w); + if (g_cfg.do_chain) + for (int d : g_cfg.depths) { + long N = pick_items(w, d, d); + 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(W.value), w, 0, N, + [&] { return bench_wide(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) { - std::fprintf(stderr, "\n── work_us=%-4d shared pool (%d thread%s) ─────────────────────────────\n", - w, pt, pt == 1 ? "" : "s"); - for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain_pool(d, w, pt)); - emit(bench_wide_pool<1>(w, pt)); - emit(bench_wide_pool<2>(w, pt)); - emit(bench_wide_pool<3>(w, pt)); - emit(bench_wide_pool<4>(w, pt)); - emit(bench_diamond_pool(w, pt)); + if (g_cfg.do_pool) { + for (int pt : g_cfg.pool_sizes) { + std::fprintf(stderr, "\n── work_us=%-4d shared pool (%d thread%s) ───────────\n", + w, pt, pt == 1 ? "" : "s"); + if (g_cfg.do_chain) + for (int d : g_cfg.depths) { + long N = pick_items(w, d, pt); + run_row("chain", d, w, pt, N, + [&] { 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(W.value), w, pt, N, + [&] { return bench_wide_pool(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); - for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain_tbb(d, w)); - emit(bench_wide_tbb<1>(w)); - emit(bench_wide_tbb<2>(w)); - emit(bench_wide_tbb<3>(w)); - emit(bench_wide_tbb<4>(w)); - emit(bench_diamond_tbb(w)); + if (g_cfg.do_tbb) { + std::fprintf(stderr, "\n── work_us=%-4d TBB flow graph ─────────────────────\n", w); + if (g_cfg.do_chain) + for (int d : g_cfg.depths) { + long N = pick_items(w, d, d); + run_row("chain_tbb", d, w, -1, N, + [&] { 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(W.value), w, -1, N, + [&] { return bench_wide_tbb(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 } } diff --git a/scripts/bench_repro_check.py b/scripts/bench_repro_check.py new file mode 100755 index 0000000..8c21d0b --- /dev/null +++ b/scripts/bench_repro_check.py @@ -0,0 +1,122 @@ +#!/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. + +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 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 parse_csv(text, metric=None): + """Return ({(topology, size, work_us, threads): value}, metric_name).""" + rows = {} + header = None + for line in text.splitlines(): + 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 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)") + + # Everything after a standalone `--` goes to bench_pipeline verbatim. + # argparse.REMAINDER would swallow this script's own flags instead. + argv = sys.argv[1:] + extra = [] + if "--" in argv: + cut = argv.index("--") + argv, extra = argv[:cut], argv[cut + 1:] + args = ap.parse_args(argv) + + passes = [] + metric = args.metric + for i in range(args.passes): + print(f"pass {i + 1}/{args.passes} ...", file=sys.stderr, flush=True) + proc = subprocess.run([args.binary] + extra, + capture_output=True, text=True) + if proc.returncode != 0: + print(proc.stderr, file=sys.stderr) + sys.exit(f"{args.binary} failed with {proc.returncode}") + rows, metric = parse_csv(proc.stdout, metric) + passes.append(rows) + + 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 <= args.tolerance + failures += not ok + label = "{}-{} w={} s={}".format(*key) + print(f"{label:<34} {med:>20.1f} {worst:>9.1f}% {'ok' if ok else 'NOISY'}") + + print("-" * 72) + if failures: + print(f"{failures}/{len(keys)} rows exceed +/-{args.tolerance:g}% — " + f"the Phase-0 gate is not met.") + return 1 + print(f"all {len(keys)} rows within +/-{args.tolerance:g}% " + f"over {args.passes} passes — Phase-0 gate met.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index f567afc..07bf131 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -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() diff --git a/tests/soak_wedge.cpp b/tests/soak_wedge.cpp new file mode 100644 index 0000000..033402e --- /dev/null +++ b/tests/soak_wedge.cpp @@ -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 + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if defined(__linux__) +#include +#endif + +using namespace kpn; +using sclock = std::chrono::steady_clock; + +static std::atomic 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, out<>>; +using PoolChainNode = PoolNode, out<>>; + +static void push_retry(Channel& 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 g_progress{0}; +static std::atomic g_iter{0}; +static std::atomic g_phase{"init"}; +static std::atomic 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(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(o.items); + auto pool = std::make_shared(o.threads); + + std::vector>> chs; + for (int i = 0; i <= o.depth; ++i) + chs.push_back(std::make_shared>(CAP)); + + std::vector> nodes; + for (int i = 0; i < o.depth; ++i) { + nodes.push_back(std::make_unique(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(o.items); + + std::vector>> chs; + for (int i = 0; i <= o.depth; ++i) + chs.push_back(std::make_shared>(CAP)); + + std::vector> nodes; + for (int i = 0; i < o.depth; ++i) { + nodes.push_back(std::make_unique(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(sclock::now() - t0).count(); + std::fprintf(stderr, "\ncompleted %d iterations in %.1fs with no wedge\n", + o.iters, secs); + return 0; +}