From 67e845b35dfc7a32a50db9637433967e0945f1ca Mon Sep 17 00:00:00 2001 From: Robert Jacobson Date: Wed, 24 Jun 2026 00:08:12 -0500 Subject: [PATCH 1/4] feat: Shutdown semantics and shutdown-time plans --- src/context.rs | 415 +++++++++++++++++++++++++++++++++++---- src/lib.rs | 9 +- src/plan.rs | 388 ------------------------------------ src/plan_queue.rs | 489 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 873 insertions(+), 428 deletions(-) delete mode 100644 src/plan.rs create mode 100644 src/plan_queue.rs diff --git a/src/context.rs b/src/context.rs index 53431806..d7073814 100644 --- a/src/context.rs +++ b/src/context.rs @@ -19,7 +19,7 @@ use crate::execution_stats::{ ExecutionStatistics, }; use crate::global_properties::get_global_property_count; -use crate::plan::{PlanId, Queue}; +use crate::plan_queue::{PlanId, PlanQueue}; use crate::{get_data_plugin_count, trace, warn, HashMap, HashMapExt}; /// The common callback used by multiple [`Context`] methods for future events @@ -53,6 +53,36 @@ impl Display for ExecutionPhase { } } +/// Tracks event-loop shutdown state and the current shutdown lifecycle phase. +/// +/// This is private implementation state, not public API. `Context::shutdown` +/// requests normal shutdown and `Context::abort` requests an immediate stop of +/// the current `execute` loop. The stopped status is deliberately cleared when a +/// later `execute` call begins. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ShutdownStatus { + /// Normal execution; no shutdown has been requested. + None, + /// Normal shutdown requested or in progress. + /// + /// In this state, callbacks still run first, but regular plans are executed + /// only if they are scheduled at `Context::current_time`. Simulation time is + /// not advanced. + Normal, + /// Drain the distinguished shutdown-time plan queue. + /// + /// Once this state is reached, regular plans are not inspected again during + /// the same execution pass, even if shutdown-time work schedules a regular + /// plan at the current simulation time. Callbacks are still executed. + ShutdownTimePlans, + /// Stop the current `execute` event loop. + /// + /// This is set by `Context::abort` and when the shutdown-time queue is + /// exhausted. Manual `execute_single_step` calls clear this state when there + /// is no callback to run. + Stopped, +} + /// A manager for the state of a discrete-event simulation /// /// Provides core simulation services including @@ -81,7 +111,7 @@ impl Display for ExecutionPhase { /// occurred and have other modules take turns reacting to these occurrences. /// pub struct Context { - plan_queue: Queue, ExecutionPhase>, + plan_queue: PlanQueue, callback_queue: VecDeque>, event_handlers: HashMap>, pub(crate) entity_store: EntityStore, @@ -89,7 +119,7 @@ pub struct Context { pub(crate) global_properties: Vec>>, current_time: Option, start_time: Option, - shutdown_requested: bool, + shutdown_status: ShutdownStatus, execution_profiler: ExecutionProfilingCollector, pub(crate) print_execution_statistics: bool, } @@ -109,7 +139,7 @@ impl Context { .collect(); Context { - plan_queue: Queue::new(), + plan_queue: PlanQueue::new(), callback_queue: VecDeque::new(), event_handlers: HashMap::new(), entity_store: EntityStore::new(), @@ -117,7 +147,7 @@ impl Context { global_properties, current_time: None, start_time: None, - shutdown_requested: false, + shutdown_status: ShutdownStatus::None, execution_profiler: ExecutionProfilingCollector::new(), print_execution_statistics: false, } @@ -217,6 +247,33 @@ impl Context { self.plan_queue.add_plan(time, Box::new(callback), phase) } + /// Add a plan to execute during shutdown-time in the normal phase. + /// + /// Shutdown-time plans execute after regular plans at the current simulation + /// time are exhausted during normal shutdown, and after natural exhaustion of + /// the regular plan queue. + /// + /// Returns a [`PlanId`] for the newly-added plan that can be used to cancel it + /// if needed. + pub fn add_shutdown_plan(&mut self, callback: impl FnOnce(&mut Context) + 'static) -> PlanId { + self.add_shutdown_plan_with_phase(callback, ExecutionPhase::Normal) + } + + /// Add a plan to execute during shutdown-time with the specified phase. + /// + /// Shutdown-time plans have no simulation time. They are ordered by phase and + /// insertion order. + /// + /// Returns a [`PlanId`] for the newly-added plan that can be used to cancel it + /// if needed. + pub fn add_shutdown_plan_with_phase( + &mut self, + callback: impl FnOnce(&mut Context) + 'static, + phase: ExecutionPhase, + ) -> PlanId { + self.plan_queue.add_shutdown_plan(Box::new(callback), phase) + } + fn evaluate_periodic_and_schedule_next( &mut self, period: f64, @@ -346,11 +403,26 @@ impl Context { .expect("TypeID does not match data plugin type. You must use the `define_data_plugin!` macro to create a data plugin.") } - /// Shutdown the simulation cleanly, abandoning all events after whatever - /// is currently executing. + /// Request normal shutdown. + /// + /// Normal shutdown stops simulation time from advancing. Execution continues + /// through queued callbacks, regular plans at the current time, and then + /// shutdown-time plans. Calling `shutdown` during shutdown-time execution + /// does not return execution to regular current-time plans. pub fn shutdown(&mut self) { trace!("shutdown context"); - self.shutdown_requested = true; + if self.shutdown_status == ShutdownStatus::None { + self.shutdown_status = ShutdownStatus::Normal; + } + } + + /// Stop the current event loop immediately. + /// + /// Abort only stops the current `execute` loop. The stopped status is cleared + /// when `execute` is called again. + pub fn abort(&mut self) { + trace!("abort context"); + self.shutdown_status = ShutdownStatus::Stopped; } /// Get the current simulation time @@ -414,19 +486,22 @@ impl Context { pub fn execute(&mut self) { trace!("entering event loop"); + if self.shutdown_status == ShutdownStatus::Stopped { + self.shutdown_status = ShutdownStatus::None; + } + if self.current_time.is_none() { self.current_time = Some(self.start_time.unwrap_or(0.0)); } // Start plan loop loop { - if self.shutdown_requested { - self.shutdown_requested = false; + if self.shutdown_status == ShutdownStatus::Stopped { + self.shutdown_status = ShutdownStatus::None; break; - } else { - self.execute_single_step(); } + self.execute_single_step(); self.execution_profiler.refresh(); } @@ -440,25 +515,59 @@ impl Context { } } - /// Executes a single step of the simulation, prioritizing tasks as follows: - /// 1. Callbacks - /// 2. Plans - /// 3. Shutdown + /// Executes a single callback, plan, or shutdown status transition. pub fn execute_single_step(&mut self) { - // If there is a callback, run it. + // Callbacks always have priority over plan selection. This remains true + // even in `Stopped` during manual stepping; `Stopped` only stops the + // `execute` loop, not the ability to explicitly step callbacks manually. if let Some(callback) = self.callback_queue.pop_front() { trace!("calling callback"); callback(self); + return; } - // There aren't any callbacks, so look at the first plan. - else if let Some(plan) = self.plan_queue.get_next_plan() { - trace!("calling plan at {:.6}", plan.time); - self.current_time = Some(plan.time); - (plan.data)(self); - } else { - trace!("No callbacks or plans; exiting event loop"); - // OK, there aren't any plans, so we're done. - self.shutdown_requested = true; + + // No callback is available, so the shutdown status determines which + // plan queue, if any, can provide the next unit of work. + match self.shutdown_status { + ShutdownStatus::None => { + // Normal execution may advance simulation time to the next + // regular plan. If no regular plans remain, natural completion + // transitions into shutdown-time plan execution. + if let Some(plan) = self.plan_queue.pop_next() { + trace!("calling plan at {:.6}", plan.time); + self.current_time = Some(plan.time); + (plan.data)(self); + } else { + self.shutdown_status = ShutdownStatus::ShutdownTimePlans; + } + } + ShutdownStatus::Normal => { + // Normal shutdown drains only regular plans scheduled at the + // current simulation time. Future regular plans must remain in + // the queue so a later `execute` call can run them. + if let Some(plan) = self.plan_queue.pop_next_at(self.get_current_time()) { + trace!("calling plan at {:.6}", plan.time); + (plan.data)(self); + } else { + self.shutdown_status = ShutdownStatus::ShutdownTimePlans; + } + } + ShutdownStatus::ShutdownTimePlans => { + // Once shutdown-time draining begins, do not return to the + // regular plan queue during this execution pass. + if let Some(plan) = self.plan_queue.pop_next_shutdown() { + trace!("calling shutdown-time plan"); + (plan.data)(self); + } else { + self.shutdown_status = ShutdownStatus::Stopped; + } + } + ShutdownStatus::Stopped => { + // `execute` exits before calling `execute_single_step` in this + // state. This arm supports manual single-step use after a prior + // stop by consuming the stopped status when no callback exists. + self.shutdown_status = ShutdownStatus::None; + } } } @@ -485,6 +594,12 @@ pub trait ContextBase: Sized { callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase, ) -> PlanId; + fn add_shutdown_plan(&mut self, callback: impl FnOnce(&mut Context) + 'static) -> PlanId; + fn add_shutdown_plan_with_phase( + &mut self, + callback: impl FnOnce(&mut Context) + 'static, + phase: ExecutionPhase, + ) -> PlanId; fn add_periodic_plan_with_phase( &mut self, period: f64, @@ -501,6 +616,7 @@ pub trait ContextBase: Sized { fn get_current_time(&self) -> f64; #[must_use] fn get_execution_statistics(&mut self) -> ExecutionStatistics; + fn abort(&mut self); } impl ContextBase for Context { delegate::delegate! { @@ -509,6 +625,8 @@ impl ContextBase for Context { fn emit_event(&mut self, event: E); fn add_plan(&mut self, time: f64, callback: impl FnOnce(&mut Context) + 'static) -> PlanId; fn add_plan_with_phase(&mut self, time: f64, callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase) -> PlanId; + fn add_shutdown_plan(&mut self, callback: impl FnOnce(&mut Context) + 'static) -> PlanId; + fn add_shutdown_plan_with_phase(&mut self, callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase) -> PlanId; fn add_periodic_plan_with_phase(&mut self, period: f64, callback: impl Fn(&mut Context) + 'static, phase: ExecutionPhase); fn cancel_plan(&mut self, plan_id: &PlanId); fn queue_callback(&mut self, callback: impl FnOnce(&mut Context) + 'static); @@ -516,6 +634,7 @@ impl ContextBase for Context { fn get_data(&self, plugin: T) -> &T::DataContainer; fn get_current_time(&self) -> f64; fn get_execution_statistics(&mut self) -> ExecutionStatistics; + fn abort(&mut self); } } } @@ -879,23 +998,29 @@ mod tests { } #[test] - fn shutdown_cancels_plans() { + fn shutdown_runs_current_time_plans_and_preserves_future_plan() { let mut context = Context::new(); add_plan(&mut context, 1.0, 1); - context.add_plan(1.5, Context::shutdown); + context.add_plan(1.5, |context| { + context.get_data_mut(ComponentA).push(2); + context.shutdown(); + }); + add_plan(&mut context, 1.5, 3); add_plan(&mut context, 2.0, 2); context.execute(); assert_eq!(context.get_current_time(), 1.5); - assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + + context.execute(); + assert_eq!(context.get_current_time(), 2.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3, 2]); } #[test] - fn shutdown_cancels_callbacks() { + fn shutdown_runs_queued_callbacks() { let mut context = Context::new(); add_plan(&mut context, 1.0, 1); context.add_plan(1.5, |context| { - // Note that we add the callback *before* we call shutdown - // but shutdown cancels everything. context.queue_callback(|context| { context.get_data_mut(ComponentA).push(3); }); @@ -903,11 +1028,11 @@ mod tests { }); context.execute(); assert_eq!(context.get_current_time(), 1.5); - assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 3]); } #[test] - fn shutdown_cancels_events() { + fn shutdown_runs_queued_events() { let mut context = Context::new(); let obs_data = Rc::new(RefCell::new(0)); let obs_data_clone = Rc::clone(&obs_data); @@ -917,7 +1042,219 @@ mod tests { context.emit_event(Event1 { data: 1 }); context.shutdown(); context.execute(); - assert_eq!(*obs_data.borrow(), 0); + assert_eq!(*obs_data.borrow(), 1); + } + + #[test] + fn shutdown_runs_regular_plans_at_current_time_all_phases() { + let mut context = Context::new(); + context.add_plan_with_phase( + 1.0, + |context| { + context.get_data_mut(ComponentA).push(1); + }, + ExecutionPhase::First, + ); + context.add_plan(1.0, |context| { + context.get_data_mut(ComponentA).push(2); + context.shutdown(); + }); + add_plan(&mut context, 1.0, 3); + add_plan_with_phase(&mut context, 1.0, 4, ExecutionPhase::Last); + add_plan(&mut context, 2.0, 5); + + context.execute(); + + assert_eq!(context.get_current_time(), 1.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3, 4]); + } + + #[test] + fn shutdown_time_plans_run_after_current_time_plans() { + let mut context = Context::new(); + context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(3); + }); + context.add_plan(1.0, |context| { + context.get_data_mut(ComponentA).push(1); + context.shutdown(); + }); + add_plan(&mut context, 1.0, 2); + + context.execute(); + + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + + #[test] + fn shutdown_plan_phase_order_is_respected() { + let mut context = Context::new(); + context.add_shutdown_plan_with_phase( + |context| { + context.get_data_mut(ComponentA).push(3); + }, + ExecutionPhase::Last, + ); + context.add_shutdown_plan_with_phase( + |context| { + context.get_data_mut(ComponentA).push(1); + }, + ExecutionPhase::First, + ); + context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(2); + }); + context.execute(); + + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + + #[test] + fn shutdown_plan_callbacks_are_drained() { + let mut context = Context::new(); + context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(1); + context.queue_callback(|context| { + context.get_data_mut(ComponentA).push(2); + }); + }); + context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(3); + }); + + context.execute(); + + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + + #[test] + fn shutdown_time_does_not_return_to_regular_queue() { + let mut context = Context::new(); + context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(1); + context.add_plan(context.get_current_time(), |context| { + context.get_data_mut(ComponentA).push(3); + }); + }); + context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(2); + }); + + context.execute(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2]); + + context.execute(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + + #[test] + fn cancel_plan_can_cancel_shutdown_plan() { + let mut context = Context::new(); + let to_cancel = context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(1); + }); + context.cancel_plan(&to_cancel); + + context.execute(); + + assert_eq!(*context.get_data_mut(ComponentA), Vec::::new()); + } + + #[test] + fn abort_inside_plan_stops_execute_loop() { + let mut context = Context::new(); + context.add_plan(1.0, |context| { + context.get_data_mut(ComponentA).push(1); + context.queue_callback(|context| { + context.get_data_mut(ComponentA).push(2); + }); + context.abort(); + }); + add_plan(&mut context, 2.0, 3); + + context.execute(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + + context.execute(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + + #[test] + fn abort_before_execute_does_not_poison_later_execute() { + let mut context = Context::new(); + context.abort(); + add_plan(&mut context, 1.0, 1); + + context.execute(); + + assert_eq!(context.get_current_time(), 1.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + } + + #[test] + fn abort_during_normal_shutdown_exits_immediately() { + let mut context = Context::new(); + context.add_plan(1.0, |context| { + context.get_data_mut(ComponentA).push(1); + context.shutdown(); + }); + context.add_plan(1.0, |context| { + context.get_data_mut(ComponentA).push(2); + context.abort(); + }); + add_plan(&mut context, 1.0, 3); + + context.execute(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2]); + + context.execute(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + + #[test] + fn shutdown_does_not_restart_stopped_status() { + let mut context = Context::new(); + context.abort(); + context.shutdown(); + add_plan(&mut context, 1.0, 1); + + context.execute(); + + assert_eq!(context.get_current_time(), 1.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + } + + #[test] + fn execute_single_step_runs_one_status_transition() { + let mut context = Context::new(); + context.add_shutdown_plan(|context| { + context.get_data_mut(ComponentA).push(1); + }); + + context.execute_single_step(); + assert_eq!(*context.get_data_mut(ComponentA), Vec::::new()); + + context.execute_single_step(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + } + + #[test] + fn execute_single_step_stopped_runs_callback_then_resets() { + let mut context = Context::new(); + context.abort(); + context.queue_callback(|context| { + context.get_data_mut(ComponentA).push(1); + }); + add_plan(&mut context, 0.0, 2); + + context.execute_single_step(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + + context.execute_single_step(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + + context.execute_single_step(); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2]); } #[test] @@ -1221,8 +1558,8 @@ mod tests { } #[test] - fn shutdown_requested_reset() { - // This test verifies that shutdown_requested is properly reset after + fn shutdown_status_reset() { + // This test verifies that shutdown_status is properly reset after // being acted upon. This allows the context to be reused after shutdown. let mut context = Context::new(); let _: PersonId = context.add_entity(with!(Person, Age(50))).unwrap(); @@ -1243,14 +1580,14 @@ mod tests { }); // Second execute - should execute the new plan - // If shutdown_requested wasn't reset, this would immediately break + // If shutdown_status wasn't reset, this would immediately break // without executing the plan, leaving population at 1. context.execute(); assert_eq!(context.get_current_time(), 2.0); assert_eq!( context.get_entity_count::(), 2, - "If this fails, shutdown_requested was not properly reset" + "If this fails, shutdown_status was not properly reset" ); } } diff --git a/src/lib.rs b/src/lib.rs index 8df6961d..a3c0a232 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -69,7 +69,14 @@ pub use network::{ContextNetworkExt, Edge, EdgeType}; pub mod macros; -pub mod plan; +pub mod plan_queue; +/// Compatibility re-exports for plan-related public API. +/// +/// The plan queue implementation lives in [`crate::plan_queue`]. This module is +/// retained so existing `ixa::plan::PlanId` imports continue to work. +pub mod plan { + pub use crate::plan_queue::PlanId; +} pub mod random; pub use random::{ContextRandomExt, RngId}; diff --git a/src/plan.rs b/src/plan.rs deleted file mode 100644 index a0cf47de..00000000 --- a/src/plan.rs +++ /dev/null @@ -1,388 +0,0 @@ -//! A priority queue that stores arbitrary data sorted by time and priority -//! -//! Defines a [`Queue`]`` that is intended to store a queue of items of type -//! `T` - sorted by `f64` time and definable priority `P` - called 'plans'. -//! This queue has methods for adding plans, cancelling plans, and retrieving -//! the earliest plan in the queue. Adding a plan is *O*(log(*n*)) while -//! cancellation and retrieval are *O*(1). -//! -//! This queue is used by [`Context`](crate::Context) to store future events where some callback -//! closure `FnOnce(&mut Context)` will be executed at a given point in time. - -use std::cmp::Ordering; -use std::collections::BinaryHeap; - -use crate::{trace, HashMap, HashMapExt}; - -/// A priority queue that stores arbitrary data sorted by time -/// -/// Items of type `T` are stored in order by `f64` time and called [`Plan`]``. -/// Plans can have priorities given by some specified orderable type `P`. -/// When plans are created they are sequentially assigned a [`PlanId`] that is a -/// wrapped `u64`. If two plans are scheduled for the same time then the plan -/// with the lowest priority is placed earlier. If two plans have the same time -/// and priority then the plan that is scheduled first (i.e., that has the -/// lowest id) is placed earlier. -/// -/// The time, plan id, and priority are stored in a binary heap of [`PlanSchedule`]`

