Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 17 Packages Wiki Activity Actions 255

jouleclaw-energy: advance the scripted RAPL tree in step with the probe #228

Merged
dcharlot merged 1 commit from fix/rapl-probe-test-in-step into main 2026-10-13 01:02:51 -04:00
Conversation 0 Commits 1 Files changed 2 +61 -32
2 changed files with 61 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

27
jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/liveness.rs
View file

@ -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<F>(open: impl FnOnce() -> Option<(usize, F)>, tick_cpu: Duration) -> CounterLiveness
where
F: FnMut() -> Option<Vec<u64>>,
{
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<F>(
open: impl FnOnce() -> Option<(usize, F)>,
tick_cpu: Duration,
mut before_burn: impl FnMut(),
) -> CounterLiveness
where
F: FnMut() -> Option<Vec<u64>>,
{
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

66
jouleclaw/jouleclaw-rs/crates/jouleclaw-energy/src/rapl.rs
View file

@ -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:?}");
}
}