diff --git a/TODO.md b/TODO.md index 099ebed..3ef6cdb 100644 --- a/TODO.md +++ b/TODO.md @@ -48,6 +48,14 @@ it (CLAUDE.md §4 "Optional external dependencies"). ### Performance - [ ] LOD system for >1M point scenes (octree-based) +### Ingestion contract (`docs/ingestion_contract.md`, landed 2026-07-11) +- [x] Normative viz-plane ingestion contract + budget table (I1–I7) +- [x] Benchmark suite `tests/benchmarks/ingestion/` + `scripts/bench_ingestion.sh` +- [ ] I3 gap: overflow/overwrite/refusal counters on `RingBuffer` / `DoubleBuffer` +- [ ] I7 gap: CI regression gate against the pinned reference report +- [ ] I2 gap: slot pre-sizing for `DataStream`/`DoubleBuffer` (heap-owning payloads) +- [ ] I5: re-run observer A/B on a hardware-GL host to pin the invisibility number + ### Extension - [ ] Plugin system (defer until concrete user) diff --git a/docs/ingestion_contract.md b/docs/ingestion_contract.md new file mode 100644 index 0000000..8fcc99d --- /dev/null +++ b/docs/ingestion_contract.md @@ -0,0 +1,109 @@ +# QuickViz Viz-Plane Ingestion Contract + +- Status: **v0.1 — normative; budgets pinned from the first benchmark run (2026-07-11)** +- Benchmark suite: `tests/benchmarks/ingestion/` (run: `./scripts/bench_ingestion.sh`) +- Reference report: `tests/benchmarks/ingestion/results/2026-07-11_i7-8700_xvfb-llvmpipe.json` + +## Why this contract exists + +QuickViz is the third stage of the XMotion observability chain: telemetry instrumentation (~100 ns class, xmTelemetry S1 gate) → transport (xmMessaging publish 64 B p50 ~100–140 ns, M9 gate) → **visualization ingestion (this contract)**. The design intent for the whole chain is high-performance, low-overhead telemetry and data visualization: MHz-class ingestion with published nanosecond overhead, data ingestion fully decoupled from a fixed-rate render thread — *the robot must not be able to tell it is being watched*. The first two stages hold this bar with measured, regression-gated numbers; this document makes the same claims measurable for the viz plane. + +This contract governs the **ingestion side only**. Rendering throughput (fps, draw calls, GPU upload strategies) is a separate concern with its own notes under `docs/notes/`. + +## Definitions + +- **Producer** — a robot/application thread (control loop, sensor callback, ROS2 subscriber, xmMessaging take-loop) that has a fresh data sample and wants it visualized. Producers are latency-critical; the viz plane is not allowed to perturb them. +- **Ingestion boundary** — the set of library calls a producer makes to hand a sample to the viz plane. Today these are exactly: + - `RingBuffer::Write(const T&)` — lossless-until-full time-series path consumed by plot widgets (`src/core/include/core/buffer/ring_buffer.hpp:182`); + - `DataStream::Push(...)` → `DoubleBuffer::Write(...)` — latest-only path for scene/streaming data (`src/core/include/core/data_stream.hpp:56`, `src/core/include/core/buffer/double_buffer.hpp:26`); + - `BufferRegistry::AddBuffer/GetBuffer` — **wiring-time only**; never called per sample (`src/core/include/core/buffer/buffer_registry.hpp:43`). +- **Render plane** — everything at frame cadence on the GL main thread: `Viewer::Show()` loop (`src/viewer/src/viewer.cpp:374`), panel `Draw()` including `RtLinePlotWidget::Draw()` draining ring buffers (`src/plot/src/rt_line_plot_widget.cpp:61`), `SceneManager` pre-draw callbacks pulling streams (`src/scene/src/scene_manager.cpp:121`), and renderable setters such as `PointCloud::SetPoints` (`src/scene/src/renderable/point_cloud.cpp:270`). Renderable setters are **render-plane calls, not ingestion calls**: they are unsynchronized and must only run on the render thread (via a pre-draw callback), as `sample/streaming_demo/main.cpp` demonstrates. + +## Current ingestion architecture (facts, as of 2026-07-11) + +**Path A — plot time series.** Producer writes `RtLinePlotWidget::DataPoint` (8 B) into a `RingBuffer` registered by name in `BufferRegistry`. Per sample: one `std::lock_guard` acquire/release plus an 8-byte copy into a member `std::array` — no allocation (`ring_buffer.hpp:182–194, 207`). Capacity `N` is a compile-time power of two (default 1024); usable capacity is `N-1`. On full: with `enable_overwrite_` (constructor default `true`, `ring_buffer.hpp:74`) the read index is advanced — **drop-oldest, silent, uncounted**; with overwrite disabled, `Write` returns 0 — **refusal, uncounted**. The render thread drains in `RtLinePlotWidget::Draw` (`rt_line_plot_widget.cpp:69–79`): `GetOccupiedSize()` then one `Read(pt)` per sample — one mutex acquisition per drained sample on the *same* mutex the producer takes — into a `ScrollingPlotBuffer` (fixed capacity 2048, overwriting, `scrolling_plot_buffer.cpp:35–42`) which ImPlot reads. + +**Path B — latest-only stream.** Producer calls `DataStream::Push`, which is `DoubleBuffer::Write`: one mutex acquire, full `T` copy/move-assign into one of two slots, index swap, `ready` flag store, `notify_one()` (`double_buffer.hpp:26–36`). **Latest-only overwrite, silent, uncounted.** The render thread calls `TryPull` once per frame (`TryRead`, `double_buffer.hpp:52–62`): non-blocking, copies the whole `T` out *while holding the same mutex*. `DoubleBuffer::Read` (blocking condition-variable wait, `double_buffer.hpp:38`) exists but is not part of the render path and must never be. + +**Path C — events (excluded).** `ThreadSafeQueue` (`src/core/include/core/event/thread_safe_queue.hpp:49`) is an **unbounded** `std::queue` with per-push allocation, feeding `AsyncEventDispatcher`. It is an eventing utility, **not a data ingestion path**: using it for sample streams violates I2 and I6 below by construction. + +Blocking analysis: neither `Write` ever waits on a condition — the only producer-side wait is mutex contention with the render-plane drain. For Path A the render thread holds the lock for one 8-byte copy at a time; for Path B it holds the lock for a full `sizeof(T)`-copy, so producer worst-case wait scales with the payload. A *stalled* render thread (stuck in a panel `Draw`, vsync-blocked, hidden window) holds no ingestion lock, so a full render stall cannot stall a producer — measured below. + +## Requirements (normative) + +Requirement IDs are `I1`–`I7`. Each is testable; the benchmark suite implements the tests. "Sample" below means a scalar time-series point (8 B) or a POD pose/small-struct (≤64 B) unless stated otherwise. + +### I1 — Bounded per-sample ingestion cost + +A producer's per-sample cost at the ingestion boundary is sub-microsecond at p99, at every supported rate up to at least 100 kHz per stream. Budgets (pinned 2026-07-11 from the first reference run, with headroom over measured values; re-pin requires a recorded decision): + +| Budget | Operation | Metric | Bound | Measured (reference run) | +|---|---|---|---|---| +| B1 | scalar ring `Write`, burst (batch-64 mean) | p50 / p99.9 | ≤ 50 ns / ≤ 300 ns | 21.6 ns / 109 ns | +| B2 | pose (64 B POD) `DataStream::Push`, burst (batch-64 mean) | p50 / p99.9 | ≤ 50 ns / ≤ 300 ns | 10.3 ns / 14.3 ns | +| B3 | scalar ring `Write`, spin-paced 100 kHz, true per-op tails | p50 / p99 / p99.9 | ≤ 150 ns / ≤ 500 ns / ≤ 2 µs | 31 ns / 176 ns / 387 ns | +| B4 | pose `Push`, spin-paced 100 kHz, true per-op tails | p50 / p99 / p99.9 | ≤ 150 ns / ≤ 500 ns / ≤ 2 µs | 22 ns / 123 ns / 162 ns | + +Methodology note: sleep-paced 1 kHz rows measure ~1.1 µs p50 on the reference machine in **every** mode including fully headless — that is post-wakeup DVFS/cache-warm cost on a `powersave` governor, a property of the producer's own sleep, not of the ingestion call. Rate-independence (I4) is therefore judged on like-for-like pacing, and budgets are pinned on burst and spin-paced rows. + +### I2 — Allocation-free steady state + +After wiring time (buffer registration, declared capacities) and a declared warm-up, ingestion of trivially-copyable samples performs **zero heap allocations** on the producer thread. Proven by allocation probe (the family S1 methodology) on every measured section; a single allocation fails the benchmark run. Non-POD payloads (e.g. `DataStream`) are only allocation-free if the application keeps slot capacities stable; the library does not yet give a way to pre-size `DoubleBuffer` slots — see conformance. + +### I3 — Producer never blocks on the render plane + +No ingestion call may wait for the render plane: no condition-variable waits, no frame-paced handshakes, no unbounded lock holds. A complete render stall (frame time ≥ 1 s) must leave producer-side ingestion tails statistically unchanged. Overflow policy must be **explicit and counted**: Path A is drop-oldest (or refusal when overwrite is disabled), Path B is latest-only overwrite — both MUST expose monotonic drop/overwrite/refusal counters queryable by the application. (Counters do not exist today — this is the contract's principal gap; see conformance.) + +### I4 — Render-rate independence + +Per-sample ingestion cost is invariant to render-plane state: rendering at 60 fps, rendering stalled, window hidden, or no window at all. Bound: windowed and stalled p99 within 3× of like-paced headless p99, and absolute p99 within B3/B4 bounds regardless of mode. + +### I5 — Observer invisibility end-to-end + +The M10-A4 acceptance shape, applied to the viz plane: a 1 kHz producer loop's period-jitter tails with full-rate ingestion *and* live rendering attached are statistically indistinguishable from the same loop with ingestion compiled to no-ops and no viz plane in the process. Judged at p99 on |period − 1 ms| with both distributions measured in the same session on the same core budget. + +### I6 — Bounded memory + +Every ingestion channel has a declared, wiring-time capacity: `RingBuffer` capacity is a compile-time constant; `DoubleBuffer`/`DataStream` hold exactly two slots; `ScrollingPlotBuffer` is a fixed overwriting window. Nothing on the data path may grow with time or with produced-sample count. Proven by soak: RSS growth ≈ 0 (≤ 2 MiB drift tolerance) over ≥ 60 s at 100 kHz scalar + 1 kHz pose with live rendering. `ThreadSafeQueue` is unbounded and therefore excluded from the data plane (Path C above). + +### I7 — Measured claims + +Every budget in this contract is benchmarked in-tree (`tests/benchmarks/ingestion/`), runnable with one command (`scripts/bench_ingestion.sh`), emitting a machine-readable JSON report that embeds the hardware context (CPU model, governor, kernel, RT patch, load, display/GL mode) — a context-less report is a failing run. Tails are reported as p50/p99/p99.9/max, never means alone. Once budgets are pinned (this document), a regression beyond them fails the run; wiring that gate into CI is an open item (see conformance). + +## Accepted divergences from the transport-layer rules + +- **The render thread is a legitimate hidden thread inside the viz plane.** xmMessaging's R3 forbids library-owned threads; QuickViz *is* the render loop. The divergence is contained: the render thread may allocate, block on vsync, and stall — the contract only forbids any of that leaking across the ingestion boundary to producers. +- **A mutex at the boundary is accepted at v0.1** (transport uses lock-free seqlock/SPSC structures). Accepted because the measured tails hold the budgets with margin; if a future payload or core-count profile breaches B1–B4, the remediation list (below) already names the lock-free replacement as the fix, not a budget relaxation. +- **`notify_one()` on the Path B write side is accepted** while no render-path code uses the blocking `Read` (no waiter → no futex wake). Introducing a waiter on the render path would move wake costs into the producer and is forbidden by I3. + +## Conformance: quickviz vs this contract (2026-07-11 reference run) + +Reference hardware: i7-8700 (12 threads), `powersave` governor, Linux 5.15.0-185-generic non-RT, offscreen X (Xvfb) with llvmpipe software GL, render loop paced to ~60 fps in-process (no vsync exists offscreen; pacing panel documented in the suite). Display mode does not affect ingestion-side numbers — demonstrated by the mode-invariance results themselves. + +| Req | Status | Evidence / gap | +|---|---|---| +| I1 bounded cost | **conforms** | B1–B4 all hold with ≥2× margin (see budget table). | +| I2 alloc-free | **partial** | Zero allocations on every gated row (probe). Gap: no API to pre-size `DoubleBuffer`/`DataStream` slots for heap-owning payloads, so alloc-freedom for e.g. point clouds rests on application discipline, not a library guarantee. | +| I3 never blocks + counted overflow | **partial** | Never-blocks: proven — with the render thread stalled to ~1 fps, 100 kHz producer p99 = 84 ns (vs 126 ns at 60 fps, 176 ns headless). Gap (**principal**): drop-oldest, write-refusal, and latest-only overwrite are all silent and uncounted — no counter exists anywhere in `core/buffer`. Silent-loss violates the family's observability rule. | +| I4 render-rate independence | **conforms** | Spin-paced 100 kHz scalar p50/p99: headless 31/176 ns, 60 fps 76/126 ns, stalled 40/84 ns — invariant within noise, all within budget. | +| I5 observer invisibility | **partial (not proven on reference host)** | 1 kHz loop period jitter, no-op headless vs ingestion+60 fps rendering attached: p99 68 µs → 78 µs (+15%), but p50 1.0 µs → 7.6 µs and p99.9 115 µs → 991 µs. The degradation is CPU contention from **llvmpipe software rasterization** occupying cores (the paced I4 rows prove the ingestion boundary itself is mode-invariant at tens of ns), not boundary coupling — but the contract judges end-to-end invisibility, and on this host it does not hold. Must be re-proven on a hardware-GL host (and ideally a pinned-core / performance-governor setup) before the invisibility claim is made. | +| I6 bounded memory | **conforms (data path)** | Soak: RSS growth ≈ 0 KiB over the soak window at 100 kHz + rendering (see `soak/rss_growth` row). All data-path buffers bounded by construction. Caveat: `ThreadSafeQueue` (event path) is unbounded — kept out of the data plane by this contract, flagged for its own bound. | +| I7 measured claims | **partial** | Suite exists in-tree, one-command, JSON + hardware context, alloc-gated. Gap: not yet a CI regression gate (no pinned-reference comparison job like telemetry S1 / messaging M9). | + +## Remediation list (ordered by severity against this contract) + +1. **I3 — add overflow accounting** to `RingBuffer` (overwrite + refusal counters) and `DoubleBuffer`/`DataStream` (overwrite counter), monotonic, relaxed-atomic, queryable by the application and drainable into telemetry. This is the only *correctness*-class gap: today a robot losing plot samples cannot know it. +2. **I7 — CI regression gate**: pin `results/2026-07-11_*.json` as the reference, add a compare step (the xmMessaging `bench/compare.py` shape) to CI so budget breaches fail the pipeline. +3. **I2 — slot pre-sizing** for `DataStream`/`DoubleBuffer` (e.g. `Reserve(args...)` constructing both slots at declared capacity) so heap-owning payloads get a library-backed alloc-free steady state instead of a usage convention. +4. **I3/I1 hardening (optional, measured-first)**: replace the Path B mutex with a seqlock latest-slot for trivially-copyable `T` (the xmMessaging `LatestSlot` shape) to remove producer waits proportional to `sizeof(T)` during render-side copies. Only if a real payload profile breaches budgets — the contract forbids speculative rework. +5. **I5 — re-run the observer A/B on a hardware-GL host** (and ideally performance governor / pinned cores): the reference environment renders with llvmpipe, which burns CPU cores and perturbs the producer loop's scheduler tails (p99.9 115 µs → 991 µs) independently of the ingestion boundary. The invisibility number stays unpinned until measured where rendering does not compete for the producer's cores. + +## How to run + +```bash +./scripts/bench_ingestion.sh # full suite (~3 min), auto-xvfb when headless +./scripts/bench_ingestion.sh --smoke # quick pass +./scripts/bench_ingestion.sh --filter=paced # one group +``` + +The suite prints a human summary and writes `ingestion_bench_report.json` (schema `quickviz-ingestion-bench-v1`). Windowed groups skip with a recorded note when no display and no Xvfb are available. diff --git a/scripts/bench_ingestion.sh b/scripts/bench_ingestion.sh new file mode 100755 index 0000000..db6d9b9 --- /dev/null +++ b/scripts/bench_ingestion.sh @@ -0,0 +1,54 @@ +#!/usr/bin/env bash +# +# scripts/bench_ingestion.sh — one-command runner for the viz-plane +# ingestion benchmark suite (docs/ingestion_contract.md). +# +# Builds the bench target if needed, then runs it. When no display is +# available, the windowed groups are run under xvfb-run (offscreen X server, +# software GL) — the render mode is recorded in the JSON report's hardware +# context and does not affect ingestion-side numbers if decoupling holds, +# which is exactly what the suite verifies. +# +# Usage: +# ./scripts/bench_ingestion.sh [--smoke] [--filter=] \ +# [--out=] [--build-dir=] + +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +BUILD_DIR="${REPO_ROOT}/build" +PASSTHROUGH=() + +for arg in "$@"; do + case "${arg}" in + --build-dir=*) BUILD_DIR="${arg#--build-dir=}" ;; + *) PASSTHROUGH+=("${arg}") ;; + esac +done + +if [ ! -d "${BUILD_DIR}" ]; then + echo "-- configuring ${BUILD_DIR}" + cmake -S "${REPO_ROOT}" -B "${BUILD_DIR}" \ + -DCMAKE_BUILD_TYPE=Release -DBUILD_TESTING=ON +fi + +echo "-- building bench_ingestion" +cmake --build "${BUILD_DIR}" --target bench_ingestion -j "$(nproc)" + +BIN="${BUILD_DIR}/bin/bench_ingestion" +if [ ! -x "${BIN}" ]; then + # Fall back to wherever the build placed it. + BIN="$(find "${BUILD_DIR}" -name bench_ingestion -type f | head -n1)" +fi + +RUNNER=() +if [ -z "${DISPLAY:-}" ] && [ -z "${WAYLAND_DISPLAY:-}" ]; then + if command -v xvfb-run > /dev/null 2>&1; then + echo "-- no display detected: running under xvfb-run (offscreen X)" + RUNNER=(xvfb-run -a) + else + echo "-- WARNING: no display and no xvfb-run; windowed groups will skip" + fi +fi + +exec "${RUNNER[@]}" "${BIN}" "${PASSTHROUGH[@]}" diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 3e76051..726412d 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -146,6 +146,31 @@ else() message(STATUS "Google Benchmark not found - skipping performance benchmarks") endif() +# ============================================================================== +# Ingestion Contract Benchmarks (docs/ingestion_contract.md) +# ============================================================================== + +# Hand-rolled harness (no google-benchmark): tails + alloc probe + JSON +# report with hardware context, matching the family methodology +# (xmTelemetry S1 gate, xmMessaging M9 suite). Uses the public library API +# only. Run via scripts/bench_ingestion.sh (wraps xvfb-run when headless). +add_executable(bench_ingestion + benchmarks/ingestion/ingestion_bench.cpp +) +target_link_libraries(bench_ingestion + PRIVATE + plot +) +target_compile_options(bench_ingestion PRIVATE -Wall -Wextra) + +# Smoke run as a ctest entry (needs a display or xvfb, like the other GL +# tests). Kept out of the default label set via the "benchmark" label. +add_test(NAME bench_ingestion_smoke + COMMAND bench_ingestion --smoke + --out=${CMAKE_CURRENT_BINARY_DIR}/ingestion_bench_smoke.json +) +set_tests_properties(bench_ingestion_smoke PROPERTIES LABELS "benchmark") + # ============================================================================== # Valgrind Memory Testing # ============================================================================== diff --git a/tests/benchmarks/ingestion/alloc_probe.hpp b/tests/benchmarks/ingestion/alloc_probe.hpp new file mode 100644 index 0000000..122481a --- /dev/null +++ b/tests/benchmarks/ingestion/alloc_probe.hpp @@ -0,0 +1,89 @@ +/* + * @file alloc_probe.hpp + * @date 2026-07-11 + * @brief RAII allocation counter for the ingestion benchmark suite + * (docs/ingestion_contract.md, requirement I2). + * + * The family S1 methodology (telemetry perf tier, xmMessaging M9): replace + * the global allocation functions and count allocations made ON THE PROBING + * THREAD between AllocProbe construction and query. Include from exactly + * ONE translation unit per binary — the replacement operator new/delete + * definitions below are deliberately non-inline (one definition per + * program). + * + * Copyright (c) 2026 Ruixiang Du (rdu) + */ + +#ifndef QUICKVIZ_TESTS_BENCHMARKS_INGESTION_ALLOC_PROBE_HPP +#define QUICKVIZ_TESTS_BENCHMARKS_INGESTION_ALLOC_PROBE_HPP + +#include +#include +#include + +namespace quickviz { +namespace bench { + +inline thread_local bool g_alloc_counting = false; +inline thread_local std::uint64_t g_alloc_count = 0; + +class AllocProbe { + public: + AllocProbe() { + g_alloc_count = 0; + g_alloc_counting = true; + } + ~AllocProbe() { g_alloc_counting = false; } + AllocProbe(const AllocProbe&) = delete; + AllocProbe& operator=(const AllocProbe&) = delete; + + std::uint64_t allocations() const { return g_alloc_count; } +}; + +inline void* CountingAlloc(std::size_t size) { + if (g_alloc_counting) { + ++g_alloc_count; + } + if (void* p = std::malloc(size != 0 ? size : 1)) { + return p; + } + throw std::bad_alloc(); +} + +inline void* CountingAllocNoThrow(std::size_t size) noexcept { + if (g_alloc_counting) { + ++g_alloc_count; + } + return std::malloc(size != 0 ? size : 1); +} + +} // namespace bench +} // namespace quickviz + +// Replaceable global allocation functions. Every form is replaced (throwing +// and nothrow) so allocation/deallocation pairs stay malloc/free-consistent +// under sanitizer interceptors. +void* operator new(std::size_t size) { + return quickviz::bench::CountingAlloc(size); +} +void* operator new[](std::size_t size) { + return quickviz::bench::CountingAlloc(size); +} +void* operator new(std::size_t size, const std::nothrow_t&) noexcept { + return quickviz::bench::CountingAllocNoThrow(size); +} +void* operator new[](std::size_t size, const std::nothrow_t&) noexcept { + return quickviz::bench::CountingAllocNoThrow(size); +} +void operator delete(void* ptr) noexcept { std::free(ptr); } +void operator delete[](void* ptr) noexcept { std::free(ptr); } +void operator delete(void* ptr, std::size_t) noexcept { std::free(ptr); } +void operator delete[](void* ptr, std::size_t) noexcept { std::free(ptr); } +void operator delete(void* ptr, const std::nothrow_t&) noexcept { + std::free(ptr); +} +void operator delete[](void* ptr, const std::nothrow_t&) noexcept { + std::free(ptr); +} + +#endif // QUICKVIZ_TESTS_BENCHMARKS_INGESTION_ALLOC_PROBE_HPP diff --git a/tests/benchmarks/ingestion/bench_harness.hpp b/tests/benchmarks/ingestion/bench_harness.hpp new file mode 100644 index 0000000..298ae3b --- /dev/null +++ b/tests/benchmarks/ingestion/bench_harness.hpp @@ -0,0 +1,346 @@ +/* + * @file bench_harness.hpp + * @date 2026-07-11 + * @brief Hand-rolled measurement harness for the ingestion benchmark suite + * (docs/ingestion_contract.md, requirement I7). + * + * No google-benchmark: the XMotion family vendors none (the telemetry perf + * tier and the xmMessaging M9 suite are the same hand-rolled shape), and a + * benchmark framework would be a new external dependency for a job this + * small. What the contract actually requires is here: tail percentiles + * (p50/p99/p99.9/max — never means alone), hardware context embedded in + * every report, and a machine-readable JSON artifact. + * + * Percentiles use the nearest-rank method on the sorted sample vector. + * Burst-micro benchmarks batch back-to-back ops per timed sample + * (amortizing the ~20 ns clock read below the cost of the measured op); + * the batch size is recorded because a batched "max" is a max of batch + * MEANS. Paced benchmarks time individual calls, so their tails are true + * per-sample tails (each carries ~two clock reads of overhead). + * + * Copyright (c) 2026 Ruixiang Du (rdu) + */ + +#ifndef QUICKVIZ_TESTS_BENCHMARKS_INGESTION_BENCH_HARNESS_HPP +#define QUICKVIZ_TESTS_BENCHMARKS_INGESTION_BENCH_HARNESS_HPP + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace quickviz { +namespace bench { + +// --------------------------------------------------------------------------- +// Statistics: tails, not means. +// --------------------------------------------------------------------------- + +struct Stats { + double p50 = 0.0; + double p99 = 0.0; + double p999 = 0.0; + double max = 0.0; + double mean = 0.0; + std::size_t samples = 0; +}; + +// Nearest-rank percentile over a SORTED vector. +inline double PercentileSorted(const std::vector& sorted, double q) { + if (sorted.empty()) { + return 0.0; + } + const auto rank = static_cast( + std::ceil(q * static_cast(sorted.size()))); + const std::size_t index = rank == 0 ? 0 : rank - 1; + return sorted[std::min(index, sorted.size() - 1)]; +} + +inline Stats ComputeStats(std::vector samples) { + Stats stats; + stats.samples = samples.size(); + if (samples.empty()) { + return stats; + } + std::sort(samples.begin(), samples.end()); + double sum = 0.0; + for (double v : samples) { + sum += v; + } + stats.p50 = PercentileSorted(samples, 0.50); + stats.p99 = PercentileSorted(samples, 0.99); + stats.p999 = PercentileSorted(samples, 0.999); + stats.max = samples.back(); + stats.mean = sum / static_cast(samples.size()); + return stats; +} + +// --------------------------------------------------------------------------- +// One benchmark's report row. +// --------------------------------------------------------------------------- + +struct BenchResult { + std::string name; // e.g. "render60/ring_write_scalar/1kHz" + std::string group; // "micro" | "paced" | "render" | "observer" | "soak" + std::string mode; // "headless" | "windowed" | "stalled" + std::string unit = "ns"; + double rate_hz = 0.0; // producer pacing rate (0 = unpaced burst) + int batch = 1; // ops per timed sample (1 = true per-op tails) + Stats stats; + std::uint64_t allocations = 0; // measured-section allocations (probe) + bool alloc_gated = false; // true: allocations != 0 fails the run + double achieved_fps = -1.0; // render-plane frame rate, when windowed + std::string notes; +}; + +// --------------------------------------------------------------------------- +// Hardware context: a context-less report is a failing run (I7). +// --------------------------------------------------------------------------- + +struct HardwareContext { + std::string cpu_model = "unknown"; + std::string governor = "unknown"; + std::string kernel = "unknown"; + std::string display = "unknown"; // e.g. ":0", "xvfb", "none" + std::string gl_renderer = "unknown"; + bool preempt_rt = false; + unsigned nproc = 0; + double load_avg_1m = -1.0; + double load_avg_5m = -1.0; + double load_avg_15m = -1.0; +}; + +inline std::string ReadFirstLine(const char* path) { + std::ifstream in(path); + std::string line; + if (in && std::getline(in, line)) { + return line; + } + return {}; +} + +inline HardwareContext CaptureHardwareContext() { + HardwareContext hw; + + { + std::ifstream cpuinfo("/proc/cpuinfo"); + std::string line; + while (cpuinfo && std::getline(cpuinfo, line)) { + if (line.rfind("model name", 0) == 0) { + const auto colon = line.find(':'); + if (colon != std::string::npos) { + auto value = line.substr(colon + 1); + const auto first = value.find_first_not_of(" \t"); + hw.cpu_model = + first == std::string::npos ? value : value.substr(first); + } + break; + } + } + } + + const std::string governor = + ReadFirstLine("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor"); + if (!governor.empty()) { + hw.governor = governor; + } + + struct utsname uts {}; + if (::uname(&uts) == 0) { + hw.kernel = uts.release; + hw.preempt_rt = + std::string(uts.version).find("PREEMPT_RT") != std::string::npos; + } + if (ReadFirstLine("/sys/kernel/realtime") == "1") { + hw.preempt_rt = true; + } + + hw.nproc = std::thread::hardware_concurrency(); + + { + std::ifstream loadavg("/proc/loadavg"); + if (loadavg) { + loadavg >> hw.load_avg_1m >> hw.load_avg_5m >> hw.load_avg_15m; + } + } + + const char* display_env = std::getenv("DISPLAY"); + hw.display = (display_env != nullptr && display_env[0] != '\0') + ? display_env + : "none"; + return hw; +} + +// Current resident set size in KiB (VmRSS from /proc/self/status). +inline long ReadVmRssKiB() { + std::ifstream status("/proc/self/status"); + std::string line; + while (status && std::getline(status, line)) { + if (line.rfind("VmRSS:", 0) == 0) { + long kib = 0; + std::sscanf(line.c_str(), "VmRSS: %ld", &kib); + return kib; + } + } + return -1; +} + +// --------------------------------------------------------------------------- +// JSON report writer. Hand-written emitter: the schema is small, fixed, and +// first-party — a JSON library would be a dependency for nothing. +// --------------------------------------------------------------------------- + +inline void JsonEscapeTo(std::string& out, const std::string& value) { + for (char c : value) { + switch (c) { + case '"': + out += "\\\""; + break; + case '\\': + out += "\\\\"; + break; + case '\n': + out += "\\n"; + break; + case '\t': + out += "\\t"; + break; + default: + if (static_cast(c) >= 0x20) { + out += c; + } + break; + } + } +} + +inline std::string UtcTimestamp() { + char buffer[32] = {}; + const std::time_t now = std::time(nullptr); + std::tm tm_utc{}; + gmtime_r(&now, &tm_utc); + std::strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%SZ", &tm_utc); + return buffer; +} + +inline bool WriteJsonReport(const std::string& path, const HardwareContext& hw, + const std::vector& results, + bool smoke, + const std::vector& gate_failures) { + std::string out; + out.reserve(64 * 1024); + char num[64]; + + const auto field_str = [&](const char* key, const std::string& value, + bool trailing_comma = true) { + out += '"'; + out += key; + out += "\": \""; + JsonEscapeTo(out, value); + out += '"'; + if (trailing_comma) { + out += ", "; + } + }; + const auto field_num = [&](const char* key, double value, + bool trailing_comma = true) { + std::snprintf(num, sizeof(num), "%.1f", value); + out += '"'; + out += key; + out += "\": "; + out += num; + if (trailing_comma) { + out += ", "; + } + }; + const auto field_int = [&](const char* key, long long value, + bool trailing_comma = true) { + std::snprintf(num, sizeof(num), "%lld", value); + out += '"'; + out += key; + out += "\": "; + out += num; + if (trailing_comma) { + out += ", "; + } + }; + const auto field_bool = [&](const char* key, bool value, + bool trailing_comma = true) { + out += '"'; + out += key; + out += "\": "; + out += value ? "true" : "false"; + if (trailing_comma) { + out += ", "; + } + }; + + out += "{\n "; + field_str("schema", "quickviz-ingestion-bench-v1"); + field_str("generated_at_utc", UtcTimestamp()); + field_bool("smoke", smoke, false); + out += ",\n \"hardware\": { "; + field_str("cpu_model", hw.cpu_model); + field_int("nproc", hw.nproc); + field_str("governor", hw.governor); + field_str("kernel", hw.kernel); + field_bool("preempt_rt", hw.preempt_rt); + field_str("display", hw.display); + field_str("gl_renderer", hw.gl_renderer); + field_num("load_avg_1m", hw.load_avg_1m); + field_num("load_avg_5m", hw.load_avg_5m); + field_num("load_avg_15m", hw.load_avg_15m, false); + out += " },\n \"alloc_gate\": { "; + field_bool("passed", gate_failures.empty()); + out += "\"failures\": ["; + for (std::size_t i = 0; i < gate_failures.size(); ++i) { + out += i == 0 ? "\"" : ", \""; + JsonEscapeTo(out, gate_failures[i]); + out += '"'; + } + out += "] },\n \"benchmarks\": [\n"; + for (std::size_t i = 0; i < results.size(); ++i) { + const BenchResult& r = results[i]; + out += " { "; + field_str("name", r.name); + field_str("group", r.group); + field_str("mode", r.mode); + field_str("unit", r.unit); + field_num("rate_hz", r.rate_hz); + field_int("batch", r.batch); + field_int("samples", static_cast(r.stats.samples)); + field_num("p50", r.stats.p50); + field_num("p99", r.stats.p99); + field_num("p999", r.stats.p999); + field_num("max", r.stats.max); + field_num("mean", r.stats.mean); + field_int("allocations", static_cast(r.allocations)); + field_bool("alloc_gated", r.alloc_gated); + field_num("achieved_fps", r.achieved_fps, !r.notes.empty()); + if (!r.notes.empty()) { + field_str("notes", r.notes, false); + } + out += i + 1 < results.size() ? " },\n" : " }\n"; + } + out += " ]\n}\n"; + + std::FILE* file = std::fopen(path.c_str(), "w"); + if (file == nullptr) { + return false; + } + const bool ok = std::fwrite(out.data(), 1, out.size(), file) == out.size(); + return std::fclose(file) == 0 && ok; +} + +} // namespace bench +} // namespace quickviz + +#endif // QUICKVIZ_TESTS_BENCHMARKS_INGESTION_BENCH_HARNESS_HPP diff --git a/tests/benchmarks/ingestion/ingestion_bench.cpp b/tests/benchmarks/ingestion/ingestion_bench.cpp new file mode 100644 index 0000000..7390f2d --- /dev/null +++ b/tests/benchmarks/ingestion/ingestion_bench.cpp @@ -0,0 +1,796 @@ +/* + * @file ingestion_bench.cpp + * @date 2026-07-11 + * @brief Benchmark suite for the QuickViz viz-plane ingestion contract + * (docs/ingestion_contract.md). + * + * Measures the producer-side cost of getting a data sample into a rendered + * chart/scene, through the library's PUBLIC ingestion boundary only: + * + * - RingBuffer::Write (plot time-series path, drop-oldest) + * - DataStream::Push (scene latest-only path, DoubleBuffer) + * + * Groups (see the contract's requirement I7 and the conformance table): + * micro — burst max-rate per-sample cost, batched, alloc-gated + * paced — per-sample tails at 1 kHz / 100 kHz, headless + * render — same producers with a real Viewer + RtLinePlotWidget / + * stream-drain panel rendering at ~60 fps (I4) + * stalled — same producers with the render thread deliberately stalled + * ~1 s per frame — the producer-never-blocks proof (I3) + * observer — 1 kHz producer loop period jitter, ingestion attached with + * rendering vs ingestion compiled to no-ops (I5, the M10-A4 + * shape) + * soak — RSS growth over a long high-rate run (I6) + * + * Threading: the render loop runs on the MAIN thread (the GL rule, + * CLAUDE.md section 8); producers run on background threads, exactly like a + * real robot application. Producers stop the viewer via + * Viewer::SetWindowShouldClose(), which GLFW documents as callable from any + * thread. + * + * Alloc gate: the producer-side measured sections run under AllocProbe + * (thread-local, producer thread only — the render plane is ALLOWED to + * allocate, the ingestion boundary is not). A single allocation in a gated + * section fails the run. + * + * Copyright (c) 2026 Ruixiang Du (rdu) + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "alloc_probe.hpp" // ONE TU per binary: defines operator new/delete +#include "bench_harness.hpp" + +// Library headers are not warning-clean under -Wall -Wextra (pre-existing: +// unused parameter/variable in viewer input headers); keep the bench TU +// zero-warning without modifying src/. +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wunused-parameter" +#pragma GCC diagnostic ignored "-Wunused-variable" +#include "core/buffer/buffer_registry.hpp" +#include "core/buffer/ring_buffer.hpp" +#include "core/data_stream.hpp" +#include "plot/rt_line_plot_widget.hpp" +#include "viewer/panel.hpp" +#include "viewer/viewer.hpp" +#pragma GCC diagnostic pop + +namespace { + +using quickviz::DataStream; +using quickviz::Panel; +using quickviz::RingBuffer; +using quickviz::RtLinePlotWidget; +using quickviz::Viewer; +using quickviz::bench::AllocProbe; +using quickviz::bench::BenchResult; +using quickviz::bench::ComputeStats; + +using Clock = std::chrono::steady_clock; +using std::chrono::nanoseconds; + +// --------------------------------------------------------------------------- +// Sample types. +// --------------------------------------------------------------------------- + +// Scalar time-series point: the type RtLinePlotWidget consumes (8 bytes). +using ScalarPoint = RtLinePlotWidget::DataPoint; +static_assert(sizeof(ScalarPoint) == 8, "scalar sample is 8 bytes"); + +// Typical pose / small-struct sample (64 bytes, POD). +struct PoseSample { + double t; + double x, y, z; + double qw, qx, qy, qz; +}; +static_assert(sizeof(PoseSample) == 64, "pose sample is 64 bytes"); +static_assert(std::is_trivially_copyable_v, + "pose sample must be POD"); + +// Ring capacity: power of two, sized so a 100 kHz producer cannot wrap a +// 60 fps consumer's per-frame drain (1667 samples/frame << 8192). +constexpr std::size_t kRingCapacity = 8192; +using ScalarRing = RingBuffer; +using PoseRing = RingBuffer; + +// Optimization sink so measured writes cannot be elided. +volatile double g_sink = 0.0; + +// --------------------------------------------------------------------------- +// Run configuration + result registry. +// --------------------------------------------------------------------------- + +bool g_smoke = false; +std::string g_filter; +std::string g_out_path = "ingestion_bench_report.json"; +std::vector g_results; +std::vector g_gate_failures; + +bool Enabled(const std::string& name) { + return g_filter.empty() || name.find(g_filter) != std::string::npos; +} + +double Scaled(double full, double smoke) { return g_smoke ? smoke : full; } + +void Record(BenchResult result) { + if (result.alloc_gated && result.allocations != 0) { + g_gate_failures.push_back(result.name + ": " + + std::to_string(result.allocations) + + " allocation(s) in the measured section"); + } + std::printf( + "%-42s %-9s p50=%9.1f p99=%9.1f p99.9=%9.1f max=%10.1f %s " + "(n=%zu, batch=%d, alloc=%llu%s%s)\n", + result.name.c_str(), result.mode.c_str(), result.stats.p50, + result.stats.p99, result.stats.p999, result.stats.max, + result.unit.c_str(), result.stats.samples, result.batch, + static_cast(result.allocations), + result.alloc_gated ? " gated" : "", + result.achieved_fps >= 0.0 + ? (", fps=" + std::to_string(result.achieved_fps)).c_str() + : ""); + g_results.push_back(std::move(result)); +} + +// --------------------------------------------------------------------------- +// Measurement primitives. +// --------------------------------------------------------------------------- + +inline double NsBetween(Clock::time_point a, Clock::time_point b) { + return static_cast( + std::chrono::duration_cast(b - a).count()); +} + +struct MeasuredRun { + std::vector samples_ns; + std::uint64_t allocations = 0; +}; + +// Burst micro: batched back-to-back ops per timed sample (amortizes the +// ~20 ns clock read; each recorded value is a batch MEAN). +template +MeasuredRun MeasureBurst(std::size_t n_samples, int batch, Op&& op) { + MeasuredRun run; + run.samples_ns.reserve(n_samples); + AllocProbe probe; + for (std::size_t s = 0; s < n_samples; ++s) { + const auto t0 = Clock::now(); + for (int b = 0; b < batch; ++b) { + op(s * static_cast(batch) + static_cast(b)); + } + const auto t1 = Clock::now(); + run.samples_ns.push_back(NsBetween(t0, t1) / batch); + } + run.allocations = probe.allocations(); + return run; +} + +// Paced producer: one op per period, per-op timing (true per-sample tails, +// including ~two clock reads of overhead). Pacing uses sleep_until for slow +// rates and a spin-wait for fast ones (sleep granularity >> 10 us periods). +// If the pacer falls far behind (e.g. this thread was preempted), the +// schedule is re-anchored instead of bursting to catch up. +template +MeasuredRun MeasurePaced(double rate_hz, std::size_t n_ops, Op&& op) { + const auto period = nanoseconds(static_cast(1e9 / rate_hz)); + const bool spin = period < std::chrono::microseconds(200); + MeasuredRun run; + run.samples_ns.reserve(n_ops); + AllocProbe probe; + auto next = Clock::now() + period; + for (std::size_t i = 0; i < n_ops; ++i) { + if (spin) { + while (Clock::now() < next) { + } + } else { + std::this_thread::sleep_until(next); + } + const auto t0 = Clock::now(); + op(i); + const auto t1 = Clock::now(); + run.samples_ns.push_back(NsBetween(t0, t1)); + next += period; + if (t1 > next + 100 * period) { + next = t1 + period; // re-anchor after a long preemption + } + } + run.allocations = probe.allocations(); + return run; +} + +// --------------------------------------------------------------------------- +// Bench panels (public Panel API only — no library modifications). +// --------------------------------------------------------------------------- + +// Paces the render loop by sleeping inside Draw(): ~60 fps for the render +// group (no vsync exists on an offscreen X server), or a deliberate multi- +// hundred-millisecond stall for the stalled group. Also measures the +// achieved frame rate. +class PacerPanel : public Panel { + public: + explicit PacerPanel(nanoseconds frame_period) + : Panel("bench_pacer"), period_(frame_period) {} + + void Draw() override { + const auto now = Clock::now(); + if (frames_.fetch_add(1, std::memory_order_relaxed) == 0) { + first_frame_ns_.store(now.time_since_epoch().count(), + std::memory_order_relaxed); + next_ = now + period_; + } + last_frame_ns_.store(now.time_since_epoch().count(), + std::memory_order_relaxed); + std::this_thread::sleep_until(next_); + next_ += period_; + if (next_ < Clock::now()) { + next_ = Clock::now() + period_; + } + } + + double AchievedFps() const { + const auto frames = frames_.load(std::memory_order_relaxed); + const double span_ns = + static_cast(last_frame_ns_.load(std::memory_order_relaxed) - + first_frame_ns_.load(std::memory_order_relaxed)); + if (frames < 2 || span_ns <= 0.0) { + return -1.0; + } + return static_cast(frames - 1) * 1e9 / span_ns; + } + + std::uint64_t Frames() const { + return frames_.load(std::memory_order_relaxed); + } + + private: + nanoseconds period_; + Clock::time_point next_{}; + std::atomic frames_{0}; + std::atomic first_frame_ns_{0}; + std::atomic last_frame_ns_{0}; +}; + +// Render-plane consumer for the DataStream (scene latest-only) path: pulls +// at most one value per frame, exactly like the streaming_demo pre-draw +// callback. +class StreamDrainPanel : public Panel { + public: + explicit StreamDrainPanel(DataStream* stream) + : Panel("bench_stream_drain"), stream_(stream) {} + + void Draw() override { + PoseSample latest; + if (stream_->TryPull(latest)) { + g_sink = latest.t; + } + } + + private: + DataStream* stream_; +}; + +// --------------------------------------------------------------------------- +// Windowed scenario runner: render loop on the main thread, producer on a +// background thread. Returns false when no display is available. +// --------------------------------------------------------------------------- + +struct WindowedResult { + MeasuredRun run; + double achieved_fps = -1.0; + bool display_available = true; +}; + +enum class DrainKind { kPlotWidget, kStreamPanel, kBoth }; + +template +WindowedResult RunWindowedScenario(const std::string& title, + nanoseconds frame_period, DrainKind drain, + ScalarRing* scalar_ring, + DataStream* pose_stream, + ProducerFn producer) { + WindowedResult out; + std::unique_ptr viewer; + try { + viewer = std::make_unique(title, 640, 480); + } catch (const std::exception& e) { + std::fprintf(stderr, "[skip] cannot create window: %s\n", e.what()); + out.display_available = false; + return out; + } + + auto pacer = std::make_shared(frame_period); + viewer->AddSceneObject(pacer); + + // Plot-widget drain: register the ring under a unique name, wire the + // widget to it through the BufferRegistry (the documented plot path). + const std::string buffer_name = "bench." + title; + auto& registry = quickviz::BufferRegistry::GetInstance(); + std::shared_ptr ring_holder; + if (drain == DrainKind::kPlotWidget || drain == DrainKind::kBoth) { + // Registry stores shared_ptr; wrap the caller's ring without ownership. + ring_holder = std::shared_ptr(scalar_ring, [](ScalarRing*) {}); + registry.AddBuffer(buffer_name, ring_holder); + auto plot = std::make_shared("bench_plot_" + title); + plot->SetFixedHistory(10.0f); + plot->SetYAxisRange(-1.5f, 1.5f); + plot->AddLine("signal", buffer_name); + viewer->AddSceneObject(plot); + } + if (drain == DrainKind::kStreamPanel || drain == DrainKind::kBoth) { + viewer->AddSceneObject(std::make_shared(pose_stream)); + } + + std::thread producer_thread([&]() { + // Wait until the render loop is demonstrably live so the measurement + // covers ingestion-with-rendering, not ingestion-before-rendering. + while (pacer->Frames() == 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + out.run = producer(); + viewer->SetWindowShouldClose(); + }); + + viewer->Show(); + producer_thread.join(); + out.achieved_fps = pacer->AchievedFps(); + if (!buffer_name.empty() && ring_holder) { + registry.RemoveBuffer(buffer_name); + } + return out; +} + +// --------------------------------------------------------------------------- +// Producer ops. +// --------------------------------------------------------------------------- + +inline ScalarPoint MakeScalar(std::size_t i) { + const float t = static_cast(i) * 1e-3f; + return ScalarPoint{t, std::sin(t)}; +} + +inline PoseSample MakePose(std::size_t i) { + const double t = static_cast(i) * 1e-3; + return PoseSample{t, t * 0.1, t * 0.2, 0.0, 1.0, 0.0, 0.0, 0.0}; +} + +// --------------------------------------------------------------------------- +// Group: micro (headless burst, batched, alloc-gated). +// --------------------------------------------------------------------------- + +void BenchMicro() { + const auto n_samples = + static_cast(Scaled(20000, 2000)); + constexpr int kBatch = 64; + + if (Enabled("micro/ring_write_scalar/burst")) { + ScalarRing ring; // overwrite-on-full (drop-oldest), the plot default + auto run = MeasureBurst(n_samples, kBatch, [&](std::size_t i) { + ring.Write(MakeScalar(i)); + }); + BenchResult r; + r.name = "micro/ring_write_scalar/burst"; + r.group = "micro"; + r.mode = "headless"; + r.batch = kBatch; + r.stats = ComputeStats(std::move(run.samples_ns)); + r.allocations = run.allocations; + r.alloc_gated = true; + r.notes = "max-rate burst; ring in steady-state overwrite"; + Record(std::move(r)); + } + + if (Enabled("micro/ring_write_pose/burst")) { + PoseRing ring; + auto run = MeasureBurst(n_samples, kBatch, [&](std::size_t i) { + ring.Write(MakePose(i)); + }); + BenchResult r; + r.name = "micro/ring_write_pose/burst"; + r.group = "micro"; + r.mode = "headless"; + r.batch = kBatch; + r.stats = ComputeStats(std::move(run.samples_ns)); + r.allocations = run.allocations; + r.alloc_gated = true; + r.notes = "max-rate burst; ring in steady-state overwrite"; + Record(std::move(r)); + } + + if (Enabled("micro/stream_push_pose/burst")) { + DataStream stream; + auto run = MeasureBurst(n_samples, kBatch, [&](std::size_t i) { + stream.Push(MakePose(i)); + }); + BenchResult r; + r.name = "micro/stream_push_pose/burst"; + r.group = "micro"; + r.mode = "headless"; + r.batch = kBatch; + r.stats = ComputeStats(std::move(run.samples_ns)); + r.allocations = run.allocations; + r.alloc_gated = true; + r.notes = "latest-only overwrite, no consumer"; + Record(std::move(r)); + } +} + +// --------------------------------------------------------------------------- +// Group: paced (headless per-op tails at declared rates, alloc-gated). +// --------------------------------------------------------------------------- + +void BenchPaced() { + struct Case { + const char* name; + double rate_hz; + double seconds; + bool pose; + }; + const Case cases[] = { + {"paced/ring_write_scalar/1kHz", 1e3, Scaled(8, 1), false}, + {"paced/ring_write_scalar/100kHz", 1e5, Scaled(2, 0.25), false}, + {"paced/stream_push_pose/1kHz", 1e3, Scaled(8, 1), true}, + {"paced/stream_push_pose/100kHz", 1e5, Scaled(2, 0.25), true}, + }; + + for (const auto& c : cases) { + if (!Enabled(c.name)) { + continue; + } + const auto n_ops = static_cast(c.rate_hz * c.seconds); + MeasuredRun run; + if (c.pose) { + DataStream stream; + run = MeasurePaced(c.rate_hz, n_ops, + [&](std::size_t i) { stream.Push(MakePose(i)); }); + } else { + ScalarRing ring; + run = MeasurePaced(c.rate_hz, n_ops, + [&](std::size_t i) { ring.Write(MakeScalar(i)); }); + } + BenchResult r; + r.name = c.name; + r.group = "paced"; + r.mode = "headless"; + r.rate_hz = c.rate_hz; + r.stats = ComputeStats(std::move(run.samples_ns)); + r.allocations = run.allocations; + r.alloc_gated = true; + r.notes = "per-op timing incl. ~2 clock reads; no consumer"; + Record(std::move(r)); + } +} + +// --------------------------------------------------------------------------- +// Groups: render (60 fps) and stalled (render thread sleeping ~1 s/frame). +// --------------------------------------------------------------------------- + +void BenchWithRenderPlane(const std::string& group, nanoseconds frame_period, + const std::string& mode) { + struct Case { + std::string name; + double rate_hz; + double seconds; + bool pose; + }; + const Case cases[] = { + {group + "/ring_write_scalar/1kHz", 1e3, Scaled(8, 2), false}, + {group + "/ring_write_scalar/100kHz", 1e5, Scaled(3, 1), false}, + {group + "/stream_push_pose/1kHz", 1e3, Scaled(8, 2), true}, + }; + + for (const auto& c : cases) { + if (!Enabled(c.name)) { + continue; + } + const auto n_ops = static_cast(c.rate_hz * c.seconds); + + ScalarRing ring; + DataStream stream; + WindowedResult wr; + if (c.pose) { + wr = RunWindowedScenario( + c.name, frame_period, DrainKind::kStreamPanel, &ring, &stream, + [&]() { + return MeasurePaced(c.rate_hz, n_ops, [&](std::size_t i) { + stream.Push(MakePose(i)); + }); + }); + } else { + wr = RunWindowedScenario( + c.name, frame_period, DrainKind::kPlotWidget, &ring, &stream, + [&]() { + return MeasurePaced(c.rate_hz, n_ops, [&](std::size_t i) { + ring.Write(MakeScalar(i)); + }); + }); + } + if (!wr.display_available) { + std::fprintf(stderr, "[skip] %s: no display available\n", + c.name.c_str()); + continue; + } + BenchResult r; + r.name = c.name; + r.group = group; + r.mode = mode; + r.rate_hz = c.rate_hz; + r.stats = ComputeStats(std::move(wr.run.samples_ns)); + r.allocations = wr.run.allocations; + r.alloc_gated = true; + r.achieved_fps = wr.achieved_fps; + r.notes = c.pose ? "render plane pulls latest once per frame" + : "RtLinePlotWidget drains ring every frame"; + Record(std::move(r)); + } +} + +// --------------------------------------------------------------------------- +// Group: observer invisibility (I5). A 1 kHz producer loop with a fixed +// synthetic workload; measured quantity is the loop PERIOD JITTER +// |actual period - 1 ms| in ns. A/B: ingestion compiled to no-ops and no +// window, vs full-rate ingestion with a 60 fps render plane attached. +// --------------------------------------------------------------------------- + +double SyntheticWorkload(std::size_t seed) { + double x = static_cast(seed % 1024) * 1e-6 + 1.0; + for (int i = 0; i < 500; ++i) { + x = x * 1.000000001 + 1e-9; + } + return x; +} + +template +MeasuredRun ObserverLoop(std::size_t iterations, ScalarRing* ring, + DataStream* stream) { + constexpr auto kPeriod = std::chrono::milliseconds(1); + MeasuredRun run; + run.samples_ns.reserve(iterations); + AllocProbe probe; + auto next = Clock::now() + kPeriod; + auto prev = Clock::now(); + for (std::size_t i = 0; i < iterations; ++i) { + std::this_thread::sleep_until(next); + next += kPeriod; + const auto now = Clock::now(); + if (i > 0) { + run.samples_ns.push_back(std::fabs(NsBetween(prev, now) - 1e6)); + } + prev = now; + g_sink = SyntheticWorkload(i); + if constexpr (kIngest) { + ring->Write(MakeScalar(i)); + stream->Push(MakePose(i)); + } + if (Clock::now() > next + std::chrono::milliseconds(100)) { + next = Clock::now() + kPeriod; // re-anchor after a long preemption + } + } + run.allocations = probe.allocations(); + return run; +} + +void BenchObserver() { + const auto iterations = static_cast(Scaled(10000, 2000)); + + if (Enabled("observer/noop_headless")) { + auto run = ObserverLoop(iterations, nullptr, nullptr); + BenchResult r; + r.name = "observer/noop_headless"; + r.group = "observer"; + r.mode = "headless"; + r.rate_hz = 1e3; + r.stats = ComputeStats(std::move(run.samples_ns)); + r.allocations = run.allocations; + r.alloc_gated = true; + r.notes = "loop period jitter |period - 1ms|; ingestion compiled out"; + Record(std::move(r)); + } + + if (Enabled("observer/attached_render60")) { + ScalarRing ring; + DataStream stream; + auto wr = RunWindowedScenario( + "observer_attached", nanoseconds(16666667), DrainKind::kBoth, &ring, + &stream, [&]() { + return ObserverLoop(iterations, &ring, &stream); + }); + if (!wr.display_available) { + std::fprintf(stderr, "[skip] observer/attached_render60: no display\n"); + return; + } + BenchResult r; + r.name = "observer/attached_render60"; + r.group = "observer"; + r.mode = "windowed"; + r.rate_hz = 1e3; + r.stats = ComputeStats(std::move(wr.run.samples_ns)); + r.allocations = wr.run.allocations; + r.alloc_gated = true; + r.achieved_fps = wr.achieved_fps; + r.notes = + "loop period jitter |period - 1ms|; scalar+pose ingestion at 1 kHz, " + "plot + stream drain rendering"; + Record(std::move(r)); + } +} + +// --------------------------------------------------------------------------- +// Group: soak (I6). Scalar @ 100 kHz + pose @ 1 kHz into a rendering viewer +// for 60 s (smoke: 6 s); RSS sampled every 500 ms after a warm-up. Reported +// unit is KiB of growth relative to the post-warm-up baseline. +// --------------------------------------------------------------------------- + +void BenchSoak() { + if (!Enabled("soak/rss_growth")) { + return; + } + const double seconds = Scaled(60, 6); + const double warmup_s = seconds * 0.2; + + ScalarRing ring; + DataStream stream; + + std::unique_ptr viewer; + try { + viewer = std::make_unique("soak", 640, 480); + } catch (const std::exception& e) { + std::fprintf(stderr, "[skip] soak/rss_growth: no display (%s)\n", + e.what()); + return; + } + auto pacer = std::make_shared(nanoseconds(16666667)); + viewer->AddSceneObject(pacer); + auto& registry = quickviz::BufferRegistry::GetInstance(); + auto ring_holder = std::shared_ptr(&ring, [](ScalarRing*) {}); + registry.AddBuffer("bench.soak", ring_holder); + auto plot = std::make_shared("bench_plot_soak"); + plot->SetFixedHistory(10.0f); + plot->AddLine("signal", "bench.soak"); + viewer->AddSceneObject(plot); + viewer->AddSceneObject(std::make_shared(&stream)); + + std::atomic stop{false}; + std::thread scalar_producer([&]() { + const auto period = nanoseconds(10000); // 100 kHz + auto next = Clock::now() + period; + std::size_t i = 0; + while (!stop.load(std::memory_order_relaxed)) { + while (Clock::now() < next) { + } + next += period; + ring.Write(MakeScalar(i++)); + if (Clock::now() > next + std::chrono::milliseconds(10)) { + next = Clock::now() + period; + } + } + }); + std::thread pose_producer([&]() { + const auto period = std::chrono::milliseconds(1); // 1 kHz + auto next = Clock::now() + period; + std::size_t i = 0; + while (!stop.load(std::memory_order_relaxed)) { + std::this_thread::sleep_until(next); + next += period; + stream.Push(MakePose(i++)); + if (Clock::now() > next + std::chrono::milliseconds(100)) { + next = Clock::now() + period; + } + } + }); + + std::vector growth_kib; + growth_kib.reserve(static_cast(seconds * 2) + 4); + long rss_warm = -1; + long rss_end = -1; + std::thread sampler([&]() { + const auto t0 = Clock::now(); + while (!stop.load(std::memory_order_relaxed)) { + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + const double elapsed = + std::chrono::duration(Clock::now() - t0).count(); + const long rss = quickviz::bench::ReadVmRssKiB(); + if (elapsed < warmup_s || rss < 0) { + continue; + } + if (rss_warm < 0) { + rss_warm = rss; + } + rss_end = rss; + growth_kib.push_back(static_cast(rss - rss_warm)); + if (elapsed >= seconds) { + stop.store(true, std::memory_order_relaxed); + viewer->SetWindowShouldClose(); + } + } + }); + + viewer->Show(); + stop.store(true, std::memory_order_relaxed); + scalar_producer.join(); + pose_producer.join(); + sampler.join(); + registry.RemoveBuffer("bench.soak"); + + BenchResult r; + r.name = "soak/rss_growth"; + r.group = "soak"; + r.mode = "windowed"; + r.unit = "KiB"; + r.achieved_fps = pacer->AchievedFps(); + r.stats = ComputeStats(std::move(growth_kib)); + char note[160]; + std::snprintf(note, sizeof(note), + "RSS growth vs post-warmup baseline; %.0f s soak, " + "scalar@100kHz + pose@1kHz; rss_warm=%ld KiB, rss_end=%ld KiB", + seconds, rss_warm, rss_end); + r.notes = note; + Record(std::move(r)); +} + +void PrintUsage(const char* argv0) { + std::printf( + "Usage: %s [--smoke] [--filter=] [--out=]\n" + "Windowed groups need a display; run under xvfb-run on headless " + "machines (scripts/bench_ingestion.sh does this automatically).\n", + argv0); +} + +} // namespace + +int main(int argc, char** argv) { + for (int i = 1; i < argc; ++i) { + const std::string arg = argv[i]; + if (arg == "--smoke") { + g_smoke = true; + } else if (arg.rfind("--filter=", 0) == 0) { + g_filter = arg.substr(9); + } else if (arg.rfind("--out=", 0) == 0) { + g_out_path = arg.substr(6); + } else if (arg == "--help" || arg == "-h") { + PrintUsage(argv[0]); + return 0; + } else { + std::fprintf(stderr, "unknown argument: %s\n", arg.c_str()); + PrintUsage(argv[0]); + return 2; + } + } + + auto hw = quickviz::bench::CaptureHardwareContext(); + std::printf("quickviz ingestion bench%s — cpu: %s, governor: %s, " + "display: %s\n", + g_smoke ? " (smoke)" : "", hw.cpu_model.c_str(), + hw.governor.c_str(), hw.display.c_str()); + + BenchMicro(); + BenchPaced(); + BenchWithRenderPlane("render60", nanoseconds(16666667), "windowed"); + BenchWithRenderPlane("stalled", nanoseconds(1000000000), "stalled"); + BenchObserver(); + BenchSoak(); + + if (!quickviz::bench::WriteJsonReport(g_out_path, hw, g_results, g_smoke, + g_gate_failures)) { + std::fprintf(stderr, "failed to write report: %s\n", g_out_path.c_str()); + return 1; + } + std::printf("report: %s\n", g_out_path.c_str()); + + if (!g_gate_failures.empty()) { + std::fprintf(stderr, "ALLOC GATE FAILED:\n"); + for (const auto& f : g_gate_failures) { + std::fprintf(stderr, " %s\n", f.c_str()); + } + return 1; + } + return 0; +} diff --git a/tests/benchmarks/ingestion/results/2026-07-11_i7-8700_xvfb-llvmpipe.json b/tests/benchmarks/ingestion/results/2026-07-11_i7-8700_xvfb-llvmpipe.json new file mode 100644 index 0000000..fe42e01 --- /dev/null +++ b/tests/benchmarks/ingestion/results/2026-07-11_i7-8700_xvfb-llvmpipe.json @@ -0,0 +1,23 @@ +{ + "schema": "quickviz-ingestion-bench-v1", "generated_at_utc": "2026-07-11T03:28:17Z", "smoke": false, + "hardware": { "cpu_model": "Intel(R) Core(TM) i7-8700 CPU @ 3.20GHz", "nproc": 12, "governor": "powersave", "kernel": "5.15.0-185-generic", "preempt_rt": false, "display": ":99", "gl_renderer": "unknown", "load_avg_1m": 0.3, "load_avg_5m": 0.5, "load_avg_15m": 0.7 }, + "alloc_gate": { "passed": true, "failures": [] }, + "benchmarks": [ + { "name": "micro/ring_write_scalar/burst", "group": "micro", "mode": "headless", "unit": "ns", "rate_hz": 0.0, "batch": 64, "samples": 20000, "p50": 21.6, "p99": 43.8, "p999": 109.0, "max": 1267.6, "mean": 24.6, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "max-rate burst; ring in steady-state overwrite" }, + { "name": "micro/ring_write_pose/burst", "group": "micro", "mode": "headless", "unit": "ns", "rate_hz": 0.0, "batch": 64, "samples": 20000, "p50": 8.2, "p99": 26.7, "p999": 211.0, "max": 327.8, "mean": 9.8, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "max-rate burst; ring in steady-state overwrite" }, + { "name": "micro/stream_push_pose/burst", "group": "micro", "mode": "headless", "unit": "ns", "rate_hz": 0.0, "batch": 64, "samples": 20000, "p50": 10.3, "p99": 11.0, "p999": 14.3, "max": 327.5, "mean": 10.5, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "latest-only overwrite, no consumer" }, + { "name": "paced/ring_write_scalar/1kHz", "group": "paced", "mode": "headless", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 8000, "p50": 1125.0, "p99": 1365.0, "p999": 2383.0, "max": 17004.0, "mean": 893.7, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "per-op timing incl. ~2 clock reads; no consumer" }, + { "name": "paced/ring_write_scalar/100kHz", "group": "paced", "mode": "headless", "unit": "ns", "rate_hz": 100000.0, "batch": 1, "samples": 200000, "p50": 31.0, "p99": 176.0, "p999": 387.0, "max": 14087.0, "mean": 40.9, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "per-op timing incl. ~2 clock reads; no consumer" }, + { "name": "paced/stream_push_pose/1kHz", "group": "paced", "mode": "headless", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 8000, "p50": 1248.0, "p99": 1473.0, "p999": 2172.0, "max": 17077.0, "mean": 1026.9, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "per-op timing incl. ~2 clock reads; no consumer" }, + { "name": "paced/stream_push_pose/100kHz", "group": "paced", "mode": "headless", "unit": "ns", "rate_hz": 100000.0, "batch": 1, "samples": 200000, "p50": 22.0, "p99": 123.0, "p999": 162.0, "max": 18304.0, "mean": 26.5, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "per-op timing incl. ~2 clock reads; no consumer" }, + { "name": "render60/ring_write_scalar/1kHz", "group": "render60", "mode": "windowed", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 8000, "p50": 1011.0, "p99": 1468.0, "p999": 1946.0, "max": 16272.0, "mean": 825.3, "allocations": 0, "alloc_gated": true, "achieved_fps": 60.0, "notes": "RtLinePlotWidget drains ring every frame" }, + { "name": "render60/ring_write_scalar/100kHz", "group": "render60", "mode": "windowed", "unit": "ns", "rate_hz": 100000.0, "batch": 1, "samples": 300000, "p50": 76.0, "p99": 126.0, "p999": 1255.0, "max": 2852.0, "mean": 65.7, "allocations": 0, "alloc_gated": true, "achieved_fps": 60.0, "notes": "RtLinePlotWidget drains ring every frame" }, + { "name": "render60/stream_push_pose/1kHz", "group": "render60", "mode": "windowed", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 8000, "p50": 1209.0, "p99": 1630.0, "p999": 2224.0, "max": 16925.0, "mean": 969.3, "allocations": 0, "alloc_gated": true, "achieved_fps": 60.0, "notes": "render plane pulls latest once per frame" }, + { "name": "stalled/ring_write_scalar/1kHz", "group": "stalled", "mode": "stalled", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 8000, "p50": 1053.0, "p99": 1282.0, "p999": 1693.0, "max": 15131.0, "mean": 861.8, "allocations": 0, "alloc_gated": true, "achieved_fps": 1.0, "notes": "RtLinePlotWidget drains ring every frame" }, + { "name": "stalled/ring_write_scalar/100kHz", "group": "stalled", "mode": "stalled", "unit": "ns", "rate_hz": 100000.0, "batch": 1, "samples": 300000, "p50": 40.0, "p99": 84.0, "p999": 161.0, "max": 26708.0, "mean": 43.7, "allocations": 0, "alloc_gated": true, "achieved_fps": 1.0, "notes": "RtLinePlotWidget drains ring every frame" }, + { "name": "stalled/stream_push_pose/1kHz", "group": "stalled", "mode": "stalled", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 8000, "p50": 1247.0, "p99": 1515.0, "p999": 1946.0, "max": 2576.0, "mean": 1059.4, "allocations": 0, "alloc_gated": true, "achieved_fps": 1.0, "notes": "render plane pulls latest once per frame" }, + { "name": "observer/noop_headless", "group": "observer", "mode": "headless", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 9999, "p50": 1015.0, "p99": 67813.0, "p999": 115194.0, "max": 1416356.0, "mean": 7070.3, "allocations": 0, "alloc_gated": true, "achieved_fps": -1.0, "notes": "loop period jitter |period - 1ms|; ingestion compiled out" }, + { "name": "observer/attached_render60", "group": "observer", "mode": "windowed", "unit": "ns", "rate_hz": 1000.0, "batch": 1, "samples": 9999, "p50": 7618.0, "p99": 77714.0, "p999": 991300.0, "max": 1246339.0, "mean": 19543.8, "allocations": 0, "alloc_gated": true, "achieved_fps": 60.0, "notes": "loop period jitter |period - 1ms|; scalar+pose ingestion at 1 kHz, plot + stream drain rendering" }, + { "name": "soak/rss_growth", "group": "soak", "mode": "windowed", "unit": "KiB", "rate_hz": 0.0, "batch": 1, "samples": 97, "p50": 0.0, "p99": 0.0, "p999": 0.0, "max": 0.0, "mean": 0.0, "allocations": 0, "alloc_gated": false, "achieved_fps": 60.0, "notes": "RSS growth vs post-warmup baseline; 60 s soak, scalar@100kHz + pose@1kHz; rss_warm=114756 KiB, rss_end=114756 KiB" } + ] +}