From a5ee3c05ce392202765f92861a71e6b8bc102b17 Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Wed, 5 Aug 2026 14:38:15 +0200 Subject: [PATCH] feat(benchmark): per-node cost and bottleneck attribution for a run --benchmark reports cumulative CPU and wall time per node and names the node pacing the run. The pacing node is located from sampled channel occupancy, not from time-in-node: backpressure inflates time-in-node for everything downstream of the real bottleneck, so the obvious measure names the victim rather than the cause. Sampling starts with the network and stops before it is destroyed. Channel fill is instantaneous and everything has drained by shutdown, so a single read at the end reports an idle pipeline however congested it was. kill -USR1 dumps the table from a running or wedged process. Channel occupancy identifies a stalled node -- full input, empty output -- without a debug build or a debugger, which is the difference between diagnosing the AR-004 hang in seconds and reproducing it under gdb. Two knobs this exposes for measurement rather than sets: SAE_CV_THREADS, because OpenCV's TBB arena and KPN's thread-per-node are two schedulers unaware of each other on the same cores; and SAE_CUDA_BLOCKING_SYNC, because the default spin-wait held the embedder thread at 99.7% user time while nvidia-powerd cut the GPU's clock from 1005 to 210 MHz. Neither default changes until a measurement says it should. TRACES: VR-015 | PR-004 --- docs/benchmark.md | 164 +++++++++- src/backends/trt_backend.cpp | 38 +++ src/benchmark.hpp | 590 +++++++++++++++++++++++++++++++++++ src/config.hpp | 8 + src/main.cpp | 107 ++++++- tests/CMakeLists.txt | 5 + tests/test_benchmark.cpp | 221 +++++++++++++ 7 files changed, 1128 insertions(+), 5 deletions(-) create mode 100644 src/benchmark.hpp create mode 100644 tests/test_benchmark.cpp diff --git a/docs/benchmark.md b/docs/benchmark.md index 3beb8b8..17111bc 100644 --- a/docs/benchmark.md +++ b/docs/benchmark.md @@ -124,11 +124,169 @@ bleeding across a boundary may be the join rather than a tracking fault. | Path | Realtime factor | Sampled fps | 17-min film | |---|---|---|---| -| `build/` (TensorRT) | **2.0×** | ~10 | ~8 min | +| `build/` (TensorRT) | **8.25×** | 41.3 | **2.1 min** | | `build-ort/` (ORT) | 0.54× | 2.7 | ~32 min | -Throughput varies strongly with face density; a sparse stretch measured 8× -realtime, so quote the whole-film average, not a window. +TensorRT figure re-measured 2026-08-04 over the whole film at `--fps 5 +--min-face-px 32 --expand-gallery`: 5129 frames, 1025.4 s of film in 124.2 s +wall. Two runs agreed to 0.4% (124.2 s clean, 124.7 s under gdb). It supersedes +an earlier 2.0×; that figure predates the current tree and was not re-derived +here, so treat the gain as measured rather than explained. + +Throughput varies strongly with face density, and **a short window is not a +sample of the film**. The opening 60 s benchmarks at 23.9× — decode there costs +4-6 ms/frame against a 12.35 ms whole-film mean (n=510), because seeking forward +in VP8/WebM gets dearer the deeper you go, and there are few faces. Always quote +the whole-film average. + +### Where the time goes (VR-015) + +Measured over the whole film, 2026-08-04: + +| node | cpu_s | % of pipeline CPU | cpu/f | exec/f | stall/f | in% | out% | +|---|---|---|---|---|---|---|---| +| **embedder** | **91.0** | **60%** | 17.74 | 21.61 | 3.87 | 12 | 0 | +| **face_detector** ▶ | 41.4 | 27% | 8.07 | 24.20 | **16.14** | **99** | **0** | +| frame_source | 12.1 | 8% | 2.36 | 11.26 | 8.90 | — | 97 | +| camera_pos | 3.2 | 2% | 0.63 | 0.64 | 0.01 | 97 | 99 | +| face_aligner | 1.7 | 1% | 0.33 | 0.34 | 0.01 | 0 | 12 | +| identity_matcher | 1.3 | 1% | 0.26 | 0.34 | 0.09 | 0 | 0 | +| tracker / sink | 0.5 | <1% | — | — | — | 0 | 0 | + +**`face_detector` paces the run**: its input channel is 97.8% full while its +output is 99.4% empty — everything upstream jammed, everything downstream +starved. It occupies 5129 × 24.20 ms ≈ 124.1 s of a 124.2 s run, essentially +100% wall occupancy, yet only 33% of that is CPU. The other 16.14 ms/frame is +device wait. + +**The embedder is the larger cost but not the constraint**: 60% of all pipeline +CPU, 73% of wall as thread-busy. Whether that is real work or a spinning +`cudaStreamSynchronize` is unresolved — see the sync caveat below, which is a +one-line experiment. + +**`frame_source` is the trap this table exists to defuse.** It reports +`exec/f = 11.26 ms` against `cpu/f = 2.36 ms`, and its output channel is 97% +full: it is backpressured, not expensive. The old KPN `ema` reading made it look +like the most costly node in the pipeline at 141.899 ms/frame. + + +`--benchmark ` writes a per-node timing report and prints a table at +shutdown. `hero/run_bench.sh` is `run_trt.sh` with it switched on: + +```bash +./build/scene_analyze … --benchmark $H/bench_trt.json --output $H/pred_bench.json +``` + +**Do not read the `ema` column of the old KPN diagnostics block as a cost.** KPN +times a node across `fire_once`, which wraps the functor *and* the push to the +next channel, and a push parks when that channel is full (AR-004). A +backpressured node therefore bills its waiting to itself. On this film that +produced a genuinely inverted answer: + +``` +│ frame_source frames=5132 ema=141.899ms ← reported cost +[frame_source] decode avg=16.6127ms fps=60.19 ← actual decode +``` + +The source is not expensive; it is idle, holding a frame nobody has taken yet. +Optimising against that number means optimising the fastest node in the graph. + +The benchmark report separates the two: + +| Column | Meaning | Blind spot | +|---|---|---| +| `cpu_s`, `cpu%tot` | thread CPU time, and this node's share of all of it | a GPU wait looks like idleness | +| `cpu/f` | CPU ms per frame — backpressure cannot inflate it | as above | +| `exec/f` | wall ms per frame in the node, **including parked pushes** | overstates a blocked node | +| `stall/f` | `exec/f − cpu/f`: parked, or waiting on a device | does not say which | +| `in%`, `out%` | mean fill of the node's input and output channels | — | +| `press` | `in% − out%`; **the node marked ▶ is pacing the run** | not a cost, an ordering | + +Read `press` first: work queues up in front of the bottleneck and starves +everything after it, so the pacing node is the one with a full input and an empty +output. Then read `cpu%run` to decide the repair — a saturated thread means the +work itself must get cheaper, while an idle thread under pressure means the node +is waiting on the GPU or the disk, where batch size and engine precision are the +knobs and the C++ is not. + +Channel fills are sampled every 100 ms (`--benchmark-interval-ms`) because +`current_fill` is instantaneous: by shutdown every channel has drained, so a +single read at the end reports an idle pipeline no matter how congested it was. + +#### Check the GPU is not throttled before comparing anything + +**On this hardware, thermal state moves the result more than any code change +we are likely to make.** The same binary measured **8.25× cool and 3.12× once +heat-soaked** — a 2.6× swing — because the laptop RTX 3050 hits `SW Thermal +Slowdown` and pins the SM clock to **210 MHz out of 2100**: + +``` +$ nvidia-smi -q -d PERFORMANCE | grep -E "SW Power Cap|SW Thermal" + SW Power Cap : Active + SW Thermal Slowdown : Active +``` + +A number recorded without its clock state is not comparable to any other +number, and back-to-back full-film runs guarantee the later ones are throttled. +`run_bench.sh` now records `nvidia-smi` either side of the run into +`bench_gpu.txt`; check it before believing a regression. Let the GPU idle back +to full clock between measurements, and never A/B two runs across a heat-soak. + +This one cost real time here: a 2.7× "regression" was attributed to a code +change and reverted on that basis, when the change was innocent and the GPU had +simply warmed up between the two measurements. + +#### `cpu_s` on a GPU node is mostly spin — measured + +CUDA's default sync policy (`cudaDeviceScheduleAuto`) spin-waits before it +yields, so `cudaStreamSynchronize` charges the *calling thread's* CPU while the +GPU works. A GPU-bound node therefore reports a large `cpu_s` and reads as +CPU-bound. + +`SAE_CUDA_BLOCKING_SYNC=1` switches to a blocking wait. Measured over 300 s of +film, four cases, identical otherwise: + +| case | realtime | total CPU | embedder CPU | +|---|---|---|---| +| baseline | 3.29× | 103 s | 66 s | +| **`SAE_CUDA_BLOCKING_SYNC=1`** | 3.29× | **23 s** | **4 s** | +| `SAE_CV_THREADS=1` | 3.30× | 101 s | 66 s | +| both | 3.29× | 25 s | 5 s | + +**94% of the embedder's CPU was spin, not work**, and 78% of the pipeline's. +Throughput is unchanged, so this is free CPU — which matters for a service +sharing a box (DP-003) and makes `cpu_s` mean what it says. Prefer it for any +run where the CPU numbers are being read. + +`SAE_CV_THREADS=1` does nothing measurable: the only OpenCV-heavy node is +`face_aligner` at 1-2% of the pipeline, so the TBB arena is not worth removing +and `warpAffine` is not worth replacing. + +**Caveat: measured with the GPU clamped at 210 MHz** (see below). A device at +full clock spends less time in the sync, so the absolute spin figure will fall; +the ranking should not. + +#### `cpu_s` counts one thread — mind the TBB arena + +OpenCV 5 here is built against TBB, and every OpenCV module links it, so +`cv::parallel_for_` dispatches onto a TBB arena of `nproc − 1` workers (19 on the +20-core dev box; visible as `libtbb.so.12` frames in a thread dump). Since +`CLOCK_THREAD_CPUTIME_ID` is per-thread, work a node fans out that way is billed +to the TBB workers, **not** to the node. + +So a node using `warpAffine`, a histogram compare or a colour conversion reads +cheaper in `cpu_s` than it really is, and the missing time appears in `stall/f`, +where it looks identical to a GPU wait. `exec/f` does capture it — the functor +does not return until the parallel region joins — so the tell is a node whose +`exec/f` far exceeds its `cpu/f` **while its output channel is empty**: that is +fan-out, not blocking. + +Worth knowing for its own sake, too: 9 KPN node threads plus 19 TBB workers plus +the CUDA and NVDEC threads is heavy oversubscription on 20 cores. + +The JSON carries the same data plus the run's configuration, so two runs can be +diffed directly — which is the point, when sweeping `--embed-batch`, `--fps` or +an engine precision. ### Check you are on the GPU diff --git a/src/backends/trt_backend.cpp b/src/backends/trt_backend.cpp index 4850ad0..535ca02 100644 --- a/src/backends/trt_backend.cpp +++ b/src/backends/trt_backend.cpp @@ -19,6 +19,7 @@ #include #include +#include #include #include @@ -45,6 +46,40 @@ inline void check_cuda(cudaError_t e, const char* what) { throw CudaError(std::string(what) + ": " + cudaGetErrorString(e)); } +/// TRACES: VR-015 | PR-004 +/// Select how a thread waits for the GPU. Must run before the CUDA context is +/// created, so every engine constructor calls it and the first one wins. +/// +/// The default (`cudaDeviceScheduleAuto`) spin-waits: `cudaStreamSynchronize` +/// burns the calling thread's CPU for the whole of the device's work. Measured +/// here, the embedder thread sat at 99.7% *user* time with 0.5 s of system time +/// across 183 s — i.e. no blocking syscalls at all — while the GPU ran flat out. +/// +/// On this laptop that is not merely wasted CPU. `nvidia-powerd` arbitrates one +/// power budget across CPU and GPU, and the GPU's ceiling was observed dropping +/// from 20 W idle to 15 W under our load, with the SM clock *falling* from +/// 1005 MHz to 210 MHz once work started. Spinning may therefore be buying +/// watts away from the device the pipeline is actually waiting on. +/// +/// SAE_CUDA_BLOCKING_SYNC=1 switches to a blocking wait so the A/B needs no +/// rebuild. Default is unchanged until the measurement says otherwise. +inline void configure_cuda_sync_once() { + static const bool done = [] { + const char* env = std::getenv("SAE_CUDA_BLOCKING_SYNC"); + if (env && env[0] == '1') { + cudaError_t e = cudaSetDeviceFlags(cudaDeviceScheduleBlockingSync); + std::cerr << "[cuda] sync policy: BlockingSync" + << (e == cudaSuccess ? "" : " (FAILED — context already created)") + << "\n"; + } else { + std::cerr << "[cuda] sync policy: default (spin) — " + "set SAE_CUDA_BLOCKING_SYNC=1 to compare\n"; + } + return true; + }(); + (void)done; +} + class TrtLogger : public nvinfer1::ILogger { public: void log(Severity sev, const char* msg) noexcept override { @@ -109,6 +144,7 @@ public: (output_is_fp16_ ? 2 : 4); check_cuda(cudaMalloc(&d_input_, in_bytes), "cudaMalloc input"); check_cuda(cudaMalloc(&d_output_, out_bytes), "cudaMalloc output"); + configure_cuda_sync_once(); check_cuda(cudaStreamCreate(&stream_), "cudaStreamCreate"); context_->setTensorAddress(input_name_.c_str(), d_input_); @@ -293,6 +329,7 @@ public: out_elem_counts_[oi] = count; } + configure_cuda_sync_once(); check_cuda(cudaStreamCreate(&stream_), "cudaStreamCreate"); std::cerr << "[TrtScrfd] loaded: " << engine_path @@ -462,6 +499,7 @@ public: const std::size_t out_count = static_cast(kWindow); check_cuda(cudaMalloc(&d_input_, in_count * 4), "cudaMalloc input"); check_cuda(cudaMalloc(&d_output_, out_count * 4), "cudaMalloc output"); + configure_cuda_sync_once(); check_cuda(cudaStreamCreate(&stream_), "cudaStreamCreate"); context_->setTensorAddress(input_name_.c_str(), d_input_); context_->setTensorAddress(output_name_.c_str(), d_output_); diff --git a/src/benchmark.hpp b/src/benchmark.hpp new file mode 100644 index 0000000..02a60ba --- /dev/null +++ b/src/benchmark.hpp @@ -0,0 +1,590 @@ +#pragma once +/// TRACES: VR-015 | PR-004 +/// +/// Pipeline throughput benchmark — how much of a run is spent in each node. +/// +/// The KPN network already counts most of what an optimiser needs, and +/// `print_diagnostics()` throws nearly all of it away: it prints frames and +/// `ema` per node, and passes `elapsed_s = 0`, which zeroes throughput. Two +/// things had to change before "time per node" could be answered honestly. +/// +/// **`ema` is not a total.** It is an exponentially weighted average, so +/// `frames * ema` tracks the end of the run rather than the whole of it. On a +/// film that is a real difference — a detector costs one thing in a crowd scene +/// and another over a landscape. `NodeStats::total_exec_us` (added alongside +/// this) is the true sum. +/// +/// **Wall time inside a node is not all work.** `PoolObjectNode::fire_once` +/// times the functor *and* `push_outputs`, and `push_outputs` parks on a full +/// downstream channel (AR-004). A node that is merely backpressured therefore +/// bills the time it spent waiting to whoever is ahead of it: SuperHero's +/// `frame_source` reported 141.9 ms/frame against a decoder logging 12-18 ms. +/// Optimising against that number means optimising the fastest node in the +/// graph. +/// +/// So each node is reported three ways, and the three together are what +/// identify a cost: +/// +/// - `exec_ms` — cumulative wall time in the node, work *and* parked pushes +/// - `cpu_ms` — thread CPU time (CLOCK_THREAD_CPUTIME_ID). Backpressure +/// cannot inflate it, because a parked node holds no thread. +/// - `pressure` — mean fill of its input channels minus that of its outputs +/// +/// **`cpu_ms` cannot tell real work from a spinning GPU wait.** CUDA's default +/// sync policy (`cudaDeviceScheduleAuto`) spin-waits before yielding, so +/// `cudaStreamSynchronize` burns the calling thread's CPU while the GPU works. +/// A node that is purely GPU-bound can therefore report a high `cpu_ms` and read +/// as CPU-bound. `cudaSetDeviceFlags(cudaDeviceScheduleBlockingSync)` settles it +/// in one line: if a node's `cpu_ms` collapses under blocking sync, that CPU was +/// spin, not work. +/// +/// **`cpu_ms` counts one thread only.** `CLOCK_THREAD_CPUTIME_ID` is per-thread, +/// and OpenCV here is built against TBB, so any node whose functor goes through +/// `cv::parallel_for_` (histogram compare, `warpAffine`, colour conversion) has +/// that work executed on TBB's arena — 19 workers on a 20-core box — and billed +/// to those threads rather than to the node. Such a node reads *cheaper* than it +/// is, and the difference shows up in `stall/f` instead, indistinguishable from a +/// GPU wait. `exec_ms` does capture it, since the functor does not return until +/// the parallel region joins: a node where `exec/f` greatly exceeds `cpu/f` +/// while its output channel is empty is fanning out, not waiting. +/// +/// Work piles up *in front of* a bottleneck and starves everything *after* it, +/// so `pressure` is maximal at the node setting the pace. `cpu_ms` then says +/// which repair applies: high pressure with a saturated thread is CPU-bound and +/// the work must get cheaper, while high pressure with an idle thread is +/// waiting on a device, where batch size and engine precision are the knobs. +/// +/// Occupancy has to be sampled during the run. `current_fill` is instantaneous +/// and every channel has drained by the time the network stops, so a single +/// read at the end reports an idle pipeline however congested it was. +/// +/// Nothing here is specific to this pipeline's topology: the node graph is +/// recovered from KPN's channel names, so it keeps working when the graph +/// changes. + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace sae::bench { + +// ── Edge naming ────────────────────────────────────────────────────────────── + +/// TRACES: VR-015 | PR-004 +/// KPN names a channel ": → :" (static_network.hpp). +/// Recovering the two node names from it is what keeps attribution +/// topology-agnostic: the graph is read back out of the channel names rather +/// than hard-coded here, so a new node or a re-wired branch needs no change. +/// Leaves both outputs untouched if the name does not carry an arrow. +inline void split_edge_name(const std::string& name, + std::string& producer, std::string& consumer) { + static const std::string kArrow = " \xe2\x86\x92 "; // " → " + const auto arrow = name.find(kArrow); + if (arrow == std::string::npos) return; + auto strip_port = [](std::string s) { + const auto colon = s.rfind(':'); + return colon == std::string::npos ? s : s.substr(0, colon); + }; + producer = strip_port(name.substr(0, arrow)); + consumer = strip_port(name.substr(arrow + kArrow.size())); +} + +// ── Channel occupancy, time-averaged ───────────────────────────────────────── + +/// TRACES: VR-015 | PR-004 +/// One channel's fill level integrated over the run. `peak_fill` is already +/// cumulative in `ChannelStats`, but a peak cannot distinguish "full once" from +/// "full throughout", and those are opposite diagnoses. A mean can. +struct ChannelOccupancy { + std::string name; // "src:0 → dst:0", as KPN names it + std::string producer; // node name left of the arrow + std::string consumer; // node name right of the arrow + std::size_t capacity{0}; + std::uint64_t samples{0}; + double fill_sum{0.0}; + std::uint64_t samples_full{0}; + std::uint64_t samples_empty{0}; + + // Final-snapshot totals (monotonic counters, so the last read is the total). + std::size_t peak_fill{0}; + std::uint64_t pushes{0}; + std::uint64_t pops{0}; + std::uint64_t drops{0}; + std::uint64_t overflows{0}; + std::uint64_t bytes_pushed{0}; + + double mean_fill() const { return samples ? fill_sum / static_cast(samples) : 0.0; } + double mean_fill_pct() const { return capacity ? 100.0 * mean_fill() / static_cast(capacity) : 0.0; } + double peak_pct() const { return capacity ? 100.0 * static_cast(peak_fill) / static_cast(capacity) : 0.0; } + double full_pct() const { return samples ? 100.0 * static_cast(samples_full) / static_cast(samples) : 0.0; } + double empty_pct() const { return samples ? 100.0 * static_cast(samples_empty) / static_cast(samples) : 0.0; } + double bandwidth_mbs(double wall_s) const { + return wall_s > 0.0 ? static_cast(bytes_pushed) / wall_s / 1e6 : 0.0; + } +}; + +// ── Per-node attributed cost ───────────────────────────────────────────────── + +/// TRACES: VR-015 | PR-004 +struct NodeCost { + std::string name; + std::uint64_t frames{0}; + + // Cumulative wall time in the node — the answer to "where did the run go", + // but only for a node that is not backpressured; it includes parked pushes. + double exec_ms{0.0}; + double exec_ms_per_frame{0.0}; // true mean, not the EMA + double exec_share{0.0}; // exec_ms / wall_ms, 0..1 + double ema_exec_ms{0.0}; // KPN's EMA, kept for continuity with the old report + double max_exec_ms{0.0}; + + // Thread CPU time: excludes sleeping, parking and waiting on a device, so it + // is the one number backpressure cannot inflate. + double cpu_ms{0.0}; + double cpu_ms_per_frame{0.0}; + double cpu_share{0.0}; // cpu_ms / wall_ms — thread saturation, 0..1 + double cpu_pct_of_pipeline{0.0}; // this node's share of all nodes' CPU time + + // Per frame, time inside the node not spent on its own CPU: parked on a + // full output channel, or waiting on the GPU. `pressure` separates those — + // a backpressured node has a full output, a device-bound one does not. + double stall_ms_per_frame{0.0}; + + // Queue occupancy either side of the node, in percent of capacity. + double in_fill_pct{0.0}; + double out_fill_pct{0.0}; + double pressure{0.0}; // in − out; maximal at the pacing node + bool has_input{false}; + bool has_output{false}; + + double queue_wait_ms{0.0}; + bool is_bottleneck{false}; + + /// TRACES: VR-015, AR-004 | PR-004 + /// Live scheduling state, so a wedged run says *why* it is wedged rather + /// than only that it is. With `queued=0, wake=1` a wake was recorded and + /// never consumed; with `queued=0, wake=0` and a full input, no wake was + /// ever generated. Those are different bugs in different files. + bool queued{false}; + bool wake_pending{false}; +}; + +/// TRACES: VR-015 | PR-004 +/// Attribute cost to nodes from a KPN node snapshot plus sampled channel +/// occupancy. Pure — no clocks, no threads, no network — so the ranking is +/// unit-testable on CI hardware that can never run the pipeline itself. +/// +/// A node with several inputs takes the **minimum** input fill: it can only run +/// once every input has data, so the emptiest one gates it, and a full sibling +/// channel means that channel's producer is blocked rather than this node being +/// slow. A node with several outputs takes the **maximum** output fill, since +/// parking on any one branch stops the node. +/// +/// Terminals are the infinite-reservoir limit of the same rule: a source has +/// unlimited work available (input treated as 100% full) and a sink unlimited +/// drain (output treated as empty), so both stay rankable against the interior +/// nodes instead of dropping out of the comparison. +inline std::vector attribute_cost( + const std::vector& nodes, + const std::vector& channels, + double wall_sec) +{ + const double wall_ms = wall_sec * 1000.0; + + double cpu_total = 0.0; + for (const auto& n : nodes) cpu_total += n.total_cpu_ms; + + std::vector out; + out.reserve(nodes.size()); + + for (const auto& n : nodes) { + NodeCost c; + c.name = n.name; + c.frames = n.frames_processed; + c.exec_ms = n.total_exec_ms; + c.ema_exec_ms = n.ema_exec_ms; + c.max_exec_ms = n.max_exec_ms; + c.cpu_ms = n.total_cpu_ms; + c.queue_wait_ms = n.queue_wait_ms; + c.queued = n.queued; + c.wake_pending = n.wake_pending; + + c.exec_ms_per_frame = c.frames ? c.exec_ms / static_cast(c.frames) : 0.0; + c.cpu_ms_per_frame = c.frames ? c.cpu_ms / static_cast(c.frames) : 0.0; + c.exec_share = wall_ms > 0.0 ? c.exec_ms / wall_ms : 0.0; + c.cpu_share = wall_ms > 0.0 ? c.cpu_ms / wall_ms : 0.0; + c.cpu_pct_of_pipeline = cpu_total > 0.0 ? 100.0 * c.cpu_ms / cpu_total : 0.0; + c.stall_ms_per_frame = c.exec_ms_per_frame - c.cpu_ms_per_frame; + if (c.stall_ms_per_frame < 0.0) c.stall_ms_per_frame = 0.0; + + double in_min = 0.0; bool have_in = false; + double out_max = 0.0; bool have_out = false; + for (const auto& ch : channels) { + if (ch.consumer == n.name) { + const double f = ch.mean_fill_pct(); + if (!have_in || f < in_min) in_min = f; + have_in = true; + } + if (ch.producer == n.name) { + const double f = ch.mean_fill_pct(); + if (!have_out || f > out_max) out_max = f; + have_out = true; + } + } + c.has_input = have_in; + c.has_output = have_out; + c.in_fill_pct = have_in ? in_min : 100.0; // source: always has work + c.out_fill_pct = have_out ? out_max : 0.0; // sink: never blocks + c.pressure = c.in_fill_pct - c.out_fill_pct; + out.push_back(std::move(c)); + } + + // Rank, but only among nodes that actually ran: a node with zero frames has + // no cost to attribute and its neighbouring channels never moved. + auto best = out.end(); + for (auto it = out.begin(); it != out.end(); ++it) { + if (it->frames == 0) continue; + if (best == out.end() || it->pressure > best->pressure) best = it; + } + if (best != out.end()) best->is_bottleneck = true; + + return out; +} + +/// TRACES: VR-015 | PR-004 +/// One line of plain English about the winning node, since the point of the +/// report is to say what to change next. A saturated thread means the node's +/// own work is the limit; an idle thread under pressure means it is waiting on +/// a device, and those are different repairs. +inline std::string verdict(const std::vector& costs) { + for (const auto& c : costs) { + if (!c.is_bottleneck) continue; + std::ostringstream os; + os << std::fixed << c.name << " sets the pace: "; + + // A source's 100% input is the infinite-reservoir convention, not a + // measured queue — saying "work is backed up in front of it" would be + // asserting something no counter observed. + if (!c.has_input) + os << "nothing downstream is waiting on it (output " + << std::setprecision(1) << c.out_fill_pct + << "% full), so the pipeline is running as fast as this node can feed it. "; + else if (!c.has_output) + os << std::setprecision(1) << c.in_fill_pct + << "% full input and nothing to block on, so it is the drain. "; + else + os << std::setprecision(1) << c.in_fill_pct << "% full input, " + << c.out_fill_pct << "% full output. "; + + os << std::setprecision(2) << c.cpu_ms_per_frame << " ms/frame on CPU. "; + + if (c.cpu_share >= 0.85) + os << "CPU-bound — its thread is busy " << std::setprecision(0) + << (100.0 * c.cpu_share) << "% of the run, so the work itself has to get" + " cheaper or be split across more threads."; + else if (c.cpu_share <= 0.35 && c.stall_ms_per_frame > c.cpu_ms_per_frame) + os << "Device-bound — its thread is busy only " << std::setprecision(0) + << (100.0 * c.cpu_share) << "% of the run and it spends " + << std::setprecision(2) << c.stall_ms_per_frame + << " ms/frame off-CPU, so it is waiting on the GPU or the disk: batch size," + " engine precision and the decode path are the knobs, not the C++."; + else + os << "Mixed — thread busy " << std::setprecision(0) << (100.0 * c.cpu_share) + << "% of the run, " << std::setprecision(2) << c.stall_ms_per_frame + << " ms/frame off-CPU."; + return os.str(); + } + return "no node processed a frame — nothing to attribute"; +} + +// ── Recorder ───────────────────────────────────────────────────────────────── + +/// TRACES: VR-015 | PR-004 +/// Samples the live network on a timer and emits the report at the end. +/// +/// The sampler only reads relaxed atomics, so it does not perturb what it +/// measures — which matters, since this exists to be trusted as a timing +/// measurement. +class BenchmarkRecorder { +public: + using Sampler = std::function; + + explicit BenchmarkRecorder(int sample_interval_ms = 100) + : interval_(std::chrono::milliseconds(sample_interval_ms)) {} + + ~BenchmarkRecorder() { stop(); } + + void start(Sampler sampler) { + sampler_ = std::move(sampler); + running_.store(true, std::memory_order_release); + thread_ = std::thread([this] { + while (running_.load(std::memory_order_acquire)) { + accumulate(sampler_()); + std::this_thread::sleep_for(interval_); + } + }); + } + + /// Stops sampling and latches the final counter values. Call while the + /// network object is still alive: the monotonic counters stay valid after + /// `net.stop()`, but they die with the object. + void stop() { + if (!running_.exchange(false, std::memory_order_acq_rel)) return; + if (thread_.joinable()) thread_.join(); + if (sampler_) { + final_ = sampler_(); + // Occupancy is deliberately NOT accumulated from this last read: + // the pipeline has drained by now, and folding an idle sample into + // the mean biases every channel toward "never congested". + for (const auto& ch : final_.channels) { + auto& occ = occupancy_[ch.name]; + if (occ.name.empty()) { // a channel that never moved + occ.name = ch.name; + occ.capacity = ch.capacity; + split_edge_name(ch.name, occ.producer, occ.consumer); + } + occ.peak_fill = ch.peak_fill; + occ.pushes = ch.pushes; + occ.pops = ch.pops; + occ.drops = ch.drops; + occ.overflows = ch.overflows; + occ.bytes_pushed = ch.bytes_pushed; + } + } + stopped_ = true; + } + + bool has_data() const { return stopped_ && !final_.nodes.empty(); } + double wall_sec() const { return final_.elapsed_s; } + + std::vector channels() const { + std::vector v; + v.reserve(occupancy_.size()); + for (const auto& [_, occ] : occupancy_) v.push_back(occ); + return v; + } + + std::vector costs() const { + return attribute_cost(final_.nodes, channels(), final_.elapsed_s); + } + + /// TRACES: VR-015 | PR-004 + /// Machine-readable report, for sweeping configurations and diffing runs. + /// `film_sec` is the last timestamp the pipeline reached, so + /// `realtime_factor` answers what the optimiser is really asking: seconds + /// of film per second of wall clock. It is 0 for a topology with no result + /// sink (the dump-only path), and the field is then omitted rather than + /// reported as zero throughput. + nlohmann::json to_json(const nlohmann::json& run_config, double film_sec) const { + using nlohmann::json; + const double wall = final_.elapsed_s; + const auto chans = channels(); + const auto cost = attribute_cost(final_.nodes, chans, wall); + + json j; + j["schema_version"] = 1; + j["config"] = run_config; + + json summary; + summary["wall_sec"] = wall; + summary["sample_count"] = sample_count_; + summary["sample_interval_ms"] = interval_.count(); + if (film_sec > 0.0) { + summary["film_sec"] = film_sec; + summary["realtime_factor"] = wall > 0.0 ? film_sec / wall : 0.0; + } + for (const auto& c : cost) + if (c.is_bottleneck) { summary["bottleneck"] = c.name; break; } + summary["verdict"] = verdict(cost); + j["summary"] = summary; + + json jnodes = json::array(); + for (const auto& c : cost) { + jnodes.push_back({ + {"name", c.name}, + {"frames", c.frames}, + {"fps", wall > 0.0 ? c.frames / wall : 0.0}, + {"exec_ms", c.exec_ms}, + {"exec_ms_per_frame", c.exec_ms_per_frame}, + {"exec_share", c.exec_share}, + {"ema_exec_ms", c.ema_exec_ms}, + {"max_exec_ms", c.max_exec_ms}, + {"cpu_ms", c.cpu_ms}, + {"cpu_ms_per_frame", c.cpu_ms_per_frame}, + {"cpu_share", c.cpu_share}, + {"cpu_pct_of_pipeline", c.cpu_pct_of_pipeline}, + {"stall_ms_per_frame", c.stall_ms_per_frame}, + {"queue_wait_ms", c.queue_wait_ms}, + {"in_fill_pct", c.in_fill_pct}, + {"out_fill_pct", c.out_fill_pct}, + {"pressure", c.pressure}, + {"is_bottleneck", c.is_bottleneck}, + {"queued", c.queued}, + {"wake_pending", c.wake_pending}, + }); + } + j["nodes"] = std::move(jnodes); + + json jch = json::array(); + for (const auto& ch : chans) { + jch.push_back({ + {"name", ch.name}, + {"producer", ch.producer}, + {"consumer", ch.consumer}, + {"capacity", ch.capacity}, + {"mean_fill", ch.mean_fill()}, + {"mean_fill_pct", ch.mean_fill_pct()}, + {"peak_fill", ch.peak_fill}, + {"peak_pct", ch.peak_pct()}, + {"full_pct", ch.full_pct()}, + {"empty_pct", ch.empty_pct()}, + {"pushes", ch.pushes}, + {"pops", ch.pops}, + {"drops", ch.drops}, + {"overflows", ch.overflows}, + {"mb_per_sec", ch.bandwidth_mbs(wall)}, + }); + } + j["channels"] = std::move(jch); + return j; + } + + /// TRACES: VR-015 | PR-004 + /// Human-readable form of the same data, so a run is legible without + /// opening the JSON. + void print(std::ostream& os, double film_sec) const { + print_impl(os, final_, film_sec); + } + + /// TRACES: VR-015, AR-004 | PR-004 + /// Dump the report from a LIVE snapshot, mid-run, without stopping anything. + /// + /// A report that only exists at shutdown is no use against the failure this + /// pipeline actually has: a wedged run never reaches shutdown, so the one + /// moment the numbers matter most is the one moment they were unavailable. + /// Channel occupancy names the stalled node directly — it is the one whose + /// input is full and whose output is empty — which is otherwise a debug-build + /// and a gdb session away. + /// + /// Safe to call from the wait loop while the pipeline is running or hung: it + /// takes the same lock-free snapshot the sampler does. + void dump_live(std::ostream& os, double film_sec) const { + if (!sampler_) { os << "[benchmark] no sampler — run with --benchmark\n"; return; } + print_impl(os, sampler_(), film_sec); + } + +private: + void print_impl(std::ostream& os, const kpn::NetworkSnapshot& snap, + double film_sec) const { + const double wall = snap.elapsed_s; + const auto chans = channels(); + const auto cost = attribute_cost(snap.nodes, chans, wall); + + os << "\n┌─ Pipeline benchmark (VR-015) ──────────────────────────────────────────────\n"; + os << "│ wall " << std::fixed << std::setprecision(1) << wall << "s"; + if (film_sec > 0.0) + os << " film " << film_sec << "s realtime x" << std::setprecision(2) + << (wall > 0.0 ? film_sec / wall : 0.0); + os << " samples " << sample_count_ << "\n│\n"; + + os << "│ node frames cpu_s cpu%run cpu%tot cpu/f" + " exec/f stall/f in% out% press q/w\n"; + for (const auto& c : cost) { + os << "│ " << (c.is_bottleneck ? "▶ " : " ") << std::left << std::setw(16) + << c.name << std::right + << std::setw(7) << c.frames + << std::setw(10) << std::setprecision(1) << (c.cpu_ms / 1000.0) + << std::setw(9) << std::setprecision(0) << (100.0 * c.cpu_share) + << std::setw(9) << std::setprecision(0) << c.cpu_pct_of_pipeline + << std::setw(8) << std::setprecision(2) << c.cpu_ms_per_frame + << std::setw(8) << std::setprecision(2) << c.exec_ms_per_frame + << std::setw(9) << std::setprecision(2) << c.stall_ms_per_frame + << std::setw(7) << std::setprecision(0) << c.in_fill_pct + << std::setw(7) << std::setprecision(0) << c.out_fill_pct + << std::setw(8) << std::setprecision(1) << c.pressure + << " " << int(c.queued) << "/" << int(c.wake_pending) + << "\n"; + } + + /// TRACES: VR-015, AR-004 | PR-004 + // Fires only when the scheduling state is actually wrong, so a healthy + // run stays quiet and a wedged one names the fault — instead of leaving + // it to be reconstructed under a debugger that suppresses the bug. + for (const auto& c : cost) { + if (c.queued || !c.has_input) continue; + if (c.wake_pending) + os << "│ !! " << c.name << " idle with a wake outstanding" + " (queued=0 wake=1): the wake was recorded and never" + " consumed — submit/release handshake.\n"; + else if (c.in_fill_pct > 50.0) + os << "│ !! " << c.name << " idle with a " + << std::setprecision(0) << c.in_fill_pct + << "% full input and no wake pending: the wake was never" + " generated — channel edge detection.\n"; + } + + os << "│\n│ channel cap mean% peak% full%" + " empty% MB/s\n"; + for (const auto& ch : chans) { + os << "│ " << std::left << std::setw(36) << ch.name << std::right + << std::setw(5) << ch.capacity + << std::setw(7) << std::setprecision(1) << ch.mean_fill_pct() + << std::setw(7) << ch.peak_pct() + << std::setw(7) << ch.full_pct() + << std::setw(7) << ch.empty_pct() + << std::setw(9) << std::setprecision(1) << ch.bandwidth_mbs(wall) + << "\n"; + } + os << "│\n│ " << verdict(cost) << "\n"; + os << "└────────────────────────────────────────────────────────────────────────────\n"; + os << " cpu_s / cpu%tot is where the run's compute actually went. exec/f is wall\n" + " time in the node INCLUDING time parked on a full output channel, so it\n" + " overstates a backpressured node — compare it against cpu/f, which cannot\n" + " be inflated that way. press = input fill − output fill, and locates the\n" + " node that work is queueing up in front of.\n" + " cpu_s counts THIS node's thread only: work OpenCV fans out via TBB is\n" + " billed to the TBB arena, so a node using cv::parallel_for_ reads cheaper\n" + " than it is and the difference surfaces in stall/f.\n"; + } + +private: + void accumulate(const kpn::NetworkSnapshot& snap) { + ++sample_count_; + for (const auto& ch : snap.channels) { + auto& occ = occupancy_[ch.name]; + if (occ.name.empty()) { + occ.name = ch.name; + occ.capacity = ch.capacity; + split_edge_name(ch.name, occ.producer, occ.consumer); + } + occ.fill_sum += static_cast(ch.current_fill); + ++occ.samples; + if (ch.capacity && ch.current_fill >= ch.capacity) ++occ.samples_full; + if (ch.current_fill == 0) ++occ.samples_empty; + } + } + + std::chrono::milliseconds interval_; + Sampler sampler_; + std::thread thread_; + std::atomic running_{false}; + bool stopped_{false}; + std::uint64_t sample_count_{0}; + std::map occupancy_; + kpn::NetworkSnapshot final_{}; +}; + +} // namespace sae::bench diff --git a/src/config.hpp b/src/config.hpp index 5ae8061..0a8a9fe 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -32,6 +32,14 @@ struct Config { // scripts/optimizer/SCHEMA.md) for offline threshold-sweep replay via sae_kpn. std::string dump_embeddings_path; + /// TRACES: VR-015 | PR-004 + // When set, write a per-node timing and bottleneck report here (src/ + // benchmark.hpp) and print it at shutdown. Costs one background thread + // reading relaxed atomics on a timer, so it is safe to leave on, but a + // measurement run should still be isolated (nothing else on the GPU). + std::string benchmark_path; + int benchmark_interval_ms{100}; // channel-occupancy sampling period + // ── Sampling ───────────────────────────────────────────────────────────── float sample_fps{1.0f}; // frames to analyse per second of movie float max_decode_fps{0.f}; // wall-clock cap on source decode rate (0 = uncapped) diff --git a/src/main.cpp b/src/main.cpp index 964d24d..04c3c58 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -44,10 +44,15 @@ // --expand-band-hi

