diff --git a/Cargo.lock b/Cargo.lock index 3a1877c..e7d4066 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -86,6 +86,9 @@ version = "0.1.0" [[package]] name = "apsis-event" version = "0.1.0" +dependencies = [ + "apsis-core", +] [[package]] name = "apsis-frames" diff --git a/crates/sim/apsis-event/Cargo.toml b/crates/sim/apsis-event/Cargo.toml index b1420ab..3bc217e 100644 --- a/crates/sim/apsis-event/Cargo.toml +++ b/crates/sim/apsis-event/Cargo.toml @@ -11,3 +11,4 @@ publish = false workspace = true [dependencies] +apsis-core.workspace = true diff --git a/crates/sim/apsis-event/src/event.rs b/crates/sim/apsis-event/src/event.rs new file mode 100644 index 0000000..60c32ad --- /dev/null +++ b/crates/sim/apsis-event/src/event.rs @@ -0,0 +1,111 @@ +//! The event type and its identifier. + +use core::fmt; + +/// An identifier for a scheduled event. +/// +/// The identifier combines the time at which the event is due with the order +/// in which it was scheduled. Two events that fall at the same instant are +/// therefore distinguishable, which is what makes the delivery order +/// repeatable. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct EventId { + /// The scheduled time, in seconds from the start of the simulation. + /// + /// The value is stored as its bit pattern so that the derived ordering on + /// the identifier matches the ordering on time for non-negative values, + /// which is the only range a simulation uses. + time_bits: u64, + /// The order in which the event was scheduled. + sequence: u64, +} + +impl EventId { + /// Creates an identifier from a time and a sequence number. + /// + /// The time must be finite and non-negative. A caller that has a time + /// outside that range cannot use this type, which is deliberate: a + /// simulation clock that goes backwards or becomes infinite has no + /// meaningful order. + pub fn new(time: f64, sequence: u64) -> Option { + if !time.is_finite() || time < 0.0 { + return None; + } + Some(Self { + time_bits: time.to_bits(), + sequence, + }) + } + + /// Creates an identifier from a time already validated as usable. + /// + /// This is the crate internal constructor used by the queue, which has + /// already checked the time before it stored the entry. + pub(crate) fn from_valid(time: f64, sequence: u64) -> Self { + Self { + time_bits: time.to_bits(), + sequence, + } + } + + /// Returns the scheduled time, in seconds. + pub fn time(self) -> f64 { + f64::from_bits(self.time_bits) + } + + /// Returns the order in which the event was scheduled. + pub const fn sequence(self) -> u64 { + self.sequence + } +} + +impl fmt::Display for EventId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "event {} at {} s", self.sequence, self.time()) + } +} + +/// One scheduled event. +/// +/// The payload is the caller's: it names what should happen. The queue does +/// not interpret it, so the same queue serves any model. +#[derive(Debug, Clone, PartialEq)] +pub struct Event { + /// The identifier of this event. + pub id: EventId, + /// The scheduled time, in seconds from the start of the simulation. + pub time: f64, + /// The payload describing what should happen. + pub payload: T, +} + +impl Event { + /// Returns `true` when the event is due at or before a time. + pub fn is_due(&self, now: f64) -> bool { + self.time <= now + } + + /// Returns a copy of this event with a different payload. + pub fn with_payload(self, payload: U) -> Event { + Event { + id: self.id, + time: self.time, + payload, + } + } + + /// Returns a copy of this event scheduled at a different time. + /// + /// Returns `None` when the new time is not usable, which leaves the + /// original event available for the caller to keep using. + pub fn reschedule(&self, time: f64, sequence: u64) -> Option> + where + T: Clone, + { + Some(Event { + id: EventId::new(time, sequence)?, + time, + payload: self.payload.clone(), + }) + } +} diff --git a/crates/sim/apsis-event/src/lib.rs b/crates/sim/apsis-event/src/lib.rs index 7b44f0d..8324e8b 100644 --- a/crates/sim/apsis-event/src/lib.rs +++ b/crates/sim/apsis-event/src/lib.rs @@ -1 +1,40 @@ //! Simulation event types and event queue. +//! +//! A discrete-event simulation advances by jumping from one event to the +//! next rather than by stepping through every instant. This crate provides +//! the queue that makes that possible: a priority queue ordered by time, +//! with a well defined order among events that fall at the same instant. +//! +//! The tie-break matters more than it first appears. Two events scheduled +//! for the same instant must be delivered in a repeatable order, or the +//! simulation produces a different result on each run and a cross-check +//! against another implementation is impossible. Each event therefore +//! carries a sequence number assigned when it is scheduled, and events at +//! equal times are ordered by that number, so the order is the order in +//! which they were scheduled. That is the convention used by every +//! discrete-event simulator. +//! +//! Times are in seconds from the start of the simulation, so the queue holds +//! no knowledge of calendars or time scales. +//! +//! ``` +//! use apsis_event::EventQueue; +//! +//! let mut queue = EventQueue::new(); +//! queue.schedule(10.0, "second"); +//! queue.schedule(5.0, "first"); +//! +//! // The earliest event comes out first. +//! let event = queue.pop().expect("the queue is not empty"); +//! assert_eq!(event.time, 5.0); +//! assert_eq!(event.payload, "first"); +//! ``` + +#![forbid(unsafe_code)] +#![deny(missing_docs)] + +mod event; +mod queue; + +pub use event::{Event, EventId}; +pub use queue::{EventQueue, QueueState}; diff --git a/crates/sim/apsis-event/src/queue.rs b/crates/sim/apsis-event/src/queue.rs new file mode 100644 index 0000000..99fa958 --- /dev/null +++ b/crates/sim/apsis-event/src/queue.rs @@ -0,0 +1,226 @@ +//! The event queue. + +use crate::event::{Event, EventId}; +use std::cmp::Ordering; +use std::collections::BinaryHeap; + +/// A heap entry ordered so that the earliest event is the greatest. +/// +/// The standard library heap is a maximum heap, so the comparison is +/// reversed on time. Ties on time are broken by the sequence number, and +/// that comparison is also reversed so that the earlier sequence is the +/// greater element. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Scheduled { + time_bits: u64, + sequence: u64, +} + +impl Scheduled { + fn time(self) -> f64 { + f64::from_bits(self.time_bits) + } +} + +impl Ord for Scheduled { + fn cmp(&self, other: &Self) -> Ordering { + // Reverse the time so that the smallest time is the greatest heap + // element. For equal times, the smaller sequence is the greater + // element, so an earlier scheduled event is delivered first. + other + .time_bits + .cmp(&self.time_bits) + .then_with(|| other.sequence.cmp(&self.sequence)) + } +} + +impl PartialOrd for Scheduled { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +/// What the queue currently holds. +/// +/// The state is exposed so that a simulation loop can decide whether to +/// advance the clock or to stop without popping an event first. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct QueueState { + /// The number of pending events. + pub pending: usize, + /// The time of the earliest pending event, if there is one. + pub next_time: Option, + /// The number of events delivered so far. + pub delivered: u64, +} + +impl QueueState { + /// Returns `true` when no events are pending. + pub const fn is_idle(&self) -> bool { + self.pending == 0 + } +} + +/// A priority queue of events ordered by time and then by schedule order. +/// +/// The queue assigns the sequence numbers, so a caller cannot accidentally +/// create two events with the same identifier. Sequence numbers start at +/// zero and increase by one with every event scheduled, including events +/// scheduled from inside the handling of another event. +#[derive(Debug, Clone)] +pub struct EventQueue { + pending: BinaryHeap, + payloads: Vec>, + next_sequence: u64, + delivered: u64, +} + +impl Default for EventQueue { + fn default() -> Self { + Self::new() + } +} + +impl EventQueue { + /// Creates an empty queue. + pub const fn new() -> Self { + Self { + pending: BinaryHeap::new(), + payloads: Vec::new(), + next_sequence: 0, + delivered: 0, + } + } + + /// Returns the number of pending events. + pub fn len(&self) -> usize { + self.pending.len() + } + + /// Returns `true` when no events are pending. + pub fn is_empty(&self) -> bool { + self.pending.is_empty() + } + + /// Returns the number of events delivered so far. + pub const fn delivered(&self) -> u64 { + self.delivered + } + + /// Returns what the queue currently holds. + pub fn state(&self) -> QueueState { + QueueState { + pending: self.pending.len(), + next_time: self.pending.peek().map(|entry| entry.time()), + delivered: self.delivered, + } + } + + /// Returns the time of the earliest pending event. + /// + /// Returns `None` when the queue is empty. + pub fn next_time(&self) -> Option { + self.pending.peek().map(|entry| entry.time()) + } + + /// Schedules an event at a time. + /// + /// The sequence number is assigned by the queue, so the delivery order + /// among events at the same instant is the order they were scheduled. + /// + /// Returns the identifier of the new event, or `None` when the time is + /// not usable: a simulation clock runs forward from zero, so a negative + /// or non-finite time is refused rather than stored. + pub fn schedule(&mut self, time: f64, payload: T) -> Option { + let sequence = self.next_sequence; + let id = EventId::new(time, sequence)?; + self.next_sequence = self.next_sequence.wrapping_add(1); + self.pending.push(Scheduled { + time_bits: time.to_bits(), + sequence, + }); + // The payload slots are indexed by sequence, so gaps are filled with + // `None` when a caller has popped an event without taking its + // payload. + let index = sequence as usize; + if index < self.payloads.len() { + self.payloads[index] = Some(payload); + } else { + while self.payloads.len() < index { + self.payloads.push(None); + } + self.payloads.push(Some(payload)); + } + Some(id) + } + + /// Schedules an event at a time relative to a base instant. + /// + /// This is the form a handler uses when it schedules a follow-up: the + /// delay is relative to the current simulation time. + /// + /// Returns `None` when the resulting time is not usable. + pub fn schedule_after(&mut self, now: f64, delay: f64, payload: T) -> Option { + if !delay.is_finite() || delay < 0.0 { + return None; + } + self.schedule(now + delay, payload) + } + + /// Removes and returns the earliest event. + /// + /// Returns `None` when the queue is empty. The payload is taken from the + /// queue, so the slot is left empty; scheduling a later event reuses the + /// slot. + pub fn pop(&mut self) -> Option> { + let entry = self.pending.pop()?; + let index = entry.sequence as usize; + let payload = self + .payloads + .get_mut(index) + .and_then(Option::take) + .expect("a pending entry always has its payload"); + self.delivered = self.delivered.wrapping_add(1); + Some(Event { + id: EventId::from_valid(entry.time(), entry.sequence), + time: entry.time(), + payload, + }) + } + + /// Returns a reference to the payload of the earliest event without + /// removing it. + pub fn peek(&self) -> Option<&T> { + let entry = self.pending.peek()?; + self.payloads + .get(entry.sequence as usize) + .and_then(Option::as_ref) + } + + /// Removes every pending event and returns how many were discarded. + /// + /// The delivered count is left alone, because it records history rather + /// than queue contents. + pub fn clear(&mut self) -> usize { + let count = self.pending.len(); + self.pending.clear(); + for slot in &mut self.payloads { + *slot = None; + } + count + } + + /// Returns every pending event in delivery order, leaving the queue + /// empty. + /// + /// The order is the order the events would have been delivered in, which + /// is what a caller wants when it records the remaining schedule at the + /// end of a run. + pub fn drain(&mut self) -> Vec> { + let mut events = Vec::with_capacity(self.pending.len()); + while let Some(event) = self.pop() { + events.push(event); + } + events + } +} diff --git a/crates/sim/apsis-event/tests/queue.rs b/crates/sim/apsis-event/tests/queue.rs new file mode 100644 index 0000000..0b75afd --- /dev/null +++ b/crates/sim/apsis-event/tests/queue.rs @@ -0,0 +1,279 @@ +//! Tests for the event queue. + +use apsis_event::{Event, EventId, EventQueue}; + +#[test] +fn an_empty_queue_reports_itself_idle() { + let queue: EventQueue = EventQueue::new(); + assert!(queue.is_empty()); + assert_eq!(queue.len(), 0); + assert_eq!(queue.next_time(), None); + assert_eq!(queue.peek(), None); + assert_eq!(queue.delivered(), 0); + assert!(queue.state().is_idle()); + assert_eq!(queue.state().next_time, None); +} + +#[test] +fn events_are_delivered_in_time_order() { + let mut queue = EventQueue::new(); + queue.schedule(30.0, "third"); + queue.schedule(10.0, "first"); + queue.schedule(20.0, "second"); + + assert_eq!(queue.len(), 3); + assert_eq!(queue.next_time(), Some(10.0)); + assert_eq!(queue.peek(), Some(&"first")); + + let mut order = Vec::new(); + while let Some(event) = queue.pop() { + order.push(event.payload); + } + assert_eq!(order, vec!["first", "second", "third"]); + assert!(queue.is_empty()); + assert_eq!(queue.delivered(), 3); +} + +#[test] +fn events_at_the_same_time_are_delivered_in_schedule_order() { + // The tie-break is what makes a run reproducible. + let mut queue = EventQueue::new(); + queue.schedule(10.0, 'a'); + queue.schedule(10.0, 'b'); + queue.schedule(10.0, 'c'); + + let order: Vec = queue + .drain() + .into_iter() + .map(|event| event.payload) + .collect(); + assert_eq!(order, vec!['a', 'b', 'c']); +} + +#[test] +fn the_tie_break_holds_across_a_heap_rebuild() { + // Scheduling in a different order must still deliver in schedule order, + // because the sequence number, not the insertion position, decides. + let mut queue = EventQueue::new(); + let mut ids = Vec::new(); + for index in 0..20 { + ids.push(queue.schedule(5.0, index).expect("the time is usable")); + } + let delivered: Vec = queue + .drain() + .into_iter() + .map(|event| event.payload) + .collect(); + assert_eq!(delivered, (0..20).collect::>()); + + // The identifiers are also in schedule order. + for (position, id) in ids.iter().enumerate() { + assert_eq!(id.sequence(), position as u64); + } +} + +#[test] +fn an_event_carries_its_identifier_and_time() { + let mut queue = EventQueue::new(); + let id = queue.schedule(12.5, "payload").expect("the time is usable"); + assert_eq!(id.time(), 12.5); + assert_eq!(id.sequence(), 0); + + let event = queue.pop().expect("the queue is not empty"); + assert_eq!(event.id, id); + assert_eq!(event.time, 12.5); + assert_eq!(event.payload, "payload"); +} + +#[test] +fn scheduling_a_negative_or_non_finite_time_is_refused() { + let mut queue = EventQueue::new(); + for time in [-1.0, f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { + assert_eq!(queue.schedule(time, "x"), None, "time {time}"); + } + // The queue is unchanged and the sequence number was not consumed. + assert!(queue.is_empty()); + let id = queue.schedule(0.0, "zero").expect("zero is usable"); + assert_eq!(id.sequence(), 0); +} + +#[test] +fn a_follow_up_event_is_scheduled_relative_to_now() { + let mut queue = EventQueue::new(); + queue + .schedule_after(100.0, 30.0, "follow up") + .expect("the delay is usable"); + let event = queue.pop().expect("the queue is not empty"); + assert_eq!(event.time, 130.0); +} + +#[test] +fn a_follow_up_with_a_bad_delay_is_refused() { + let mut queue = EventQueue::new(); + assert_eq!(queue.schedule_after(0.0, -1.0, "x"), None); + assert_eq!(queue.schedule_after(0.0, f64::NAN, "x"), None); + assert_eq!(queue.schedule_after(0.0, f64::INFINITY, "x"), None); + assert!(queue.is_empty()); + // A zero delay schedules at the current instant. + let id = queue.schedule_after(5.0, 0.0, "now").expect("usable"); + assert_eq!(id.time(), 5.0); +} + +#[test] +fn events_can_be_scheduled_from_a_handler() { + // A handler that schedules its successor, which is the pattern a + // discrete-event simulation uses for a periodic process. + let mut queue: EventQueue = EventQueue::new(); + queue.schedule(0.0, 0).expect("usable"); + let mut ticks = 0; + while let Some(event) = queue.pop() { + ticks += 1; + if event.payload < 3 { + queue + .schedule_after(event.time, 10.0, event.payload + 1) + .expect("the delay is usable"); + } + } + assert_eq!(ticks, 4); + assert_eq!(queue.delivered(), 4); +} + +#[test] +fn clearing_discards_the_pending_events() { + let mut queue = EventQueue::new(); + for index in 0..5 { + queue.schedule(f64::from(index), index).expect("usable"); + } + queue.pop().expect("the queue is not empty"); + assert_eq!(queue.len(), 4); + assert_eq!(queue.delivered(), 1); + + assert_eq!(queue.clear(), 4); + assert!(queue.is_empty()); + // The delivered count records history and is not reset. + assert_eq!(queue.delivered(), 1); + assert_eq!(queue.pop(), None); +} + +#[test] +fn draining_returns_the_events_in_delivery_order() { + let mut queue = EventQueue::new(); + queue.schedule(3.0, 'c'); + queue.schedule(1.0, 'a'); + queue.schedule(2.0, 'b'); + + let events = queue.drain(); + let payloads: Vec = events.iter().map(|event| event.payload).collect(); + assert_eq!(payloads, vec!['a', 'b', 'c']); + assert!(queue.is_empty()); + assert_eq!(queue.delivered(), 3); +} + +#[test] +fn the_queue_state_reports_what_is_pending() { + let mut queue = EventQueue::new(); + queue.schedule(10.0, "a").expect("usable"); + queue.schedule(20.0, "b").expect("usable"); + let state = queue.state(); + assert_eq!(state.pending, 2); + assert_eq!(state.next_time, Some(10.0)); + assert_eq!(state.delivered, 0); + assert!(!state.is_idle()); + + queue.pop(); + let state = queue.state(); + assert_eq!(state.pending, 1); + assert_eq!(state.next_time, Some(20.0)); + assert_eq!(state.delivered, 1); +} + +#[test] +fn a_payload_slot_is_reused_after_a_pop() { + // The queue indexes payloads by sequence, so a popped slot must be + // reusable without disturbing a later event. + let mut queue = EventQueue::new(); + queue.schedule(1.0, "first").expect("usable"); + queue.schedule(2.0, "second").expect("usable"); + let first = queue.pop().expect("non-empty"); + assert_eq!(first.payload, "first"); + let second = queue.pop().expect("non-empty"); + assert_eq!(second.payload, "second"); + assert_eq!(queue.pop(), None); +} + +#[test] +fn the_identifier_orders_by_time_then_sequence() { + // The derived ordering on the identifier is the delivery order, which is + // what lets a caller sort identifiers to reproduce a run. + let earlier = EventId::new(1.0, 5).expect("usable"); + let later_time = EventId::new(2.0, 0).expect("usable"); + let later_sequence = EventId::new(1.0, 6).expect("usable"); + assert!(earlier < later_time); + assert!(earlier < later_sequence); + assert!(EventId::new(1.0, 0).expect("usable") < earlier); +} + +#[test] +fn an_identifier_reports_itself() { + let id = EventId::new(3.5, 7).expect("usable"); + assert_eq!(id.time(), 3.5); + assert_eq!(id.sequence(), 7); + assert!(id.to_string().contains("7")); + assert!(id.to_string().contains("3.5")); +} + +#[test] +fn an_event_reports_whether_it_is_due() { + let event = Event { + id: EventId::new(5.0, 0).expect("usable"), + time: 5.0, + payload: "x", + }; + assert!(!event.is_due(4.999)); + assert!(event.is_due(5.0)); + assert!(event.is_due(6.0)); +} + +#[test] +fn an_event_can_be_rescheduled() { + let event = Event { + id: EventId::new(5.0, 0).expect("usable"), + time: 5.0, + payload: "x", + }; + let moved = event.reschedule(9.0, 1).expect("the new time is usable"); + assert_eq!(moved.time, 9.0); + assert_eq!(moved.id.sequence(), 1); + assert_eq!(moved.payload, "x"); + // The original is unchanged. + assert_eq!(event.time, 5.0); + + assert!(event.reschedule(-1.0, 2).is_none()); + assert!(event.reschedule(f64::NAN, 3).is_none()); +} + +#[test] +fn an_event_can_change_its_payload() { + let event = Event { + id: EventId::new(5.0, 0).expect("usable"), + time: 5.0, + payload: 1_u32, + }; + let changed = event.with_payload("text"); + assert_eq!(changed.payload, "text"); + assert_eq!(changed.time, 5.0); +} + +#[test] +fn a_zero_time_event_is_the_earliest_possible() { + let mut queue = EventQueue::new(); + queue.schedule(1.0, "later").expect("usable"); + queue.schedule(0.0, "first").expect("usable"); + assert_eq!(queue.pop().expect("non-empty").payload, "first"); +} + +#[test] +fn the_queue_default_is_empty() { + let queue: EventQueue = EventQueue::default(); + assert!(queue.is_empty()); +}