Jan-Peter von Hunnius
08f53d68fe
Introduces a VirtualClock-based timing system replacing wall-clock Instant-based timing for hardware-accurate simulation. New modules: - virtual_clock.rs: Shared virtual time source with condvar sync - real_time_governor.rs: Thread that advances virtual time - component_timing.rs: Per-component timing constraints Key changes: - Component trait: added update_at_time(), timing(), set_virtual_clock_optional() - All Intel 400x components migrated to virtual-time-aware operation - ConfigurableSystem: start() returns Result, spawns governor thread - GUI: timing controls (speed slider, mode selector, step button) - Console: speed/mode/step/time commands - Backward compatible: legacy fallback for components without VirtualClock Reviewed-by: code review found 2 warnings (clock resume() dead code, reset() safety docs) — both fixed.
1025 lines
40 KiB
Rust
1025 lines
40 KiB
Rust
//! Virtual Clock for simulation timing synchronization.
|
||
//!
|
||
//! The `VirtualClock` provides a shared virtual time source that all component threads
|
||
//! synchronize against. Instead of using wall-clock time (`Instant::now()`) for timing
|
||
//! decisions, components use virtual time — a monotonically increasing nanosecond counter
|
||
//! that is independent of OS scheduling granularity.
|
||
//!
|
||
//! # Architecture
|
||
//!
|
||
//! ```text
|
||
//! Governor Thread Component Thread 1 Component Thread N
|
||
//! | | |
|
||
//! advance_by(tick_size) wait_until(target) wait_until(target)
|
||
//! | | |
|
||
//! v v v
|
||
//! [VirtualClock] <-- notify_all wakes all waiting threads
|
||
//! time_ns: AtomicU64
|
||
//! ```
|
||
//!
|
||
//! # How It Works
|
||
//!
|
||
//! 1. The `RealTimeGovernor` advances virtual time by `tick_size` nanoseconds
|
||
//! 2. All component threads that called `wait_until()` are woken via `condvar.notify_all()`
|
||
//! 3. Each component checks if virtual time has reached its target
|
||
//! 4. If yes: process the update, schedule next wake-up
|
||
//! 5. If no: go back to waiting
|
||
//!
|
||
//! This models real hardware where each chip operates independently but shares
|
||
//! a common timeline. Components can race each other, just like real chips.
|
||
|
||
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, AtomicUsize, Ordering};
|
||
use std::sync::{Condvar, Mutex};
|
||
use std::time::{Duration, Instant};
|
||
|
||
/// Convert a Duration to u64 nanoseconds, panicking if it exceeds u64::MAX (~584 years).
|
||
///
|
||
/// This is a safety check to prevent silent truncation when converting from
|
||
/// `Duration::as_nanos()` (u128) to u64. In practice, durations exceeding
|
||
/// u64::MAX nanoseconds are impossible in simulation, so a panic indicates a bug.
|
||
fn duration_to_nanos(d: Duration) -> u64 {
|
||
u64::try_from(d.as_nanos()).expect("Duration exceeds u64::MAX nanoseconds")
|
||
}
|
||
|
||
/// Simulation mode controlling how virtual time advances.
|
||
#[derive(Debug, Clone, Copy, PartialEq)]
|
||
#[repr(u8)]
|
||
pub enum SimulationMode {
|
||
/// Virtual time advances at a rate determined by `speed_multiplier`.
|
||
/// At 1.0x, one virtual nanosecond maps to one wall-clock nanosecond.
|
||
/// At 0.1x, one virtual nanosecond takes 10 wall-clock nanoseconds.
|
||
RealTime = 0,
|
||
|
||
/// Virtual time advances as fast as the host CPU allows.
|
||
/// No wall-clock sleeping — pure computation speed.
|
||
MaxSpeed = 1,
|
||
|
||
/// Virtual time advances only when the user triggers a step.
|
||
/// Each step advances by `tick_size` nanoseconds.
|
||
SingleStep = 2,
|
||
}
|
||
|
||
impl SimulationMode {
|
||
fn from_u8(value: u8) -> Self {
|
||
match value {
|
||
0 => SimulationMode::RealTime,
|
||
1 => SimulationMode::MaxSpeed,
|
||
2 => SimulationMode::SingleStep,
|
||
_ => {
|
||
debug_assert!(false, "Invalid SimulationMode value: {}", value);
|
||
SimulationMode::RealTime
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Granularity for single-step mode.
|
||
///
|
||
/// TODO: Integrate into `trigger_step()` / `consume_step()` to allow stepping
|
||
/// by clock phase, cycle, or instruction granularity.
|
||
#[derive(Debug, Clone, Copy, PartialEq)]
|
||
#[allow(dead_code)] // Planned for future use in single-step control
|
||
pub enum StepGranularity {
|
||
/// Advance by one clock phase (e.g., PHI1 high → PHI1 low).
|
||
/// For a 750 kHz clock, this is ~666 ns of virtual time.
|
||
ClockPhase,
|
||
|
||
/// Advance by one full clock cycle (PHI1 high → PHI1 low → PHI2 high → PHI2 low).
|
||
/// For a 750 kHz clock, this is ~1333 ns of virtual time.
|
||
ClockCycle,
|
||
|
||
/// Advance by one complete CPU instruction (8 or 16 clock cycles).
|
||
/// For a 750 kHz clock, this is ~10.6–21.3 µs of virtual time.
|
||
Instruction,
|
||
}
|
||
|
||
/// Internal state protected by the condvar mutex.
|
||
struct ClockState {
|
||
/// Flag indicating the simulation is paused.
|
||
paused: bool,
|
||
/// Flag indicating the simulation has been stopped (cancelled).
|
||
/// Used to wake threads in `wait_for_start()` that are blocking
|
||
/// before the simulation was ever started.
|
||
stopped: bool,
|
||
}
|
||
|
||
/// Shared virtual clock for simulation timing synchronization.
|
||
///
|
||
/// This struct is wrapped in `Arc` and shared between:
|
||
/// - The `RealTimeGovernor` thread (advances time)
|
||
/// - All component threads (wait for time to reach their scheduled update)
|
||
///
|
||
/// # Thread Safety
|
||
///
|
||
/// All time values are stored as `AtomicU64` for lock-free reads.
|
||
/// The `Condvar` is used to efficiently wake waiting threads when time advances.
|
||
///
|
||
/// # Lock Ordering
|
||
///
|
||
/// The `VirtualClock` has an internal `state_mutex` used for condvar synchronization.
|
||
/// When integrating with external mutexes (e.g., `parking_lot::Mutex<ConfigurableSystem>`),
|
||
/// the following lock ordering must be observed to prevent deadlocks:
|
||
///
|
||
/// **Order: External mutex → VirtualClock.state_mutex**
|
||
///
|
||
/// ```text
|
||
/// ConfigurableSystem mutex (parking_lot)
|
||
/// └→ VirtualClock.state_mutex (std::sync)
|
||
/// └→ No further locks acquired
|
||
/// ```
|
||
///
|
||
/// This ordering is safe because:
|
||
/// - The GUI thread acquires `ConfigurableSystem` first, then reads VirtualClock atomics
|
||
/// - Component threads never acquire `ConfigurableSystem`; they only access VirtualClock
|
||
/// - The governor thread only accesses VirtualClock
|
||
///
|
||
/// **Do NOT** acquire `ConfigurableSystem` mutex while holding `VirtualClock.state_mutex`.
|
||
///
|
||
/// # Example
|
||
///
|
||
/// ```rust,ignore
|
||
/// use std::sync::Arc;
|
||
/// use std::time::Duration;
|
||
/// use rusty_emu::virtual_clock::VirtualClock;
|
||
///
|
||
/// let clock = Arc::new(VirtualClock::new(Duration::from_nanos(333)));
|
||
///
|
||
/// // In governor thread:
|
||
/// clock.advance_by(Duration::from_nanos(333));
|
||
///
|
||
/// // In component thread:
|
||
/// let current_time = clock.wait_until(Duration::from_nanos(666));
|
||
/// // Component can now process its update at virtual time 666ns
|
||
/// ```
|
||
pub struct VirtualClock {
|
||
/// Current virtual time in nanoseconds (monotonically increasing).
|
||
time_ns: AtomicU64,
|
||
|
||
/// Virtual time advance per tick in nanoseconds.
|
||
tick_size_ns: AtomicU64,
|
||
|
||
/// Speed multiplier stored as fixed-point: actual_value = raw / 1_000_000_000.
|
||
/// This allows atomic updates without floating-point.
|
||
/// Examples:
|
||
/// 1.0x = 1_000_000_000
|
||
/// 0.1x = 100_000_000
|
||
/// 0.001x = 1_000_000
|
||
/// 10.0x = 10_000_000_000
|
||
speed_multiplier_fp: AtomicU64,
|
||
|
||
/// Current simulation mode (RealTime, MaxSpeed, SingleStep).
|
||
mode: AtomicU8,
|
||
|
||
/// Whether the simulation is running (used to signal component threads to exit).
|
||
running: AtomicBool,
|
||
|
||
/// Single-step trigger: set to true by user, consumed by governor.
|
||
step_pending: AtomicBool,
|
||
|
||
/// Counter for batching notify_all calls. Only every NOTIFY_INTERVAL-th
|
||
/// call to advance_by() actually notifies waiting threads. This reduces
|
||
/// mutex contention in MaxSpeed mode where advance_by() is called in a
|
||
/// tight loop.
|
||
notify_counter: AtomicU64,
|
||
|
||
/// Debug-only: tracks which thread last called advance_by() to detect
|
||
/// concurrent callers. 0 means no owner yet. Each thread gets a unique
|
||
/// ID via a global counter (see advance_by).
|
||
#[cfg(debug_assertions)]
|
||
advance_owner: AtomicUsize,
|
||
|
||
/// Condvar for notifying component threads when virtual time advances.
|
||
time_advanced: Condvar,
|
||
|
||
/// Mutex for condvar wait.
|
||
state_mutex: Mutex<ClockState>,
|
||
|
||
}
|
||
|
||
impl VirtualClock {
|
||
/// Create a new VirtualClock with the specified tick size.
|
||
///
|
||
/// # Parameters
|
||
/// - `tick_size`: Virtual time advance per tick. For a 750 kHz two-phase clock,
|
||
/// use `Duration::from_nanos(333)` (1/4 of the 1333ns cycle).
|
||
///
|
||
/// # Returns
|
||
/// A new VirtualClock initialized at virtual time 0, speed 1.0x, RealTime mode.
|
||
pub fn new(tick_size: Duration) -> Self {
|
||
VirtualClock {
|
||
time_ns: AtomicU64::new(0),
|
||
tick_size_ns: AtomicU64::new(duration_to_nanos(tick_size)),
|
||
speed_multiplier_fp: AtomicU64::new(1_000_000_000), // 1.0x
|
||
mode: AtomicU8::new(SimulationMode::RealTime as u8),
|
||
running: AtomicBool::new(false),
|
||
step_pending: AtomicBool::new(false),
|
||
notify_counter: AtomicU64::new(0),
|
||
#[cfg(debug_assertions)]
|
||
advance_owner: AtomicUsize::new(0),
|
||
time_advanced: Condvar::new(),
|
||
state_mutex: Mutex::new(ClockState { paused: false, stopped: false }),
|
||
}
|
||
}
|
||
|
||
// =========================================================================
|
||
// Time Read Operations (lock-free)
|
||
// =========================================================================
|
||
|
||
/// Get the current virtual time.
|
||
///
|
||
/// This is lock-free and can be called from any thread at any time.
|
||
#[inline]
|
||
pub fn now(&self) -> Duration {
|
||
Duration::from_nanos(self.time_ns.load(Ordering::Acquire))
|
||
}
|
||
|
||
/// Get the current virtual time as raw nanoseconds.
|
||
///
|
||
/// Useful for high-performance comparisons where `Duration` overhead matters.
|
||
#[inline]
|
||
pub fn now_ns(&self) -> u64 {
|
||
self.time_ns.load(Ordering::Acquire)
|
||
}
|
||
|
||
/// Get the current tick size.
|
||
pub fn tick_size(&self) -> Duration {
|
||
Duration::from_nanos(self.tick_size_ns.load(Ordering::Acquire))
|
||
}
|
||
|
||
/// Get the current tick size as raw nanoseconds.
|
||
///
|
||
/// Useful for high-performance calculations where `Duration` overhead matters.
|
||
#[inline]
|
||
pub fn tick_size_ns(&self) -> u64 {
|
||
self.tick_size_ns.load(Ordering::Acquire)
|
||
}
|
||
|
||
/// Get the total number of ticks elapsed since start.
|
||
///
|
||
/// Computed from `time_ns / tick_size_ns` for consistency — avoids
|
||
/// maintaining a separate counter that could diverge from the actual time.
|
||
///
|
||
/// **Note:** This is an approximate value. The two atomic loads (`time_ns`
|
||
/// and `tick_size_ns`) are not performed atomically. If `set_tick_size()`
|
||
/// is called concurrently from another thread, the result may be briefly
|
||
/// inconsistent. For precise timing decisions, use `wait_until()` instead.
|
||
pub fn tick_count(&self) -> u64 {
|
||
let time = self.time_ns.load(Ordering::Acquire);
|
||
let tick = self.tick_size_ns.load(Ordering::Acquire);
|
||
if tick == 0 { return 0; }
|
||
time / tick
|
||
}
|
||
|
||
// =========================================================================
|
||
// Time Wait Operations (blocking)
|
||
// =========================================================================
|
||
|
||
/// Block the calling thread until virtual time reaches at least `target`.
|
||
///
|
||
/// This is the primary synchronization mechanism for component threads.
|
||
/// The thread will sleep efficiently (via condvar) until the governor
|
||
/// advances virtual time past the target.
|
||
///
|
||
/// # Parameters
|
||
/// - `target`: The minimum virtual time to wait for.
|
||
///
|
||
/// # Returns
|
||
/// The actual virtual time when the thread was woken (>= target).
|
||
///
|
||
/// # Cancellation
|
||
/// Returns `None` if the simulation is stopped while waiting.
|
||
///
|
||
/// # Pause Behavior
|
||
/// When the simulation is paused, this method blocks indefinitely (the condvar
|
||
/// wait will not return until `resume()` or `stop()` is called). If `resume()`
|
||
/// is called, the thread re-enters the wait loop. If `stop()` is called, the
|
||
/// thread returns `None`. This means paused components do not consume CPU — they
|
||
/// efficiently sleep until the simulation is resumed or stopped.
|
||
pub fn wait_until(&self, target: Duration) -> Option<Duration> {
|
||
// NOTE: All component threads are woken on every tick via notify_all(),
|
||
// even if their individual target hasn't been reached. For large numbers
|
||
// of components, this causes O(N) spurious wakeups per tick. With
|
||
// NOTIFY_INTERVAL=10 batching at 750 kHz, this means ~300K wakeups/sec.
|
||
// Acceptable for current component counts (~10), but if this becomes a
|
||
// bottleneck, consider per-target notification (e.g., min-heap of targets)
|
||
// or timed wait batches.
|
||
let target_ns = duration_to_nanos(target);
|
||
|
||
// Fast path: check if time has already reached target (lock-free)
|
||
let current = self.time_ns.load(Ordering::Acquire);
|
||
if current >= target_ns {
|
||
// SeqCst to match stop()'s SeqCst store — ensures we honor the
|
||
// "Returns None if the simulation is stopped" contract even on
|
||
// the fast path.
|
||
if !self.running.load(Ordering::SeqCst) {
|
||
return None;
|
||
}
|
||
return Some(Duration::from_nanos(current));
|
||
}
|
||
|
||
// Slow path: acquire mutex to safely wait on condvar.
|
||
// All checks are done under the mutex to prevent missed-wakeup races
|
||
// with stop() — if stop() calls notify_all() while we hold the lock,
|
||
// we'll see the running=false on the next iteration.
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
|
||
loop {
|
||
// Fast path: check if time has reached target (common case).
|
||
// Acquire is sufficient for time_ns: we need to see the latest value
|
||
// written by advance_by()'s Release store. The condvar wait already
|
||
// provides the necessary synchronization guarantee (the mutex ensures
|
||
// we can't miss a notify_all from advance_by).
|
||
let current = self.time_ns.load(Ordering::Acquire);
|
||
if current >= target_ns {
|
||
return Some(Duration::from_nanos(current));
|
||
}
|
||
// SeqCst to match stop()'s SeqCst store — ensures we see the
|
||
// stopped state even if notify_all races with our wait().
|
||
if !self.running.load(Ordering::SeqCst) {
|
||
return None;
|
||
}
|
||
// If paused, stay sleeping efficiently (no CPU waste).
|
||
// When resume() notifies, we'll re-check conditions.
|
||
if guard.paused {
|
||
guard = self.time_advanced.wait(guard).unwrap_or_else(|e| e.into_inner());
|
||
continue;
|
||
}
|
||
guard = self.time_advanced.wait(guard).unwrap_or_else(|e| e.into_inner());
|
||
}
|
||
}
|
||
|
||
/// Non-blocking check: has virtual time reached the target?
|
||
///
|
||
/// Use this for polling-based component loops that prefer to do useful work
|
||
/// rather than block.
|
||
#[inline]
|
||
pub fn has_reached(&self, target: Duration) -> bool {
|
||
self.time_ns.load(Ordering::Acquire) >= duration_to_nanos(target)
|
||
}
|
||
|
||
/// Wait until virtual time reaches target, with a timeout.
|
||
///
|
||
/// Useful for component threads that need to periodically check for
|
||
/// external events (e.g., GUI updates) even if virtual time hasn't advanced.
|
||
///
|
||
/// # Returns
|
||
/// - `Ok(Duration)`: Virtual time reached the target
|
||
/// - `Err(())`: Timed out or simulation stopped
|
||
pub fn wait_until_with_timeout(&self, target: Duration, timeout: Duration) -> Result<Duration, ()> {
|
||
let target_ns = duration_to_nanos(target);
|
||
let deadline = Instant::now() + timeout;
|
||
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
|
||
loop {
|
||
// If paused, stay sleeping (timeout still applies).
|
||
// When resume() notifies, we'll re-check conditions.
|
||
if guard.paused {
|
||
// SeqCst to match stop()'s SeqCst store
|
||
if !self.running.load(Ordering::SeqCst) {
|
||
return Err(());
|
||
}
|
||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||
if remaining.is_zero() {
|
||
return Err(());
|
||
}
|
||
let result = self.time_advanced.wait_timeout(guard, remaining);
|
||
match result {
|
||
Ok((new_guard, timeout_result)) => {
|
||
guard = new_guard;
|
||
if timeout_result.timed_out() {
|
||
return Err(());
|
||
}
|
||
}
|
||
Err(poisoned) => {
|
||
(guard, _) = poisoned.into_inner();
|
||
}
|
||
}
|
||
continue;
|
||
}
|
||
let current = self.time_ns.load(Ordering::Acquire);
|
||
if current >= target_ns {
|
||
return Ok(Duration::from_nanos(current));
|
||
}
|
||
// SeqCst to match stop()'s SeqCst store
|
||
if !self.running.load(Ordering::SeqCst) {
|
||
return Err(());
|
||
}
|
||
|
||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||
if remaining.is_zero() {
|
||
return Err(());
|
||
}
|
||
let result = self.time_advanced.wait_timeout(guard, remaining);
|
||
match result {
|
||
Ok((new_guard, timeout_result)) => {
|
||
guard = new_guard;
|
||
if timeout_result.timed_out() {
|
||
return Err(());
|
||
}
|
||
}
|
||
Err(poisoned) => {
|
||
// Mutex was poisoned — recover the guard and re-check conditions
|
||
(guard, _) = poisoned.into_inner();
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// =========================================================================
|
||
// Time Advance Operations (called by governor only)
|
||
// =========================================================================
|
||
|
||
/// Advance virtual time by the specified duration.
|
||
///
|
||
/// This is called by the `RealTimeGovernor` to move simulation time forward.
|
||
/// All component threads waiting in `wait_until()` will be woken.
|
||
///
|
||
/// # Parameters
|
||
/// - `delta`: Amount of virtual time to advance.
|
||
///
|
||
/// # Visibility
|
||
/// This method is `pub(crate)` because it must only be called by the
|
||
/// governor thread. Calling it from component threads would corrupt
|
||
/// the simulation timeline.
|
||
pub(crate) fn advance_by(&self, delta: Duration) {
|
||
let delta_ns = duration_to_nanos(delta);
|
||
if delta_ns == 0 {
|
||
return;
|
||
}
|
||
|
||
// Debug-only guard: advance_by must only be called from a single thread
|
||
// (the governor). Concurrent calls would produce correct fetch_add results
|
||
// but could wake components at intermediate time values.
|
||
#[cfg(debug_assertions)]
|
||
{
|
||
use std::cell::Cell;
|
||
use std::sync::atomic::AtomicUsize;
|
||
// Global counter for assigning unique per-thread IDs.
|
||
static NEXT_THREAD_ID: AtomicUsize = AtomicUsize::new(1);
|
||
thread_local! {
|
||
static THREAD_ID: Cell<usize> = const { Cell::new(0) };
|
||
}
|
||
let tid = THREAD_ID.with(|t| {
|
||
let id = t.get();
|
||
if id == 0 {
|
||
let assigned = NEXT_THREAD_ID.fetch_add(1, Ordering::Relaxed);
|
||
t.set(assigned);
|
||
assigned
|
||
} else {
|
||
id
|
||
}
|
||
});
|
||
// Per-instance owner tracking: different VirtualClock instances
|
||
// (e.g., in different tests) can have different owner threads.
|
||
let prev = self.advance_owner.load(Ordering::SeqCst);
|
||
debug_assert!(
|
||
prev == 0 || prev == tid,
|
||
"advance_by() called from multiple threads — only the governor should advance time"
|
||
);
|
||
self.advance_owner.store(tid, Ordering::SeqCst);
|
||
}
|
||
|
||
// SeqCst ensures that woken component threads always see the latest time_ns value,
|
||
// even on weakly-ordered architectures (ARM, RISC-V). The Release/Acquire pair that
|
||
// was here previously is sufficient on x86-64 (TSO) but could allow stale reads on
|
||
// weaker memory models when combined with the NOTIFY_INTERVAL batching — a thread
|
||
// woken by a later notify_all() might not see the cumulative effect of intermediate
|
||
// Release stores. SeqCst provides total ordering at negligible cost on x86-64.
|
||
self.time_ns.fetch_add(delta_ns, Ordering::SeqCst);
|
||
|
||
// Batch notifications: only acquire mutex and call notify_all() every
|
||
// NOTIFY_INTERVAL ticks. This reduces syscall overhead from ~3M/s to ~300K/s
|
||
// at 750 kHz simulation speed, while keeping wakeup latency under ~3.3µs.
|
||
let count = self.notify_counter.fetch_add(1, Ordering::Relaxed);
|
||
if count % Self::NOTIFY_INTERVAL == 0 {
|
||
// Hold the mutex while notifying to prevent a missed-wakeup race:
|
||
// Without this, a waiter could load time_ns (see old value), then we
|
||
// update time_ns AND call notify_all(), then the waiter enters wait()
|
||
// — missing the notification entirely.
|
||
let _guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
// We use notify_all (not notify_one) because multiple components
|
||
// may be waiting for different target times, and any of them might
|
||
// now have their target satisfied.
|
||
self.time_advanced.notify_all();
|
||
}
|
||
}
|
||
|
||
/// Number of advance_by() calls between notify_all() calls.
|
||
/// Reduces mutex/syscall overhead in MaxSpeed mode where ticks happen
|
||
/// in a tight loop. Components may observe slightly stale time values
|
||
/// (up to NOTIFY_INTERVAL ticks behind), but this is acceptable because:
|
||
/// - Components that need precise timing use wait_until() which always
|
||
/// wakes on the next notification
|
||
/// - The maximum staleness is NOTIFY_INTERVAL × tick_size (e.g., 10 × 333ns = 3.33µs)
|
||
const NOTIFY_INTERVAL: u64 = 10;
|
||
|
||
/// Advance virtual time by one tick size.
|
||
///
|
||
/// Convenience method equivalent to `self.advance_by(self.tick_size())`.
|
||
///
|
||
/// # Visibility
|
||
/// This method is `pub(crate)` because it must only be called by the
|
||
/// governor thread. Calling it from component threads would corrupt
|
||
/// the simulation timeline.
|
||
pub(crate) fn advance_one_tick(&self) {
|
||
let tick_ns = self.tick_size_ns.load(Ordering::Acquire);
|
||
self.advance_by(Duration::from_nanos(tick_ns));
|
||
}
|
||
|
||
// =========================================================================
|
||
// Speed Control
|
||
// =========================================================================
|
||
|
||
/// Get the current speed multiplier.
|
||
///
|
||
/// Returns the multiplier as a floating-point value:
|
||
/// - 1.0 = real-time
|
||
/// - 0.1 = 10× slower than real-time
|
||
/// - 10.0 = 10× faster than real-time (limited by host CPU)
|
||
pub fn speed_multiplier(&self) -> f64 {
|
||
self.speed_multiplier_fp.load(Ordering::Acquire) as f64 / 1_000_000_000.0
|
||
}
|
||
|
||
/// Set the speed multiplier.
|
||
///
|
||
/// # Parameters
|
||
/// - `multiplier`: Speed multiplier. Must be > 0.0.
|
||
/// - 0.001 = 1000× slower than real-time
|
||
/// - 1.0 = real-time
|
||
/// - 100.0 = 100× faster than real-time (limited by host CPU)
|
||
///
|
||
/// # Panics
|
||
/// Panics if `multiplier` is not positive (<= 0.0).
|
||
pub fn set_speed_multiplier(&self, multiplier: f64) {
|
||
assert!(multiplier > 0.0, "Speed multiplier must be positive, got {}", multiplier);
|
||
let fp_value = (multiplier * 1_000_000_000.0).round().clamp(1.0, u64::MAX as f64) as u64;
|
||
// Verify round-trip to detect precision loss from f64 → u64 fixed-point conversion.
|
||
// This catches values that cannot be represented in our fixed-point format.
|
||
let reconstructed = fp_value as f64 / 1_000_000_000.0;
|
||
assert!(
|
||
(reconstructed - multiplier).abs() <= 0.000_000_001,
|
||
"Speed multiplier {} cannot be represented with sufficient precision (stored as {}). \
|
||
Use a value with at most ~9 significant decimal digits.",
|
||
multiplier, reconstructed
|
||
);
|
||
self.speed_multiplier_fp.store(fp_value, Ordering::Release);
|
||
}
|
||
|
||
/// Get the simulation mode.
|
||
pub fn mode(&self) -> SimulationMode {
|
||
SimulationMode::from_u8(self.mode.load(Ordering::Acquire))
|
||
}
|
||
|
||
/// Set the simulation mode.
|
||
pub fn set_mode(&self, mode: SimulationMode) {
|
||
self.mode.store(mode as u8, Ordering::Release);
|
||
// Hold mutex during notify to prevent missed-wakeup race with wait_until()
|
||
{
|
||
let _guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
self.time_advanced.notify_all();
|
||
}
|
||
}
|
||
|
||
/// Set the tick size.
|
||
///
|
||
/// This changes the virtual time advance per tick.
|
||
/// Can be called while the simulation is running.
|
||
///
|
||
/// # Panics
|
||
/// Panics if `tick_size` is zero. A zero tick size would cause division-by-zero
|
||
/// in `tick_count()` and produce no time advancement in the governor loop.
|
||
pub fn set_tick_size(&self, tick_size: Duration) {
|
||
let ns = duration_to_nanos(tick_size);
|
||
assert!(ns > 0, "tick_size must be non-zero");
|
||
self.tick_size_ns.store(ns, Ordering::Release);
|
||
}
|
||
|
||
// =========================================================================
|
||
// Single-Step Control
|
||
// =========================================================================
|
||
|
||
/// Trigger a single step in SingleStep mode.
|
||
///
|
||
/// Returns `false` if a step is already pending (not yet consumed by governor).
|
||
pub fn trigger_step(&self) -> bool {
|
||
// Only allow stepping in SingleStep mode
|
||
if self.mode() != SimulationMode::SingleStep {
|
||
return false;
|
||
}
|
||
// Try to set step_pending from false to true
|
||
self.step_pending
|
||
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
|
||
.is_ok()
|
||
}
|
||
|
||
/// Consume a pending step trigger (called by governor).
|
||
///
|
||
/// Returns `true` if a step was pending and has been consumed.
|
||
pub fn consume_step(&self) -> bool {
|
||
self.step_pending
|
||
.compare_exchange(true, false, Ordering::SeqCst, Ordering::SeqCst)
|
||
.is_ok()
|
||
}
|
||
|
||
/// Check if a step is pending without consuming it.
|
||
pub fn is_step_pending(&self) -> bool {
|
||
self.step_pending.load(Ordering::Acquire)
|
||
}
|
||
|
||
// =========================================================================
|
||
// Lifecycle Control
|
||
// =========================================================================
|
||
|
||
/// Start the simulation.
|
||
///
|
||
/// Sets the running flag and wakes all waiting threads.
|
||
/// This method is idempotent — calling it when already running has no effect
|
||
/// beyond waking currently waiting threads.
|
||
pub fn start(&self) {
|
||
let was_stopped = self.running.compare_exchange(
|
||
false, true, Ordering::SeqCst, Ordering::SeqCst
|
||
).is_ok();
|
||
if was_stopped {
|
||
// Hold mutex during notify to prevent missed-wakeup race with wait_until()
|
||
let _guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
self.time_advanced.notify_all();
|
||
}
|
||
}
|
||
|
||
/// Stop the simulation.
|
||
///
|
||
/// Clears the running flag and wakes all waiting threads so they can exit.
|
||
/// Also sets the internal `stopped` flag so that threads blocked in
|
||
/// `wait_for_start()` can detect cancellation and return `false`.
|
||
pub fn stop(&self) {
|
||
self.running.store(false, Ordering::SeqCst);
|
||
// Hold mutex during notify to prevent missed-wakeup race with wait_until()
|
||
{
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
guard.stopped = true;
|
||
self.time_advanced.notify_all();
|
||
}
|
||
}
|
||
|
||
/// Check if the simulation is running.
|
||
#[inline]
|
||
pub fn is_running(&self) -> bool {
|
||
self.running.load(Ordering::SeqCst)
|
||
}
|
||
|
||
/// Pause the simulation (virtual time stops advancing).
|
||
///
|
||
/// The governor will stop advancing time but component threads remain alive.
|
||
/// Waiting threads are notified so they can re-check the pause state.
|
||
pub fn pause(&self) {
|
||
{
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
guard.paused = true;
|
||
// Hold mutex during notify to prevent missed-wakeup race with wait_until()
|
||
self.time_advanced.notify_all();
|
||
}
|
||
}
|
||
|
||
/// Resume the simulation from paused state.
|
||
pub fn resume(&self) {
|
||
{
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
guard.paused = false;
|
||
// Hold mutex during notify to prevent missed-wakeup race with wait_until()
|
||
self.time_advanced.notify_all();
|
||
}
|
||
}
|
||
|
||
/// Check if the simulation is paused.
|
||
pub fn is_paused(&self) -> bool {
|
||
self.state_mutex
|
||
.lock()
|
||
.map(|g| g.paused)
|
||
.unwrap_or(false)
|
||
}
|
||
|
||
/// Block the calling thread until the simulation is unpaused.
|
||
///
|
||
/// This is used by the governor to efficiently wait (zero CPU) while paused,
|
||
/// instead of busy-polling with `thread::sleep`. Returns immediately if not
|
||
/// currently paused.
|
||
///
|
||
/// # Returns
|
||
/// - `true`: Simulation was resumed (or was never paused).
|
||
/// - `false`: Simulation was stopped while paused.
|
||
pub fn wait_for_resume(&self) -> bool {
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
while guard.paused {
|
||
// SeqCst to match stop()'s SeqCst store
|
||
if !self.running.load(Ordering::SeqCst) {
|
||
return false;
|
||
}
|
||
guard = self.time_advanced.wait(guard).unwrap_or_else(|e| e.into_inner());
|
||
}
|
||
true
|
||
}
|
||
|
||
/// Block the calling thread until the simulation has been started.
|
||
///
|
||
/// This is used by component threads to safely wait for the governor to call
|
||
/// `start()` before entering their main loop. Without this, components that call
|
||
/// `wait_until()` before the governor starts will see `running == false` and
|
||
/// immediately return `None`.
|
||
///
|
||
/// # Returns
|
||
/// - `true`: Simulation was started.
|
||
/// - `false`: Simulation was stopped before starting (should not normally happen).
|
||
///
|
||
/// # Deadlock Prevention
|
||
/// If `stop()` is called before `start()`, this method returns `false` instead
|
||
/// of blocking forever. This prevents deadlock when the simulation is torn down
|
||
/// during initialization.
|
||
///
|
||
/// # Panic Safety
|
||
/// If the governor thread panics before calling `start()` and no other thread
|
||
/// calls `stop()`, component threads will block forever in this method. To
|
||
/// prevent this, ensure `stop()` is called in the governor's cleanup path
|
||
/// (e.g., via `Drop` or a panic hook), or use `wait_until_with_timeout()`
|
||
/// instead for time-bounded waiting.
|
||
pub fn wait_for_start(&self) -> bool {
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
loop {
|
||
// SeqCst to match start()'s SeqCst compare_exchange
|
||
if self.running.load(Ordering::SeqCst) {
|
||
return true;
|
||
}
|
||
if guard.stopped {
|
||
return false;
|
||
}
|
||
guard = self.time_advanced.wait(guard).unwrap_or_else(|e| e.into_inner());
|
||
}
|
||
}
|
||
|
||
/// Reset the virtual clock to time zero and clear all state.
|
||
///
|
||
/// This is equivalent to calling `stop()` followed by a full state reset.
|
||
/// After `reset()`, the clock is in the same state as a freshly created one
|
||
/// (except `tick_size_ns` is preserved as a config parameter).
|
||
///
|
||
/// Safe to call whether the clock is running or stopped — if running, it will
|
||
/// be stopped first (setting `running = false` under the mutex to prevent
|
||
/// TOCTOU races with `start()`).
|
||
pub fn reset(&self) {
|
||
// Hold the state mutex while stopping AND performing the stores.
|
||
// This prevents a TOCTOU race where another thread calls start() between
|
||
// dropping the mutex and the subsequent atomic stores.
|
||
let mut guard = self.state_mutex.lock().unwrap_or_else(|e| e.into_inner());
|
||
// Stop the clock under the mutex so start() cannot interleave
|
||
self.running.store(false, Ordering::SeqCst);
|
||
guard.paused = false;
|
||
guard.stopped = false;
|
||
|
||
// Perform stores while still holding the mutex so that start()
|
||
// (which also acquires the mutex) cannot interleave.
|
||
self.time_ns.store(0, Ordering::Release);
|
||
self.step_pending.store(false, Ordering::Release);
|
||
// Reset to initial state — callers expect a clean slate after reset().
|
||
self.speed_multiplier_fp.store(1_000_000_000, Ordering::Release); // 1.0×
|
||
self.mode.store(SimulationMode::RealTime as u8, Ordering::Release);
|
||
// Reset notification counter to ensure the first advance_by() after
|
||
// restart triggers an immediate notify_all(). Without this, the counter
|
||
// retains its pre-reset value, shifting the batching phase unpredictably.
|
||
self.notify_counter.store(0, Ordering::Release);
|
||
// tick_size_ns is intentionally NOT reset — it's a config parameter
|
||
// that should persist across resets (e.g., set during system construction).
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use std::sync::Arc;
|
||
use std::thread;
|
||
use std::time::Duration;
|
||
|
||
#[test]
|
||
fn test_virtual_clock_creation() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
assert_eq!(clock.now(), Duration::ZERO);
|
||
assert_eq!(clock.tick_size(), Duration::from_nanos(333));
|
||
assert!((clock.speed_multiplier() - 1.0).abs() < 0.001);
|
||
assert_eq!(clock.mode(), SimulationMode::RealTime);
|
||
assert!(!clock.is_running());
|
||
}
|
||
|
||
#[test]
|
||
fn test_advance_by() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
clock.advance_by(Duration::from_nanos(100));
|
||
assert_eq!(clock.now(), Duration::from_nanos(100));
|
||
|
||
clock.advance_by(Duration::from_nanos(200));
|
||
assert_eq!(clock.now(), Duration::from_nanos(300));
|
||
}
|
||
|
||
#[test]
|
||
fn test_advance_one_tick() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
clock.advance_one_tick();
|
||
assert_eq!(clock.now(), Duration::from_nanos(333));
|
||
|
||
clock.advance_one_tick();
|
||
assert_eq!(clock.now(), Duration::from_nanos(666));
|
||
}
|
||
|
||
#[test]
|
||
fn test_has_reached() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
assert!(!clock.has_reached(Duration::from_nanos(100)));
|
||
clock.advance_by(Duration::from_nanos(100));
|
||
assert!(clock.has_reached(Duration::from_nanos(100)));
|
||
assert!(!clock.has_reached(Duration::from_nanos(200)));
|
||
}
|
||
|
||
#[test]
|
||
fn test_speed_multiplier() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
clock.set_speed_multiplier(0.5);
|
||
assert!((clock.speed_multiplier() - 0.5).abs() < 0.001);
|
||
|
||
clock.set_speed_multiplier(10.0);
|
||
assert!((clock.speed_multiplier() - 10.0).abs() < 0.01);
|
||
|
||
clock.set_speed_multiplier(0.001);
|
||
assert!((clock.speed_multiplier() - 0.001).abs() < 0.0001);
|
||
}
|
||
|
||
#[test]
|
||
fn test_simulation_mode() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
clock.set_mode(SimulationMode::MaxSpeed);
|
||
assert_eq!(clock.mode(), SimulationMode::MaxSpeed);
|
||
|
||
clock.set_mode(SimulationMode::SingleStep);
|
||
assert_eq!(clock.mode(), SimulationMode::SingleStep);
|
||
|
||
clock.set_mode(SimulationMode::RealTime);
|
||
assert_eq!(clock.mode(), SimulationMode::RealTime);
|
||
}
|
||
|
||
#[test]
|
||
fn test_single_step() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
clock.set_mode(SimulationMode::SingleStep);
|
||
|
||
// Can't step in non-SingleStep mode
|
||
clock.set_mode(SimulationMode::RealTime);
|
||
assert!(!clock.trigger_step());
|
||
|
||
clock.set_mode(SimulationMode::SingleStep);
|
||
|
||
// Trigger a step
|
||
assert!(clock.trigger_step());
|
||
assert!(clock.is_step_pending());
|
||
|
||
// Can't trigger again while pending
|
||
assert!(!clock.trigger_step());
|
||
|
||
// Consume the step
|
||
assert!(clock.consume_step());
|
||
assert!(!clock.is_step_pending());
|
||
|
||
// Can't consume when none pending
|
||
assert!(!clock.consume_step());
|
||
}
|
||
|
||
#[test]
|
||
fn test_lifecycle() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
assert!(!clock.is_running());
|
||
|
||
clock.start();
|
||
assert!(clock.is_running());
|
||
|
||
clock.pause();
|
||
assert!(clock.is_paused());
|
||
|
||
clock.resume();
|
||
assert!(!clock.is_paused());
|
||
|
||
clock.stop();
|
||
assert!(!clock.is_running());
|
||
}
|
||
|
||
#[test]
|
||
fn test_reset() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
clock.advance_by(Duration::from_nanos(1000));
|
||
assert_eq!(clock.now(), Duration::from_nanos(1000));
|
||
|
||
clock.reset();
|
||
assert_eq!(clock.now(), Duration::ZERO);
|
||
}
|
||
|
||
#[test]
|
||
fn test_wait_until_immediate() {
|
||
let clock = Arc::new(VirtualClock::new(Duration::from_nanos(333)));
|
||
clock.start();
|
||
|
||
// If time has already passed, wait_until should return immediately
|
||
clock.advance_by(Duration::from_nanos(1000));
|
||
|
||
let result = clock.wait_until(Duration::from_nanos(500));
|
||
assert!(result.is_some());
|
||
assert!(result.unwrap() >= Duration::from_nanos(500));
|
||
}
|
||
|
||
#[test]
|
||
fn test_wait_until_blocks_then_returns() {
|
||
let clock = Arc::new(VirtualClock::new(Duration::from_nanos(333)));
|
||
clock.start();
|
||
|
||
let clock_clone = Arc::clone(&clock);
|
||
let handle = thread::spawn(move || {
|
||
// This should block until time reaches 1000ns
|
||
clock_clone.wait_until(Duration::from_nanos(1000))
|
||
});
|
||
|
||
// Give the thread time to start waiting
|
||
thread::sleep(Duration::from_millis(50));
|
||
|
||
// Advance time past the target
|
||
clock.advance_by(Duration::from_nanos(1500));
|
||
|
||
let result = handle.join().expect("Thread panicked");
|
||
assert!(result.is_some());
|
||
assert!(result.unwrap() >= Duration::from_nanos(1000));
|
||
}
|
||
|
||
#[test]
|
||
fn test_wait_until_returns_none_on_stop() {
|
||
let clock = Arc::new(VirtualClock::new(Duration::from_nanos(333)));
|
||
clock.start();
|
||
|
||
let clock_clone = Arc::clone(&clock);
|
||
let handle = thread::spawn(move || {
|
||
// This should block until stopped
|
||
clock_clone.wait_until(Duration::from_nanos(1_000_000))
|
||
});
|
||
|
||
// Give the thread time to start waiting
|
||
thread::sleep(Duration::from_millis(50));
|
||
|
||
// Stop the simulation
|
||
clock.stop();
|
||
|
||
let result = handle.join().expect("Thread panicked");
|
||
assert!(result.is_none());
|
||
}
|
||
|
||
#[test]
|
||
fn test_multiple_waiters() {
|
||
let clock = Arc::new(VirtualClock::new(Duration::from_nanos(100)));
|
||
clock.start();
|
||
|
||
let clock1 = Arc::clone(&clock);
|
||
let clock2 = Arc::clone(&clock);
|
||
let clock3 = Arc::clone(&clock);
|
||
|
||
let h1 = thread::spawn(move || clock1.wait_until(Duration::from_nanos(300)));
|
||
let h2 = thread::spawn(move || clock2.wait_until(Duration::from_nanos(200)));
|
||
let h3 = thread::spawn(move || clock3.wait_until(Duration::from_nanos(500)));
|
||
|
||
// Advance past all targets
|
||
thread::sleep(Duration::from_millis(10));
|
||
clock.advance_by(Duration::from_nanos(600));
|
||
|
||
assert!(h1.join().unwrap().is_some());
|
||
assert!(h2.join().unwrap().is_some());
|
||
assert!(h3.join().unwrap().is_some());
|
||
}
|
||
|
||
#[test]
|
||
fn test_tick_count() {
|
||
let clock = VirtualClock::new(Duration::from_nanos(333));
|
||
|
||
assert_eq!(clock.tick_count(), 0);
|
||
|
||
clock.advance_one_tick();
|
||
assert_eq!(clock.tick_count(), 1);
|
||
|
||
clock.advance_one_tick();
|
||
assert_eq!(clock.tick_count(), 2);
|
||
|
||
// advance_by(100ns) with tick_size=333ns → total 766ns / 333 = 2 (integer division)
|
||
clock.advance_by(Duration::from_nanos(100));
|
||
assert_eq!(clock.tick_count(), 2); // 766 / 333 = 2
|
||
|
||
// One more full tick → total 1099ns / 333 = 3
|
||
clock.advance_one_tick();
|
||
assert_eq!(clock.tick_count(), 3); // 1099 / 333 = 3
|
||
}
|
||
}
|