store admission ceiling, P(same person) (default: 0.95) // --expand-min-anchor accepted frames before a track confirms (default: 3) // --expand-debug-dir

dump promoted mugshots + embeddings here (SAE_DEBUG) +// --benchmark write a per-node timing + bottleneck report (JSON) and +// print it at shutdown. Says where the run's time went +// and which node is pacing it. See src/benchmark.hpp. +// --benchmark-interval-ms channel-occupancy sampling period (default: 100) // (SAE_DEBUG only) // --debug-dir debug frames output dir (default: debug_frames) // --crop-context bbox expansion factor for context crops (default: 1.5) +#include "benchmark.hpp" #include "config.hpp" #include "types.hpp" #include "gallery/embedder_stamp.hpp" @@ -71,9 +76,14 @@ #include +#include // cv::setNumThreads (SAE_CV_THREADS) + #include #include +#include +#include #include +#include #include #include #include @@ -96,6 +106,20 @@ static constexpr std::size_t kSceneJoinDepth = 256; /// actually worked. static std::shared_ptr scene_stats; +/// TRACES: VR-015, AR-004 | PR-004 +/// Set by SIGUSR1, serviced by the wait loop. `kill -USR1 ` on a running +/// or WEDGED run prints the benchmark table immediately — channel occupancy +/// names the stalled node (full input, empty output) without a debug build or a +/// debugger, which is the difference between diagnosing the AR-004 hang in +/// seconds and reproducing it under gdb. +/// +/// The handler only stores a flag; all printing happens on the main thread, +/// since nothing in the report is async-signal-safe. +static std::atomic g_dump_request{false}; +extern "C" void sae_on_dump_signal(int) { + g_dump_request.store(true, std::memory_order_relaxed); +} + static Config parse_args(int argc, char** argv) { Config cfg; cfg.detector_model = kDefaultDetectorModel; @@ -114,6 +138,8 @@ static Config parse_args(int argc, char** argv) { else if (arg("--gallery")) cfg.gallery_path = next(); else if (arg("--output")) cfg.output_path = next(); else if (arg("--dump-embeddings")) cfg.dump_embeddings_path = next(); + else if (arg("--benchmark")) cfg.benchmark_path = next(); + else if (arg("--benchmark-interval-ms")) cfg.benchmark_interval_ms = std::stoi(next()); else if (arg("--fps")) cfg.sample_fps = std::stof(next()); else if (arg("--max-decode-fps")) cfg.max_decode_fps = std::stof(next()); else if (arg("--start")) cfg.start_sec = std::stod(next()); @@ -174,6 +200,21 @@ static Config parse_args(int argc, char** argv) { // ── Main ────────────────────────────────────────────────────────────────────── int main(int argc, char** argv) { + /// TRACES: VR-015 | PR-004 + // OpenCV here is built against TBB, so cv::parallel_for_ opens an arena of + // nproc-1 workers (19 on a 20-core box) *on top of* KPN's one thread per + // node. Two schedulers, neither aware of the other, on the same cores. + // + // SAE_CV_THREADS=1 hands concurrency entirely to KPN, which is where this + // pipeline's parallelism is supposed to come from. Worth measuring rather + // than assuming: TBB fan-out inside warpAffine is free speed when the + // pipeline is otherwise idle, so this can cut either way. Unset = default. + if (const char* t = std::getenv("SAE_CV_THREADS")) { + const int n = std::atoi(t); + cv::setNumThreads(n); + std::cerr << "[opencv] cv::setNumThreads(" << n << ")\n"; + } + Config cfg; try { cfg = parse_args(argc, argv); @@ -236,7 +277,15 @@ int main(int argc, char** argv) { // AR-016: a film ends with faces on screen and those tracks have not timed // out. Without this flush the closing scene's cast is silently never // emitted — a loss that reads as a recognition miss, not a bookkeeping bug. - sink_fn.set_pre_write_hook([registry](double last_ts) { registry->flush(last_ts); }); + /// TRACES: VR-015 | PR-004 + // Last timestamp the pipeline reached, latched on the way out. It is what + // turns wall-clock seconds into the number that matters — seconds of film + // per second of run — and the sink is the only node that knows it. + std::atomic film_sec{0.0}; + sink_fn.set_pre_write_hook([registry, &film_sec](double last_ts) { + film_sec.store(last_ts, std::memory_order_release); + registry->flush(last_ts); + }); #ifdef SAE_DEBUG DebugRendererFunc debug_fn {cfg}; #endif @@ -302,8 +351,23 @@ int main(int argc, char** argv) { return false; }); + /// TRACES: VR-015 | PR-004 + // Sampling must start with the network and stop before it is destroyed: + // channel fill is instantaneous, and by the time a run ends everything + // has drained, so a single read at shutdown reports an idle pipeline no + // matter how congested it was. + sae::bench::BenchmarkRecorder bench{cfg.benchmark_interval_ms}; + const bool benchmarking = !cfg.benchmark_path.empty(); + std::cerr << "[main] starting pipeline…\n"; net.start(); + if (benchmarking) { + bench.start([&net] { return net.network_snapshot(); }); + std::signal(SIGUSR1, sae_on_dump_signal); + std::cerr << "[benchmark] sampling every " << cfg.benchmark_interval_ms + << "ms — `kill -USR1 " << getpid() + << "` to dump the table now (works while hung)\n"; + } // Wait until BOTH terminal branches finish: result_sink (face pipeline) // and, when enabled, scene_detector (the dense TransNetV2 branch, which @@ -311,12 +375,51 @@ int main(int argc, char** argv) { // pre-set true when scene detection is disabled. while ((!done.load(std::memory_order_acquire) || !scene_done.load(std::memory_order_acquire)) && - !node_crashed.load(std::memory_order_acquire)) + !node_crashed.load(std::memory_order_acquire)) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); + /// TRACES: VR-015, AR-004 | PR-004 + if (g_dump_request.exchange(false, std::memory_order_relaxed)) + bench.dump_live(std::cerr, film_sec.load(std::memory_order_acquire)); + } + + // Latch the counters before stop(): they stay readable afterwards, but + // only while the network object is alive, and this keeps the numbers + // describing the run rather than the teardown. + if (benchmarking) bench.stop(); net.stop(); net.print_diagnostics(); + /// TRACES: VR-015 | PR-004 + if (benchmarking && bench.has_data()) { + const double film = film_sec.load(std::memory_order_acquire); + bench.print(std::cerr, film); + + nlohmann::json run_cfg{ + {"movie", cfg.movie_path}, + {"gallery", cfg.gallery_path}, + {"gallery_actors", gallery.actors.size()}, + {"sample_fps", cfg.sample_fps}, + {"min_face_px", cfg.min_face_px}, + {"max_faces", cfg.max_faces}, + {"embed_batch", cfg.embed_batch_size}, + {"expand_gallery", cfg.expand_gallery}, + {"scene_detect", cfg.scene_detect}, + {"detector_engine", cfg.detector_engine}, + {"arcface_engine", cfg.arcface_engine}, + {"detector_model", cfg.detector_model}, + {"arcface_model", cfg.arcface_model}, + }; + std::ofstream bf(cfg.benchmark_path); + if (bf) { + bf << bench.to_json(run_cfg, film).dump(2) << "\n"; + std::cerr << "[benchmark] wrote " << cfg.benchmark_path << "\n"; + } else { + std::cerr << "[benchmark] ERROR: could not write " + << cfg.benchmark_path << "\n"; + } + } + /// TRACES: AR-004 | SR-002 // A dropped frame does not degrade a result, it silently changes one — // the output is a claim about footage that was never analysed, and diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 5f47259..d6e628c 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -27,6 +27,7 @@ add_executable(sae_tests test_replay_fixtures.cpp test_embedding_dump.cpp test_audio_signature.cpp + test_benchmark.cpp ${CMAKE_SOURCE_DIR}/src/backends/gemm_backend.cpp ${CMAKE_SOURCE_DIR}/src/gallery/gallery_store.cpp ${CMAKE_SOURCE_DIR}/src/audio_signature.cpp @@ -70,6 +71,10 @@ target_compile_definitions(sae_tests PRIVATE target_link_libraries(sae_tests PRIVATE Catch2::Catch2WithMain nlohmann_json::nlohmann_json + # VR-015: test_benchmark.cpp includes src/benchmark.hpp, which reads KPN's + # diagnostics structs. Header-only — no KPN network is constructed here, so + # the cost attribution stays testable on CI's GPU-free N100. + kpn ffmpeg_libs ${OpenCV_LIBS} ${HDF5_CXX_LIBRARIES}) diff --git a/tests/test_benchmark.cpp b/tests/test_benchmark.cpp new file mode 100644 index 0000000..0355fdd --- /dev/null +++ b/tests/test_benchmark.cpp @@ -0,0 +1,221 @@ +// Cost attribution for the pipeline benchmark. +// +// TRACES: VR-015 | UT-120, UT-121, UT-122, UT-123, UT-124 | PR-004 +// +// `attribute_cost` is pure — no clock, no thread, no network — precisely so the +// ranking can be tested on CI hardware that can never run the pipeline. These +// are T1 tests: they build the snapshots the KPN network would have produced +// and assert which node gets blamed. +// +// The case that matters is UT-121. On SuperHero the real run reported +// `frame_source ema=141.899ms` against a decoder logging 12-18 ms, because +// `fire_once` bills time parked pushing into a full downstream channel to the +// node doing the pushing. Any metric that ranks nodes by wall time inside the +// node picks the source — the fastest node in the graph — as the thing to +// optimise. That is the mistake this file exists to prevent regressing. + +#include "benchmark.hpp" + +#include +#include + +#include +#include + +using Catch::Matchers::WithinAbs; +using Catch::Matchers::WithinRel; +using namespace sae::bench; + +namespace { + +/// A KPN node snapshot with only the fields attribution reads. +kpn::NodeSnapshot node(std::string name, std::uint64_t frames, + double ema_ms, double cpu_ms, double exec_ms) { + kpn::NodeSnapshot s{}; + s.name = std::move(name); + s.frames_processed = frames; + s.ema_exec_ms = ema_ms; + s.total_cpu_ms = cpu_ms; + s.total_exec_ms = exec_ms; + return s; +} + +/// A channel whose mean fill is `fill_pct` of `capacity`. +ChannelOccupancy chan(const std::string& producer, const std::string& consumer, + std::size_t capacity, double fill_pct) { + ChannelOccupancy c; + c.name = producer + ":0 \xe2\x86\x92 " + consumer + ":0"; + c.producer = producer; + c.consumer = consumer; + c.capacity = capacity; + c.samples = 1000; + c.fill_sum = static_cast(c.samples) * static_cast(capacity) * fill_pct / 100.0; + return c; +} + +// Returned by value: a reference into `v` bound to a name built from a string +// literal trips -Wdangling-reference, and the struct is small enough not to care. +NodeCost by_name(const std::vector& v, std::string_view name) { + for (const auto& c : v) if (c.name == name) return c; + throw std::runtime_error("no such node: " + std::string(name)); +} + +NodeCost bottleneck(const std::vector& v) { + for (const auto& c : v) if (c.is_bottleneck) return c; + throw std::runtime_error("no bottleneck flagged"); +} + +} // namespace + +// UT-120 — the node work queues up in front of is the one blamed. +TEST_CASE("attribute_cost blames the node with a full input and an empty output", + "[benchmark][VR-015]") { + // source → mid → sink. mid is slow: its input backs up, its output drains. + const auto nodes = std::vector{ + node("source", 1000, 10.0, 2'000.0, 10'000.0), + node("mid", 1000, 50.0, 45'000.0, 50'000.0), + node("sink", 1000, 0.5, 400.0, 500.0), + }; + const auto channels = std::vector{ + chan("source", "mid", 32, 95.0), // full: work piling up in front of mid + chan("mid", "sink", 16, 2.0), // empty: mid starves everything after + }; + + const auto costs = attribute_cost(nodes, channels, /*wall_sec=*/50.0); + + CHECK(bottleneck(costs).name == "mid"); + CHECK(by_name(costs, "mid").pressure > by_name(costs, "source").pressure); + CHECK(by_name(costs, "mid").pressure > by_name(costs, "sink").pressure); +} + +// UT-121 — the SuperHero regression: a backpressured source reports a huge +// wall time per frame and must NOT be mistaken for the bottleneck. +TEST_CASE("a backpressured source is not blamed for the time it spent parked", + "[benchmark][VR-015]") { + // Numbers taken from the real SuperHero TRT run: the source reports + // 141.9 ms/frame inside fire_once while its decoder logs ~15 ms, because + // the remaining ~127 ms is spent parked on a full output channel. + const auto nodes = std::vector{ + node("frame_source", 5132, 141.899, 77'000.0, 728'000.0), + node("face_detector", 5129, 6.830, 480'000.0, 35'000.0), + node("result_sink", 5129, 0.080, 410.0, 410.0), + }; + const auto channels = std::vector{ + chan("frame_source", "face_detector", 32, 99.0), // source blocked on this + chan("face_detector", "result_sink", 16, 1.0), + }; + + const auto costs = attribute_cost(nodes, channels, /*wall_sec=*/500.0); + + const auto& src = by_name(costs, "frame_source"); + const auto& det = by_name(costs, "face_detector"); + + // The trap: by wall time inside the node, the source looks 20x costlier. + REQUIRE(src.exec_ms_per_frame > det.exec_ms_per_frame * 10.0); + // The fix: it is not blamed, because its own output channel is the thing + // that is full — it is waiting, not working. + CHECK_FALSE(src.is_bottleneck); + CHECK(bottleneck(costs).name == "face_detector"); + // And CPU time, which parking cannot inflate, agrees: the detector burns + // 480 s of thread time against the source's 77 s. + CHECK(det.cpu_ms > src.cpu_ms); + CHECK(det.cpu_pct_of_pipeline > src.cpu_pct_of_pipeline); +} + +// UT-122 — terminals stay rankable via the infinite-reservoir convention. +TEST_CASE("a source with an empty output is blamed; a sink with a full input is too", + "[benchmark][VR-015]") { + SECTION("starved pipeline: the source cannot keep up") { + const auto nodes = std::vector{ + node("source", 100, 90.0, 9'000.0, 9'000.0), + node("mid", 100, 1.0, 100.0, 100.0), + }; + // Nothing ever accumulates: the source is the constraint. + const auto costs = attribute_cost(nodes, {chan("source", "mid", 32, 1.0)}, 10.0); + CHECK(bottleneck(costs).name == "source"); + // Source has no input channel, so it is treated as always having work. + CHECK_THAT(by_name(costs, "source").in_fill_pct, WithinAbs(100.0, 1e-9)); + } + + SECTION("congested pipeline: the sink cannot drain") { + const auto nodes = std::vector{ + node("mid", 100, 1.0, 100.0, 100.0), + node("sink", 100, 90.0, 9'000.0, 9'000.0), + }; + const auto costs = attribute_cost(nodes, {chan("mid", "sink", 16, 98.0)}, 10.0); + CHECK(bottleneck(costs).name == "sink"); + // Sink has no output channel, so it is treated as never blocking. + CHECK_THAT(by_name(costs, "sink").out_fill_pct, WithinAbs(0.0, 1e-9)); + } +} + +// UT-123 — the per-node time figures are the ones an optimiser would act on. +TEST_CASE("cost shares are computed against wall clock and pipeline total", + "[benchmark][VR-015]") { + const auto nodes = std::vector{ + node("a", 100, 1.0, 30'000.0, 40'000.0), // 30 s CPU + node("b", 100, 1.0, 10'000.0, 12'000.0), // 10 s CPU + }; + const auto costs = attribute_cost(nodes, {chan("a", "b", 8, 50.0)}, /*wall_sec=*/50.0); + + const auto& a = by_name(costs, "a"); + CHECK_THAT(a.cpu_ms_per_frame, WithinRel(300.0, 1e-9)); // 30 s / 100 frames + CHECK_THAT(a.exec_ms_per_frame, WithinRel(400.0, 1e-9)); + CHECK_THAT(a.cpu_share, WithinRel(0.6, 1e-9)); // 30 s of a 50 s run + CHECK_THAT(a.exec_share, WithinRel(0.8, 1e-9)); + CHECK_THAT(a.cpu_pct_of_pipeline, WithinRel(75.0, 1e-9)); // 30 of 40 s total + // Time inside the node that was not spent on its own CPU: parked, or on GPU. + CHECK_THAT(a.stall_ms_per_frame, WithinRel(100.0, 1e-9)); + + // A node that never ran cannot be the bottleneck, and contributes no cost. + const auto idle = std::vector{ + node("ran", 10, 1.0, 100.0, 100.0), + node("idle", 0, 0.0, 0.0, 0.0), + }; + const auto idle_costs = attribute_cost(idle, {}, 10.0); + CHECK(bottleneck(idle_costs).name == "ran"); + CHECK_FALSE(by_name(idle_costs, "idle").is_bottleneck); +} + +// UT-124 — the node graph is recovered from KPN's channel names, which is what +// keeps attribution working when the topology changes. +TEST_CASE("channel names split back into producer and consumer", + "[benchmark][VR-015]") { + std::string p, c; + split_edge_name("frame_source:0 \xe2\x86\x92 camera_pos:0", p, c); + CHECK(p == "frame_source"); + CHECK(c == "camera_pos"); + + // Multi-port nodes: the port index is stripped, the node name is not. + split_edge_name("detector:2 \xe2\x86\x92 aligner:1", p, c); + CHECK(p == "detector"); + CHECK(c == "aligner"); + + // A name with no arrow leaves both untouched rather than inventing an edge. + std::string q = "unset", r = "unset"; + split_edge_name("not an edge", q, r); + CHECK(q == "unset"); + CHECK(r == "unset"); +} + +// A node with several inputs is gated by its emptiest one, and blocked by its +// fullest output — the multi-branch case the scene-detect topology creates. +TEST_CASE("multi-port nodes take min input fill and max output fill", + "[benchmark][VR-015]") { + const auto nodes = std::vector{ + node("join", 100, 1.0, 1'000.0, 1'000.0), + }; + const auto channels = std::vector{ + chan("up_a", "join", 32, 99.0), // full, but... + chan("up_b", "join", 32, 4.0), // ...this one gates the node + chan("join", "down_a", 16, 10.0), + chan("join", "down_b", 16, 80.0), // parking on this stops the node + }; + + const auto costs = attribute_cost(nodes, channels, 10.0); + const auto& j = by_name(costs, "join"); + + CHECK_THAT(j.in_fill_pct, WithinAbs( 4.0, 1e-9)); + CHECK_THAT(j.out_fill_pct, WithinAbs(80.0, 1e-9)); + CHECK(j.pressure < 0.0); // starved, not congested +}