` -/// objects. The data payload of the event is stored in a hash map by plan id. -/// Plan cancellation occurs by removing the corresponding entry from the data -/// hash map. -pub struct Queue { - queue: BinaryHeap>, - data_map: HashMap, - /// The number of plans that have been added; equivalently, the next plan ID that - /// will be issued. - plan_counter: u64, - /// Tracks the high water mark of plans in flight (scheduled but not yet executed). - /// This is the max of `self.queue.len()`, not of `self.data_map.len()`. - #[cfg(feature = "profiling")] - pub(crate) max_plans_in_flight: u64, - #[cfg(feature = "profiling")] - pub(crate) max_memory_in_use: u64, -} - -impl Queue { - /// Create a new empty `Queue` - #[must_use] - pub fn new() -> Queue { - Queue { - queue: BinaryHeap::new(), - data_map: HashMap::new(), - plan_counter: 0, - #[cfg(feature = "profiling")] - max_plans_in_flight: 0, - #[cfg(feature = "profiling")] - max_memory_in_use: 0, - } - } - - /// Add a plan to the queue at the specified time - /// - /// Returns a [`PlanId`] for the newly-added plan that can be used to cancel it - /// if needed. - pub fn add_plan(&mut self, time: f64, data: T, priority: P) -> PlanId { - trace!("adding plan at {time}"); - // Add plan to queue, store data, and increment counter - let plan_id = self.plan_counter; - self.queue.push(PlanSchedule { - plan_id, - time, - priority, - }); - self.data_map.insert(plan_id, data); - self.plan_counter += 1; - #[cfg(feature = "profiling")] - { - self.max_plans_in_flight = self.max_plans_in_flight.max(self.queue.len() as u64); - self.max_memory_in_use = self - .max_memory_in_use - .max(self.estimated_memory_in_use() as u64); - } - - PlanId(plan_id) - } - - /// Cancel a plan that has been added to the queue - pub fn cancel_plan(&mut self, plan_id: &PlanId) -> Option { - trace!("cancel plan {plan_id:?}"); - // Delete the plan from the map, but leave in the queue - // It will be skipped when the plan is popped from the queue - self.data_map.remove(&plan_id.0) - } - - #[must_use] - pub fn is_empty(&self) -> bool { - self.queue.is_empty() - } - - #[must_use] - pub fn next_time(&self) -> Option { - self.queue.peek().map(|e| e.time) - } - - #[allow(dead_code)] - pub(crate) fn clear(&mut self) { - self.data_map.clear(); - self.queue.clear(); - self.plan_counter = 0; - } - - #[must_use] - #[allow(dead_code)] - pub(crate) fn peek(&self) -> Option<(&PlanSchedule

, &T)> { - // Iterate over queue until we find a plan with data or queue is empty - for entry in &self.queue { - // Skip plans that have been cancelled and thus have no data - if let Some(data) = self.data_map.get(&entry.plan_id) { - return Some((entry, data)); - } - } - None - } - - /// Retrieve the earliest plan in the queue - /// - /// Returns the next plan if it exists or else `None` if the queue is empty - pub fn get_next_plan(&mut self) -> Option> { - trace!("getting next plan"); - loop { - // Pop from queue until we find a plan with data or queue is empty - match self.queue.pop() { - Some(entry) => { - // Skip plans that have been cancelled and thus have no data - if let Some(data) = self.data_map.remove(&entry.plan_id) { - return Some(Plan { - time: entry.time, - data, - }); - } - } - None => { - return None; - } - } - } - } - - /// Returns a list of length `at_most`, or unbounded if `at_most=0`, of active scheduled - /// [`PlanSchedule`]s ordered as they are in the queue itself. - #[must_use] - pub fn list_schedules(&self, at_most: usize) -> Vec<&PlanSchedule

> { - let mut items = vec![]; - - // Iterate over queue until we find a plan with data or queue is empty - for entry in &self.queue { - // Skip plans that have been cancelled and thus have no data - if self.data_map.contains_key(&entry.plan_id) { - items.push(entry); - if items.len() == at_most { - break; - } - } - } - items - } - - #[doc(hidden)] - pub(crate) fn remaining_plan_count(&self) -> usize { - self.queue.len() - } - - #[cfg(feature = "profiling")] - fn estimated_memory_in_use(&self) -> usize { - let queue_bytes = self.queue.capacity() * size_of::>(); - - let map_entry_bytes = self.data_map.capacity() * size_of::<(u64, T)>(); - - queue_bytes + map_entry_bytes - } -} - -impl Default for Queue { - fn default() -> Self { - Self::new() - } -} - -/// A time, id, and priority object used to order plans in the [`Queue`]`` -/// -/// [`PlanSchedule`] objects are sorted in increasing order of time, priority and then -/// plan id -#[derive(PartialEq, Debug)] -pub struct PlanSchedule { - pub plan_id: u64, - pub time: f64, - pub priority: P, -} - -#[allow(clippy::expl_impl_clone_on_copy)] // Clippy false positive -impl Clone for PlanSchedule

{ - fn clone(&self) -> Self { - PlanSchedule { - priority: self.priority.clone(), - ..*self - } - } -} - -impl Copy for PlanSchedule

{} - -impl Eq for PlanSchedule

{} - -impl PartialOrd for PlanSchedule

{ - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - -/// Entry objects are ordered in increasing order by time, priority, and then -/// plan id -impl Ord for PlanSchedule

{ - fn cmp(&self, other: &Self) -> Ordering { - let time_ordering = self.time.partial_cmp(&other.time).unwrap().reverse(); - match time_ordering { - // Break time ties in order of priority and then plan id - Ordering::Equal => { - let priority_ordering = self - .priority - .partial_cmp(&other.priority) - .unwrap() - .reverse(); - match priority_ordering { - Ordering::Equal => self.plan_id.cmp(&other.plan_id).reverse(), - _ => priority_ordering, - } - } - _ => time_ordering, - } - } -} - -/// A unique identifier for a plan added to a [`Queue`]`` -#[derive(Clone, Copy, Debug, Hash, Eq, PartialEq)] -pub struct PlanId(pub(crate) u64); - -/// A plan that holds data of type `T` intended to be used at the specified time -pub struct Plan { - pub time: f64, - pub data: T, -} - -#[cfg(test)] -#[allow(clippy::float_cmp)] -mod tests { - use super::Queue; - - #[test] - fn empty_queue() { - let mut plan_queue = Queue::<(), ()>::new(); - assert!(plan_queue.get_next_plan().is_none()); - } - - #[test] - fn add_plans() { - let mut plan_queue = Queue::new(); - plan_queue.add_plan(1.0, 1, ()); - plan_queue.add_plan(3.0, 3, ()); - plan_queue.add_plan(2.0, 2, ()); - assert!(!plan_queue.is_empty()); - - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.0); - assert_eq!(next_plan.data, 1); - - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 2.0); - assert_eq!(next_plan.data, 2); - - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 3.0); - assert_eq!(next_plan.data, 3); - - assert!(plan_queue.is_empty()); - assert!(plan_queue.get_next_plan().is_none()); - } - - #[test] - fn add_plans_at_same_time_with_same_priority() { - let mut plan_queue = Queue::new(); - plan_queue.add_plan(1.0, 1, ()); - plan_queue.add_plan(1.0, 2, ()); - assert!(!plan_queue.is_empty()); - - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.0); - assert_eq!(next_plan.data, 1); - - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.0); - assert_eq!(next_plan.data, 2); - - assert!(plan_queue.is_empty()); - assert!(plan_queue.get_next_plan().is_none()); - } - - #[test] - fn add_plans_at_same_time_with_different_priority() { - let mut plan_queue = Queue::new(); - plan_queue.add_plan(1.0, 1, 1); - plan_queue.add_plan(1.0, 2, 0); - - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.0); - assert_eq!(next_plan.data, 2); - - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.0); - assert_eq!(next_plan.data, 1); - - assert!(plan_queue.is_empty()); - assert!(plan_queue.get_next_plan().is_none()); - } - - #[test] - fn add_and_cancel_plans() { - let mut plan_queue = Queue::new(); - plan_queue.add_plan(1.0, 1, ()); - let plan_to_cancel = plan_queue.add_plan(2.0, 2, ()); - plan_queue.add_plan(3.0, 3, ()); - plan_queue.cancel_plan(&plan_to_cancel); - assert!(!plan_queue.is_empty()); - - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.0); - assert_eq!(next_plan.data, 1); - - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 3.0); - assert_eq!(next_plan.data, 3); - - assert!(plan_queue.is_empty()); - assert!(plan_queue.get_next_plan().is_none()); - } - - #[test] - fn add_and_get_plans() { - let mut plan_queue = Queue::new(); - plan_queue.add_plan(1.0, 1, ()); - plan_queue.add_plan(2.0, 2, ()); - assert!(!plan_queue.is_empty()); - - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.0); - assert_eq!(next_plan.data, 1); - - plan_queue.add_plan(1.5, 3, ()); - - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 1.5); - assert_eq!(next_plan.data, 3); - - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.get_next_plan().unwrap(); - assert_eq!(next_plan.time, 2.0); - assert_eq!(next_plan.data, 2); - - assert!(plan_queue.is_empty()); - assert!(plan_queue.get_next_plan().is_none()); - } - - #[test] - fn cancel_invalid_plan() { - let mut plan_queue = Queue::new(); - let plan_to_cancel = plan_queue.add_plan(1.0, (), ()); - // is_empty just checks for a plan existing, not whether it is valid/has data - assert!(!plan_queue.is_empty()); - plan_queue.get_next_plan(); - assert!(plan_queue.is_empty()); - let result = plan_queue.cancel_plan(&plan_to_cancel); - assert!(result.is_none()); - } -} diff --git a/src/plan_queue.rs b/src/plan_queue.rs new file mode 100644 index 00000000..83152c28 --- /dev/null +++ b/src/plan_queue.rs @@ -0,0 +1,489 @@ +//! A priority queue that stores scheduled simulation plans. +//! +//! Defines [`PlanQueue`], which stores regular time-ordered plans and +//! shutdown-time plans. Both queues share a single [`PlanId`] allocator and +//! cancellation map. + +use std::cmp::Ordering; +use std::collections::BinaryHeap; + +use crate::context::{Context, ExecutionPhase}; +use crate::{trace, HashMap, HashMapExt}; + +type Callback = dyn FnOnce(&mut Context); +type BoxedCallback = Box; + +/// A priority queue that stores scheduled plans. +/// +/// Regular plans are ordered by simulation time, execution phase, and plan ID. +/// Shutdown-time plans are ordered by execution phase and plan ID; their stored +/// time is only an internal constant and has no simulation-time meaning. +pub(crate) struct PlanQueue { + queue: BinaryHeap, + shutdown_queue: BinaryHeap, + data_map: HashMap, + /// The next plan ID that will be issued. + next_plan_id: u64, + /// Tracks the high water mark of plans in flight (scheduled but not yet executed). + /// This is the max of the two heap lengths, not of `self.data_map.len()`. + #[cfg(feature = "profiling")] + pub(crate) max_plans_in_flight: u64, + #[cfg(feature = "profiling")] + pub(crate) max_memory_in_use: u64, +} + +impl PlanQueue { + /// Create a new empty `PlanQueue`. + #[must_use] + pub(crate) fn new() -> PlanQueue { + PlanQueue { + queue: BinaryHeap::new(), + shutdown_queue: BinaryHeap::new(), + data_map: HashMap::new(), + next_plan_id: 0, + #[cfg(feature = "profiling")] + max_plans_in_flight: 0, + #[cfg(feature = "profiling")] + max_memory_in_use: 0, + } + } + + /// Add a regular plan to the queue at the specified time. + /// + /// Returns a [`PlanId`] for the newly-added plan that can be used to cancel it + /// if needed. + pub(crate) fn add_plan( + &mut self, + time: f64, + callback: BoxedCallback, + phase: ExecutionPhase, + ) -> PlanId { + trace!("adding plan at {time}"); + let plan_id = self.next_plan_id; + self.queue.push(PlanSchedule { + plan_id, + time, + phase, + }); + self.data_map.insert(plan_id, callback); + self.next_plan_id += 1; + self.update_profiling_high_water_marks(); + + PlanId(plan_id) + } + + /// Add a shutdown-time plan. + /// + /// Shutdown-time plans have no simulation time. They are ordered by phase and + /// plan ID. + pub(crate) fn add_shutdown_plan( + &mut self, + callback: BoxedCallback, + phase: ExecutionPhase, + ) -> PlanId { + trace!("adding shutdown-time plan"); + let plan_id = self.next_plan_id; + self.shutdown_queue.push(PlanSchedule { + plan_id, + time: 0.0, + phase, + }); + self.data_map.insert(plan_id, callback); + self.next_plan_id += 1; + self.update_profiling_high_water_marks(); + + PlanId(plan_id) + } + + /// Cancel a plan that has been added to either queue. + pub(crate) fn cancel_plan(&mut self, plan_id: &PlanId) -> Option { + trace!("cancel plan {plan_id:?}"); + // Delete the plan from the map, but leave in the heap. It will be skipped + // when its heap entry reaches the root. + self.data_map.remove(&plan_id.0) + } + + #[must_use] + pub(crate) fn is_empty(&mut self) -> bool { + self.next_time().is_none() + } + + #[must_use] + pub(crate) fn next_time(&mut self) -> Option { + Self::discard_canceled_roots(&mut self.queue, &self.data_map); + self.queue.peek().map(|e| e.time) + } + + #[allow(dead_code)] + pub(crate) fn clear(&mut self) { + self.data_map.clear(); + self.queue.clear(); + self.shutdown_queue.clear(); + self.next_plan_id = 0; + } + + /// Retrieve the earliest regular plan in the queue. + /// + /// Returns the next plan if it exists or else `None` if the regular queue is + /// empty. + pub(crate) fn pop_next(&mut self) -> Option { + trace!("getting next plan"); + Self::pop_next_from_heap(&mut self.queue, &mut self.data_map) + } + + /// Retrieve the earliest regular plan only if it is scheduled at `time`. + /// + /// Returns `None` without removing a future plan if the next regular plan is + /// later than `time`. + pub(crate) fn pop_next_at(&mut self, time: f64) -> Option { + Self::discard_canceled_roots(&mut self.queue, &self.data_map); + match self.queue.peek() { + Some(entry) if entry.time == time => { + Self::pop_next_from_heap(&mut self.queue, &mut self.data_map) + } + _ => None, + } + } + + /// Retrieve the next shutdown-time plan. + /// + /// Returns the next shutdown-time plan if it exists or else `None` if the + /// shutdown-time queue is empty. + pub(crate) fn pop_next_shutdown(&mut self) -> Option { + trace!("getting next shutdown-time plan"); + Self::pop_next_from_heap(&mut self.shutdown_queue, &mut self.data_map) + } + + #[doc(hidden)] + pub(crate) fn remaining_plan_count(&self) -> usize { + self.queue.len() + } + + fn discard_canceled_roots( + heap: &mut BinaryHeap, + data_map: &HashMap, + ) { + while heap + .peek() + .is_some_and(|entry| !data_map.contains_key(&entry.plan_id)) + { + heap.pop(); + } + } + + fn pop_next_from_heap( + heap: &mut BinaryHeap, + data_map: &mut HashMap, + ) -> Option { + loop { + { + let entry = heap.pop()?; + if let Some(data) = data_map.remove(&entry.plan_id) { + return Some(Plan { + time: entry.time, + data, + }); + } + } + } + } + + fn update_profiling_high_water_marks(&mut self) { + #[cfg(feature = "profiling")] + { + let plans_in_flight = self.queue.len() + self.shutdown_queue.len(); + self.max_plans_in_flight = self.max_plans_in_flight.max(plans_in_flight as u64); + self.max_memory_in_use = self + .max_memory_in_use + .max(self.estimated_memory_in_use() as u64); + } + } + + #[cfg(feature = "profiling")] + fn estimated_memory_in_use(&self) -> usize { + let queue_bytes = + (self.queue.capacity() + self.shutdown_queue.capacity()) * size_of::(); + + let map_entry_bytes = self.data_map.capacity() * size_of::<(u64, BoxedCallback)>(); + + queue_bytes + map_entry_bytes + } +} + +impl Default for PlanQueue { + fn default() -> Self { + Self::new() + } +} + +/// A time, id, and phase object used to order plans in a [`PlanQueue`]. +/// +/// Regular [`PlanSchedule`] objects are sorted in increasing order of time, +/// phase, and then plan id. Shutdown-time schedules all have the same internal +/// time and are therefore sorted by phase and then plan id. +#[derive(PartialEq, Debug, Clone, Copy)] +pub(crate) struct PlanSchedule { + pub plan_id: u64, + pub time: f64, + pub phase: ExecutionPhase, +} + +impl Eq for PlanSchedule {} + +impl PartialOrd for PlanSchedule { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +/// Entry objects are ordered in increasing order by time, phase, and then plan id. +impl Ord for PlanSchedule { + fn cmp(&self, other: &Self) -> Ordering { + let time_ordering = self.time.partial_cmp(&other.time).unwrap().reverse(); + match time_ordering { + Ordering::Equal => { + let phase_ordering = self.phase.partial_cmp(&other.phase).unwrap().reverse(); + match phase_ordering { + Ordering::Equal => self.plan_id.cmp(&other.plan_id).reverse(), + _ => phase_ordering, + } + } + _ => time_ordering, + } + } +} + +/// A unique identifier for a plan added to a [`PlanQueue`]. +#[derive(Clone, Copy, Debug, Hash, Eq, PartialEq)] +pub struct PlanId(pub(crate) u64); + +/// A plan that holds a callback intended to be executed at the specified time. +pub(crate) struct Plan { + pub time: f64, + pub data: BoxedCallback, +} + +#[cfg(test)] +#[allow(clippy::float_cmp)] +mod tests { + use std::cell::RefCell; + use std::rc::Rc; + + use super::PlanQueue; + use crate::context::{Context, ExecutionPhase}; + + fn callback(value: u32, observed: Rc>>) -> Box { + Box::new(move |_| observed.borrow_mut().push(value)) + } + + fn run_plan(plan: super::Plan, context: &mut Context) { + (plan.data)(context); + } + + #[test] + fn empty_queue() { + let mut plan_queue = PlanQueue::new(); + assert!(plan_queue.pop_next().is_none()); + } + + #[test] + fn add_plans() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + plan_queue.add_plan( + 3.0, + callback(3, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + plan_queue.add_plan( + 2.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + assert!(!plan_queue.is_empty()); + + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 1.0); + run_plan(next_plan, &mut context); + + assert!(!plan_queue.is_empty()); + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 2.0); + run_plan(next_plan, &mut context); + + assert!(!plan_queue.is_empty()); + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 3.0); + run_plan(next_plan, &mut context); + + assert!(plan_queue.pop_next().is_none()); + assert_eq!(*observed.borrow(), vec![1, 2, 3]); + } + + #[test] + fn add_plans_at_same_time_with_same_phase() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + plan_queue.add_plan( + 1.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 1.0); + run_plan(next_plan, &mut context); + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 1.0); + run_plan(next_plan, &mut context); + + assert!(plan_queue.pop_next().is_none()); + assert_eq!(*observed.borrow(), vec![1, 2]); + } + + #[test] + fn add_plans_at_same_time_with_different_phase() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + plan_queue.add_plan( + 1.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::First, + ); + + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 1.0); + run_plan(next_plan, &mut context); + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 1.0); + run_plan(next_plan, &mut context); + + assert!(plan_queue.pop_next().is_none()); + assert_eq!(*observed.borrow(), vec![2, 1]); + } + + #[test] + fn cancel_plan() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + let plan_to_cancel = plan_queue.add_plan( + 2.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + plan_queue.add_plan( + 3.0, + callback(3, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + plan_queue.cancel_plan(&plan_to_cancel); + + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 1.0); + run_plan(next_plan, &mut context); + + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 3.0); + run_plan(next_plan, &mut context); + + assert!(plan_queue.pop_next().is_none()); + assert_eq!(*observed.borrow(), vec![1, 3]); + } + + #[test] + fn next_time_ignores_canceled_root() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut plan_queue = PlanQueue::new(); + let plan_to_cancel = plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + plan_queue.add_plan( + 2.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + + plan_queue.cancel_plan(&plan_to_cancel); + + assert_eq!(plan_queue.next_time(), Some(2.0)); + } + + #[test] + fn pop_next_at_leaves_future_plan_in_queue() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_plan( + 2.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + + assert!(plan_queue.pop_next_at(1.0).is_none()); + + let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(next_plan.time, 2.0); + run_plan(next_plan, &mut context); + assert_eq!(*observed.borrow(), vec![2]); + } + + #[test] + fn shutdown_plans_use_phase_and_fifo_order() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_shutdown_plan(callback(3, Rc::clone(&observed)), ExecutionPhase::Last); + plan_queue.add_shutdown_plan(callback(1, Rc::clone(&observed)), ExecutionPhase::First); + plan_queue.add_shutdown_plan(callback(2, Rc::clone(&observed)), ExecutionPhase::Normal); + plan_queue.add_shutdown_plan(callback(4, Rc::clone(&observed)), ExecutionPhase::Last); + + while let Some(plan) = plan_queue.pop_next_shutdown() { + run_plan(plan, &mut context); + } + + assert_eq!(*observed.borrow(), vec![1, 2, 3, 4]); + } + + #[test] + fn plan_ids_are_shared_between_regular_and_shutdown_queues() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut plan_queue = PlanQueue::new(); + let regular_id = plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + ); + let shutdown_id = + plan_queue.add_shutdown_plan(callback(2, Rc::clone(&observed)), ExecutionPhase::Normal); + + assert_ne!(regular_id, shutdown_id); + assert_eq!(regular_id.0, 0); + assert_eq!(shutdown_id.0, 1); + } +} From a169937291bec100bf679c17fb3d6e9be3a8ae67 Mon Sep 17 00:00:00 2001 From: Robert Jacobson Date: Wed, 24 Jun 2026 12:46:31 -0500 Subject: [PATCH 2/4] fix: Fix AI slop in `PlanQueue` implementation. --- src/plan_queue.rs | 91 ++++++++++++++++++++++++++++------------------- 1 file changed, 54 insertions(+), 37 deletions(-) diff --git a/src/plan_queue.rs b/src/plan_queue.rs index 83152c28..eb8b8b59 100644 --- a/src/plan_queue.rs +++ b/src/plan_queue.rs @@ -108,12 +108,21 @@ impl PlanQueue { self.next_time().is_none() } + /// Return the time the next plan is scheduled for, if there is one. #[must_use] pub(crate) fn next_time(&mut self) -> Option { - Self::discard_canceled_roots(&mut self.queue, &self.data_map); + // First trim any cancelled plans. We want the time of the next legitimate plan. + while self + .queue + .peek() + .is_some_and(|entry| !self.data_map.contains_key(&entry.plan_id)) + { + self.queue.pop(); + } self.queue.peek().map(|e| e.time) } + /// Completely empties the queue, including the plans scheduled at shutdown time. #[allow(dead_code)] pub(crate) fn clear(&mut self) { self.data_map.clear(); @@ -128,7 +137,17 @@ impl PlanQueue { /// empty. pub(crate) fn pop_next(&mut self) -> Option { trace!("getting next plan"); - Self::pop_next_from_heap(&mut self.queue, &mut self.data_map) + loop { + // Return `None` if `pop` fails. + let entry = self.queue.pop()?; + // Discard any cancelled plans we encounter. + if let Some(data) = self.data_map.remove(&entry.plan_id) { + return Some(Plan { + time: entry.time, + data, + }); + } + } } /// Retrieve the earliest regular plan only if it is scheduled at `time`. @@ -136,12 +155,29 @@ impl PlanQueue { /// Returns `None` without removing a future plan if the next regular plan is /// later than `time`. pub(crate) fn pop_next_at(&mut self, time: f64) -> Option { - Self::discard_canceled_roots(&mut self.queue, &self.data_map); - match self.queue.peek() { - Some(entry) if entry.time == time => { - Self::pop_next_from_heap(&mut self.queue, &mut self.data_map) + loop { + match self.queue.peek() { + // Trim any cancelled plans + Some(entry) if !self.data_map.contains_key(&entry.plan_id) => { + self.queue.pop(); + } + + // Return only if the plan is scheduled for the given time + Some(entry) if entry.time == time => { + let entry = self.queue.pop().expect("peeked entry must exist"); + let data = self + .data_map + .remove(&entry.plan_id) + .expect("live plan must have callback"); + return Some(Plan { + time: entry.time, + data, + }); + } + + // There are no plans scheduled at the given time + _ => return None, } - _ => None, } } @@ -151,7 +187,17 @@ impl PlanQueue { /// shutdown-time queue is empty. pub(crate) fn pop_next_shutdown(&mut self) -> Option { trace!("getting next shutdown-time plan"); - Self::pop_next_from_heap(&mut self.shutdown_queue, &mut self.data_map) + loop { + // Return `None` if `pop` fails. + let entry = self.shutdown_queue.pop()?; + // Discard any cancelled plans we encounter. + if let Some(data) = self.data_map.remove(&entry.plan_id) { + return Some(Plan { + time: entry.time, + data, + }); + } + } } #[doc(hidden)] @@ -159,35 +205,6 @@ impl PlanQueue { self.queue.len() } - fn discard_canceled_roots( - heap: &mut BinaryHeap, - data_map: &HashMap, - ) { - while heap - .peek() - .is_some_and(|entry| !data_map.contains_key(&entry.plan_id)) - { - heap.pop(); - } - } - - fn pop_next_from_heap( - heap: &mut BinaryHeap, - data_map: &mut HashMap, - ) -> Option { - loop { - { - let entry = heap.pop()?; - if let Some(data) = data_map.remove(&entry.plan_id) { - return Some(Plan { - time: entry.time, - data, - }); - } - } - } - } - fn update_profiling_high_water_marks(&mut self) { #[cfg(feature = "profiling")] { From 215eaa9cfef2c010498f07f1b6389cd281476d5b Mon Sep 17 00:00:00 2001 From: Robert Jacobson Date: Wed, 24 Jun 2026 23:48:07 -0500 Subject: [PATCH 3/4] feat: Passive plans --- docs/book/src/topics/reports.md | 5 + examples/births-deaths/README.md | 2 +- src/context.rs | 185 +++++++++++++++++---- src/entity/context_extension.rs | 18 +-- src/plan_queue.rs | 265 +++++++++++++++++++++++-------- src/plugin_context.rs | 10 ++ 6 files changed, 385 insertions(+), 100 deletions(-) diff --git a/docs/book/src/topics/reports.md b/docs/book/src/topics/reports.md index d28c3a27..3c767342 100644 --- a/docs/book/src/topics/reports.md +++ b/docs/book/src/topics/reports.md @@ -409,6 +409,11 @@ pub fn init(context: &mut Context) { } ``` +Periodic plans are passive. They reschedule themselves after each report, but +they do not keep the simulation running after active plans are exhausted. This +means a periodic report can run at the final active simulation time, while later +periodic report callbacks remain queued unless more active work is scheduled. + The implementation of `write_aggregate_sir_report_item` is straightforward: We fetch the values from the data plugin, construct an instance of `AggregateSIRReportItem`, and "send" it to the report. diff --git a/examples/births-deaths/README.md b/examples/births-deaths/README.md index ffc0964c..e12370b9 100644 --- a/examples/births-deaths/README.md +++ b/examples/births-deaths/README.md @@ -33,7 +33,7 @@ scheduled for each individual. Once the person recovers, these plans are removed from the data structure. However, if the person dies before recovering, these plans are canceled at the time of death. The infection status of recovered individuals remains as recovered for the rest of the simulation, which will stop -when the plan queue is empty. +when no active plans remain and shutdown work is complete. ## Population manager diff --git a/src/context.rs b/src/context.rs index d7073814..9beff9cf 100644 --- a/src/context.rs +++ b/src/context.rs @@ -232,6 +232,55 @@ impl Context { time: f64, callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase, + ) -> PlanId { + self.add_plan_with_phase_and_activity(time, callback, phase, true) + } + + /// Add a passive plan to the future event list at the specified time in the + /// normal phase. + /// + /// Passive plans execute like regular plans but do not keep the simulation + /// timeline alive. + /// + /// Returns a [`PlanId`] for the newly-added plan that can be used to cancel it + /// if needed. + /// # Panics + /// + /// Panics if time is in the past, infinite, or NaN. + pub fn add_passive_plan( + &mut self, + time: f64, + callback: impl FnOnce(&mut Context) + 'static, + ) -> PlanId { + self.add_passive_plan_with_phase(time, callback, ExecutionPhase::Normal) + } + + /// Add a passive plan to the future event list at the specified time and + /// with the specified phase. + /// + /// Passive plans execute like regular plans but do not keep the simulation + /// timeline alive. + /// + /// Returns a [`PlanId`] for the newly-added plan that can be used to cancel it + /// if needed. + /// # Panics + /// + /// Panics if time is in the past, infinite, or NaN. + pub fn add_passive_plan_with_phase( + &mut self, + time: f64, + callback: impl FnOnce(&mut Context) + 'static, + phase: ExecutionPhase, + ) -> PlanId { + self.add_plan_with_phase_and_activity(time, callback, phase, false) + } + + fn add_plan_with_phase_and_activity( + &mut self, + time: f64, + callback: impl FnOnce(&mut Context) + 'static, + phase: ExecutionPhase, + is_active: bool, ) -> PlanId { let current = self.get_current_time(); assert!(!time.is_nan(), "Time {time} is invalid: cannot be NaN"); @@ -244,7 +293,8 @@ impl Context { "Time {time} is invalid: cannot be less than the current time ({}). Consider calling set_start_time() before scheduling plans.", current ); - self.plan_queue.add_plan(time, Box::new(callback), phase) + self.plan_queue + .add_plan(time, Box::new(callback), phase, is_active) } /// Add a plan to execute during shutdown-time in the normal phase. @@ -286,19 +336,22 @@ impl Context { period ); callback(self); - if !self.plan_queue.is_empty() { - let next_time = self.get_current_time() + period; - self.add_plan_with_phase( - next_time, - move |context| context.evaluate_periodic_and_schedule_next(period, callback, phase), - phase, - ); - } + let next_time = self.get_current_time() + period; + self.add_passive_plan_with_phase( + next_time, + move |context| context.evaluate_periodic_and_schedule_next(period, callback, phase), + phase, + ); } - /// Add a plan with specified priority to the future event list, and - /// continuously repeat the plan at the specified period, stopping - /// only once there are no other plans scheduled. + /// Add a passive periodic plan with specified priority to the future event + /// list. + /// + /// Periodic plans reschedule themselves after every run. They do not keep + /// the simulation timeline alive: when no active plans remain, normal + /// shutdown begins, and only passive plans at the final current time can + /// still run during that execution pass. Future passive periodic plans + /// remain queued and may run if later active work is scheduled. /// /// Notes: /// * The first periodic plan is scheduled at time `0.0`. If `set_start_time` was @@ -319,7 +372,7 @@ impl Context { "Period must be greater than 0" ); - self.add_plan_with_phase( + self.add_passive_plan_with_phase( 0.0, move |context| context.evaluate_periodic_and_schedule_next(period, callback, phase), phase, @@ -340,12 +393,6 @@ impl Context { } } - #[doc(hidden)] - #[allow(dead_code)] - pub(crate) fn remaining_plan_count(&self) -> usize { - self.plan_queue.remaining_plan_count() - } - /// Add a `Callback` to the queue to be executed before the next plan pub fn queue_callback(&mut self, callback: impl FnOnce(&mut Context) + 'static) { trace!("queuing callback"); @@ -482,7 +529,8 @@ impl Context { self.start_time } - /// Execute the simulation until the plan and callback queues are empty + /// Execute the simulation until active plans are exhausted and shutdown + /// work is complete. pub fn execute(&mut self) { trace!("entering event loop"); @@ -531,14 +579,15 @@ impl Context { match self.shutdown_status { ShutdownStatus::None => { // Normal execution may advance simulation time to the next - // regular plan. If no regular plans remain, natural completion - // transitions into shutdown-time plan execution. - if let Some(plan) = self.plan_queue.pop_next() { + // regular plan only while active regular work remains. Once no + // active regular plans remain, enter normal shutdown to drain + // current-time regular work without advancing time. + if let Some(plan) = self.plan_queue.pop_next_if_active() { trace!("calling plan at {:.6}", plan.time); self.current_time = Some(plan.time); (plan.data)(self); } else { - self.shutdown_status = ShutdownStatus::ShutdownTimePlans; + self.shutdown_status = ShutdownStatus::Normal; } } ShutdownStatus::Normal => { @@ -594,6 +643,17 @@ pub trait ContextBase: Sized { callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase, ) -> PlanId; + fn add_passive_plan( + &mut self, + time: f64, + callback: impl FnOnce(&mut Context) + 'static, + ) -> PlanId; + fn add_passive_plan_with_phase( + &mut self, + time: f64, + callback: impl FnOnce(&mut Context) + 'static, + phase: ExecutionPhase, + ) -> PlanId; fn add_shutdown_plan(&mut self, callback: impl FnOnce(&mut Context) + 'static) -> PlanId; fn add_shutdown_plan_with_phase( &mut self, @@ -625,6 +685,8 @@ impl ContextBase for Context { fn emit_event(&mut self, event: E); fn add_plan(&mut self, time: f64, callback: impl FnOnce(&mut Context) + 'static) -> PlanId; fn add_plan_with_phase(&mut self, time: f64, callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase) -> PlanId; + fn add_passive_plan(&mut self, time: f64, callback: impl FnOnce(&mut Context) + 'static) -> PlanId; + fn add_passive_plan_with_phase(&mut self, time: f64, callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase) -> PlanId; fn add_shutdown_plan(&mut self, callback: impl FnOnce(&mut Context) + 'static) -> PlanId; fn add_shutdown_plan_with_phase(&mut self, callback: impl FnOnce(&mut Context) + 'static, phase: ExecutionPhase) -> PlanId; fn add_periodic_plan_with_phase(&mut self, period: f64, callback: impl Fn(&mut Context) + 'static, phase: ExecutionPhase); @@ -714,6 +776,27 @@ mod tests { ) } + fn add_passive_plan(context: &mut Context, time: f64, value: u32) -> PlanId { + context.add_passive_plan(time, move |context| { + context.get_data_mut(ComponentA).push(value); + }) + } + + fn add_passive_plan_with_phase( + context: &mut Context, + time: f64, + value: u32, + phase: ExecutionPhase, + ) -> PlanId { + context.add_passive_plan_with_phase( + time, + move |context| { + context.get_data_mut(ComponentA).push(value); + }, + phase, + ) + } + #[test] #[should_panic(expected = "Time inf is invalid")] fn infinite_plan_time() { @@ -1147,6 +1230,49 @@ mod tests { assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); } + #[test] + fn passive_only_initial_time_runs_during_normal_shutdown() { + let mut context = Context::new(); + add_passive_plan(&mut context, 0.0, 1); + add_passive_plan(&mut context, 1.0, 2); + + context.execute(); + + assert_eq!(context.get_current_time(), 0.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + } + + #[test] + fn passive_plans_at_final_active_time_run_across_phases() { + let mut context = Context::new(); + add_passive_plan_with_phase(&mut context, 1.0, 1, ExecutionPhase::First); + add_plan(&mut context, 1.0, 2); + add_passive_plan_with_phase(&mut context, 1.0, 3, ExecutionPhase::Last); + + context.execute(); + + assert_eq!(context.get_current_time(), 1.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + + #[test] + fn passive_future_plan_survives_until_later_active_work() { + let mut context = Context::new(); + add_plan(&mut context, 1.0, 1); + add_passive_plan(&mut context, 2.0, 2); + + context.execute(); + + assert_eq!(context.get_current_time(), 1.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1]); + + add_plan(&mut context, 2.0, 3); + context.execute(); + + assert_eq!(context.get_current_time(), 2.0); + assert_eq!(*context.get_data_mut(ComponentA), vec![1, 2, 3]); + } + #[test] fn cancel_plan_can_cancel_shutdown_plan() { let mut context = Context::new(); @@ -1234,6 +1360,9 @@ mod tests { context.execute_single_step(); assert_eq!(*context.get_data_mut(ComponentA), Vec::::new()); + context.execute_single_step(); + assert_eq!(*context.get_data_mut(ComponentA), Vec::::new()); + context.execute_single_step(); assert_eq!(*context.get_data_mut(ComponentA), vec![1]); } @@ -1261,8 +1390,8 @@ mod tests { #[allow(clippy::cast_sign_loss)] #[allow(clippy::cast_possible_truncation)] fn periodic_plan_self_schedules() { - // checks whether the person properties report schedules itself - // based on whether there are plans in the queue + // checks whether the periodic plan schedules itself passively without + // keeping execution alive after active plans are exhausted. let mut context = Context::new(); context.add_periodic_plan_with_phase( 1.0, @@ -1275,9 +1404,9 @@ mod tests { context.add_plan(1.0, move |_context| {}); context.add_plan(1.5, move |_context| {}); context.execute(); - assert_eq!(context.get_current_time(), 2.0); + assert_eq!(context.get_current_time(), 1.5); - assert_eq!(*context.get_data(ComponentA), vec![0, 1, 2]); // time 0.0, 1.0, and 2.0 + assert_eq!(*context.get_data(ComponentA), vec![0, 1]); // time 0.0 and 1.0 } // Tests for negative time handling diff --git a/src/entity/context_extension.rs b/src/entity/context_extension.rs index 7b8e2ea4..2bb44cda 100644 --- a/src/entity/context_extension.rs +++ b/src/entity/context_extension.rs @@ -76,12 +76,8 @@ fn handle_periodic_value_change_count_event( let _ = std::mem::replace(slot.get_mut(), counter); } - if context.remaining_plan_count() == 0 { - return; - } - let next_time = context.get_current_time() + period; - context.add_plan_with_phase( + context.add_passive_plan_with_phase( next_time, move |context| { handle_periodic_value_change_count_event::( @@ -135,9 +131,11 @@ pub trait ContextEntitiesExt { /// Also panics if `period` is not finite and strictly positive. /// /// Recording starts at `ExecutionPhase::First` at simulation start time. The - /// first report runs at simulation start time in `ExecutionPhase::Last`, then at - /// each subsequent `start_time + k * period`. After the handler returns, the - /// matched counter is cleared. + /// report callbacks are passive plans: they do not keep the simulation + /// timeline alive. Reports run at simulation start time in + /// `ExecutionPhase::Last`, then at each subsequent `start_time + k * period` + /// that is reached while active work remains or during final-time shutdown. + /// After the handler returns, the matched counter is cleared. /// /// ```rust,ignore /// context.track_periodic_value_change_counts::( @@ -409,7 +407,7 @@ impl ContextEntitiesExt for Context { // We defer the first handler plan until now because it needs // `counter_id`, and it must run in `ExecutionPhase::Last`. - context.add_plan_with_phase( + context.add_passive_plan_with_phase( context.get_current_time(), move |context| { handle_periodic_value_change_count_event::( @@ -1437,6 +1435,7 @@ mod tests { context.set_property(person, CounterValue(1)); context.set_property(person, CounterValue(2)); }); + context.add_plan(1.0, |_| {}); context.execute(); assert_eq!(*observed.borrow(), vec![(0, 0), (1, 1)]); @@ -1469,6 +1468,7 @@ mod tests { context.add_plan(1.5, move |context| { context.set_property(person, CounterValue(1)); }); + context.add_plan(2.0, |_| {}); context.execute(); assert_eq!(*observed.borrow(), vec![0, 1, 0]); diff --git a/src/plan_queue.rs b/src/plan_queue.rs index eb8b8b59..6a146aef 100644 --- a/src/plan_queue.rs +++ b/src/plan_queue.rs @@ -13,6 +13,11 @@ use crate::{trace, HashMap, HashMapExt}; type Callback = dyn FnOnce(&mut Context); type BoxedCallback = Box; +struct QueuedPlan { + callback: BoxedCallback, + is_active: bool, +} + /// A priority queue that stores scheduled plans. /// /// Regular plans are ordered by simulation time, execution phase, and plan ID. @@ -21,7 +26,8 @@ type BoxedCallback = Box; pub(crate) struct PlanQueue { queue: BinaryHeap, shutdown_queue: BinaryHeap, - data_map: HashMap, + data_map: HashMap, + active_plan_count: usize, /// The next plan ID that will be issued. next_plan_id: u64, /// Tracks the high water mark of plans in flight (scheduled but not yet executed). @@ -40,6 +46,7 @@ impl PlanQueue { queue: BinaryHeap::new(), shutdown_queue: BinaryHeap::new(), data_map: HashMap::new(), + active_plan_count: 0, next_plan_id: 0, #[cfg(feature = "profiling")] max_plans_in_flight: 0, @@ -57,6 +64,7 @@ impl PlanQueue { time: f64, callback: BoxedCallback, phase: ExecutionPhase, + is_active: bool, ) -> PlanId { trace!("adding plan at {time}"); let plan_id = self.next_plan_id; @@ -65,7 +73,16 @@ impl PlanQueue { time, phase, }); - self.data_map.insert(plan_id, callback); + self.data_map.insert( + plan_id, + QueuedPlan { + callback, + is_active, + }, + ); + if is_active { + self.active_plan_count += 1; + } self.next_plan_id += 1; self.update_profiling_high_water_marks(); @@ -88,7 +105,13 @@ impl PlanQueue { time: 0.0, phase, }); - self.data_map.insert(plan_id, callback); + self.data_map.insert( + plan_id, + QueuedPlan { + callback, + is_active: false, + }, + ); self.next_plan_id += 1; self.update_profiling_high_water_marks(); @@ -100,26 +123,26 @@ impl PlanQueue { trace!("cancel plan {plan_id:?}"); // Delete the plan from the map, but leave in the heap. It will be skipped // when its heap entry reaches the root. - self.data_map.remove(&plan_id.0) - } - - #[must_use] - pub(crate) fn is_empty(&mut self) -> bool { - self.next_time().is_none() + self.data_map.remove(&plan_id.0).map(|queued_plan| { + if queued_plan.is_active { + self.active_plan_count -= 1; + } + queued_plan.callback + }) } /// Return the time the next plan is scheduled for, if there is one. #[must_use] pub(crate) fn next_time(&mut self) -> Option { - // First trim any cancelled plans. We want the time of the next legitimate plan. - while self - .queue - .peek() - .is_some_and(|entry| !self.data_map.contains_key(&entry.plan_id)) - { + while let Some(entry) = self.queue.peek() { + // We only want to report the time if the plan has not been canceled. + if self.data_map.contains_key(&entry.plan_id) { + return Some(entry.time); + } + // Trim the canceled plan. self.queue.pop(); } - self.queue.peek().map(|e| e.time) + None } /// Completely empties the queue, including the plans scheduled at shutdown time. @@ -128,23 +151,40 @@ impl PlanQueue { self.data_map.clear(); self.queue.clear(); self.shutdown_queue.clear(); + self.active_plan_count = 0; self.next_plan_id = 0; } - /// Retrieve the earliest regular plan in the queue. + /// Retrieve the earliest regular plan only if at least one active regular + /// plan is scheduled. /// - /// Returns the next plan if it exists or else `None` if the regular queue is - /// empty. - pub(crate) fn pop_next(&mut self) -> Option { + /// The active-plan check controls whether normal execution may continue. + /// The returned plan is simply the next regular plan by time, phase, and + /// plan ID; it may be active or passive. If no live active regular plan is + /// scheduled, this returns `None` without removing passive plans from the + /// regular queue. + pub(crate) fn pop_next_if_active(&mut self) -> Option { + if self.active_plan_count == 0 { + return None; + } + trace!("getting next plan"); loop { - // Return `None` if `pop` fails. - let entry = self.queue.pop()?; + // The `pop` should be infallible when the active plan count is positive unless the + // queue invariants have been violated. + let entry = self + .queue + .pop() + .expect("active plan count was positive but no regular plan was available"); + // Discard any cancelled plans we encounter. - if let Some(data) = self.data_map.remove(&entry.plan_id) { + if let Some(queued_plan) = self.data_map.remove(&entry.plan_id) { + if queued_plan.is_active { + self.active_plan_count -= 1; + } return Some(Plan { time: entry.time, - data, + data: queued_plan.callback, }); } } @@ -164,14 +204,18 @@ impl PlanQueue { // Return only if the plan is scheduled for the given time Some(entry) if entry.time == time => { - let entry = self.queue.pop().expect("peeked entry must exist"); - let data = self + // Pop is infallible here. + let entry = self.queue.pop().unwrap(); + let queued_plan = self .data_map .remove(&entry.plan_id) .expect("live plan must have callback"); + if queued_plan.is_active { + self.active_plan_count -= 1; + } return Some(Plan { - time: entry.time, - data, + time, + data: queued_plan.callback, }); } @@ -187,22 +231,18 @@ impl PlanQueue { /// shutdown-time queue is empty. pub(crate) fn pop_next_shutdown(&mut self) -> Option { trace!("getting next shutdown-time plan"); - loop { - // Return `None` if `pop` fails. - let entry = self.shutdown_queue.pop()?; - // Discard any cancelled plans we encounter. - if let Some(data) = self.data_map.remove(&entry.plan_id) { - return Some(Plan { - time: entry.time, - data, - }); + std::iter::from_fn(|| self.shutdown_queue.pop()).find_map(|entry| { + // If there's no `data_map` entry, the plan has been canceled, so discard + // and pop another plan. + let queued_plan = self.data_map.remove(&entry.plan_id)?; + if queued_plan.is_active { + self.active_plan_count -= 1; } - } - } - - #[doc(hidden)] - pub(crate) fn remaining_plan_count(&self) -> usize { - self.queue.len() + Some(Plan { + time: entry.time, + data: queued_plan.callback, + }) + }) } fn update_profiling_high_water_marks(&mut self) { @@ -221,7 +261,7 @@ impl PlanQueue { let queue_bytes = (self.queue.capacity() + self.shutdown_queue.capacity()) * size_of::(); - let map_entry_bytes = self.data_map.capacity() * size_of::<(u64, BoxedCallback)>(); + let map_entry_bytes = self.data_map.capacity() * size_of::<(u64, QueuedPlan)>(); queue_bytes + map_entry_bytes } @@ -300,7 +340,7 @@ mod tests { #[test] fn empty_queue() { let mut plan_queue = PlanQueue::new(); - assert!(plan_queue.pop_next().is_none()); + assert!(plan_queue.pop_next_if_active().is_none()); } #[test] @@ -312,34 +352,37 @@ mod tests { 1.0, callback(1, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.add_plan( 3.0, callback(3, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.add_plan( 2.0, callback(2, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); - assert!(!plan_queue.is_empty()); + assert_eq!(plan_queue.next_time(), Some(1.0)); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 1.0); run_plan(next_plan, &mut context); - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(plan_queue.next_time(), Some(2.0)); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 2.0); run_plan(next_plan, &mut context); - assert!(!plan_queue.is_empty()); - let next_plan = plan_queue.pop_next().unwrap(); + assert_eq!(plan_queue.next_time(), Some(3.0)); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 3.0); run_plan(next_plan, &mut context); - assert!(plan_queue.pop_next().is_none()); + assert!(plan_queue.pop_next_if_active().is_none()); assert_eq!(*observed.borrow(), vec![1, 2, 3]); } @@ -352,21 +395,23 @@ mod tests { 1.0, callback(1, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.add_plan( 1.0, callback(2, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 1.0); run_plan(next_plan, &mut context); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 1.0); run_plan(next_plan, &mut context); - assert!(plan_queue.pop_next().is_none()); + assert!(plan_queue.pop_next_if_active().is_none()); assert_eq!(*observed.borrow(), vec![1, 2]); } @@ -379,21 +424,23 @@ mod tests { 1.0, callback(1, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.add_plan( 1.0, callback(2, Rc::clone(&observed)), ExecutionPhase::First, + true, ); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 1.0); run_plan(next_plan, &mut context); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 1.0); run_plan(next_plan, &mut context); - assert!(plan_queue.pop_next().is_none()); + assert!(plan_queue.pop_next_if_active().is_none()); assert_eq!(*observed.borrow(), vec![2, 1]); } @@ -406,31 +453,121 @@ mod tests { 1.0, callback(1, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); let plan_to_cancel = plan_queue.add_plan( 2.0, callback(2, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.add_plan( 3.0, callback(3, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.cancel_plan(&plan_to_cancel); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 1.0); run_plan(next_plan, &mut context); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 3.0); run_plan(next_plan, &mut context); - assert!(plan_queue.pop_next().is_none()); + assert!(plan_queue.pop_next_if_active().is_none()); assert_eq!(*observed.borrow(), vec![1, 3]); } + #[test] + fn passive_only_plans_do_not_pop_during_active_execution() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + false, + ); + + assert_eq!(plan_queue.active_plan_count, 0); + assert!(plan_queue.pop_next_if_active().is_none()); + assert_eq!(plan_queue.next_time(), Some(1.0)); + assert!(observed.borrow().is_empty()); + } + + #[test] + fn passive_plan_can_pop_before_later_active_plan() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + false, + ); + plan_queue.add_plan( + 2.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::Normal, + true, + ); + + assert_eq!(plan_queue.active_plan_count, 1); + let next_plan = plan_queue.pop_next_if_active().unwrap(); + assert_eq!(next_plan.time, 1.0); + run_plan(next_plan, &mut context); + assert_eq!(plan_queue.active_plan_count, 1); + + let next_plan = plan_queue.pop_next_if_active().unwrap(); + assert_eq!(next_plan.time, 2.0); + run_plan(next_plan, &mut context); + assert_eq!(plan_queue.active_plan_count, 0); + + assert_eq!(*observed.borrow(), vec![1, 2]); + } + + #[test] + fn canceling_active_plan_decrements_active_count() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut plan_queue = PlanQueue::new(); + let active = plan_queue.add_plan( + 1.0, + callback(1, Rc::clone(&observed)), + ExecutionPhase::Normal, + true, + ); + let passive = plan_queue.add_plan( + 2.0, + callback(2, Rc::clone(&observed)), + ExecutionPhase::Normal, + false, + ); + + assert_eq!(plan_queue.active_plan_count, 1); + plan_queue.cancel_plan(&passive); + assert_eq!(plan_queue.active_plan_count, 1); + plan_queue.cancel_plan(&active); + assert_eq!(plan_queue.active_plan_count, 0); + } + + #[test] + fn shutdown_plans_do_not_affect_active_count() { + let observed = Rc::new(RefCell::new(Vec::new())); + let mut context = Context::new(); + let mut plan_queue = PlanQueue::new(); + plan_queue.add_shutdown_plan(callback(1, Rc::clone(&observed)), ExecutionPhase::Normal); + + assert_eq!(plan_queue.active_plan_count, 0); + let next_plan = plan_queue.pop_next_shutdown().unwrap(); + run_plan(next_plan, &mut context); + assert_eq!(plan_queue.active_plan_count, 0); + assert_eq!(*observed.borrow(), vec![1]); + } + #[test] fn next_time_ignores_canceled_root() { let observed = Rc::new(RefCell::new(Vec::new())); @@ -439,11 +576,13 @@ mod tests { 1.0, callback(1, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.add_plan( 2.0, callback(2, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); plan_queue.cancel_plan(&plan_to_cancel); @@ -460,11 +599,12 @@ mod tests { 2.0, callback(2, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); assert!(plan_queue.pop_next_at(1.0).is_none()); - let next_plan = plan_queue.pop_next().unwrap(); + let next_plan = plan_queue.pop_next_if_active().unwrap(); assert_eq!(next_plan.time, 2.0); run_plan(next_plan, &mut context); assert_eq!(*observed.borrow(), vec![2]); @@ -495,6 +635,7 @@ mod tests { 1.0, callback(1, Rc::clone(&observed)), ExecutionPhase::Normal, + true, ); let shutdown_id = plan_queue.add_shutdown_plan(callback(2, Rc::clone(&observed)), ExecutionPhase::Normal); diff --git a/src/plugin_context.rs b/src/plugin_context.rs index 640ccaaf..43810bac 100644 --- a/src/plugin_context.rs +++ b/src/plugin_context.rs @@ -74,6 +74,16 @@ mod test_plugin_context { self.add_plan(1.0, |context| { assert_eq!(context.get_my_data(), 42); }); + self.add_passive_plan(1.0, |context| { + assert_eq!(context.get_my_data(), 42); + }); + self.add_passive_plan_with_phase( + 1.0, + |context| { + assert_eq!(context.get_my_data(), 100); + }, + crate::ExecutionPhase::Last, + ); self.add_periodic_plan_with_phase( 1.0, |context| { From 28415e6554d14dabd3ee98ebed7276c6b434fd54 Mon Sep 17 00:00:00 2001 From: Robert Jacobson Date: Fri, 26 Jun 2026 12:32:04 -0500 Subject: [PATCH 4/4] docs: Simulation execution and shutdown chapter of The Ixa Book --- docs/book/src/SUMMARY.md | 1 + docs/book/src/first_model/next-steps.md | 4 +- docs/book/src/first_model/setup.md | 3 +- .../src/topics/burn-in-and-negative-time.md | 2 + .../book/src/topics/execution-and-shutdown.md | 345 ++++++++++++++++++ docs/book/src/topics/reports.md | 8 +- docs/book/src/topics/topics.md | 1 + 7 files changed, 358 insertions(+), 6 deletions(-) create mode 100644 docs/book/src/topics/execution-and-shutdown.md diff --git a/docs/book/src/SUMMARY.md b/docs/book/src/SUMMARY.md index 438920c4..a34f9aa1 100644 --- a/docs/book/src/SUMMARY.md +++ b/docs/book/src/SUMMARY.md @@ -14,6 +14,7 @@ - [Properties](topics/properties.md) - [Indexing](topics/indexing.md) - [Burn-in Periods and Negative Time](topics/burn-in-and-negative-time.md) + - [Execution and Shutdown](topics/execution-and-shutdown.md) - [Handling Errors](topics/handling-errors.md) - [Performance and Profiling](topics/performance.md) - [Profiling Module](topics/profiling-module.md) diff --git a/docs/book/src/first_model/next-steps.md b/docs/book/src/first_model/next-steps.md index 6c10b361..95448fbc 100644 --- a/docs/book/src/first_model/next-steps.md +++ b/docs/book/src/first_model/next-steps.md @@ -14,7 +14,9 @@ in its entirety. 1. Currently the simulation runs until `MAX_TIME` even if every single person has been infected and has recovered. Add a check somewhere that calls `context.shutdown()` if there is no more work for the simulation to do. Where - should this check live? _Hint: Use `context.query_entity_count`._ + should this check live? _Hint: Use `context.query_entity_count`._ See + [Common Shutdown Patterns](../topics/execution-and-shutdown.md#common-shutdown-patterns) + for context on explicit stop conditions. 2. Analyze the data output by the incident reporter. Plot the number of people with each `InfectionStatus` on the same axis to see how they change over the course of the simulation. Are the curves what we expect to see given our diff --git a/docs/book/src/first_model/setup.md b/docs/book/src/first_model/setup.md index 47b99424..9725f281 100644 --- a/docs/book/src/first_model/setup.md +++ b/docs/book/src/first_model/setup.md @@ -130,7 +130,8 @@ The `run_with_args()` function does the following: 3. Finally, it kicks off the simulation by executing `context.execute()`. Of course, our model doesn't actually do anything or even contain any data, so `context.execute()` checks that there is no work to do and immediately - returns. + returns. (See + [Execution and Shutdown](../topics/execution-and-shutdown.md).) If there is an error at any stage, `run_with_args()` will return an error result. The Rust compiler will complain if we do not handle the returned result, diff --git a/docs/book/src/topics/burn-in-and-negative-time.md b/docs/book/src/topics/burn-in-and-negative-time.md index 8a843a20..ef145360 100644 --- a/docs/book/src/topics/burn-in-and-negative-time.md +++ b/docs/book/src/topics/burn-in-and-negative-time.md @@ -19,6 +19,8 @@ A common but fragile solution is to run a separate “initialization” simulati Ixa provides a simpler mechanism: treat burn-in as part of the same execution. Instead of resetting time or stitching simulations together, you allow the timeline to extend into negative values. You then designate `0.0` as the beginning of your analysis window. Negative time is not a special execution mode in Ixa. It is simply earlier simulation time. The event queue, scheduling rules, and execution semantics are identical before and after `0.0`. What may differ is your model logic. You may choose to disable transmission, suppress reporting, use alternate parameters, or run simplified dynamics during burn-in. These differences arise from your code, not from special treatment by the framework. +For details about the execution lifecycle itself, see +[Execution and Shutdown](execution-and-shutdown.md). ## The Core Pattern diff --git a/docs/book/src/topics/execution-and-shutdown.md b/docs/book/src/topics/execution-and-shutdown.md new file mode 100644 index 00000000..4c69395a --- /dev/null +++ b/docs/book/src/topics/execution-and-shutdown.md @@ -0,0 +1,345 @@ +# Execution and Shutdown + +This chapter details everything you might need to know about how simulations start, run, and stop. + +## Syntax Summary + +Running a manually constructed `Context` instance: + +```rust +let mut context = Context::new(); +// Initialize the model before execution. + +// Start the event loop... +context.execute(); +// ...or single-step by executing one callback or plan: +context.execute_single_step(); +``` + +Shut down a running simulation from within model code: + +```rust +// Trigger normal shutdown cycle and exit the event loop +context.shutdown(); +// Skip normal shutdown cycle and exit the event loop immediately +context.abort(); +``` + +## The Execution Lifecycle + +A single simulation in Ixa is encapsulated by a `Context` instance. At the highest level, the lifecycle of a simulation +is: + +1. **Construct a new `Context` instance.** Let's call it `context`. +2. **Initialize `context`.** Configure model behavior and initial state, load any initial entity data, register event + subscribers, schedule initial plans, etc. +3. **Start the event loop.** This "runs" the simulation, performing work scheduled on the timeline or queued on the job + queue, evolving the simulation state and driving the simulation forward. +4. **Exit the event loop.** Either the work queue is exhausted or client code manually triggers a stop. + +### Beginning Execution + +Executing a simulation, that is, starting the event loop, is as simple as calling +`context.execute()`: + +```rust +let mut context = Context::new(); +// Perform initialization steps here, before execution. +context.execute(); +``` + +The only important thing to know is that `context.execute()` does not return until the event loop exits. + +The preferred pattern is to use `run_with_args` (or `run_with_custom_args`) to configure and run your `Context` instance +using command-line options. In that case, you don't even have to construct a new `Context` instance or call +`context.execute()` yourself. The `run_with_args` function: + +1. creates a new `Context` instance; +2. initializes the new `Context` instance according to the command line parameters; +3. calls the initialization function / closure you provided it, passing in the partially initialized `context`; +4. if your closure returns `Ok(())`, calls `context.execute()`; otherwise returns your error. + +In code: + +```rust +// Constructs, initializes, and (if there are no errors) executes a `Context`: +let result = run_with_args(|context: &mut Context, _args, _| { + // Any additional custom initialization of `context` goes here. + Ok(()) +}); +// If initialization failed, an error is returned. +// Otherwise, execution is run to completion. +``` + +See the chapter [Your First Model](../first_model/your-first-model.md) for a complete example. + +### During Normal Event-loop Execution + +Once `context.execute()` starts, Ixa repeatedly performs the next available unit +of work. During ordinary execution, this means: + +1. If a callback is queued, run the next callback. +2. Otherwise, run the next plan from the future event list. +3. If there is no remaining active plan to keep the event loop alive, begin the + shutdown lifecycle. + +This callback-before-plan ordering is important. A callback represents work +that should happen as soon as control returns to the event loop, before the next +scheduled plan is chosen. This is also how event handlers run: when code emits +an event, Ixa queues the matching handlers as callbacks rather than running them +inline inside the code that emitted the event. + +Plans are different from callbacks because they are scheduled at a particular simulation time. When the event loop +chooses a plan, Ixa advances the current simulation time to that plan's scheduled time and then runs the plan. Callbacks +do not advance simulation time; they run at whatever the current simulation time already is. + +When multiple plans are scheduled for the same time, Ixa uses execution phases to order them. Plans in +`ExecutionPhase::First` run before plans in `ExecutionPhase::Normal`, and plans in `ExecutionPhase::Last` run after +normal plans. Plans with the same time and same phase run in the order they were scheduled. Model code should avoid +abusing `ExecutionPhase` to force ordering of otherwise normal simulation behavior. + +> [!TIP] First and Last Execution Phases +> +> As a general rule, `ExecutionPhase::First` and `ExecutionPhase::Last` should be used for out-of-simulation +> administrative tasks, such as collecting statistics or reporting tasks. If normal in-simulation behavior needs to have a +> strict ordering of tasks scheduled at the same simulation time, the model should be structured to ensure that ordering +> in another way, for example by executing those tasks from within a single plan instead of in separate plans scheduled +> for the same time. + +The event loop is not just burning through a fixed list of work created during initialization. Most models schedule some +initial plans after constructing `context` and before calling `context.execute()`, so execution has somewhere to begin. Once +the event loop is running, the queue is usually dynamic: plans and event subscribers run model code, that model code +changes simulation state, and those changes often schedule more plans or emit events. Emitted events queue callbacks for +subscribed handlers, and those callbacks can themselves schedule plans, emit more events, or queue additional callbacks. +In this way, the model continually creates the future work that drives the simulation forward. + +This process is usually probabilistic. For example, a model might randomly sample how long an infection lasts and then +schedule a future plan that changes the infected person to recovered at the sampled time. In that sense, the future plan +queue is often generated probabilistically as the simulation runs. Even so, a simulation should be exactly reproducible +when run with the same random seed; see the [`random` module](random-module.md) chapter for details. + +The event loop keeps repeating this process until execution starts to end. That +can happen automatically when Ixa detects that no active plans remain, or manually when model code calls +`context.shutdown()` or `context.abort()`. The following sections describe those ending behaviors in more detail. + +### Ending Execution + +Execution ends when the event loop enters a shutdown path. Ixa can do this +automatically when active work is exhausted, or model code can request it +explicitly. + +#### Automatic Shutdown + +The built-in automatic shutdown condition is active-plan exhaustion. During +normal event-loop execution, Ixa only advances simulation time while at least +one active plan remains scheduled. When there are no active plans left, Ixa +stops advancing simulation time and begins the normal shutdown lifecycle. + +This means that not every scheduled plan is responsible for keeping the +simulation alive. Ixa distinguishes active plans from passive plans. Active +plans represent simulation-driving work: they evolve state, create future +simulation work, or otherwise indicate that the model still has forward +progress to make. Plans scheduled with `context.add_plan()` and +`context.add_plan_with_phase()` are active. + +Passive plans are scheduled on the same future event list, use the same +simulation-time and execution-phase ordering rules, and run like other plans +while active work remains. The difference is that passive plans do not keep +execution alive by themselves. They are intended for observational or +administrative work: reporting, statistics collection, and other tasks that can +be skipped if the simulation has otherwise finished. + +Periodic plans are passive. This is important because a periodic plan +reschedules itself every time it runs. If periodic plans were active, a periodic +reporter or statistics collector could keep the simulation alive forever even +after all simulation-driving work had finished. Because periodic plans are +passive, they can run while active work is keeping the timeline alive without +preventing automatic shutdown. + +Passive plans can remain queued after automatic shutdown begins. A passive plan +scheduled at the final current simulation time can still run as part of the +normal shutdown lifecycle, but a passive plan scheduled for a later time will +not cause Ixa to advance time just to run it. That future passive plan remains +queued and may run later if model code schedules new active work and calls +`context.execute()` again. + +#### Normal Shutdown + +Normal shutdown is the orderly way for execution to end. It happens automatically when active plans are exhausted, and +it can also be requested manually by calling `context.shutdown()` from model code. + +Calling `context.shutdown()` does not stop execution at that exact line of code. Instead, it tells the event loop to +finish the normal shutdown cycle and then return from `context.execute()`. The key rule is that normal shutdown stops +simulation time from advancing. Ixa may still run work that is already due at the current simulation time, but it will +not move forward to a later simulation time during that execution pass. + +##### What Runs During Normal Shutdown + +Normal shutdown preserves the usual callback-before-plan rule. If callbacks are already queued, or if shutdown work +queues new callbacks, those callbacks run before the next plan is selected. + +After callbacks are drained, Ixa runs regular plans scheduled exactly at `context.get_current_time()`. These are plans +that are due at the current simulation time, not future plans. Once no more current-time regular plans are available, +Ixa runs shutdown-time plans. After shutdown-time plans and any callbacks they queue are exhausted, `context.execute()` +returns. + +The overall order is: + +1. queued callbacks; +2. regular plans at the current simulation time; +3. shutdown-time plans, with callbacks still drained before the next shutdown-time plan; +4. return from `context.execute()`. + +##### Current-Time Plans Across Phases + +During normal shutdown, Ixa drains all regular plans scheduled at the current simulation time, across execution phases. +This matters when `context.shutdown()` is called by one plan while other plans are scheduled for the same time. + +For example, suppose a normal-phase plan at time `10.0` calls `context.shutdown()`. Ixa will not advance to time `11.0`, +but it will still run the remaining plans scheduled for time `10.0`, including plans in `ExecutionPhase::Last`. The +callback queue is drained between execution of plans as usual. + +##### Shutdown-Time Plans + +Shutdown-time plans are plans that are meant to run at the end of execution rather than at a specific simulation time. +They are scheduled with `context.add_shutdown_plan()` or `context.add_shutdown_plan_with_phase()`. + +Shutdown-time plans run after current-time regular plans have been exhausted. They do not advance simulation +time; during a shutdown-time plan, `context.get_current_time()` still returns the last regular simulation time. + +Ixa orders shutdown-time plans the same way it orders regular plans at a shared simulation time: by execution phase, +then by scheduling order. + +Once Ixa has begun running shutdown-time plans, it does not return to the regular plan queue during that same execution +pass. If a shutdown-time plan schedules a regular plan, even at the current simulation time, that regular plan remains +unexecuted in the queue until `context.execute()` is called again. + +Shutdown-time plans are useful for finalization work that should happen after ordinary current-time simulation work has +finished but before `context.execute()` returns. Common use cases include: + +- Writing a final "totals" row to a summary report, such as total infections, peak prevalence, or final population size. +- Emitting a final partial-period report when shutdown happens between regular periodic report times. If shutdown occurs + partway through a reporting period, the next periodic report time will never come. A shutdown-time plan can write this + final partial-period report. +- Flushing model-owned buffers or cached output that has not yet been written through Ixa's + [reporting system](reports.md). +- Running final consistency checks, such as verifying conservation relationships or checking that no model-specific + invariants were violated. +- Recording model-specific shutdown metadata, such as whether the run ended by max time, extinction, or another + model-defined condition. +- Finalizing accumulators whose meaning depends on the full run, such as time-at-risk totals, person-time denominators, + or cumulative exposure measures. + +##### What Remains Queued After Shutdown + +Normal shutdown does not clear plans scheduled in the future, that is, plans scheduled after the current simulation +time. Also, if during shutdown-time execution a plan is scheduled for the current simulation time, that plan will remain +queued without being executed as well, because once shutdown-time plans start executing, the event loop never returns +to the regular plan queue before exiting. + +This is intentional. A `Context` can be executed again after `context.execute()` returns. If model code later schedules +new active work and calls `context.execute()` again, queued future work can still run according to the usual event-loop +rules. However, this would be unusual. The more typical case is for `context.execute()` to only ever be called once. + +#### Aborting Execution + +Calling `context.abort()` stops the current `context.execute()` event loop immediately. Unlike normal shutdown, aborting +does not drain queued callbacks, does not run remaining plans at the current simulation time, and does not run +shutdown-time plans. It is the escape hatch for cases where the model should stop now rather than complete the normal +shutdown cycle. + +An abort only affects the current execution pass. It does not clear the future event list or permanently poison the +`Context`. If queued work remains, a later call to `context.execute()` can continue from that state. In typical model +code, however, `abort()` should be reserved for exceptional cases where skipping normal shutdown work is intentional. +If `context.shutdown()` is normal shutdown, then `context.abort()` is abnormal shutdown. + +#### Common Shutdown Patterns + +Even though Ixa automatically begins shutdown when no active plans remain, most models should still define an explicit +stop condition. An explicit stop condition makes the intended run horizon clear and protects the model from running +longer than intended if some part of the simulation keeps producing active work. + +A recommended best practice is to define a fixed end time scheduled during initialization: + +```rust +const MAX_TIME: f64 = 365.0; + +context.add_passive_plan(MAX_TIME, |context| { + context.shutdown(); +}); +``` + +This does not prevent the model from ending earlier by active-plan exhaustion. It simply guarantees that, if the +simulation is still running at `MAX_TIME`, normal shutdown begins then. The end time should usually be far enough in the +future to cover the intended simulation horizon. + +Models can also define their own shutdown triggers in ordinary model code. For example, a model might shut down once a +target condition has been reached: + +```rust +fn maybe_stop(context: &mut Context) { + if no_infections_remain(context) { + context.shutdown(); + } +} +``` + +That trigger can be called from whichever callbacks or plans observe a relevant state change. Once `context.shutdown()` +is called, Ixa observes the normal shutdown cycle, exiting the event loop before simulation time can progress. + +## Execution Outside of the Event-loop + +Most models call `context.execute()` once and let the event loop run to completion. Ixa also exposes the lower-level +`context.execute_single_step()` method for less common cases where model code needs to step through work manually. +Never call these methods from in-simulation code, that is, code that executes from within an active event loop. They +should only be called outside of a running event loop. + +### Executing Again After `execute()` Returns + +A `Context` still exists after `context.execute()` returns. Its model state remains in place, its current simulation +time remains where execution stopped, and any queued work that was not run during the previous execution pass remains +queued. + +That means it is possible to call `context.execute()` again. This is not the typical way to structure a model run, but +it can be useful for interactive workflows, tests, debugging, or specialized control code that intentionally runs a +simulation in stages. + +The main thing to remember is that a later execution pass continues from the current state. It does not reset the +`Context`, clear queued work, or move simulation time backward. For example, if normal shutdown stops at time `10.0` +and leaves an active plan queued for time `12.0`, a later call to `context.execute()` can continue to that future plan. +Similarly, if `context.abort()` stops execution while callbacks or plans remain queued, a later call to +`context.execute()` can continue with that queued work. + +Passive plans require special care. A future passive plan can remain queued after execution stops, but a passive plan +does not keep the event loop alive. If only passive future work remains, calling `context.execute()` again will not make +Ixa advance time just to run it. To reach that passive work in a later execution pass, the model must also schedule +active work that keeps the timeline alive until the relevant time. + +[Negative-time burn-in](burn-in-and-negative-time.md) and restarting the event loop are not interchangeable. Burn-in +runs as a single, uninterrupted `context.execute()` call that simply starts before `0.0` and crosses into the analysis +window without resetting state. This is what you want whenever you need an initialization period for things like +population structure, immunity, or infection states to stabilize before measurement begins. Calling `context.execute()` +a second time is a separate pattern that only makes sense when the event loop must stop and hand control back to your +driver code, typically for tests, debugging, interactive workflows, or staged runs that inspect or modify the `Context` +between passes. In short: reach for burn-in for initialization, and use repeated `execute()` calls only when you +specifically need out-of-simulation low-level control. + +### Single-Step Execution + +The `context.execute_single_step()` method exposes one iteration of the event-loop state machine. One call runs at most +one queued callback, one plan, or one shutdown status transition. This is the primitive that `context.execute()` uses +internally: `execute()` repeatedly calls `execute_single_step()` until the event loop reaches a stopped state. + +Single stepping is most useful for tests, debugging, visualization, and interactive control code. It lets external code +observe the `Context` between units of event-loop work instead of waiting for the whole simulation to finish. + +The same ordering rules still apply. If a callback is queued, a single step runs that callback before selecting a plan. +If no callback is queued, a single step may run the next regular plan, move from active-plan exhaustion into normal +shutdown, run a current-time shutdown plan, move into shutdown-time execution, run a shutdown-time plan, or complete the +stopped-state transition. + +The main subtlety is that not every single step runs user model code. Some calls only advance the execution state +machine. For example, if there is no active work left but a shutdown-time plan is queued, one call may enter normal +shutdown, another may move from current-time regular plans to shutdown-time plans, and a later call may actually run the +shutdown-time plan. This is expected: single stepping exposes the intermediate lifecycle states that `context.execute()` +usually hides by looping until execution is complete. diff --git a/docs/book/src/topics/reports.md b/docs/book/src/topics/reports.md index 3c767342..5cadd363 100644 --- a/docs/book/src/topics/reports.md +++ b/docs/book/src/topics/reports.md @@ -409,10 +409,10 @@ pub fn init(context: &mut Context) { } ``` -Periodic plans are passive. They reschedule themselves after each report, but -they do not keep the simulation running after active plans are exhausted. This -means a periodic report can run at the final active simulation time, while later -periodic report callbacks remain queued unless more active work is scheduled. +Periodic plans are passive. They reschedule themselves after each report, but they do not keep the simulation running +after active plans are exhausted. This means a periodic report can run at the final active simulation time, while later +periodic report callbacks remain queued without being executed unless more active work is scheduled. See [Execution and +Shutdown](execution-and-shutdown.md) for details about active and passive plans. The implementation of `write_aggregate_sir_report_item` is straightforward: We fetch the values from the data plugin, construct an instance of diff --git a/docs/book/src/topics/topics.md b/docs/book/src/topics/topics.md index 1dfd0fc8..3d479f7c 100644 --- a/docs/book/src/topics/topics.md +++ b/docs/book/src/topics/topics.md @@ -3,6 +3,7 @@ - [Properties](properties.md) - [Indexing](indexing.md) - [Burn-in Periods and Negative Time](burn-in-and-negative-time.md) +- [Execution and Shutdown](execution-and-shutdown.md) - [Handling Errors](handling-errors.md) - [Performance and Profiling](performance.md) - [Profiling Module](profiling-module.md)