diff --git a/src/causal.rs b/src/causal.rs index 96a23cf..7fd4b2b 100644 --- a/src/causal.rs +++ b/src/causal.rs @@ -397,6 +397,20 @@ mod inner { end(); } + /// Test support: grow the global delay ledger directly, as if target-site + /// samples had attributed `cycles` — deterministic driver for the timer + /// virtual-time tests. Calibrates the TSC eagerly so conversion later + /// never stalls a scheduler loop. + pub fn inject_delay_cycles_for_test(cycles: u64) { + let _ = tsc_hz(); + GLOBAL_DELAY.fetch_add(cycles, Ordering::Relaxed); + } + + /// Convert ledger cycles to wall time at the measured TSC rate. + pub fn cycles_to_duration(cycles: u64) -> Duration { + Duration::from_secs_f64(cycles as f64 / tsc_hz()) + } + /// Controller parameters: which speedups to try per site, and the /// experiment/cooldown windows. pub struct ExperimentPlan { diff --git a/src/timer.rs b/src/timer.rs index de2ddb2..ecf1fd4 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -102,6 +102,12 @@ pub struct Entry { seq: u64, pub pid: Pid, pub reason: Reason, + /// RFC 007 virtual time: the global delay ledger reading when this entry + /// was (re-)queued. `pop_due` shifts the effective deadline by any delay + /// injected since, so timers dilate together with the causally-delayed + /// workload instead of firing early in virtual terms. + #[cfg(feature = "smarm-causal")] + delay_stamp: u64, } impl PartialEq for Entry { @@ -166,7 +172,14 @@ impl Timers { let seq = self.next_seq; self.next_seq = self.next_seq.wrapping_add(1); self.armed.insert(seq); - self.heap.push(Reverse(Entry { deadline, seq, pid, reason: Reason::Send { fire } })); + self.heap.push(Reverse(Entry { + deadline, + seq, + pid, + reason: Reason::Send { fire }, + #[cfg(feature = "smarm-causal")] + delay_stamp: crate::causal::global_delay_cycles(), + })); TimerId(seq) } @@ -183,7 +196,14 @@ impl Timers { pub fn insert(&mut self, deadline: Instant, pid: Pid, reason: Reason) { let seq = self.next_seq; self.next_seq = self.next_seq.wrapping_add(1); - self.heap.push(Reverse(Entry { deadline, seq, pid, reason })); + self.heap.push(Reverse(Entry { + deadline, + seq, + pid, + reason, + #[cfg(feature = "smarm-causal")] + delay_stamp: crate::causal::global_delay_cycles(), + })); } pub fn is_empty(&self) -> bool { @@ -210,22 +230,58 @@ impl Timers { /// one is silently dropped here (its `seq` was already removed from /// `armed` by [`cancel`](Self::cancel)). Returning it removes it from /// `armed`, so a later `cancel` of a fired timer reports `false`. + /// + /// RFC 007 virtual time (feature `smarm-causal`): before an entry fires, + /// any global delay injected since it was (re-)queued is added to its + /// deadline; an entry whose *effective* deadline hasn't passed is pushed + /// back with the shifted deadline and a fresh stamp, so it keeps chasing + /// delay injected while it waits. Consequences, both benign: + /// [`peek_deadline`](Self::peek_deadline) may under-report (raw deadline + /// earlier than effective), costing at most one spurious scheduler wake + /// per injected chunk; and a shift never converts wall time — with zero + /// debt the path is byte-identical to the featureless one. pub fn pop_due(&mut self, now: Instant) -> Vec { let mut out = Vec::new(); + #[cfg(feature = "smarm-causal")] + let global = crate::causal::global_delay_cycles(); while let Some(r) = self.heap.peek() { - if r.0.deadline <= now { - let entry = match self.heap.pop() { - Some(e) => e.0, - None => panic!("smarm: timer heap pop after peek returned None (core corrupt)"), - }; - if matches!(entry.reason, Reason::Send { .. }) && !self.armed.remove(&entry.seq) { - // Cancelled before it came due: discard, do not deliver. - continue; - } - out.push(entry); - } else { + if r.0.deadline > now { break; } + #[allow(unused_mut)] + let mut entry = match self.heap.pop() { + Some(e) => e.0, + None => panic!("smarm: timer heap pop after peek returned None (core corrupt)"), + }; + if matches!(entry.reason, Reason::Send { .. }) && !self.armed.contains(&entry.seq) { + // Cancelled before it came due: discard, do not deliver. + // (Checked before any shift so a cancelled entry is never + // re-queued just to be discarded later.) + continue; + } + #[cfg(feature = "smarm-causal")] + { + let debt = global.saturating_sub(entry.delay_stamp); + if debt > 0 { + let shifted = entry + .deadline + .checked_add(crate::causal::cycles_to_duration(debt)) + .unwrap_or(entry.deadline); + if shifted > now { + // Not due in virtual time: re-queue at the shifted + // deadline, stamped, keeping `seq` (and thus `Send` + // cancellation identity) intact. + entry.deadline = shifted; + entry.delay_stamp = global; + self.heap.push(Reverse(entry)); + continue; + } + } + } + if matches!(entry.reason, Reason::Send { .. }) { + self.armed.remove(&entry.seq); + } + out.push(entry); } out } diff --git a/tests/causal.rs b/tests/causal.rs index 85d07c7..be41d1c 100644 --- a/tests/causal.rs +++ b/tests/causal.rs @@ -73,6 +73,7 @@ fn site_guard_nesting_restores() { /// catches up to the global one) instead of running delay-free. #[test] fn virtual_speedup_ledger() { + let _s = serial(); let global_out = Arc::new(AtomicU64::new(0)); let absorbed_out = Arc::new(AtomicU64::new(0)); let g_out = global_out.clone(); @@ -137,6 +138,7 @@ fn virtual_speedup_ledger() { /// resume made all bystanders delay-immune, so experiments moved nothing.) #[test] fn runnable_bystander_pays_delay() { + let _s = serial(); let baseline = Arc::new(AtomicU64::new(0)); let during = Arc::new(AtomicU64::new(0)); let b_out = baseline.clone(); @@ -218,6 +220,7 @@ fn runnable_bystander_pays_delay() { /// baseline Coz relies on, and it doubles as the overhead sanity check. #[test] fn zero_speedup_injects_nothing() { + let _s = serial(); smarm::init(smarm::Config::exact(2)).run(|| { let stop = Arc::new(AtomicBool::new(false)); let stop2 = stop.clone(); @@ -299,6 +302,7 @@ fn impact_from_synthetic_results() { /// section with experiment / throughput-point lines. #[test] fn controller_produces_report() { + let _s = serial(); smarm::init(smarm::Config::exact(2)).run(|| { let stop = Arc::new(AtomicBool::new(false)); let stop2 = stop.clone(); @@ -341,6 +345,7 @@ fn controller_produces_report() { /// (+83.5% where theory says +100%). #[test] fn site_boundaries_flush_tail() { + let _s = serial(); fn tsc() -> u64 { // SAFETY: rdtsc has no preconditions on x86_64. unsafe { core::arch::x86_64::_rdtsc() } @@ -388,3 +393,152 @@ fn site_boundaries_flush_tail() { "boundary flush attributed {injected} of {spent} spent cycles (want ~50%)" ); } + +// --------------------------------------------------------------------------- +// RFC 007 timer virtual time: injected delay dilates virtual time for +// everyone, so a pending timer's *effective* deadline is its raw deadline +// plus whatever delay was injected while it was armed. Without this, a +// sleep or receive-timeout fires "early" in virtual terms and timeout-driven +// behaviour speeds up relative to the dilated world. +// +// These tests (and every ledger-delta test above) share the process-global +// delay ledger, so they serialize on `serial()`. +// --------------------------------------------------------------------------- + +fn serial() -> std::sync::MutexGuard<'static, ()> { + static M: std::sync::Mutex<()> = std::sync::Mutex::new(()); + M.lock().unwrap_or_else(std::sync::PoisonError::into_inner) +} + +/// A pending timer must not fire at its raw deadline if delay was injected +/// after it was armed; it fires once wall time covers raw + injected. +#[test] +fn timer_deadline_shifts_with_injected_delay() { + let _s = serial(); + use smarm::pid::Pid; + use smarm::timer::Timers; + use std::time::Instant; + + let hz = smarm::causal::tsc_hz(); + let mut t = Timers::new(); + let now = Instant::now(); + t.insert_sleep(now + Duration::from_millis(50), Pid::new(0, 0), 1); + + // 100ms of debt lands while the timer is pending. + smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64); + + // Raw deadline passed, effective deadline not: nothing fires, entry kept. + assert!(t.pop_due(now + Duration::from_millis(60)).is_empty()); + assert!(!t.is_empty(), "shifted entry must be re-queued, not dropped"); + + // Past raw + injected (with margin) it must fire. Chase in case a + // parallel test injected more debt meanwhile. + let mut fire_at = now + Duration::from_millis(170); + let mut fired = false; + for _ in 0..40 { + if !t.pop_due(fire_at).is_empty() { + fired = true; + break; + } + fire_at += Duration::from_millis(50); + } + assert!(fired, "timer never fired after covering injected delay"); +} + +/// With no debt accrued since insertion, behaviour is byte-identical to +/// before: due entries fire at their raw deadline. +#[test] +fn timer_zero_debt_pops_at_raw_deadline() { + let _s = serial(); + use smarm::pid::Pid; + use smarm::timer::Timers; + use std::time::Instant; + + let mut t = Timers::new(); + let now = Instant::now(); + t.insert_sleep(now - Duration::from_millis(1), Pid::new(0, 0), 1); + assert_eq!(t.pop_due(now).len(), 1); + assert!(t.is_empty()); +} + +/// Re-queueing a shifted entry must preserve its identity: a `send_after` +/// cancelled *after* being shifted past its raw deadline must still cancel +/// (return true) and must never deliver. +#[test] +fn send_after_cancel_survives_reheap() { + let _s = serial(); + use smarm::pid::Pid; + use smarm::timer::Timers; + use std::sync::atomic::AtomicBool; + use std::time::Instant; + + let hz = smarm::causal::tsc_hz(); + let mut t = Timers::new(); + let now = Instant::now(); + let delivered = Arc::new(AtomicBool::new(false)); + let d2 = delivered.clone(); + let id = t.insert_send( + now + Duration::from_millis(20), + Pid::new(0, 0), + Box::new(move || d2.store(true, Ordering::Relaxed)), + ); + + smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64); + + // Raw deadline passed: shifted, not fired. + assert!(t.pop_due(now + Duration::from_millis(30)).is_empty()); + // Cancellation still works on the re-queued entry. + assert!(t.cancel(id), "cancel lost track of a re-queued send_after"); + // Even far past the effective deadline nothing is delivered. + for e in t.pop_due(now + Duration::from_secs(3600)) { + if let smarm::timer::Reason::Send { fire } = e.reason { + fire(); + } + } + assert!( + !delivered.load(Ordering::Relaxed), + "cancelled send_after delivered after re-heap" + ); +} + +/// End-to-end through the runtime: an actor mid-`sleep` pays delay injected +/// while it is parked by sleeping *longer* — which is exactly what makes the +/// park-gated resume-credit fast-forward correct rather than forgiving. +#[test] +fn sleeping_actor_pays_injected_delay() { + let _s = serial(); + use std::time::Instant; + + let slept = Arc::new(AtomicU64::new(0)); + let s_out = slept.clone(); + + smarm::init(smarm::Config::exact(1)).run(move || { + let hz = smarm::causal::tsc_hz(); + let s2 = s_out.clone(); + let sleeper = smarm::spawn(move || { + let t0 = Instant::now(); + smarm::sleep(Duration::from_millis(50)); + s2.store(t0.elapsed().as_millis() as u64, Ordering::Relaxed); + }); + + // Let the sleeper arm its timer, then inject 150ms of debt from a + // plain OS thread — no slot, so nothing spin-absorbs it here and the + // only path by which it can slow the sleeper is the timer shift. + smarm::sleep(Duration::from_millis(10)); + std::thread::spawn(move || { + smarm::causal::inject_delay_cycles_for_test((hz * 0.150) as u64); + }) + .join() + .unwrap(); + + sleeper.join().unwrap(); + }); + + let ms = slept.load(Ordering::Relaxed); + // Raw sleep 50ms + ~150ms debt injected mid-sleep => ~200ms effective. + // Loose lower bound; the pre-fix behaviour is ~50ms. + assert!( + ms >= 150, + "sleeper paid nothing: slept {ms}ms, expected >= 150ms (raw 50 + injected 150)" + ); +}