alee/rutster
0
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(sim): TickLagGauge -- per-tick wall-clock measurement (slice-4½ S6)

Browse Source 
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<TickLagStats> 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<Vec<Duration>>
which would lock the vector per tick + add ordering overhead significant to the per-tick
microseconds budget.

ConcurrencyRunner creates one shared Arc<TickLagStats> 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 <himself@adlee.work>
This commit is contained in:
A.D.Lee
2026-07-05 03:26:23 -04:00
parent 6da8e4095a
commit da63d58f61
4 changed files with 288 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
crates/rutster-sim/src/concurrency.rs
View File

@@ -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(),
}
}
}

1
crates/rutster-sim/src/lib.rs
View File

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

38
crates/rutster-sim/src/runner.rs
View File

@@ -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<TickLagStats>) -> 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;
}

244
crates/rutster-sim/src/tick_lag.rs
View File

@@ -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<TickLagStats>`
//! 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<Vec<Duration>>)
//!
//! 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<Vec<Duration>>` 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<Self>` (the constructor returns `Arc<Self>`)
///
/// 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<Self>` 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<Self>`).
pub struct TickLagStats {
max_tick_lag_micros: AtomicU64,
tick_overruns: AtomicU64,
total_ticks: AtomicU64,
}
impl TickLagStats {
/// Construct a fresh `Arc<TickLagStats>` with all counters zeroed.
pub fn new() -> Arc<Self> {
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<TickLagStats>` 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<TickLagStats>,
}
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<TickLagStats> {
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);
}
}