diff --git a/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/liveness.rs b/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/liveness.rs index cae0c2e61..887ff1dd0 100644 --- a/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/liveness.rs +++ b/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/liveness.rs @@ -249,11 +249,36 @@ where /// this thread's own CPU time ([`burn_thread_cpu`]), the monotonic clock and /// [`LIVENESS_BUDGET`]. Runs on the calling thread. pub(crate) fn probe(open: impl FnOnce() -> Option<(usize, F)>, tick_cpu: Duration) -> CounterLiveness +where + F: FnMut() -> Option>, +{ + probe_stepping(open, tick_cpu, || {}) +} + +/// [`probe`], with `before_burn` run at the start of every tick, ahead of the +/// burn. The real probe passes nothing. A test over a counter it scripts +/// advances the counter there, so that every interval the probe samples holds +/// one advance however the host schedules threads: a writer thread on a timer +/// is starved on a loaded host, and the probe then rightly judges a live +/// scripted counter frozen. +pub(crate) fn probe_stepping( + open: impl FnOnce() -> Option<(usize, F)>, + tick_cpu: Duration, + mut before_burn: impl FnMut(), +) -> CounterLiveness where F: FnMut() -> Option>, { let epoch = Instant::now(); - probe_with(open, || burn_thread_cpu(tick_cpu, PROBE_TICK_WALL_CAP), move || epoch.elapsed(), LIVENESS_BUDGET) + probe_with( + open, + || { + before_burn(); + burn_thread_cpu(tick_cpu, PROBE_TICK_WALL_CAP) + }, + move || epoch.elapsed(), + LIVENESS_BUDGET, + ) } /// This thread's CPU time so far (`CLOCK_THREAD_CPUTIME_ID`), or `None` where diff --git a/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/rapl.rs b/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/rapl.rs index 9be892dbc..5b7ce966c 100644 --- a/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/rapl.rs +++ b/jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/rapl.rs @@ -331,7 +331,15 @@ fn cached(set: &[PathBuf], probe: impl FnOnce() -> CounterLiveness) -> CounterLi /// caller is the cache in [`package_liveness`]. #[cfg_attr(not(target_os = "linux"), allow(dead_code))] fn probe_packages(packages: &[Package]) -> CounterLiveness { - crate::liveness::probe(|| open_sampler(packages), PROBE_TICK) + probe_packages_stepping(packages, || {}) +} + +/// [`probe_packages`], with `before_burn` run at the start of every tick +/// ([`crate::liveness::probe_stepping`]): how a test advances a fake tree in +/// step with the probe. +#[cfg_attr(not(target_os = "linux"), allow(dead_code))] +fn probe_packages_stepping(packages: &[Package], before_burn: impl FnMut()) -> CounterLiveness { + crate::liveness::probe_stepping(|| open_sampler(packages), PROBE_TICK, before_burn) } /// The probe's opener for `packages`: one channel per package, sampled by @@ -409,7 +417,7 @@ impl EnergyMeter for RaplCounter { #[cfg(test)] mod tests { use super::*; - use crate::liveness::{probe_with, LIVENESS_BUDGET}; + use crate::liveness::{probe_with, LIVENESS_BUDGET, MIN_LIVE_WINDOW, PROBE_CAP_BUDGETS, PROBE_TICK_WALL_CAP}; use crate::liveness::tests::EVERY; use std::cell::Cell; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -759,42 +767,38 @@ mod tests { assert!(t.elapsed() >= LIVENESS_BUDGET, "a frozen verdict took only {:?}", t.elapsed()); } - /// The production probe over a two-socket tree another thread keeps - /// advancing, as the kernel advances live MSRs: live, well inside the budget, - /// when both packages advance; frozen when package 1 holds while package 0 - /// and `psys` advance. + /// The production probe over a two-socket tree that advances once in every + /// tick, as the kernel advances live MSRs between two reads: live when + /// both packages advance; frozen when package 1 holds while package 0 and + /// `psys` advance. The tree is advanced in step with the probe + /// ([`probe_packages_stepping`]), so each interval it samples holds one + /// advance on any host. A writer thread on a 2 ms timer, as this test once + /// used, is starved on a loaded host, and the probe then rightly judged + /// the tree frozen. #[test] fn the_production_probe_needs_every_package_to_keep_advancing() { - use std::sync::atomic::AtomicBool; - use std::sync::Arc; - let run = |tag: &str, moving: &'static [&'static str]| { - let pc = Arc::new(two_socket_server(tag)); - let stop = Arc::new(AtomicBool::new(false)); - let writer = { - let (pc, stop) = (Arc::clone(&pc), Arc::clone(&stop)); - std::thread::spawn(move || { - let mut v = 100_000_000u64; - while !stop.load(Ordering::SeqCst) { - v += 90_000; // ~45 W at a 2 ms cadence - for dir in moving { - pc.set(dir, v); - } - std::thread::sleep(Duration::from_millis(2)); - } - }) - }; + let run = |tag: &str, moving: &[&str]| { + let pc = two_socket_server(tag); let packages = packages_under(pc.root()); + let mut v = 100_000_000u64; let t = std::time::Instant::now(); - let liveness = probe_packages(&packages); - let took = t.elapsed(); - stop.store(true, Ordering::SeqCst); - writer.join().expect("writer thread"); - (liveness, took) + let liveness = probe_packages_stepping(&packages, || { + v += 1_125_000; // 45 W over a 25 ms tick + for dir in moving { + pc.set(dir, v); + } + }); + (liveness, t.elapsed()) }; let (liveness, took) = run("advancing-both", &["intel-rapl:0", "intel-rapl:1"]); assert_eq!(liveness, CounterLiveness::Live); - assert!(took >= Duration::from_millis(300) && took < LIVENESS_BUDGET, "{took:?}"); - let (liveness, _) = run("advancing-one", &["intel-rapl:0", "intel-rapl:2"]); + // No sooner than the rule's window, and inside the probe's own cap. + // That is all a loaded host leaves certain: a tick there may take up + // to its wall cap, so "inside one budget" is a claim about the host. + let cap = LIVENESS_BUDGET * PROBE_CAP_BUDGETS + PROBE_TICK_WALL_CAP; + assert!(took >= MIN_LIVE_WINDOW && took < cap, "{took:?}"); + let (liveness, took) = run("advancing-one", &["intel-rapl:0", "intel-rapl:2"]); assert_eq!(liveness, CounterLiveness::Frozen, "package 1 held its value"); + assert!(took >= LIVENESS_BUDGET, "a frozen verdict took only {took:?}"); } }