ApsisRS
/
apsis-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

sim: add discrete event simulation engine

This commit is contained in:
ApsisRS contributors 2026-09-26 22:28:23 +08:00
parent 5f064fd422
commit 861ec120f4
6 changed files with 749 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Cargo.lock generated
View File

@ -309,6 +309,10 @@ version = "0.1.0"
[[package]] [[package]]
name = "apsis-sim-engine" name = "apsis-sim-engine"
version = "0.1.0" version = "0.1.0"
dependencies = [
"apsis-core",
"apsis-event",
]
[[package]] [[package]]
name = "apsis-task" name = "apsis-task"

2
crates/sim/apsis-sim-engine/Cargo.toml
View File

@ -11,3 +11,5 @@ publish = false
workspace = true workspace = true
[dependencies] [dependencies]
apsis-core.workspace = true
apsis-event.workspace = true

225
crates/sim/apsis-sim-engine/src/engine.rs Normal file
View File

@ -0,0 +1,225 @@
//! The engine itself.
use crate::outcome::{EngineOutcome, EngineState, StopReason};
use apsis_event::{Event, EventId, EventQueue};
/// A discrete-event simulation engine.
///
/// The engine owns the event queue and the clock. A run repeatedly takes the
/// earliest event and passes it to a handler, which may schedule more
/// events. The engine stops when the queue empties or when one of the limits
/// is reached.
#[derive(Debug, Clone)]
pub struct Engine<T> {
queue: EventQueue<T>,
now: f64,
executed: u64,
stop_requested: bool,
time_limit: Option<f64>,
event_limit: Option<u64>,
}
impl<T> Default for Engine<T> {
fn default() -> Self {
Self::new()
}
}
impl<T> Engine<T> {
/// Creates an engine with the clock at zero and no limits.
///
/// An engine with no limits runs until the queue empties, which is the
/// right default for a model that terminates. A model that schedules
/// without bound needs a limit, which is set with
/// [`Engine::with_time_limit`] or [`Engine::with_event_limit`].
pub const fn new() -> Self {
Self {
queue: EventQueue::new(),
now: 0.0,
executed: 0,
stop_requested: false,
time_limit: None,
event_limit: None,
}
}
/// Sets the latest simulation time the engine will process.
///
/// An event scheduled beyond the limit is left in the queue rather than
/// executed, so the caller can examine what remained. A limit that is
/// negative or not finite disables the limit.
pub const fn with_time_limit(mut self, limit: f64) -> Self {
self.time_limit = if limit.is_finite() && limit >= 0.0 {
Some(limit)
} else {
None
};
self
}
/// Sets the greatest number of events the engine will execute.
///
/// A limit of zero disables the rule, because a run that executes no
/// events is never useful.
pub const fn with_event_limit(mut self, limit: u64) -> Self {
self.event_limit = if limit == 0 { None } else { Some(limit) };
self
}
/// Returns the current simulation time, in seconds.
pub const fn now(&self) -> f64 {
self.now
}
/// Sets the current simulation time.
///
/// This is used to start a run at an offset, for example at a scenario
/// epoch rather than at zero. It does not reschedule pending events, so a
/// caller that moves the clock forward past a pending event will execute
/// that event late. The engine has no way to detect the mistake, so the
/// caller is responsible for keeping the clock consistent with the
/// queue.
pub fn set_now(&mut self, time: f64) {
if time.is_finite() && time >= 0.0 {
self.now = time;
}
}
/// Returns the number of events executed so far.
pub const fn executed(&self) -> u64 {
self.executed
}
/// Returns the number of pending events.
pub fn pending(&self) -> usize {
self.queue.len()
}
/// Returns `true` when no events are pending.
pub fn is_idle(&self) -> bool {
self.queue.is_empty()
}
/// Returns the time of the next pending event, if there is one.
pub fn next_time(&self) -> Option<f64> {
self.queue.next_time()
}
/// Returns what the engine currently holds.
pub fn state(&self) -> EngineState {
EngineState {
now: self.now,
pending: self.queue.len(),
executed: self.executed,
next_time: self.queue.next_time(),
}
}
/// Schedules an event at a time.
///
/// Returns the identifier of the new event, or `None` when the time is
/// not usable.
pub fn schedule(&mut self, time: f64, payload: T) -> Option<EventId> {
self.queue.schedule(time, payload)
}
/// Schedules an event at a delay from the current simulation time.
///
/// Returns the identifier of the new event, or `None` when the delay is
/// not usable.
pub fn schedule_after(&mut self, delay: f64, payload: T) -> Option<EventId> {
self.queue.schedule_after(self.now, delay, payload)
}
/// Asks the engine to stop after the current event.
///
/// The handler calls this when its own model says the run is over, for
/// example when a stopping condition that the queue cannot express is
/// met. The pending events are left in the queue.
pub fn request_stop(&mut self) {
self.stop_requested = true;
}
/// Clears the stop request, so that a subsequent run continues.
pub fn clear_stop(&mut self) {
self.stop_requested = false;
}
/// Removes every pending event and returns how many were discarded.
pub fn clear(&mut self) -> usize {
self.queue.clear()
}
/// Returns the queue, for a caller that needs to inspect or manipulate it
/// directly.
pub const fn queue(&self) -> &EventQueue<T> {
&self.queue
}
/// Returns the queue for modification.
pub const fn queue_mut(&mut self) -> &mut EventQueue<T> {
&mut self.queue
}
/// Runs the engine until it stops.
///
/// Each iteration takes the earliest event, advances the clock to its
/// time, and passes it to the handler. The handler may schedule further
/// events, including at the same instant, which are delivered after the
/// events already scheduled for that instant.
///
/// Returns the outcome of the run. The clock is left at the time of the
/// last event executed, not at a limit, so a caller that records the
/// clock sees the last instant the model actually reached.
pub fn run<F>(&mut self, mut handler: F) -> EngineOutcome
where
F: FnMut(&mut Engine<T>, Event<T>),
{
let started_at = self.now;
let mut executed_this_run = 0_u64;
self.stop_requested = false;
let reason = loop {
// The event limit is checked before taking an event, so a run
// that reaches the limit leaves the queue untouched for the next
// call.
if let Some(limit) = self.event_limit {
if self.executed >= limit {
break StopReason::EventLimit;
}
}
let Some(event) = self.queue.pop() else {
break StopReason::QueueEmpty;
};
// The time limit leaves the event in the queue rather than
// discarding it, so the caller can inspect what remained. The
// event was already popped, so it is put back.
if let Some(limit) = self.time_limit {
if event.time > limit {
self.queue.schedule(event.time, event.payload);
break StopReason::TimeLimit;
}
}
self.now = event.time;
self.executed = self.executed.wrapping_add(1);
executed_this_run += 1;
handler(self, event);
if self.stop_requested {
break StopReason::Requested;
}
};
EngineOutcome {
reason,
executed: executed_this_run,
started_at,
ended_at: self.now,
remaining: self.queue.len(),
}
}
}

