mohr/nexosim
1
0
Fork 0
forked from ROMEO/nexosim
Code Pull Requests Activity

Return both simulation and scheduler at init

Browse Source
This commit is contained in:
Serge Barral 2024-11-15 16:52:24 +01:00
parent f4686af49a
commit 0f1d876aed
20 changed files with 101 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
asynchronix-util/examples/observables.rs
View File

@ -182,7 +182,10 @@ fn main() -> Result<(), SimulationError> {
let t0 = MonotonicTime::EPOCH;
// Assembly and initialization.
let mut simu = SimInit::new().add_model(proc, proc_mbox, "proc").init(t0)?;
let mut simu = SimInit::new()
.add_model(proc, proc_mbox, "proc")
.init(t0)?
.0;
// ----------
// Simulation.

4
asynchronix/examples/assembly.rs
View File

@ -133,12 +133,10 @@ fn main() -> Result<(), SimulationError> {
let t0 = MonotonicTime::EPOCH;
// Assembly and initialization.
let mut simu = SimInit::new()
let (mut simu, scheduler) = SimInit::new()
.add_model(assembly, assembly_mbox, "assembly")
.init(t0)?;
let scheduler = simu.scheduler();
// ----------
// Simulation.
// ----------

4
asynchronix/examples/espresso_machine.rs
View File

@ -366,14 +366,12 @@ fn main() -> Result<(), SimulationError> {
let t0 = MonotonicTime::EPOCH;
// Assembly and initialization.
let mut simu = SimInit::new()
let (mut simu, scheduler) = SimInit::new()
.add_model(controller, controller_mbox, "controller")
.add_model(pump, pump_mbox, "pump")
.add_model(tank, tank_mbox, "tank")
.init(t0)?;
let scheduler = simu.scheduler();
// ----------
// Simulation.
// ----------

3
asynchronix/examples/external_input.rs
View File

@ -210,7 +210,8 @@ fn main() -> Result<(), SimulationError> {
let mut simu = SimInit::new()
.add_model(listener, listener_mbox, "listener")
.set_clock(AutoSystemClock::new())
.init(t0)?;
.init(t0)?
.0;
// ----------
// Simulation.

3
asynchronix/examples/power_supply.rs
View File

@ -144,7 +144,8 @@ fn main() -> Result<(), SimulationError> {
.add_model(load1, load1_mbox, "load1")
.add_model(load2, load2_mbox, "load2")
.add_model(load3, load3_mbox, "load3")
.init(t0)?;
.init(t0)?
.0;
// ----------
// Simulation.

4
asynchronix/examples/stepper_motor.rs
View File

@ -206,13 +206,11 @@ fn main() -> Result<(), asynchronix::simulation::SimulationError> {
let t0 = MonotonicTime::EPOCH;
// Assembly and initialization.
let mut simu = SimInit::new()
let (mut simu, scheduler) = SimInit::new()
.add_model(driver, driver_mbox, "driver")
.add_model(motor, motor_mbox, "motor")
.init(t0)?;
let scheduler = simu.scheduler();
// ----------
// Simulation.
// ----------

3
asynchronix/src/grpc/run.rs
View File

@ -102,9 +102,10 @@ impl simulation_server::Simulation for GrpcSimulationService {
let (reply, bench) = self.initializer().init(request);
if let Some((simulation, endpoint_registry)) = bench {
if let Some((simulation, scheduler, endpoint_registry)) = bench {
*self.controller() = ControllerService::Started {
simulation,
scheduler,
event_source_registry: endpoint_registry.event_source_registry,
query_source_registry: endpoint_registry.query_source_registry,
key_registry: KeyRegistry::default(),

14
asynchronix/src/grpc/services.rs
View File

@ -8,7 +8,7 @@ use prost_types::Timestamp;
use tai_time::MonotonicTime;
use super::codegen::simulation::{Error, ErrorCode};
use crate::simulation::ExecutionError;
use crate::simulation::{ExecutionError, SchedulingError};
pub(crate) use controller_service::ControllerService;
pub(crate) use init_service::InitService;
@ -47,6 +47,18 @@ fn map_execution_error(error: ExecutionError) -> Error {
to_error(error_code, error_message)
}
/// Map a `SchedulingError` to a Protobuf error.
fn map_scheduling_error(error: SchedulingError) -> Error {
let error_code = match error {
SchedulingError::InvalidScheduledTime => ErrorCode::InvalidDeadline,
SchedulingError::NullRepetitionPeriod => ErrorCode::InvalidPeriod,
};
let error_message = error.to_string();
to_error(error_code, error_message)
}
/// Attempts a cast from a `MonotonicTime` to a protobuf `Timestamp`.
///
/// This will fail if the time is outside the protobuf-specified range for

13
asynchronix/src/grpc/services/controller_service.rs
View File

@ -4,12 +4,13 @@ use prost_types::Timestamp;
use crate::grpc::key_registry::{KeyRegistry, KeyRegistryId};
use crate::registry::{EventSourceRegistry, QuerySourceRegistry};
use crate::simulation::Simulation;
use crate::simulation::{Scheduler, Simulation};
use super::super::codegen::simulation::*;
use super::{
map_execution_error, monotonic_to_timestamp, simulation_not_started_error,
timestamp_to_monotonic, to_error, to_positive_duration, to_strictly_positive_duration,
map_execution_error, map_scheduling_error, monotonic_to_timestamp,
simulation_not_started_error, timestamp_to_monotonic, to_error, to_positive_duration,
to_strictly_positive_duration,
};
/// Protobuf-based simulation manager.
@ -24,6 +25,7 @@ pub(crate) enum ControllerService {
NotStarted,
Started {
simulation: Simulation,
scheduler: Scheduler,
event_source_registry: EventSourceRegistry,
query_source_registry: QuerySourceRegistry,
key_registry: KeyRegistry,
@ -147,6 +149,7 @@ impl ControllerService {
let reply = match self {
Self::Started {
simulation,
scheduler,
event_source_registry,
key_registry,
..
@ -224,7 +227,9 @@ impl ControllerService {
}
});
simulation.process(action).map_err(map_execution_error)?;
scheduler
.schedule(deadline, action)
.map_err(map_scheduling_error)?;
Ok(key_id)
}(),

7
asynchronix/src/grpc/services/init_service.rs
View File

@ -2,8 +2,7 @@ use ciborium;
use serde::de::DeserializeOwned;
use crate::registry::EndpointRegistry;
use crate::simulation::SimInit;
use crate::simulation::Simulation;
use crate::simulation::{Scheduler, SimInit, Simulation};
use super::{map_execution_error, timestamp_to_monotonic, to_error};
@ -51,7 +50,7 @@ impl InitService {
pub(crate) fn init(
&mut self,
request: InitRequest,
) -> (InitReply, Option<(Simulation, EndpointRegistry)>) {
) -> (InitReply, Option<(Simulation, Scheduler, EndpointRegistry)>) {
let start_time = request.time.unwrap_or_default();
let reply = (self.sim_gen)(&request.cfg)
@ -73,7 +72,7 @@ impl InitService {
sim_init
.init(start_time)
.map_err(map_execution_error)
.map(|sim| (sim, registry))
.map(|(sim, sched)| (sim, sched, registry))
})
});

3
asynchronix/src/lib.rs
View File

@ -325,7 +325,8 @@
//! # .add_model(multiplier2, multiplier2_mbox, "multiplier2")
//! # .add_model(delay1, delay1_mbox, "delay1")
//! # .add_model(delay2, delay2_mbox, "delay2")
//! # .init(t0)?;
//! # .init(t0)?
//! # .0;
//! // Send a value to the first multiplier.
//! simu.process_event(Multiplier::input, 21.0, &input_address)?;
//!

17
asynchronix/src/simulation.rs
View File

@ -107,7 +107,7 @@
//! # impl Model for ModelB {};
//! # let modelA_addr = Mailbox::<ModelA>::new().address();
//! # let modelB_addr = Mailbox::<ModelB>::new().address();
//! # let mut simu = SimInit::new().init(MonotonicTime::EPOCH)?;
//! # let mut simu = SimInit::new().init(MonotonicTime::EPOCH)?.0;
//! simu.process_event(
//! |m: &mut ModelA| {
//! m.output.connect(ModelB::input, modelB_addr);
@ -164,11 +164,11 @@ thread_local! { pub(crate) static CURRENT_MODEL_ID: Cell<ModelId> = const { Cell
///
/// A [`Simulation`] object also manages an event scheduling queue and
/// simulation time. The scheduling queue can be accessed from the simulation
/// itself, but also from models via the optional
/// [`&mut Context`](crate::model::Context) argument of input and replier port
/// methods. Likewise, simulation time can be accessed with the
/// [`Simulation::time()`] method, or from models with the
/// [`LocalScheduler::time()`](crate::simulation::LocalScheduler::time) method.
/// itself, but also from models via the optional [`&mut
/// Context`](crate::model::Context) argument of input and replier port methods.
/// Likewise, simulation time can be accessed with the [`Simulation::time()`]
/// method, or from models with the
/// [`Context::time()`](crate::simulation::Context::time) method.
///
/// Events and queries can be scheduled immediately, *i.e.* for the current
/// simulation time, using [`process_event()`](Simulation::process_event) and
@ -276,11 +276,6 @@ impl Simulation {
self.step_until_unchecked(target_time)
}
/// Returns an owned scheduler handle.
pub fn scheduler(&self) -> Scheduler {
Scheduler::new(self.scheduler_queue.clone(), self.time.reader())
}
/// Processes an action immediately, blocking until completion.
///
/// Simulation time remains unchanged. The periodicity of the action, if

8
asynchronix/src/simulation/scheduler.rs
View File

@ -69,8 +69,6 @@ impl Scheduler {
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`LocalScheduler::schedule_event`](LocalScheduler::schedule_event).
pub fn schedule_event<M, F, T, S>(
&self,
deadline: impl Deadline,
@ -95,8 +93,6 @@ impl Scheduler {
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`LocalScheduler::schedule_keyed_event`](LocalScheduler::schedule_keyed_event).
pub fn schedule_keyed_event<M, F, T, S>(
&self,
deadline: impl Deadline,
@ -121,8 +117,6 @@ impl Scheduler {
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`LocalScheduler::schedule_periodic_event`](LocalScheduler::schedule_periodic_event).
pub fn schedule_periodic_event<M, F, T, S>(
&self,
deadline: impl Deadline,
@ -155,8 +149,6 @@ impl Scheduler {
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`LocalScheduler::schedule_keyed_periodic_event`](LocalScheduler::schedule_keyed_periodic_event).
pub fn schedule_keyed_periodic_event<M, F, T, S>(
&self,
deadline: impl Deadline,

14
asynchronix/src/simulation/sim_init.rs
View File

@ -11,7 +11,8 @@ use crate::util::priority_queue::PriorityQueue;
use crate::util::sync_cell::SyncCell;
use super::{
add_model, ExecutionError, Mailbox, GlobalScheduler, SchedulerQueue, Signal, Simulation,
add_model, ExecutionError, GlobalScheduler, Mailbox, Scheduler, SchedulerQueue, Signal,
Simulation,
};
/// Builder for a multi-threaded, discrete-event simulation.
@ -146,10 +147,17 @@ impl SimInit {
/// Builds a simulation initialized at the specified simulation time,
/// executing the [`Model::init()`](crate::model::Model::init) method on all
/// model initializers.
pub fn init(mut self, start_time: MonotonicTime) -> Result<Simulation, ExecutionError> {
///
/// The simulation object and its associated scheduler are returned upon
/// success.
pub fn init(
mut self,
start_time: MonotonicTime,
) -> Result<(Simulation, Scheduler), ExecutionError> {
self.time.write(start_time);
self.clock.synchronize(start_time);
let scheduler = Scheduler::new(self.scheduler_queue.clone(), self.time.reader());
let mut simulation = Simulation::new(
self.executor,
self.scheduler_queue,
@ -162,7 +170,7 @@ impl SimInit {
);
simulation.run()?;
Ok(simulation)
Ok((simulation, scheduler))
}
}

15
asynchronix/tests/integration/model_scheduling.rs
View File

@ -36,7 +36,8 @@ fn model_schedule_event(num_threads: usize) {
let mut simu = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "")
.init(t0)
.unwrap();
.unwrap()
.0;
simu.process_event(TestModel::trigger, (), addr).unwrap();
simu.step().unwrap();
@ -82,7 +83,8 @@ fn model_cancel_future_keyed_event(num_threads: usize) {
let mut simu = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "")
.init(t0)
.unwrap();
.unwrap()
.0;
simu.process_event(TestModel::trigger, (), addr).unwrap();
simu.step().unwrap();
@ -129,7 +131,8 @@ fn model_cancel_same_time_keyed_event(num_threads: usize) {
let mut simu = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "")
.init(t0)
.unwrap();
.unwrap()
.0;
simu.process_event(TestModel::trigger, (), addr).unwrap();
simu.step().unwrap();
@ -172,7 +175,8 @@ fn model_schedule_periodic_event(num_threads: usize) {
let mut simu = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "")
.init(t0)
.unwrap();
.unwrap()
.0;
simu.process_event(TestModel::trigger, (), addr).unwrap();
@ -224,7 +228,8 @@ fn model_cancel_periodic_event(num_threads: usize) {
let mut simu = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "")
.init(t0)
.unwrap();
.unwrap()
.0;
simu.process_event(TestModel::trigger, (), addr).unwrap();

3
asynchronix/tests/integration/simulation_clock_sync.rs
View File

@ -39,14 +39,13 @@ fn clock_sync(
let addr = mbox.address();
let t0 = MonotonicTime::EPOCH;
let mut simu = SimInit::with_num_threads(num_threads)
let (mut simu, scheduler) = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "test")
.set_clock(clock)
.set_clock_tolerance(clock_tolerance)
.init(t0)
.unwrap();
let scheduler = simu.scheduler();
let mut delta = Duration::ZERO;
for tick_ms in ticks_ms {
let tick = Duration::from_millis(*tick_ms);

12
asynchronix/tests/integration/simulation_deadlock.rs
View File

@ -45,7 +45,8 @@ fn deadlock_on_mailbox_overflow(num_threads: usize) {
let mut simu = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, MODEL_NAME)
.init(t0)
.unwrap();
.unwrap()
.0;
match simu.process_event(TestModel::activate_output, (), addr) {
Err(ExecutionError::Deadlock(deadlock_info)) => {
@ -80,7 +81,8 @@ fn deadlock_on_query_loopback(num_threads: usize) {
let mut simu = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, MODEL_NAME)
.init(t0)
.unwrap();
.unwrap()
.0;
match simu.process_query(TestModel::activate_requestor, (), addr) {
Err(ExecutionError::Deadlock(deadlock_info)) => {
@ -124,7 +126,8 @@ fn deadlock_on_transitive_query_loopback(num_threads: usize) {
.add_model(model1, mbox1, MODEL1_NAME)
.add_model(model2, mbox2, MODEL2_NAME)
.init(t0)
.unwrap();
.unwrap()
.0;
match simu.process_query(TestModel::activate_requestor, (), addr1) {
Err(ExecutionError::Deadlock(deadlock_info)) => {
@ -181,7 +184,8 @@ fn deadlock_on_multiple_query_loopback(num_threads: usize) {
.add_model(model1, mbox1, MODEL1_NAME)
.add_model(model2, mbox2, MODEL2_NAME)
.init(t0)
.unwrap();
.unwrap()
.0;
match simu.process_query(TestModel::activate_requestor, (), addr0) {
Err(ExecutionError::Deadlock(deadlock_info)) => {

2
asynchronix/tests/integration/simulation_panic.rs
View File

@ -48,7 +48,7 @@ fn model_panic(num_threads: usize) {
// Run the simulation.
let t0 = MonotonicTime::EPOCH;
let mut simu = siminit.init(t0).unwrap();
let mut simu = siminit.init(t0).unwrap().0;
match simu.process_event(TestModel::countdown_in, INIT_COUNTDOWN, addr0) {
Err(ExecutionError::Panic { model, payload }) => {

47
asynchronix/tests/integration/simulation_scheduling.rs
View File

@ -6,7 +6,7 @@ use std::time::Duration;
use asynchronix::model::Context;
use asynchronix::model::Model;
use asynchronix::ports::{EventBuffer, Output};
use asynchronix::simulation::{Address, Mailbox, SimInit, Simulation};
use asynchronix::simulation::{Address, Mailbox, Scheduler, SimInit, Simulation};
use asynchronix::time::MonotonicTime;
const MT_NUM_THREADS: usize = 4;
@ -32,7 +32,12 @@ impl<T: Clone + Send + 'static> Model for PassThroughModel<T> {}
fn passthrough_bench<T: Clone + Send + 'static>(
num_threads: usize,
t0: MonotonicTime,
) -> (Simulation, Address<PassThroughModel<T>>, EventBuffer<T>) {
) -> (
Simulation,
Scheduler,
Address<PassThroughModel<T>>,
EventBuffer<T>,
) {
// Bench assembly.
let mut model = PassThroughModel::new();
let mbox = Mailbox::new();
@ -41,19 +46,17 @@ fn passthrough_bench<T: Clone + Send + 'static>(
model.output.connect_sink(&out_stream);
let addr = mbox.address();
let simu = SimInit::with_num_threads(num_threads)
let (simu, scheduler) = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "")
.init(t0)
.unwrap();
(simu, addr, out_stream)
(simu, scheduler, addr, out_stream)
}
fn schedule_events(num_threads: usize) {
let t0 = MonotonicTime::EPOCH;
let (mut simu, addr, mut output) = passthrough_bench(num_threads, t0);
let scheduler = simu.scheduler();
let (mut simu, scheduler, addr, mut output) = passthrough_bench(num_threads, t0);
// Queue 2 events at t0+3s and t0+2s, in reverse order.
scheduler
@ -92,9 +95,7 @@ fn schedule_events(num_threads: usize) {
fn schedule_keyed_events(num_threads: usize) {
let t0 = MonotonicTime::EPOCH;
let (mut simu, addr, mut output) = passthrough_bench(num_threads, t0);
let scheduler = simu.scheduler();
let (mut simu, scheduler, addr, mut output) = passthrough_bench(num_threads, t0);
let event_t1 = scheduler
.schedule_keyed_event(
@ -133,9 +134,7 @@ fn schedule_keyed_events(num_threads: usize) {
fn schedule_periodic_events(num_threads: usize) {
let t0 = MonotonicTime::EPOCH;
let (mut simu, addr, mut output) = passthrough_bench(num_threads, t0);
let scheduler = simu.scheduler();
let (mut simu, scheduler, addr, mut output) = passthrough_bench(num_threads, t0);
// Queue 2 periodic events at t0 + 3s + k*2s.
scheduler
@ -172,9 +171,7 @@ fn schedule_periodic_events(num_threads: usize) {
fn schedule_periodic_keyed_events(num_threads: usize) {
let t0 = MonotonicTime::EPOCH;
let (mut simu, addr, mut output) = passthrough_bench(num_threads, t0);
let scheduler = simu.scheduler();
let (mut simu, scheduler, addr, mut output) = passthrough_bench(num_threads, t0);
// Queue 2 periodic events at t0 + 3s + k*2s.
scheduler
@ -292,6 +289,7 @@ fn timestamp_bench(
clock: impl Clock + 'static,
) -> (
Simulation,
Scheduler,
Address<TimestampModel>,
EventBuffer<(Instant, SystemTime)>,
) {
@ -303,13 +301,13 @@ fn timestamp_bench(
model.stamp.connect_sink(&stamp_stream);
let addr = mbox.address();
let simu = SimInit::with_num_threads(num_threads)
let (simu, scheduler) = SimInit::with_num_threads(num_threads)
.add_model(model, mbox, "")
.set_clock(clock)
.init(t0)
.unwrap();
(simu, addr, stamp_stream)
(simu, scheduler, addr, stamp_stream)
}
#[cfg(not(miri))]
@ -335,9 +333,7 @@ fn system_clock_from_instant(num_threads: usize) {
let clock = SystemClock::from_instant(simulation_ref, wall_clock_ref);
let (mut simu, addr, mut stamp) = timestamp_bench(num_threads, t0, clock);
let scheduler = simu.scheduler();
let (mut simu, scheduler, addr, mut stamp) = timestamp_bench(num_threads, t0, clock);
// Queue a single event at t0 + 0.1s.
scheduler
@ -391,9 +387,7 @@ fn system_clock_from_system_time(num_threads: usize) {
let clock = SystemClock::from_system_time(simulation_ref, wall_clock_ref);
let (mut simu, addr, mut stamp) = timestamp_bench(num_threads, t0, clock);
let scheduler = simu.scheduler();
let (mut simu, scheduler, addr, mut stamp) = timestamp_bench(num_threads, t0, clock);
// Queue a single event at t0 + 0.1s.
scheduler
@ -435,11 +429,10 @@ fn auto_system_clock(num_threads: usize) {
let t0 = MonotonicTime::EPOCH;
const TOLERANCE: f64 = 0.005; // [s]
let (mut simu, addr, mut stamp) = timestamp_bench(num_threads, t0, AutoSystemClock::new());
let (mut simu, scheduler, addr, mut stamp) =
timestamp_bench(num_threads, t0, AutoSystemClock::new());
let instant_t0 = Instant::now();
let scheduler = simu.scheduler();
// Queue a periodic event at t0 + 0.2s + k*0.2s.
scheduler
.schedule_periodic_event(

6
asynchronix/tests/integration/simulation_timeout.rs
View File

@ -51,7 +51,8 @@ fn timeout_untriggered(num_threads: usize) {
.add_model(model, mbox, "test")
.set_timeout(Duration::from_secs(1))
.init(t0)
.unwrap();
.unwrap()
.0;
assert!(simu.process_event(TestModel::input, (), addr).is_ok());
}
@ -69,7 +70,8 @@ fn timeout_triggered(num_threads: usize) {
.add_model(model, mbox, "test")
.set_timeout(Duration::from_secs(1))
.init(t0)
.unwrap();
.unwrap()
.0;
assert!(matches!(
simu.process_event(TestModel::input, (), addr),