diff --git a/crates/grill-perf/src/process.rs b/crates/grill-perf/src/process.rs index 6e23326..0409ca2 100644 --- a/crates/grill-perf/src/process.rs +++ b/crates/grill-perf/src/process.rs @@ -55,7 +55,7 @@ pub mod program_signals { static ACTIVE: Mutex<()> = Mutex::new(()); static CANCELLED: AtomicBool = AtomicBool::new(false); - extern "C" fn cancel(_: libc::c_int) { + pub(super) extern "C" fn cancel(_: libc::c_int) { CANCELLED.store(true, Ordering::SeqCst); } @@ -433,8 +433,9 @@ mod tests { if cancel { // The leader is already reaped, so cancellation must interrupt // normal-exit drainage rather than the running-program path. - // SAFETY: the collector guard has installed the SIGINT handler. - assert_eq!(unsafe { libc::raise(libc::SIGINT) }, 0); + // A raised SIGINT would reach whichever handler a concurrent + // test installed last, so invoke the guard's handler directly. + program_signals::cancel(libc::SIGINT); let completed = completion.recv_timeout(Duration::from_secs(5)).is_ok(); drop((stdout, stderr)); assert!(completed, "cancellation waited for pipe EOF"); diff --git a/crates/grill-perf/tests/support/systemone.rs b/crates/grill-perf/tests/support/systemone.rs index e2323c4..11cbabf 100644 --- a/crates/grill-perf/tests/support/systemone.rs +++ b/crates/grill-perf/tests/support/systemone.rs @@ -404,21 +404,44 @@ fn inspect_detects_changed_retained_response_bytes() { #[test] fn compare_reports_a_declared_change_under_the_observed_envelope() { let temp = Temp::new(); - let slow = DecisionServer::new(|request| { - thread::sleep(Duration::from_millis(40)); - immediate(request) - }); let fast = DecisionServer::new(immediate); - // Hosted macOS runners oversleep the 40 ms fixture delay by up to ~150 ms, so a + // Hosted macOS runners oversleep the fixture delay by up to ~150 ms, so a // tight reference-spread bound tests the host timer, not the comparison. let mut wide = plan_json(); for gate in wide["gates"].as_array_mut().unwrap() { gate["max_reference_spread_bps"] = json!(1000000); } let plan = fixtures(&temp, "compare", &wide); - successful(&capture_server(&temp, &slow, &plan, 0, "a")); + // The candidate is captured first so every reference request can outlast + // its slowest burst, however long load stretched that burst. successful(&capture(&temp, &fast.endpoint, &plan, 1, "b")); + let waves = |arm: &str| { + (0..) + .map(|index| temp.path(&format!("{arm}/wave-{index:06}.json"))) + .take_while(|path| path.exists()) + .map(value) + .collect::>() + }; + let slowest_burst = |waves: &[Value]| { + waves + .iter() + .map(|wave| wave["makespan_us"].as_u64().unwrap()) + .max() + .unwrap() + }; + let b = waves("b"); + let delay = Arc::new(std::sync::atomic::AtomicU64::new(slowest_burst(&b))); + let server_delay = delay.clone(); + let slow = DecisionServer::new(move |request| { + thread::sleep(Duration::from_micros(server_delay.load(Ordering::SeqCst))); + immediate(request) + }); + successful(&capture_server(&temp, &slow, &plan, 0, "a")); + let a = waves("a"); + // A2 must extend A's envelope even when scheduling stretches A's capture. + delay.store(slowest_burst(&a), Ordering::SeqCst); successful(&capture_server(&temp, &slow, &plan, 2, "a2")); + let a2 = waves("a2"); let output = cli() .args(["systemone", "compare", "--plan"]) .arg(&plan) @@ -438,11 +461,41 @@ fn compare_reports_a_declared_change_under_the_observed_envelope() { assert_eq!(report["acquisitions"].as_array().unwrap().len(), 3); let gates = report["gates"].as_array().unwrap(); assert_eq!(gates.len(), 2); - for gate in gates { + for (gate, declared) in gates.iter().zip(wide["gates"].as_array().unwrap()) { assert_eq!(gate["counts"], json!([3, 3, 3])); assert_eq!(gate["outcome"], json!("PASS")); - assert!(gate["reference"].is_array()); - assert!(gate["candidate"].is_array()); + let extrema = |arms: &[&[Value]]| { + let mut samples = Vec::new(); + for wave in arms + .iter() + .flat_map(|waves| waves.iter()) + .filter(|wave| wave["phase"] == "measured" && wave["cell"] == declared["cell"]) + { + assert_eq!(wave["complete"], json!(true)); + match declared["metric"].as_str().unwrap() { + "latency_us" => { + for attempt in wave["attempts"].as_array().unwrap() { + samples.push((attempt["latency_us"].as_u64().unwrap(), 1)); + } + } + "decisions_per_second" => samples.push(( + wave["decisions"].as_u64().unwrap() * 1_000_000, + wave["makespan_us"].as_u64().unwrap(), + )), + metric => panic!("unexpected fixture metric {metric}"), + } + } + samples.sort_unstable_by(|&(an, ad), &(bn, bd)| { + (u128::from(an) * u128::from(bd)).cmp(&(u128::from(bn) * u128::from(ad))) + }); + [samples[0], *samples.last().unwrap()].map(|(numerator, denominator)| { + json!({"numerator": numerator, "denominator": denominator}) + }) + }; + let reference = extrema(&[&a, &a2]); + assert_ne!(reference, extrema(&[&a]), "A2 must extend the envelope"); + assert_eq!(gate["reference"], json!(reference)); + assert_eq!(gate["candidate"], json!(extrema(&[&b]))); } let latency = &gates[0]["candidate"][1]; let reference = &gates[0]["reference"][0];