50
crates/sim/apsis-sim-engine/src/lib.rs
View File

@ -1 +1,51 @@
//! Discrete-event simulation engine. //! Discrete-event simulation engine.
//!
//! The engine advances a clock by jumping from one event to the next. It
//! holds the event queue, the current simulation time, and the statistics a
//! caller needs to judge a run. What an event means is the caller's: it
//! supplies a handler that receives the event and may schedule more events,
//! and the engine only sequences the calls.
//!
//! The engine keeps the simulation deterministic. Events at the same instant
//! are delivered in the order they were scheduled, which the event queue
//! guarantees, so a run with the same inputs produces the same output. That
//! is what makes a result reproducible and a cross-check against another
//! implementation meaningful.
//!
//! The engine stops for one of four reasons, reported as a
//! [`StopReason`](crate::StopReason):
//! the queue empties, the simulation reaches its time limit, the caller asks
//! it to stop, or an event limit is reached. The last two exist so that a
//! model that schedules events forever cannot run without bound.
//!
//! ```
//! use apsis_sim_engine::Engine;
//!
//! let mut engine = Engine::new();
//! engine.schedule(10.0, "hello");
//!
//! let mut seen = Vec::new();
//! let outcome = engine.run(|engine, event| {
//! seen.push(event.payload);
//! // The handler schedules the next event, which the engine delivers
//! // after this one. It schedules it only once, so the queue drains.
//! if event.payload == "hello" {
//! engine.schedule(20.0, "world");
//! }
//! });
//!
//! assert_eq!(seen, vec!["hello", "world"]);
//! assert_eq!(outcome.executed, 2);
//! assert_eq!(engine.now(), 20.0);
//! ```
#![forbid(unsafe_code)]
#![deny(missing_docs)]
mod engine;
mod outcome;
pub use engine::Engine;
pub use outcome::{EngineOutcome, EngineState, StopReason};
pub use apsis_event::{Event, EventId, EventQueue, QueueState};

