feat(causal): wall-anchored timers — controller windows keep fixed wall length (RFC 007)
New timer anchor: insert_sleep_wall / scheduler::sleep_wall (exported) opt a Sleep entry out of the RFC 007 virtual-time shift, so it fires at its raw deadline regardless of injected delay. Featureless config is unchanged (the API exists but is identical to sleep). The causal controller's window/cooldown sleeps and the tsc_hz calibration sleep use it on the actor path (the OS-thread path was already wall). This fixes the controller's own sleeps dilating under its own injection — experiment windows stretched ~2x at 50% speedup (337ms -> 646ms injected/ window). Deltas were rate-normalized so results were unbiased; this fixes sweep cost, not bias. The general wall-anchored-timer-semantics jar item (user-facing opt-out) remains open; this lands the substrate. Test: wall_timer_ignores_injected_delay — wall entry fires at raw deadline while a virtual sibling in the same heap shifts. 13/13 causal, 34/34 binaries both feature configs.
This commit is contained in:
+12
-2
@@ -457,7 +457,12 @@ mod inner {
|
|||||||
if preempt::current_slot_ptr().is_null() {
|
if preempt::current_slot_ptr().is_null() {
|
||||||
std::thread::sleep(d);
|
std::thread::sleep(d);
|
||||||
} else {
|
} else {
|
||||||
crate::scheduler::sleep(d);
|
// Wall-anchored: the controller's window/cooldown sleeps
|
||||||
|
// *define* the experiment's wall length; letting them chase
|
||||||
|
// the delay it is itself injecting would stretch every window
|
||||||
|
// (observed ~2x at 50% speedup). Deltas are rate-normalized
|
||||||
|
// either way — this fixes cost, not bias.
|
||||||
|
crate::scheduler::sleep_wall(d);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Calibrate before any window so report rendering never has to sleep.
|
// Calibrate before any window so report rendering never has to sleep.
|
||||||
@@ -527,7 +532,12 @@ mod inner {
|
|||||||
if preempt::current_slot_ptr().is_null() {
|
if preempt::current_slot_ptr().is_null() {
|
||||||
std::thread::sleep(d);
|
std::thread::sleep(d);
|
||||||
} else {
|
} else {
|
||||||
crate::scheduler::sleep(d);
|
// Wall-anchored: calibration divides TSC delta by *wall*
|
||||||
|
// elapsed; a virtual sleep dilated by concurrent injection
|
||||||
|
// would still measure correctly (elapsed() is wall) but
|
||||||
|
// waste window time — and must never depend on the ledger
|
||||||
|
// it exists to convert.
|
||||||
|
crate::scheduler::sleep_wall(d);
|
||||||
}
|
}
|
||||||
(preempt::rdtsc().wrapping_sub(c0)) as f64 / t0.elapsed().as_secs_f64()
|
(preempt::rdtsc().wrapping_sub(c0)) as f64 / t0.elapsed().as_secs_f64()
|
||||||
})
|
})
|
||||||
|
|||||||
+1
-1
@@ -80,7 +80,7 @@ pub use registry::{
|
|||||||
};
|
};
|
||||||
pub use runtime::{init, Config, Runtime};
|
pub use runtime::{init, Config, Runtime};
|
||||||
pub use scheduler::{
|
pub use scheduler::{
|
||||||
block_on_io, cancel_timer, request_stop, run, self_pid, send_after, send_after_named, sleep,
|
block_on_io, cancel_timer, request_stop, run, self_pid, send_after, send_after_named, sleep, sleep_wall,
|
||||||
spawn, spawn_addr, spawn_under, wait_readable, wait_readable_timeout, wait_writable,
|
spawn, spawn_addr, spawn_under, wait_readable, wait_readable_timeout, wait_writable,
|
||||||
wait_writable_timeout, yield_now, FdArm, JoinError, JoinHandle,
|
wait_writable_timeout, yield_now, FdArm, JoinError, JoinHandle,
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -403,6 +403,28 @@ pub fn sleep(duration: std::time::Duration) {
|
|||||||
park_current();
|
park_current();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Like [`sleep`], but wall-anchored: under causal profiling (feature
|
||||||
|
/// `smarm-causal`) the deadline is honoured in wall time instead of chasing
|
||||||
|
/// injected virtual delay. Identical to [`sleep`] without the feature. For
|
||||||
|
/// measurement machinery whose durations define wall time (the causal
|
||||||
|
/// controller's windows, TSC calibration) — workload code wants [`sleep`].
|
||||||
|
pub fn sleep_wall(duration: std::time::Duration) {
|
||||||
|
let me = match current_pid() {
|
||||||
|
Some(pid) => pid,
|
||||||
|
None => panic!("sleep_wall() called outside an actor"),
|
||||||
|
};
|
||||||
|
let _np = NoPreempt::enter();
|
||||||
|
let epoch = begin_wait();
|
||||||
|
let deadline = crate::timer::deadline_from_now(duration);
|
||||||
|
with_runtime(|inner| {
|
||||||
|
match inner.timers.lock() {
|
||||||
|
Ok(mut timers) => timers.insert_sleep_wall(deadline, me, epoch),
|
||||||
|
Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
|
||||||
|
}
|
||||||
|
});
|
||||||
|
park_current();
|
||||||
|
}
|
||||||
|
|
||||||
pub fn insert_wait_timer(
|
pub fn insert_wait_timer(
|
||||||
deadline: std::time::Instant,
|
deadline: std::time::Instant,
|
||||||
pid: Pid,
|
pid: Pid,
|
||||||
|
|||||||
+40
-15
@@ -108,6 +108,13 @@ pub struct Entry {
|
|||||||
/// workload instead of firing early in virtual terms.
|
/// workload instead of firing early in virtual terms.
|
||||||
#[cfg(feature = "smarm-causal")]
|
#[cfg(feature = "smarm-causal")]
|
||||||
delay_stamp: u64,
|
delay_stamp: u64,
|
||||||
|
/// RFC 007: a wall-anchored entry opts out of the virtual-time shift —
|
||||||
|
/// its deadline is honoured in wall time regardless of injected delay.
|
||||||
|
/// Used by the causal controller's own measurement/cooldown sleeps so
|
||||||
|
/// experiment windows keep a fixed wall length; ordinary workload timers
|
||||||
|
/// stay virtual (`false`).
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
wall: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl PartialEq for Entry {
|
impl PartialEq for Entry {
|
||||||
@@ -158,6 +165,19 @@ impl Timers {
|
|||||||
self.insert(deadline, pid, Reason::Sleep { epoch });
|
self.insert(deadline, pid, Reason::Sleep { epoch });
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Insert a *wall-anchored* `Sleep` timer: fires at `deadline` in wall
|
||||||
|
/// time even while causal profiling (feature `smarm-causal`) is injecting
|
||||||
|
/// virtual delay — it never chases the delay ledger. Without the feature
|
||||||
|
/// this is identical to [`insert_sleep`](Self::insert_sleep).
|
||||||
|
///
|
||||||
|
/// Intended for measurement machinery (the causal controller's window and
|
||||||
|
/// cooldown sleeps, TSC calibration) whose durations *define* wall time
|
||||||
|
/// rather than participate in the workload. Workload code should use the
|
||||||
|
/// ordinary virtual-anchored timers.
|
||||||
|
pub fn insert_sleep_wall(&mut self, deadline: Instant, pid: Pid, epoch: u32) {
|
||||||
|
self.push(deadline, pid, Reason::Sleep { epoch }, true);
|
||||||
|
}
|
||||||
|
|
||||||
/// Arm a cancellable `send_after` timer: run `fire` at `deadline` unless
|
/// Arm a cancellable `send_after` timer: run `fire` at `deadline` unless
|
||||||
/// [`cancel`](Self::cancel)led first. `pid` is informational only (the
|
/// [`cancel`](Self::cancel)led first. `pid` is informational only (the
|
||||||
/// destination, or who armed it — useful for introspection); it is *not*
|
/// destination, or who armed it — useful for introspection); it is *not*
|
||||||
@@ -169,18 +189,8 @@ impl Timers {
|
|||||||
pid: Pid,
|
pid: Pid,
|
||||||
fire: Box<dyn FnOnce() + Send>,
|
fire: Box<dyn FnOnce() + Send>,
|
||||||
) -> TimerId {
|
) -> TimerId {
|
||||||
let seq = self.next_seq;
|
self.armed.insert(self.next_seq);
|
||||||
self.next_seq = self.next_seq.wrapping_add(1);
|
TimerId(self.push(deadline, pid, Reason::Send { fire }, false))
|
||||||
self.armed.insert(seq);
|
|
||||||
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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Cancel an armed `send_after` timer. Returns `true` if the timer was
|
/// Cancel an armed `send_after` timer. Returns `true` if the timer was
|
||||||
@@ -192,8 +202,18 @@ impl Timers {
|
|||||||
self.armed.remove(&id.0)
|
self.armed.remove(&id.0)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Insert an arbitrary timer entry.
|
/// Insert an arbitrary (virtual-anchored) timer entry.
|
||||||
pub fn insert(&mut self, deadline: Instant, pid: Pid, reason: Reason) {
|
pub fn insert(&mut self, deadline: Instant, pid: Pid, reason: Reason) {
|
||||||
|
self.push(deadline, pid, reason, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Common insertion path. `wall` selects the RFC 007 anchor (see
|
||||||
|
/// [`insert_sleep_wall`](Self::insert_sleep_wall)); it is accepted — and
|
||||||
|
/// ignored — without the `smarm-causal` feature so callers don't fork.
|
||||||
|
/// Returns the entry's `seq`.
|
||||||
|
fn push(&mut self, deadline: Instant, pid: Pid, reason: Reason, wall: bool) -> u64 {
|
||||||
|
#[cfg(not(feature = "smarm-causal"))]
|
||||||
|
let _ = wall;
|
||||||
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 {
|
self.heap.push(Reverse(Entry {
|
||||||
@@ -203,7 +223,10 @@ impl Timers {
|
|||||||
reason,
|
reason,
|
||||||
#[cfg(feature = "smarm-causal")]
|
#[cfg(feature = "smarm-causal")]
|
||||||
delay_stamp: crate::causal::global_delay_cycles(),
|
delay_stamp: crate::causal::global_delay_cycles(),
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
wall,
|
||||||
}));
|
}));
|
||||||
|
seq
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn is_empty(&self) -> bool {
|
pub fn is_empty(&self) -> bool {
|
||||||
@@ -239,7 +262,9 @@ impl Timers {
|
|||||||
/// [`peek_deadline`](Self::peek_deadline) may under-report (raw deadline
|
/// [`peek_deadline`](Self::peek_deadline) may under-report (raw deadline
|
||||||
/// earlier than effective), costing at most one spurious scheduler wake
|
/// earlier than effective), costing at most one spurious scheduler wake
|
||||||
/// per injected chunk; and a shift never converts wall time — with zero
|
/// per injected chunk; and a shift never converts wall time — with zero
|
||||||
/// debt the path is byte-identical to the featureless one.
|
/// debt the path is byte-identical to the featureless one. Wall-anchored
|
||||||
|
/// entries ([`insert_sleep_wall`](Self::insert_sleep_wall)) are exempt
|
||||||
|
/// from the shift and always fire at their raw deadline.
|
||||||
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")]
|
#[cfg(feature = "smarm-causal")]
|
||||||
@@ -260,7 +285,7 @@ impl Timers {
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
#[cfg(feature = "smarm-causal")]
|
#[cfg(feature = "smarm-causal")]
|
||||||
{
|
if !entry.wall {
|
||||||
let debt = global.saturating_sub(entry.delay_stamp);
|
let debt = global.saturating_sub(entry.delay_stamp);
|
||||||
if debt > 0 {
|
if debt > 0 {
|
||||||
let shifted = entry
|
let shifted = entry
|
||||||
|
|||||||
@@ -461,6 +461,34 @@ fn timer_zero_debt_pops_at_raw_deadline() {
|
|||||||
assert!(t.is_empty());
|
assert!(t.is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A wall-anchored timer (RFC 007 controller fix) fires at its raw deadline
|
||||||
|
/// regardless of injected delay, while a virtual sibling in the same heap
|
||||||
|
/// still shifts. This is what keeps the causal controller's experiment
|
||||||
|
/// windows a fixed wall length even under heavy injection.
|
||||||
|
#[test]
|
||||||
|
fn wall_timer_ignores_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_wall(now + Duration::from_millis(50), Pid::new(0, 0), 1);
|
||||||
|
t.insert_sleep(now + Duration::from_millis(50), Pid::new(1, 0), 1);
|
||||||
|
|
||||||
|
// 100ms of debt lands while both are pending.
|
||||||
|
smarm::causal::inject_delay_cycles_for_test((hz * 0.100) as u64);
|
||||||
|
|
||||||
|
// Just past the raw deadline: the wall entry fires, the virtual one is
|
||||||
|
// re-queued at its shifted deadline.
|
||||||
|
let due = t.pop_due(now + Duration::from_millis(60));
|
||||||
|
assert_eq!(due.len(), 1, "exactly the wall entry must fire at raw deadline");
|
||||||
|
assert_eq!(due[0].pid, Pid::new(0, 0));
|
||||||
|
assert!(!t.is_empty(), "virtual sibling must remain queued, shifted");
|
||||||
|
}
|
||||||
|
|
||||||
/// Re-queueing a shifted entry must preserve its identity: a `send_after`
|
/// Re-queueing a shifted entry must preserve its identity: a `send_after`
|
||||||
/// cancelled *after* being shifted past its raw deadline must still cancel
|
/// cancelled *after* being shifted past its raw deadline must still cancel
|
||||||
/// (return true) and must never deliver.
|
/// (return true) and must never deliver.
|
||||||
|
|||||||
Reference in New Issue
Block a user