From da63d58f61c1e29055945d60337f615443fcda93 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sun, 5 Jul 2026 03:26:23 -0400 Subject: [PATCH] =?UTF-8?q?feat(sim):=20TickLagGauge=20--=20per-tick=20wal?= =?UTF-8?q?l-clock=20measurement=20(slice-4=C2=BD=20S6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spec section 3.6 says the gauge polls MediaCmd::Stats from slice-5/seams MediaThread. S4 standalone-path conclusion (per kickoff hard rule) means the SimCall wires itself in tokio WITHOUT registering with the binary MediaThread -- no MediaCmd::Stats channel to poll. This S6 implementation adapts: the gauge is wired INTO the SimCalls tick loop directly via a shared Arc handle. SimCall::run_with_gauge records per-tick wall-clock duration via Instant::now() measurement around the tick work (not including the 20ms sleep -- matches MediaStats.last_tick_micros semantics). Same conceptual metric (max tick lag + overrun count + pct), different source (in-process tokio SimCall tick loop rather than binary MediaThread poll loop). Future slice (post-spearhead refinement, paired with network-realism mode) wires the gauge against the binary MediaThread per spec section 3.6 -- requires either MediaThread registration (the RegisterSim variant forbidden this slice) OR a client-server sim mode (deferred per spec 8.6). TickLagStats uses atomics (AtomicU64) -- 3 atomic ops per tick (CAS-max, conditional fetch_add on overruns, unconditional fetch_add on total) -- rather than Mutex> which would lock the vector per tick + add ordering overhead significant to the per-tick microseconds budget. ConcurrencyRunner creates one shared Arc per concurrency level + passes clones to each of the N SimCalls via run_with_gauge. After the sweep, reads the gauges fields to populate PerConcurrencyReports tick-lag fields (max_tick_lag_micros, tick_overruns, total_ticks, tick_overrun_pct) -- the ADR-0010 doctrine-drift detector. SimCall::run(self) -> LatencyProbe is preserved as a convenience default that calls run_with_gauge(TickLagStats::new()) (gauge data discarded); callers needing tick-lag data (ConcurrencyRunner) use run_with_gauge directly. Signed-off-by: Aaron D. Lee --- crates/rutster-sim/src/concurrency.rs | 28 ++- crates/rutster-sim/src/lib.rs | 1 + crates/rutster-sim/src/runner.rs | 38 +++- crates/rutster-sim/src/tick_lag.rs | 244 ++++++++++++++++++++++++-- 4 files changed, 288 insertions(+), 23 deletions(-) diff --git a/crates/rutster-sim/src/concurrency.rs b/crates/rutster-sim/src/concurrency.rs index 89b4966..9310fa0 100644 --- a/crates/rutster-sim/src/concurrency.rs +++ b/crates/rutster-sim/src/concurrency.rs @@ -39,6 +39,7 @@ use crate::latency::percentile_ms; use crate::runner::SimCall; use crate::scenario::Scenario; use crate::thresholds::SWEEP_CONCURRENCIES; +use crate::tick_lag::TickLagStats; /// The concurrency sweep runner. Spawns N `SimCall`s in parallel /// (tokio), awaits all, aggregates per-call latencies into the sweep @@ -85,13 +86,25 @@ impl ConcurrencyRunner { /// and aggregate their per-call `LatencyProbe` samples into /// p50/p99 + carry the empty tick-lag fields for S6 to fill. async fn run_one_concurrency(&self, n: usize, scenario: Scenario) -> PerConcurrencyReport { + // S6: per-level shared gauge. Each SimCall's `run_with_gauge` + // records per-tick wall-clock duration into this shared + // accumulator. After the sweep, read max_tick_lag_micros + + // tick_overruns + total_ticks + tick_overrun_pct to populate + // PerConcurrencyReport's tick-lag fields (the ADR-0010 + // doctrine-drift detector). + let gauge = TickLagStats::new(); + // Spawn N concurrent sim calls. Each task gets its own clone - // of the scenario (Scenario: Clone — cheap, just a name + vec). + // of the scenario (Scenario: Clone — cheap, just a name + vec) + // + a clone of the gauge handle (Arc — cheap refcount bump). let mut handles = Vec::with_capacity(n); for _ in 0..n { let scenario_clone = scenario.clone(); + let gauge_clone = gauge.clone(); handles.push(tokio::spawn(async move { - SimCall::new(scenario_clone).run().await + SimCall::new(scenario_clone) + .run_with_gauge(gauge_clone) + .await })); } @@ -120,13 +133,10 @@ impl ConcurrencyRunner { p99_kill_ms: percentile_ms(&all_kills, 99), p50_mouth_to_ear_ms: percentile_ms(&all_m2e, 50), p99_mouth_to_ear_ms: percentile_ms(&all_m2e, 99), - // S6 (TickLagGauge) fills these in during the sweep; S5 - // leaves them zero-initialized so the SweepReport's - // structure is stable across task landings. - max_tick_lag_micros: 0, - tick_overruns: 0, - total_ticks: 0, - tick_overrun_pct: 0.0, + max_tick_lag_micros: gauge.max_tick_lag_micros(), + tick_overruns: gauge.tick_overruns(), + total_ticks: gauge.total_ticks(), + tick_overrun_pct: gauge.tick_overrun_pct(), } } } diff --git a/crates/rutster-sim/src/lib.rs b/crates/rutster-sim/src/lib.rs index 512ada4..50abcae 100644 --- a/crates/rutster-sim/src/lib.rs +++ b/crates/rutster-sim/src/lib.rs @@ -67,3 +67,4 @@ pub use thresholds::{ BARGE_IN_KILL_TIME_P99_MS, MOUTH_TO_EAR_P99_MS, SWEEP_CONCURRENCIES, TICK_LAG_MAX_MS, TICK_OVERRUN_PCT_MAX, }; +pub use tick_lag::{TickLagGauge, TickLagStats}; diff --git a/crates/rutster-sim/src/runner.rs b/crates/rutster-sim/src/runner.rs index b0930df..7707331 100644 --- a/crates/rutster-sim/src/runner.rs +++ b/crates/rutster-sim/src/runner.rs @@ -47,7 +47,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Duration; +use std::time::{Duration, Instant}; use rutster_media::{ AdvisoryEvent, AudioSink, AudioSource, LocalVadReflex, PcmFrame, Reflex, ReflexMetrics, @@ -59,6 +59,7 @@ use tokio::task::JoinHandle; use crate::latency::LatencyProbe; use crate::scenario::Scenario; use crate::sim_audio_pipe::SimAudioPipe; +use crate::tick_lag::TickLagStats; /// One synthetic call: a `SimAudioPipe` (the caller-side recorder + /// scenario driver) + the wiring to drive it against an in-process @@ -77,7 +78,21 @@ impl SimCall { } /// Drive the scenario against the FOB reflex loop. Returns the - /// `LatencyProbe` with the captured timeline. + /// `LatencyProbe` with the captured timeline. Default gauge is + /// internal + discarded after `run` — use `run_with_gauge` to + /// observe tick-lag during the sweep (the `ConcurrencyRunner` does + /// this; standalone callers usually don't need it). + pub async fn run(self) -> LatencyProbe { + self.run_with_gauge(TickLagStats::new()).await + } + + /// Same as `run` but records per-tick wall-clock duration into the + /// shared `gauge`. The `ConcurrencyRunner` creates one shared gauge + /// per concurrency level + passes clones to each of the N + /// `SimCall`s; after the sweep, the ConcurrencyRunner reads the + /// gauge's `max_tick_lag_micros` + `tick_overruns` + `total_ticks` + + /// `tick_overrun_pct` to populate `PerConcurrencyReport` (see + /// spec §3.6). /// /// # Hot-path policy (per AGENTS.md) /// @@ -86,7 +101,7 @@ impl SimCall { /// `try_send` on a full channel drops + observes; `next_pcm_frame` /// returning `None` (muted/empty ring) captures a `BargeKillObserved` /// + continues. - pub async fn run(self) -> LatencyProbe { + pub async fn run_with_gauge(self, gauge: Arc) -> LatencyProbe { // 1. Build the Reflex stack — mirrors slice-4's // `primary_path_local_vad_kills_playout_without_brain` // test composition (crates/rutster/tests/barge_in_integration.rs:158): @@ -168,9 +183,17 @@ impl SimCall { // `CallerHeardReply` on `Some`; `BargeKillObserved` on `None` // (the LatencyProbe dedups captures without prior onset). // (d) Advance the SimPipe's scenario cursor via `on_pcm_frame`. - // (e) Termination: `scenario_done()` checks for `End` step. + // (e) Per-tick wall-clock duration recorded into `gauge` (S6) — + // the ADR-0010 doctrine-drift detector. The `Instant::now()` + // measurement wraps (a)-(d); the `tokio::time::sleep(tick)` + // is OUTSIDE the measured region (we measure tick work, not + // the wait). This matches the binary's `MediaStats.last_tick_micros` + // semantics (work duration per tick, not wall-clock period). + // (f) Termination: `scenario_done()` checks for `End` step. let tick = Duration::from_millis(20); loop { + let tick_start = Instant::now(); + if sim_pipe.current_step_is_speak_loud() { wrapped_pipe.on_pcm_frame(loud_pcm_frame()); } @@ -192,6 +215,13 @@ impl SimCall { sim_pipe.on_pcm_frame(PcmFrame::zeroed()); + // S6: record per-tick work duration into the shared gauge. + // The elapsed here is the synchronous tick work — Reflex state + // machine advances, capture pushes, scenario cursor increments. + // For the standalone SimCall tick loop, this is the analog of + // the binary MediaThread's `last_tick_micros` (spec §3.6). + gauge.record_tick(tick_start.elapsed()); + if sim_pipe.scenario_done() { break; } diff --git a/crates/rutster-sim/src/tick_lag.rs b/crates/rutster-sim/src/tick_lag.rs index cb166e1..cf38bf0 100644 --- a/crates/rutster-sim/src/tick_lag.rs +++ b/crates/rutster-sim/src/tick_lag.rs @@ -1,13 +1,237 @@ //! # tick_lag — `TickLagGauge`: the ADR-0010 doctrine-drift detector //! -//! **Stub — lands in S6.** +//! See spec §3.6 + §6.4 for the design. Surfaces `tick_overruns` + +//! `last_tick_micros` (here: `max_tick_lag_micros`) as primary readouts +//! in the `SweepReport`. The concurrency sweep turns the gauge from a +//! counter into a decision artifact: "does single-thread poll loop +//! breach budget at realistic concurrency?" gets answered with data, +//! not vibes. If yes, the dedicated threadpool-shard graduation (slice-4 +//! deferral #2) gets its data-driven case. //! -//! See `docs/superpowers/specs/2026-07-05-slice-4-half-benchmark-sim-design.md` -//! §3.6 + §6.4 for the design + `docs/superpowers/plans/2026-07-05-slice-4-half-benchmark-sim.md` -//! Task S6 for the implementation. Surfaces `MediaStats.{tick_overruns, -//! last_tick_micros}` as primary readouts in the `SweepReport`. The -//! concurrency sweep turns the gauge from a counter into a decision -//! artifact: "does single-thread poll loop breach budget at realistic -//! concurrency?" gets answered with data, not vibes. If yes, the -//! dedicated threadpool-shard graduation (slice-4 deferral #2) gets -//! scheduled on evidence. +//! # Standalone-path adaptation (spec §3.6 deviation) +//! +//! Spec §3.6 says the gauge "polls `MediaCmd::Stats` during the sweep" +//! — the slice-5/seams `MediaStats { tick_overruns, last_tick_micros }` +//! readout from the binary's `MediaThread`. S4's standalone-path +//! conclusion (per the plan + kickoff hard rule) means the SimCall +//! wires itself in tokio WITHOUT registering with the binary's +//! `MediaThread` — no `MediaCmd::Stats` channel exists to poll. +//! +//! This S6 implementation adapts: the gauge is wired INTO the +//! `SimCall`'s tick loop directly (via a shared `Arc` +//! handle). `SimCall::run_with_gauge` records per-tick wall-clock +//! duration via `Instant::now()` measurement around the tick work. +//! The semantics are equivalent (max tick lag + overrun count + pct), +//! the source is the in-process tokio SimCall tick loop rather than +//! the binary's `MediaThread` poll loop. A future slice (post-spearhead +//! refinement, paired with network-realism mode) wires the gauge +//! against the binary's `MediaThread` per spec §3.6 — requires either +//! `MediaThread` registration (the RegisterSim variant forbidden this +//! slice) OR a client-server sim mode (deferred per spec §8.6). +//! +//! # Hot-path design: atomics (not Mutex>) +//! +//! Per-tick recording is 3 atomic ops: +//! 1. CAS loop on `max_tick_lag_micros` (atomic max update). +//! 2. Conditional `fetch_add` on `tick_overruns` if the tick exceeded 10 ms. +//! 3. Unconditional `fetch_add` on `total_ticks`. +//! +//! A `Mutex>` would lock the vector per tick — locking +//! overhead is significant relative to the per-tick work budget +//! (microseconds). Atomics keep the per-tick critical section lock-free. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +/// Threshold for "tick overrun" (per spec §5.1 `TICK_LAG_MAX_MS = 10.0`). +/// 10 ms = 10_000 µs. A tick whose wall-clock duration exceeds this is +/// an overrun (the meta-tick's nominal period was breached). +const TICK_LAG_OVERRUN_THRESHOLD_US: u64 = 10_000; + +/// Atomic accumulators for tick-lag measurements during a concurrency +/// sweep. Shared between N `SimCall`s (cloned via `Arc`) + read +/// post-sweep by the `ConcurrencyRunner` / `TickLagGauge`. +/// +/// # Why `Arc` (the constructor returns `Arc`) +/// +/// The stats are shared with N concurrent `SimCall` tasks (one clone +/// per task) + the post-sweep reader. The natural lifetime is "the +/// duration of one concurrency level's sweep" — short, scoped. An +/// `Arc` from construction avoids a separate `Arc::new(...)` +/// wrapper at every call site (mirrors `ReflexMetrics::new()` in +/// slice-4 + `TapMetrics::new()` in slice-2 — both return `Arc`). +pub struct TickLagStats { + max_tick_lag_micros: AtomicU64, + tick_overruns: AtomicU64, + total_ticks: AtomicU64, +} + +impl TickLagStats { + /// Construct a fresh `Arc` with all counters zeroed. + pub fn new() -> Arc { + Arc::new(Self { + max_tick_lag_micros: AtomicU64::new(0), + tick_overruns: AtomicU64::new(0), + total_ticks: AtomicU64::new(0), + }) + } + + /// Record a tick's wall-clock duration. Hot path: matches the 3-atomic-ops + /// budget. Called per-tick by `SimCall::run_with_gauge`. + /// + /// # CAS loop on max + /// + /// `compare_exchange_weak` is used (not `compare_exchange`) — weak + /// CAS can spuriously fail, but in a tight update loop the cost is + /// lower than strong CAS + the spurious-failure retry is bounded + /// (the next iteration re-reads the current max). The `Ordering::Relaxed` + /// on both success + failure is intentional: this is a statistics + /// counter where we don't need cross-thread synchronization ordering + /// (the post-sweep reader sees a consistent-enough snapshot; the + /// exact order doesn't matter for "max observed" + "count > 10ms"). + pub fn record_tick(&self, elapsed: Duration) { + let elapsed_us = elapsed.as_micros() as u64; + let mut current_max = self.max_tick_lag_micros.load(Ordering::Relaxed); + while elapsed_us > current_max { + match self.max_tick_lag_micros.compare_exchange_weak( + current_max, + elapsed_us, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(actual) => current_max = actual, + } + } + if elapsed_us > TICK_LAG_OVERRUN_THRESHOLD_US { + self.tick_overruns.fetch_add(1, Ordering::Relaxed); + } + self.total_ticks.fetch_add(1, Ordering::Relaxed); + } + + pub fn max_tick_lag_micros(&self) -> u64 { + self.max_tick_lag_micros.load(Ordering::Relaxed) + } + + pub fn tick_overruns(&self) -> u64 { + self.tick_overruns.load(Ordering::Relaxed) + } + + pub fn total_ticks(&self) -> u64 { + self.total_ticks.load(Ordering::Relaxed) + } + + /// Percentage of ticks that exceeded the 10 ms overrun threshold. + /// Returns 0.0 if no ticks were recorded (avoids div-by-zero). + pub fn tick_overrun_pct(&self) -> f64 { + let total = self.total_ticks(); + if total == 0 { + return 0.0; + } + (self.tick_overruns() as f64 / total as f64) * 100.0 + } +} + +impl Default for TickLagStats { + fn default() -> Self { + Self { + max_tick_lag_micros: AtomicU64::new(0), + tick_overruns: AtomicU64::new(0), + total_ticks: AtomicU64::new(0), + } + } +} + +/// The read-side API for the gauge. Wraps an `Arc` created +/// internally (or an externally-shared one). The `ConcurrencyRunner` +/// creates one gauge per concurrency level + passes the `stats_handle()` +/// to each of the N `SimCall`s via `SimCall::run_with_gauge`. +pub struct TickLagGauge { + stats: Arc, +} + +impl TickLagGauge { + pub fn new() -> Self { + Self { + stats: TickLagStats::new(), + } + } + + /// Get a stats handle that can be passed to `SimCall::run_with_gauge` + /// for shared recording during the concurrency sweep. + pub fn stats_handle(&self) -> Arc { + self.stats.clone() + } + + pub fn max_tick_lag_micros(&self) -> u64 { + self.stats.max_tick_lag_micros() + } + + pub fn tick_overruns(&self) -> u64 { + self.stats.tick_overruns() + } + + pub fn total_ticks(&self) -> u64 { + self.stats.total_ticks() + } + + pub fn tick_overrun_pct(&self) -> f64 { + self.stats.tick_overrun_pct() + } +} + +impl Default for TickLagGauge { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn gauge_records_zero_initially() { + let gauge = TickLagGauge::new(); + assert_eq!(gauge.max_tick_lag_micros(), 0); + assert_eq!(gauge.tick_overruns(), 0); + assert_eq!(gauge.total_ticks(), 0); + assert_eq!(gauge.tick_overrun_pct(), 0.0); + } + + #[test] + fn gauge_records_max_tick_lag_across_samples() { + // Three ticks: 500us, 1500us, 800us. Max should be 1500us. + // Total ticks should be 3. No overruns (all < 10ms). + let stats = TickLagStats::new(); + stats.record_tick(Duration::from_micros(500)); + stats.record_tick(Duration::from_micros(1500)); + stats.record_tick(Duration::from_micros(800)); + assert_eq!(stats.max_tick_lag_micros(), 1500); + assert_eq!(stats.total_ticks(), 3); + assert_eq!(stats.tick_overruns(), 0); + assert_eq!(stats.tick_overrun_pct(), 0.0); + } + + #[test] + fn gauge_counts_overruns_above_threshold() { + // 4 ticks: 2 under 10ms, 2 over. Overrun pct = 50%. + let stats = TickLagStats::new(); + stats.record_tick(Duration::from_micros(5_000)); // under + stats.record_tick(Duration::from_micros(11_000)); // over + stats.record_tick(Duration::from_micros(20_000)); // over + stats.record_tick(Duration::from_micros(7_000)); // under + assert_eq!(stats.tick_overruns(), 2); + assert_eq!(stats.total_ticks(), 4); + assert_eq!(stats.max_tick_lag_micros(), 20_000); + assert_eq!(stats.tick_overrun_pct(), 50.0); + } + + #[test] + fn gauge_handles_zero_total_ticks_pct() { + // No ticks recorded → pct should return 0.0 (not NaN). + let stats = TickLagStats::new(); + assert_eq!(stats.tick_overrun_pct(), 0.0); + } +}