119
crates/sim/apsis-sim-engine/src/outcome.rs Normal file
View File

@ -0,0 +1,119 @@
//! What a run produced and what the engine currently holds.
use core::fmt;
/// Why a run stopped.
///
/// The reason is reported rather than inferred from the event count, because
/// the same count can arise from an empty queue or from a limit, and a
/// caller that records the outcome of a run needs to distinguish them.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum StopReason {
/// No events were left when the engine looked for the next one.
QueueEmpty,
/// The time of the next event is beyond the simulation time limit.
TimeLimit,
/// The handler set a flag asking the engine to stop.
Requested,
/// The event count reached the configured limit.
EventLimit,
}
impl StopReason {
/// Returns `true` when the run ended because the model had no more work
/// to do rather than because a limit intervened.
pub const fn is_natural(self) -> bool {
matches!(self, Self::QueueEmpty)
}
/// Returns a short name for the reason.
pub const fn name(self) -> &'static str {
match self {
Self::QueueEmpty => "queue empty",
Self::TimeLimit => "time limit",
Self::Requested => "requested",
Self::EventLimit => "event limit",
}
}
}
impl fmt::Display for StopReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.name())
}
}
/// What a run produced.
///
/// The outcome records the counts and the clock, so a caller can record the
/// result of a run without reaching back into the engine.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct EngineOutcome {
/// Why the run stopped.
pub reason: StopReason,
/// The number of events the run executed.
pub executed: u64,
/// The simulation time the run started from, in seconds.
pub started_at: f64,
/// The simulation time the run reached, in seconds.
pub ended_at: f64,
/// The number of events still pending when the run stopped.
pub remaining: usize,
}
impl EngineOutcome {
/// Returns the time the run covered, in seconds.
pub fn duration(&self) -> f64 {
self.ended_at - self.started_at
}
/// Returns the mean number of events per second of simulation time.
///
/// This measures how busy the model is rather than how long the run took
/// on the wall clock. Returns `None` when the run covered no time.
pub fn event_rate(&self) -> Option<f64> {
let duration = self.duration();
if duration > 0.0 {
Some(self.executed as f64 / duration)
} else {
None
}
}
/// Returns `true` when the run finished because the model had no more
/// work rather than because a limit stopped it.
pub fn is_natural(&self) -> bool {
self.reason.is_natural()
}
}
impl fmt::Display for EngineOutcome {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"{} event(s) from {} s to {} s, stopped by {}",
self.executed, self.started_at, self.ended_at, self.reason
)
}
}
/// What the engine currently holds.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct EngineState {
/// The current simulation time, in seconds.
pub now: f64,
/// The number of pending events.
pub pending: usize,
/// The number of events executed so far.
pub executed: u64,
/// The time of the next pending event, if there is one.
pub next_time: Option<f64>,
}
impl EngineState {
/// Returns `true` when no events are pending.
pub const fn is_idle(&self) -> bool {
self.pending == 0
}
}

349
crates/sim/apsis-sim-engine/tests/engine.rs Normal file
View File

