feat(causal): timer-heap virtual time — deadlines chase injected delay (RFC 007)
Injected delays dilate virtual time for the workload, but timer deadlines stayed wall-anchored: a sleep or receive-timeout fired early in virtual terms, so timeout/retry behaviour sped up relative to the dilated world (v1 known gap #1). Every heap Entry now carries delay_stamp — the global delay ledger at (re-)queue time, cfg-gated on smarm-causal. pop_due converts any debt accrued since the stamp to wall time (tsc_hz) and shifts the effective deadline; a not-yet-due entry is re-queued at the shifted deadline with a fresh stamp, so it keeps chasing delay injected while it waits. seq is preserved across re-queues, keeping send_after cancellation identity intact (cancelled entries are discarded before any shift). Zero debt is byte-identical to the old path; peek_deadline may under-report, costing one spurious scheduler wake per injected chunk (documented). This also makes the park-gated resume credit *correct* rather than forgiving for sleepers: a sleeping actor now physically pays its debt by sleeping longer, so the on_resume fast-forward reflects real payment (sleeping_actor_pays_injected_delay pins this end-to-end through the runtime). New test hooks: inject_delay_cycles_for_test (deterministic ledger driver, eagerly TSC-calibrating so conversion never stalls a scheduler loop) and cycles_to_duration. The ledger is process-global, so the delta-sensitive causal tests now serialize on a shared test mutex — they were racy under the parallel test harness before this, in principle.
This commit is contained in:
@@ -397,6 +397,20 @@ mod inner {
|
|||||||
end();
|
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
|
/// Controller parameters: which speedups to try per site, and the
|
||||||
/// experiment/cooldown windows.
|
/// experiment/cooldown windows.
|
||||||
pub struct ExperimentPlan {
|
pub struct ExperimentPlan {
|
||||||
|
|||||||
+64
-8
@@ -102,6 +102,12 @@ pub struct Entry {
|
|||||||
seq: u64,
|
seq: u64,
|
||||||
pub pid: Pid,
|
pub pid: Pid,
|
||||||
pub reason: Reason,
|
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 {
|
impl PartialEq for Entry {
|
||||||
@@ -166,7 +172,14 @@ impl Timers {
|
|||||||
let seq = self.next_seq;
|
let seq = self.next_seq;
|
||||||
self.next_seq = self.next_seq.wrapping_add(1);
|
self.next_seq = self.next_seq.wrapping_add(1);
|
||||||
self.armed.insert(seq);
|
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)
|
TimerId(seq)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -183,7 +196,14 @@ impl Timers {
|
|||||||
pub fn insert(&mut self, deadline: Instant, pid: Pid, reason: Reason) {
|
pub fn insert(&mut self, deadline: Instant, pid: Pid, reason: Reason) {
|
||||||
let seq = self.next_seq;
|
let seq = self.next_seq;
|
||||||
self.next_seq = self.next_seq.wrapping_add(1);
|
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 {
|
pub fn is_empty(&self) -> bool {
|
||||||
@@ -210,23 +230,59 @@ impl Timers {
|
|||||||
/// one is silently dropped here (its `seq` was already removed from
|
/// one is silently dropped here (its `seq` was already removed from
|
||||||
/// `armed` by [`cancel`](Self::cancel)). Returning it removes it from
|
/// `armed` by [`cancel`](Self::cancel)). Returning it removes it from
|
||||||
/// `armed`, so a later `cancel` of a fired timer reports `false`.
|
/// `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<Entry> {
|
pub fn pop_due(&mut self, now: Instant) -> Vec<Entry> {
|
||||||
let mut out = Vec::new();
|
let mut out = Vec::new();
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let global = crate::causal::global_delay_cycles();
|
||||||
while let Some(r) = self.heap.peek() {
|
while let Some(r) = self.heap.peek() {
|
||||||
if r.0.deadline <= now {
|
if r.0.deadline > now {
|
||||||
let entry = match self.heap.pop() {
|
break;
|
||||||
|
}
|
||||||
|
#[allow(unused_mut)]
|
||||||
|
let mut entry = match self.heap.pop() {
|
||||||
Some(e) => e.0,
|
Some(e) => e.0,
|
||||||
None => panic!("smarm: timer heap pop after peek returned None (core corrupt)"),
|
None => panic!("smarm: timer heap pop after peek returned None (core corrupt)"),
|
||||||
};
|
};
|
||||||
if matches!(entry.reason, Reason::Send { .. }) && !self.armed.remove(&entry.seq) {
|
if matches!(entry.reason, Reason::Send { .. }) && !self.armed.contains(&entry.seq) {
|
||||||
// Cancelled before it came due: discard, do not deliver.
|
// 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;
|
continue;
|
||||||
}
|
}
|
||||||
out.push(entry);
|
#[cfg(feature = "smarm-causal")]
|
||||||
} else {
|
{
|
||||||
break;
|
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
|
out
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+154
@@ -73,6 +73,7 @@ fn site_guard_nesting_restores() {
|
|||||||
/// catches up to the global one) instead of running delay-free.
|
/// catches up to the global one) instead of running delay-free.
|
||||||
#[test]
|
#[test]
|
||||||
fn virtual_speedup_ledger() {
|
fn virtual_speedup_ledger() {
|
||||||
|
let _s = serial();
|
||||||
let global_out = Arc::new(AtomicU64::new(0));
|
let global_out = Arc::new(AtomicU64::new(0));
|
||||||
let absorbed_out = Arc::new(AtomicU64::new(0));
|
let absorbed_out = Arc::new(AtomicU64::new(0));
|
||||||
let g_out = global_out.clone();
|
let g_out = global_out.clone();
|
||||||
@@ -137,6 +138,7 @@ fn virtual_speedup_ledger() {
|
|||||||
/// resume made all bystanders delay-immune, so experiments moved nothing.)
|
/// resume made all bystanders delay-immune, so experiments moved nothing.)
|
||||||
#[test]
|
#[test]
|
||||||
fn runnable_bystander_pays_delay() {
|
fn runnable_bystander_pays_delay() {
|
||||||
|
let _s = serial();
|
||||||
let baseline = Arc::new(AtomicU64::new(0));
|
let baseline = Arc::new(AtomicU64::new(0));
|
||||||
let during = Arc::new(AtomicU64::new(0));
|
let during = Arc::new(AtomicU64::new(0));
|
||||||
let b_out = baseline.clone();
|
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.
|
/// baseline Coz relies on, and it doubles as the overhead sanity check.
|
||||||
#[test]
|
#[test]
|
||||||
fn zero_speedup_injects_nothing() {
|
fn zero_speedup_injects_nothing() {
|
||||||
|
let _s = serial();
|
||||||
smarm::init(smarm::Config::exact(2)).run(|| {
|
smarm::init(smarm::Config::exact(2)).run(|| {
|
||||||
let stop = Arc::new(AtomicBool::new(false));
|
let stop = Arc::new(AtomicBool::new(false));
|
||||||
let stop2 = stop.clone();
|
let stop2 = stop.clone();
|
||||||
@@ -299,6 +302,7 @@ fn impact_from_synthetic_results() {
|
|||||||
/// section with experiment / throughput-point lines.
|
/// section with experiment / throughput-point lines.
|
||||||
#[test]
|
#[test]
|
||||||
fn controller_produces_report() {
|
fn controller_produces_report() {
|
||||||
|
let _s = serial();
|
||||||
smarm::init(smarm::Config::exact(2)).run(|| {
|
smarm::init(smarm::Config::exact(2)).run(|| {
|
||||||
let stop = Arc::new(AtomicBool::new(false));
|
let stop = Arc::new(AtomicBool::new(false));
|
||||||
let stop2 = stop.clone();
|
let stop2 = stop.clone();
|
||||||
@@ -341,6 +345,7 @@ fn controller_produces_report() {
|
|||||||
/// (+83.5% where theory says +100%).
|
/// (+83.5% where theory says +100%).
|
||||||
#[test]
|
#[test]
|
||||||
fn site_boundaries_flush_tail() {
|
fn site_boundaries_flush_tail() {
|
||||||
|
let _s = serial();
|
||||||
fn tsc() -> u64 {
|
fn tsc() -> u64 {
|
||||||
// SAFETY: rdtsc has no preconditions on x86_64.
|
// SAFETY: rdtsc has no preconditions on x86_64.
|
||||||
unsafe { core::arch::x86_64::_rdtsc() }
|
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%)"
|
"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)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user