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; +}