@ -0,0 +1,349 @@
//! Tests for the simulation engine.
use apsis_sim_engine::{Engine, StopReason};
#[test]
fn a_new_engine_is_idle_at_time_zero() {
let engine: Engine<u32> = Engine::new();
assert_eq!(engine.now(), 0.0);
assert_eq!(engine.executed(), 0);
assert!(engine.is_idle());
assert_eq!(engine.pending(), 0);
assert_eq!(engine.next_time(), None);
let state = engine.state();
assert_eq!(state.now, 0.0);
assert_eq!(state.executed, 0);
assert!(state.is_idle());
assert_eq!(state.next_time, None);
}
#[test]
fn a_run_executes_every_event_in_order() {
let mut engine = Engine::new();
engine.schedule(30.0, 3);
engine.schedule(10.0, 1);
engine.schedule(20.0, 2);
let mut seen = Vec::new();
let outcome = engine.run(|_, event| seen.push(event.payload));
assert_eq!(seen, vec![1, 2, 3]);
assert_eq!(outcome.executed, 3);
assert_eq!(outcome.reason, StopReason::QueueEmpty);
assert!(outcome.is_natural());
assert_eq!(outcome.started_at, 0.0);
assert_eq!(outcome.ended_at, 30.0);
assert_eq!(outcome.remaining, 0);
assert_eq!(engine.now(), 30.0);
assert_eq!(engine.executed(), 3);
assert!(engine.is_idle());
}
#[test]
fn a_handler_can_schedule_follow_up_events() {
// A periodic process that reschedules itself, which is the pattern a
// discrete-event model uses for a clock driven process.
let mut engine = Engine::new();
engine.schedule(0.0, 0_u32);
let mut seen = Vec::new();
let outcome = engine.run(|engine, event| {
seen.push(event.time);
if event.payload < 3 {
engine
.schedule_after(10.0, event.payload + 1)
.expect("the delay is usable");
}
});
assert_eq!(seen, vec![0.0, 10.0, 20.0, 30.0]);
assert_eq!(outcome.executed, 4);
assert_eq!(outcome.reason, StopReason::QueueEmpty);
// The clock ends at the last event, not at the delay of the last
// follow-up, which was never scheduled.
assert_eq!(outcome.ended_at, 30.0);
}
#[test]
fn events_at_the_same_instant_are_delivered_in_schedule_order() {
let mut engine = Engine::new();
engine.schedule(5.0, 'a');
engine.schedule(5.0, 'b');
engine.schedule(5.0, 'c');
let mut seen = Vec::new();
engine.run(|_, event| seen.push(event.payload));
assert_eq!(seen, vec!['a', 'b', 'c']);
}
#[test]
fn a_follow_up_scheduled_for_the_same_instant_is_delivered_after_the_pending_ones() {
// This is what makes a cascade at one instant well defined: events
// scheduled by a handler go behind the events already queued.
let mut engine = Engine::new();
engine.schedule(5.0, "first");
engine.schedule(5.0, "second");
let mut seen = Vec::new();
engine.run(|engine, event| {
seen.push(event.payload);
if event.payload == "first" {
engine
.schedule(5.0, "inserted")
.expect("the time is usable");
}
});
assert_eq!(seen, vec!["first", "second", "inserted"]);
}
#[test]
fn a_run_is_reproducible() {
// The same inputs must give the same output, which the tie-break on the
// schedule order guarantees.
let build = || {
let mut engine = Engine::new();
for index in 0..20 {
engine
.schedule(f64::from(index % 3), index)
.expect("the time is usable");
}
let mut seen = Vec::new();
engine.run(|_, event| seen.push(event.payload));
seen
};
assert_eq!(build(), build());
}
#[test]
fn a_time_limit_stops_the_run_and_leaves_the_event_pending() {
let mut engine = Engine::new().with_time_limit(15.0);
engine.schedule(10.0, 1);
engine.schedule(20.0, 2);
engine.schedule(30.0, 3);
let mut seen = Vec::new();
let outcome = engine.run(|_, event| seen.push(event.payload));
assert_eq!(seen, vec![1]);
assert_eq!(outcome.reason, StopReason::TimeLimit);
assert!(!outcome.is_natural());
// The event beyond the limit is still queued.
assert_eq!(outcome.remaining, 2);
assert_eq!(engine.next_time(), Some(20.0));
// The clock is at the last event executed, not at the limit.
assert_eq!(outcome.ended_at, 10.0);
}
#[test]
fn an_event_exactly_at_the_time_limit_is_executed() {
let mut engine = Engine::new().with_time_limit(10.0);
engine.schedule(10.0, "at limit");
let outcome = engine.run(|_, _| {});
assert_eq!(outcome.executed, 1);
assert_eq!(outcome.reason, StopReason::QueueEmpty);
}
#[test]
fn an_event_limit_stops_the_run_without_discarding_events() {
let mut engine = Engine::new().with_event_limit(2);
for index in 0..5 {
engine
.schedule(f64::from(index), index)
.expect("the time is usable");
}
let mut seen = Vec::new();
let outcome = engine.run(|_, event| seen.push(event.payload));
assert_eq!(seen, vec![0, 1]);
assert_eq!(outcome.reason, StopReason::EventLimit);
assert_eq!(outcome.executed, 2);
assert_eq!(outcome.remaining, 3);
assert_eq!(engine.executed(), 2);
// The limit is a lifetime cap rather than a per run one, so a second
// run at once stops without executing anything. Raising the limit lets
// the remaining events run.
let outcome = engine.run(|_, event| seen.push(event.payload));
assert_eq!(outcome.executed, 0);
assert_eq!(outcome.reason, StopReason::EventLimit);
let mut continued = engine.with_event_limit(10);
let outcome = continued.run(|_, event| seen.push(event.payload));
assert_eq!(seen, vec![0, 1, 2, 3, 4]);
assert_eq!(outcome.reason, StopReason::QueueEmpty);
assert_eq!(outcome.executed, 3);
}
#[test]
fn the_event_limit_counts_across_runs() {
// The limit is on the total number of events, so a run that already
// reached it does nothing.
let mut engine = Engine::new().with_event_limit(1);
engine.schedule(1.0, "a").expect("usable");
assert_eq!(engine.run(|_, _| {}).executed, 1);
assert_eq!(engine.run(|_, _| {}).reason, StopReason::EventLimit);
assert_eq!(engine.executed(), 1);
}
#[test]
fn a_stop_request_ends_the_run_after_the_current_event() {
let mut engine = Engine::new();
for index in 0..5 {
engine
.schedule(f64::from(index), index)
.expect("the time is usable");
}
let mut seen = Vec::new();
let outcome = engine.run(|engine, event| {
seen.push(event.payload);
if event.payload == 2 {
engine.request_stop();
}
});
// The event that requested the stop is the last one executed.
assert_eq!(seen, vec![0, 1, 2]);
assert_eq!(outcome.reason, StopReason::Requested);
assert_eq!(outcome.remaining, 2);
// The request is cleared by the run, so the next run continues.
let outcome = engine.run(|_, event| seen.push(event.payload));
assert_eq!(seen, vec![0, 1, 2, 3, 4]);
assert_eq!(outcome.reason, StopReason::QueueEmpty);
}
#[test]
fn a_stop_request_can_be_cleared_by_hand() {
let mut engine: Engine<u8> = Engine::new();
engine.schedule(1.0, 1).expect("usable");
engine.request_stop();
engine.clear_stop();
// The run clears the flag itself, so a clear beforehand is harmless.
assert_eq!(engine.run(|_, _| {}).executed, 1);
}
#[test]
fn an_empty_run_does_nothing() {
let mut engine: Engine<u8> = Engine::new();
let outcome = engine.run(|_, _| panic!("no handler call is expected"));
assert_eq!(outcome.executed, 0);
assert_eq!(outcome.reason, StopReason::QueueEmpty);
assert_eq!(outcome.duration(), 0.0);
assert_eq!(outcome.event_rate(), None);
}
#[test]
fn the_clock_can_be_set_to_start_at_an_offset() {
let mut engine = Engine::new();
engine.set_now(1_000.0);
assert_eq!(engine.now(), 1_000.0);
// A relative schedule is measured from the offset.
engine.schedule_after(5.0, "x").expect("usable");
assert_eq!(engine.next_time(), Some(1_005.0));
let outcome = engine.run(|_, _| {});
assert_eq!(outcome.started_at, 1_000.0);
assert_eq!(outcome.ended_at, 1_005.0);
assert_eq!(outcome.duration(), 5.0);
// An unusable time is ignored, leaving the clock alone.
engine.set_now(-1.0);
assert_eq!(engine.now(), 1_005.0);
engine.set_now(f64::NAN);
assert_eq!(engine.now(), 1_005.0);
}
#[test]
fn the_outcome_reports_the_event_rate() {
let mut engine = Engine::new();
for index in 0..10 {
engine
.schedule(f64::from(index) * 10.0, index)
.expect("usable");
}
let outcome = engine.run(|_, _| {});
// Ten events over ninety seconds, so the rate is a ninth per second.
assert_eq!(outcome.duration(), 90.0);
assert!((outcome.event_rate().expect("a positive duration") - 10.0 / 90.0).abs() < 1e-12);
}
#[test]
fn clearing_discards_the_pending_events() {
let mut engine = Engine::new();
for index in 0..4 {
engine.schedule(f64::from(index), index).expect("usable");
}
assert_eq!(engine.clear(), 4);
assert!(engine.is_idle());
assert_eq!(engine.run(|_, _| {}).executed, 0);
}
#[test]
fn the_queue_is_reachable_from_the_engine() {
let mut engine = Engine::new();
engine.schedule(1.0, "a").expect("usable");
assert_eq!(engine.queue().len(), 1);
assert_eq!(engine.queue().peek(), Some(&"a"));
// The queue may be modified directly.
engine.queue_mut().clear();
assert!(engine.is_idle());
}
#[test]
fn a_bad_time_limit_is_ignored() {
let mut engine = Engine::new().with_time_limit(-1.0);
engine.schedule(100.0, "x").expect("usable");
// The run reaches the event, because no limit applies.
assert_eq!(engine.run(|_, _| {}).executed, 1);
let mut engine = Engine::new().with_time_limit(f64::NAN);
engine.schedule(100.0, "x").expect("usable");
assert_eq!(engine.run(|_, _| {}).executed, 1);
}
#[test]
fn an_event_limit_of_zero_is_ignored() {
let mut engine = Engine::new().with_event_limit(0);
engine.schedule(1.0, "x").expect("usable");
assert_eq!(engine.run(|_, _| {}).executed, 1);
}
#[test]
fn a_negative_time_cannot_be_scheduled() {
let mut engine = Engine::new();
assert_eq!(engine.schedule(-1.0, "x"), None);
assert_eq!(engine.schedule(f64::NAN, "x"), None);
assert!(engine.is_idle());
}
#[test]
fn the_stop_reasons_report_their_names() {
assert!(StopReason::QueueEmpty.is_natural());
assert!(!StopReason::TimeLimit.is_natural());
assert!(!StopReason::Requested.is_natural());
assert!(!StopReason::EventLimit.is_natural());
assert_eq!(StopReason::QueueEmpty.name(), "queue empty");
assert_eq!(StopReason::TimeLimit.name(), "time limit");
assert_eq!(StopReason::Requested.name(), "requested");
assert_eq!(StopReason::EventLimit.name(), "event limit");
assert_eq!(StopReason::TimeLimit.to_string(), "time limit");
}
#[test]
fn the_outcome_displays_its_shape() {
let mut engine = Engine::new();
engine.schedule(5.0, "x").expect("usable");
let outcome = engine.run(|_, _| {});
let text = outcome.to_string();
assert!(text.contains("1 event"), "{text}");
assert!(text.contains("queue empty"), "{text}");
}
#[test]
fn the_engine_default_is_empty() {
let engine: Engine<u8> = Engine::default();
assert!(engine.is_idle());
assert_eq!(engine.now(), 0.0);
}