Add support for simulation timeouts

asynchronix/tests/integration/main.rs

@@ -0,0 +1,8 @@
+// Integration tests follow the organization suggested by Matklad:
+// https://matklad.github.io/2021/02/27/delete-cargo-integration-tests.html
+
+mod model_scheduling;
+mod simulation_deadlock;
+mod simulation_scheduling;
+#[cfg(not(miri))]
+mod simulation_timeout;

asynchronix/tests/integration/model_scheduling.rs

@@ -0,0 +1,320 @@
+//! Event scheduling within `Model` input methods.
+
+use std::time::Duration;
+
+use asynchronix::model::{Context, Model};
+use asynchronix::ports::{EventBuffer, Output};
+use asynchronix::simulation::{ActionKey, Mailbox, SimInit};
+use asynchronix::time::MonotonicTime;
+
+const MT_NUM_THREADS: usize = 4;
+
+fn model_schedule_event(num_threads: usize) {
+    #[derive(Default)]
+    struct TestModel {
+        output: Output<()>,
+    }
+    impl TestModel {
+        fn trigger(&mut self, _: (), context: &Context<Self>) {
+            context
+                .scheduler
+                .schedule_event(
+                    context.scheduler.time() + Duration::from_secs(2),
+                    Self::action,
+                    (),
+                )
+                .unwrap();
+        }
+        async fn action(&mut self) {
+            self.output.send(()).await;
+        }
+    }
+    impl Model for TestModel {}
+
+    let mut model = TestModel::default();
+    let mbox = Mailbox::new();
+
+    let mut output = EventBuffer::new();
+    model.output.connect_sink(&output);
+    let addr = mbox.address();
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "")
+        .init(t0)
+        .unwrap();
+
+    simu.process_event(TestModel::trigger, (), addr).unwrap();
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(2));
+    assert!(output.next().is_some());
+    simu.step().unwrap();
+    assert!(output.next().is_none());
+}
+
+fn model_cancel_future_keyed_event(num_threads: usize) {
+    #[derive(Default)]
+    struct TestModel {
+        output: Output<i32>,
+        key: Option<ActionKey>,
+    }
+    impl TestModel {
+        fn trigger(&mut self, _: (), context: &Context<Self>) {
+            context
+                .scheduler
+                .schedule_event(
+                    context.scheduler.time() + Duration::from_secs(1),
+                    Self::action1,
+                    (),
+                )
+                .unwrap();
+            self.key = context
+                .scheduler
+                .schedule_keyed_event(
+                    context.scheduler.time() + Duration::from_secs(2),
+                    Self::action2,
+                    (),
+                )
+                .ok();
+        }
+        async fn action1(&mut self) {
+            self.output.send(1).await;
+            // Cancel the call to `action2`.
+            self.key.take().unwrap().cancel();
+        }
+        async fn action2(&mut self) {
+            self.output.send(2).await;
+        }
+    }
+    impl Model for TestModel {}
+
+    let mut model = TestModel::default();
+    let mbox = Mailbox::new();
+
+    let mut output = EventBuffer::new();
+    model.output.connect_sink(&output);
+    let addr = mbox.address();
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "")
+        .init(t0)
+        .unwrap();
+
+    simu.process_event(TestModel::trigger, (), addr).unwrap();
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(1));
+    assert_eq!(output.next(), Some(1));
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(1));
+    assert!(output.next().is_none());
+}
+
+fn model_cancel_same_time_keyed_event(num_threads: usize) {
+    #[derive(Default)]
+    struct TestModel {
+        output: Output<i32>,
+        key: Option<ActionKey>,
+    }
+    impl TestModel {
+        fn trigger(&mut self, _: (), context: &Context<Self>) {
+            context
+                .scheduler
+                .schedule_event(
+                    context.scheduler.time() + Duration::from_secs(2),
+                    Self::action1,
+                    (),
+                )
+                .unwrap();
+            self.key = context
+                .scheduler
+                .schedule_keyed_event(
+                    context.scheduler.time() + Duration::from_secs(2),
+                    Self::action2,
+                    (),
+                )
+                .ok();
+        }
+        async fn action1(&mut self) {
+            self.output.send(1).await;
+            // Cancel the call to `action2`.
+            self.key.take().unwrap().cancel();
+        }
+        async fn action2(&mut self) {
+            self.output.send(2).await;
+        }
+    }
+    impl Model for TestModel {}
+
+    let mut model = TestModel::default();
+    let mbox = Mailbox::new();
+
+    let mut output = EventBuffer::new();
+    model.output.connect_sink(&output);
+    let addr = mbox.address();
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "")
+        .init(t0)
+        .unwrap();
+
+    simu.process_event(TestModel::trigger, (), addr).unwrap();
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(2));
+    assert_eq!(output.next(), Some(1));
+    assert!(output.next().is_none());
+    simu.step().unwrap();
+    assert!(output.next().is_none());
+}
+
+fn model_schedule_periodic_event(num_threads: usize) {
+    #[derive(Default)]
+    struct TestModel {
+        output: Output<i32>,
+    }
+    impl TestModel {
+        fn trigger(&mut self, _: (), context: &Context<Self>) {
+            context
+                .scheduler
+                .schedule_periodic_event(
+                    context.scheduler.time() + Duration::from_secs(2),
+                    Duration::from_secs(3),
+                    Self::action,
+                    42,
+                )
+                .unwrap();
+        }
+        async fn action(&mut self, payload: i32) {
+            self.output.send(payload).await;
+        }
+    }
+    impl Model for TestModel {}
+
+    let mut model = TestModel::default();
+    let mbox = Mailbox::new();
+
+    let mut output = EventBuffer::new();
+    model.output.connect_sink(&output);
+    let addr = mbox.address();
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "")
+        .init(t0)
+        .unwrap();
+
+    simu.process_event(TestModel::trigger, (), addr).unwrap();
+
+    // Move to the next events at t0 + 2s + k*3s.
+    for k in 0..10 {
+        simu.step().unwrap();
+        assert_eq!(
+            simu.time(),
+            t0 + Duration::from_secs(2) + k * Duration::from_secs(3)
+        );
+        assert_eq!(output.next(), Some(42));
+        assert!(output.next().is_none());
+    }
+}
+
+fn model_cancel_periodic_event(num_threads: usize) {
+    #[derive(Default)]
+    struct TestModel {
+        output: Output<()>,
+        key: Option<ActionKey>,
+    }
+    impl TestModel {
+        fn trigger(&mut self, _: (), context: &Context<Self>) {
+            self.key = context
+                .scheduler
+                .schedule_keyed_periodic_event(
+                    context.scheduler.time() + Duration::from_secs(2),
+                    Duration::from_secs(3),
+                    Self::action,
+                    (),
+                )
+                .ok();
+        }
+        async fn action(&mut self) {
+            self.output.send(()).await;
+            // Cancel the next events.
+            self.key.take().unwrap().cancel();
+        }
+    }
+    impl Model for TestModel {}
+
+    let mut model = TestModel::default();
+    let mbox = Mailbox::new();
+
+    let mut output = EventBuffer::new();
+    model.output.connect_sink(&output);
+    let addr = mbox.address();
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "")
+        .init(t0)
+        .unwrap();
+
+    simu.process_event(TestModel::trigger, (), addr).unwrap();
+
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(2));
+    assert!(output.next().is_some());
+    assert!(output.next().is_none());
+
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(2));
+    assert!(output.next().is_none());
+}
+
+#[test]
+fn model_schedule_event_st() {
+    model_schedule_event(1);
+}
+
+#[test]
+fn model_schedule_event_mt() {
+    model_schedule_event(MT_NUM_THREADS);
+}
+
+#[test]
+fn model_cancel_future_keyed_event_st() {
+    model_cancel_future_keyed_event(1);
+}
+
+#[test]
+fn model_cancel_future_keyed_event_mt() {
+    model_cancel_future_keyed_event(MT_NUM_THREADS);
+}
+
+#[test]
+fn model_cancel_same_time_keyed_event_st() {
+    model_cancel_same_time_keyed_event(1);
+}
+
+#[test]
+fn model_cancel_same_time_keyed_event_mt() {
+    model_cancel_same_time_keyed_event(MT_NUM_THREADS);
+}
+
+#[test]
+fn model_schedule_periodic_event_st() {
+    model_schedule_periodic_event(1);
+}
+
+#[test]
+fn model_schedule_periodic_event_mt() {
+    model_schedule_periodic_event(MT_NUM_THREADS);
+}
+
+#[test]
+fn model_cancel_periodic_event_st() {
+    model_cancel_periodic_event(1);
+}
+
+#[test]
+fn model_cancel_periodic_event_mt() {
+    model_cancel_periodic_event(MT_NUM_THREADS);
+}

asynchronix/tests/integration/simulation_deadlock.rs

@@ -0,0 +1,249 @@
+//! Deadlock-detection for model loops.
+
+use asynchronix::model::Model;
+use asynchronix::ports::{Output, Requestor};
+use asynchronix::simulation::{DeadlockInfo, ExecutionError, Mailbox, SimInit};
+use asynchronix::time::MonotonicTime;
+
+const MT_NUM_THREADS: usize = 4;
+
+#[derive(Default)]
+struct TestModel {
+    output: Output<()>,
+    requestor: Requestor<(), ()>,
+}
+impl TestModel {
+    async fn activate_output(&mut self) {
+        self.output.send(()).await;
+    }
+    async fn activate_requestor(&mut self) {
+        let _ = self.requestor.send(()).await;
+    }
+}
+impl Model for TestModel {}
+
+/// Overflows a mailbox by sending 2 messages in loopback for each incoming
+/// message.
+fn deadlock_on_mailbox_overflow(num_threads: usize) {
+    const MODEL_NAME: &str = "testmodel";
+    const MAILBOX_SIZE: usize = 5;
+
+    let mut model = TestModel::default();
+    let mbox = Mailbox::with_capacity(MAILBOX_SIZE);
+    let addr = mbox.address();
+
+    // Make two self-connections so that each outgoing message generates two
+    // incoming messages.
+    model
+        .output
+        .connect(TestModel::activate_output, addr.clone());
+    model
+        .output
+        .connect(TestModel::activate_output, addr.clone());
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, MODEL_NAME)
+        .init(t0)
+        .unwrap();
+
+    match simu.process_event(TestModel::activate_output, (), addr) {
+        Err(ExecutionError::Deadlock(deadlock_info)) => {
+            // We expect only 1 deadlocked model.
+            assert_eq!(deadlock_info.len(), 1);
+            // We expect the mailbox to be full.
+            assert_eq!(
+                deadlock_info[0],
+                DeadlockInfo {
+                    model_name: MODEL_NAME.into(),
+                    mailbox_size: MAILBOX_SIZE
+                }
+            )
+        }
+        _ => panic!("deadlock not detected"),
+    }
+}
+
+/// Generates a deadlock with a query loopback.
+fn deadlock_on_query_loopback(num_threads: usize) {
+    const MODEL_NAME: &str = "testmodel";
+
+    let mut model = TestModel::default();
+    let mbox = Mailbox::new();
+    let addr = mbox.address();
+
+    model
+        .requestor
+        .connect(TestModel::activate_requestor, addr.clone());
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, MODEL_NAME)
+        .init(t0)
+        .unwrap();
+
+    match simu.process_query(TestModel::activate_requestor, (), addr) {
+        Err(ExecutionError::Deadlock(deadlock_info)) => {
+            // We expect only 1 deadlocked model.
+            assert_eq!(deadlock_info.len(), 1);
+            // We expect the mailbox to have a single query.
+            assert_eq!(
+                deadlock_info[0],
+                DeadlockInfo {
+                    model_name: MODEL_NAME.into(),
+                    mailbox_size: 1,
+                }
+            );
+        }
+        _ => panic!("deadlock not detected"),
+    }
+}
+
+/// Generates a deadlock with a query loopback involving several models.
+fn deadlock_on_transitive_query_loopback(num_threads: usize) {
+    const MODEL1_NAME: &str = "testmodel1";
+    const MODEL2_NAME: &str = "testmodel2";
+
+    let mut model1 = TestModel::default();
+    let mut model2 = TestModel::default();
+    let mbox1 = Mailbox::new();
+    let mbox2 = Mailbox::new();
+    let addr1 = mbox1.address();
+    let addr2 = mbox2.address();
+
+    model1
+        .requestor
+        .connect(TestModel::activate_requestor, addr2);
+
+    model2
+        .requestor
+        .connect(TestModel::activate_requestor, addr1.clone());
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model1, mbox1, MODEL1_NAME)
+        .add_model(model2, mbox2, MODEL2_NAME)
+        .init(t0)
+        .unwrap();
+
+    match simu.process_query(TestModel::activate_requestor, (), addr1) {
+        Err(ExecutionError::Deadlock(deadlock_info)) => {
+            // We expect only 1 deadlocked model.
+            assert_eq!(deadlock_info.len(), 1);
+            // We expect the mailbox of this model to have a single query.
+            assert_eq!(
+                deadlock_info[0],
+                DeadlockInfo {
+                    model_name: MODEL1_NAME.into(),
+                    mailbox_size: 1,
+                }
+            );
+        }
+        _ => panic!("deadlock not detected"),
+    }
+}
+
+/// Generates deadlocks with query loopbacks on several models at the same time.
+fn deadlock_on_multiple_query_loopback(num_threads: usize) {
+    const MODEL0_NAME: &str = "testmodel0";
+    const MODEL1_NAME: &str = "testmodel1";
+    const MODEL2_NAME: &str = "testmodel2";
+
+    let mut model0 = TestModel::default();
+    let mut model1 = TestModel::default();
+    let mut model2 = TestModel::default();
+    let mbox0 = Mailbox::new();
+    let mbox1 = Mailbox::new();
+    let mbox2 = Mailbox::new();
+    let addr0 = mbox0.address();
+    let addr1 = mbox1.address();
+    let addr2 = mbox2.address();
+
+    model0
+        .requestor
+        .connect(TestModel::activate_requestor, addr1.clone());
+
+    model0
+        .requestor
+        .connect(TestModel::activate_requestor, addr2.clone());
+
+    model1
+        .requestor
+        .connect(TestModel::activate_requestor, addr1);
+
+    model2
+        .requestor
+        .connect(TestModel::activate_requestor, addr2);
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model0, mbox0, MODEL0_NAME)
+        .add_model(model1, mbox1, MODEL1_NAME)
+        .add_model(model2, mbox2, MODEL2_NAME)
+        .init(t0)
+        .unwrap();
+
+    match simu.process_query(TestModel::activate_requestor, (), addr0) {
+        Err(ExecutionError::Deadlock(deadlock_info)) => {
+            // We expect 2 deadlocked models.
+            assert_eq!(deadlock_info.len(), 2);
+            // We expect the mailbox of each deadlocked model to have a single
+            // query.
+            assert_eq!(
+                deadlock_info[0],
+                DeadlockInfo {
+                    model_name: MODEL1_NAME.into(),
+                    mailbox_size: 1,
+                }
+            );
+            assert_eq!(
+                deadlock_info[1],
+                DeadlockInfo {
+                    model_name: MODEL2_NAME.into(),
+                    mailbox_size: 1,
+                }
+            );
+        }
+        _ => panic!("deadlock not detected"),
+    }
+}
+
+#[test]
+fn deadlock_on_mailbox_overflow_st() {
+    deadlock_on_mailbox_overflow(1);
+}
+
+#[test]
+fn deadlock_on_mailbox_overflow_mt() {
+    deadlock_on_mailbox_overflow(MT_NUM_THREADS);
+}
+
+#[test]
+fn deadlock_on_query_loopback_st() {
+    deadlock_on_query_loopback(1);
+}
+
+#[test]
+fn deadlock_on_query_loopback_mt() {
+    deadlock_on_query_loopback(MT_NUM_THREADS);
+}
+
+#[test]
+fn deadlock_on_transitive_query_loopback_st() {
+    deadlock_on_transitive_query_loopback(1);
+}
+
+#[test]
+fn deadlock_on_transitive_query_loopback_mt() {
+    deadlock_on_transitive_query_loopback(MT_NUM_THREADS);
+}
+
+#[test]
+fn deadlock_on_multiple_query_loopback_st() {
+    deadlock_on_multiple_query_loopback(1);
+}
+
+#[test]
+fn deadlock_on_multiple_query_loopback_mt() {
+    deadlock_on_multiple_query_loopback(MT_NUM_THREADS);
+}

asynchronix/tests/integration/simulation_scheduling.rs

@@ -0,0 +1,513 @@
+//! Event scheduling from a `Simulation` instance.
+
+use std::time::Duration;
+
+#[cfg(not(miri))]
+use asynchronix::model::Context;
+use asynchronix::model::Model;
+use asynchronix::ports::{EventBuffer, Output};
+use asynchronix::simulation::{Address, Mailbox, SimInit, Simulation};
+use asynchronix::time::MonotonicTime;
+
+const MT_NUM_THREADS: usize = 4;
+
+// Input-to-output pass-through model.
+struct PassThroughModel<T: Clone + Send + 'static> {
+    pub output: Output<T>,
+}
+impl<T: Clone + Send + 'static> PassThroughModel<T> {
+    pub fn new() -> Self {
+        Self {
+            output: Output::default(),
+        }
+    }
+    pub async fn input(&mut self, arg: T) {
+        self.output.send(arg).await;
+    }
+}
+impl<T: Clone + Send + 'static> Model for PassThroughModel<T> {}
+
+/// A simple bench containing a single pass-through model (input forwarded to
+/// output) running as fast as possible.
+fn passthrough_bench<T: Clone + Send + 'static>(
+    num_threads: usize,
+    t0: MonotonicTime,
+) -> (Simulation, Address<PassThroughModel<T>>, EventBuffer<T>) {
+    // Bench assembly.
+    let mut model = PassThroughModel::new();
+    let mbox = Mailbox::new();
+
+    let out_stream = EventBuffer::new();
+    model.output.connect_sink(&out_stream);
+    let addr = mbox.address();
+
+    let simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "")
+        .init(t0)
+        .unwrap();
+
+    (simu, 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();
+
+    // Queue 2 events at t0+3s and t0+2s, in reverse order.
+    scheduler
+        .schedule_event(Duration::from_secs(3), PassThroughModel::input, (), &addr)
+        .unwrap();
+    scheduler
+        .schedule_event(
+            t0 + Duration::from_secs(2),
+            PassThroughModel::input,
+            (),
+            &addr,
+        )
+        .unwrap();
+
+    // Move to the 1st event at t0+2s.
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(2));
+    assert!(output.next().is_some());
+
+    // Schedule another event in 4s (at t0+6s).
+    scheduler
+        .schedule_event(Duration::from_secs(4), PassThroughModel::input, (), &addr)
+        .unwrap();
+
+    // Move to the 2nd event at t0+3s.
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(3));
+    assert!(output.next().is_some());
+
+    // Move to the 3rd event at t0+6s.
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(6));
+    assert!(output.next().is_some());
+    assert!(output.next().is_none());
+}
+
+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 event_t1 = scheduler
+        .schedule_keyed_event(
+            t0 + Duration::from_secs(1),
+            PassThroughModel::input,
+            1,
+            &addr,
+        )
+        .unwrap();
+
+    let event_t2_1 = scheduler
+        .schedule_keyed_event(Duration::from_secs(2), PassThroughModel::input, 21, &addr)
+        .unwrap();
+
+    scheduler
+        .schedule_event(Duration::from_secs(2), PassThroughModel::input, 22, &addr)
+        .unwrap();
+
+    // Move to the 1st event at t0+1.
+    simu.step().unwrap();
+
+    // Try to cancel the 1st event after it has already taken place and check
+    // that the cancellation had no effect.
+    event_t1.cancel();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(1));
+    assert_eq!(output.next(), Some(1));
+
+    // Cancel the second event (t0+2) before it is meant to takes place and
+    // check that we move directly to the 3rd event.
+    event_t2_1.cancel();
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(2));
+    assert_eq!(output.next(), Some(22));
+    assert!(output.next().is_none());
+}
+
+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();
+
+    // Queue 2 periodic events at t0 + 3s + k*2s.
+    scheduler
+        .schedule_periodic_event(
+            Duration::from_secs(3),
+            Duration::from_secs(2),
+            PassThroughModel::input,
+            1,
+            &addr,
+        )
+        .unwrap();
+    scheduler
+        .schedule_periodic_event(
+            t0 + Duration::from_secs(3),
+            Duration::from_secs(2),
+            PassThroughModel::input,
+            2,
+            &addr,
+        )
+        .unwrap();
+
+    // Move to the next events at t0 + 3s + k*2s.
+    for k in 0..10 {
+        simu.step().unwrap();
+        assert_eq!(
+            simu.time(),
+            t0 + Duration::from_secs(3) + k * Duration::from_secs(2)
+        );
+        assert_eq!(output.next(), Some(1));
+        assert_eq!(output.next(), Some(2));
+        assert!(output.next().is_none());
+    }
+}
+
+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();
+
+    // Queue 2 periodic events at t0 + 3s + k*2s.
+    scheduler
+        .schedule_periodic_event(
+            Duration::from_secs(3),
+            Duration::from_secs(2),
+            PassThroughModel::input,
+            1,
+            &addr,
+        )
+        .unwrap();
+    let event2_key = scheduler
+        .schedule_keyed_periodic_event(
+            t0 + Duration::from_secs(3),
+            Duration::from_secs(2),
+            PassThroughModel::input,
+            2,
+            &addr,
+        )
+        .unwrap();
+
+    // Move to the next event at t0+3s.
+    simu.step().unwrap();
+    assert_eq!(simu.time(), t0 + Duration::from_secs(3));
+    assert_eq!(output.next(), Some(1));
+    assert_eq!(output.next(), Some(2));
+    assert!(output.next().is_none());
+
+    // Cancel the second event.
+    event2_key.cancel();
+
+    // Move to the next events at t0 + 3s + k*2s.
+    for k in 1..10 {
+        simu.step().unwrap();
+        assert_eq!(
+            simu.time(),
+            t0 + Duration::from_secs(3) + k * Duration::from_secs(2)
+        );
+        assert_eq!(output.next(), Some(1));
+        assert!(output.next().is_none());
+    }
+}
+
+#[test]
+fn schedule_events_st() {
+    schedule_events(1);
+}
+
+#[test]
+fn schedule_events_mt() {
+    schedule_events(MT_NUM_THREADS);
+}
+
+#[test]
+fn schedule_keyed_events_st() {
+    schedule_keyed_events(1);
+}
+
+#[test]
+fn schedule_keyed_events_mt() {
+    schedule_keyed_events(MT_NUM_THREADS);
+}
+
+#[test]
+fn schedule_periodic_events_st() {
+    schedule_periodic_events(1);
+}
+
+#[test]
+fn schedule_periodic_events_mt() {
+    schedule_periodic_events(MT_NUM_THREADS);
+}
+
+#[test]
+fn schedule_periodic_keyed_events_st() {
+    schedule_periodic_keyed_events(1);
+}
+
+#[test]
+fn schedule_periodic_keyed_events_mt() {
+    schedule_periodic_keyed_events(MT_NUM_THREADS);
+}
+
+#[cfg(not(miri))]
+use std::time::{Instant, SystemTime};
+
+#[cfg(not(miri))]
+use asynchronix::time::{AutoSystemClock, Clock, SystemClock};
+
+// Model that outputs timestamps at init and each time its input is triggered.
+#[cfg(not(miri))]
+#[derive(Default)]
+struct TimestampModel {
+    pub stamp: Output<(Instant, SystemTime)>,
+}
+#[cfg(not(miri))]
+impl TimestampModel {
+    pub async fn trigger(&mut self) {
+        self.stamp.send((Instant::now(), SystemTime::now())).await;
+    }
+}
+#[cfg(not(miri))]
+impl Model for TimestampModel {
+    async fn init(mut self, _: &Context<Self>) -> asynchronix::model::InitializedModel<Self> {
+        self.stamp.send((Instant::now(), SystemTime::now())).await;
+        self.into()
+    }
+}
+
+/// A simple bench containing a single timestamping model with a custom clock.
+#[cfg(not(miri))]
+fn timestamp_bench(
+    num_threads: usize,
+    t0: MonotonicTime,
+    clock: impl Clock + 'static,
+) -> (
+    Simulation,
+    Address<TimestampModel>,
+    EventBuffer<(Instant, SystemTime)>,
+) {
+    // Bench assembly.
+    let mut model = TimestampModel::default();
+    let mbox = Mailbox::new();
+
+    let stamp_stream = EventBuffer::new();
+    model.stamp.connect_sink(&stamp_stream);
+    let addr = mbox.address();
+
+    let simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "")
+        .set_clock(clock)
+        .init(t0)
+        .unwrap();
+
+    (simu, addr, stamp_stream)
+}
+
+#[cfg(not(miri))]
+fn system_clock_from_instant(num_threads: usize) {
+    let t0 = MonotonicTime::EPOCH;
+    const TOLERANCE: f64 = 0.005; // [s]
+
+    // The reference simulation time is set in the past of t0 so that the
+    // simulation starts in the future when the reference wall clock time is
+    // close to the wall clock time when the simulation in initialized.
+    let simulation_ref_offset = 0.3; // [s] must be greater than any `instant_offset`.
+    let simulation_ref = t0 - Duration::from_secs_f64(simulation_ref_offset);
+
+    // Test reference wall clock times in the near past and near future.
+    for wall_clock_offset in [-0.1, 0.1] {
+        // The clock reference is the current time offset by `instant_offset`.
+        let wall_clock_init = Instant::now();
+        let wall_clock_ref = if wall_clock_offset >= 0.0 {
+            wall_clock_init + Duration::from_secs_f64(wall_clock_offset)
+        } else {
+            wall_clock_init - Duration::from_secs_f64(-wall_clock_offset)
+        };
+
+        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();
+
+        // Queue a single event at t0 + 0.1s.
+        scheduler
+            .schedule_event(
+                Duration::from_secs_f64(0.1),
+                TimestampModel::trigger,
+                (),
+                &addr,
+            )
+            .unwrap();
+
+        // Check the stamps.
+        for expected_time in [
+            simulation_ref_offset + wall_clock_offset,
+            simulation_ref_offset + wall_clock_offset + 0.1,
+        ] {
+            let measured_time = (stamp.next().unwrap().0 - wall_clock_init).as_secs_f64();
+            assert!(
+                (expected_time - measured_time).abs() <= TOLERANCE,
+                "Expected t = {:.6}s +/- {:.6}s, measured t = {:.6}s",
+                expected_time,
+                TOLERANCE,
+                measured_time,
+            );
+
+            simu.step().unwrap();
+        }
+    }
+}
+
+#[cfg(not(miri))]
+fn system_clock_from_system_time(num_threads: usize) {
+    let t0 = MonotonicTime::EPOCH;
+    const TOLERANCE: f64 = 0.005; // [s]
+
+    // The reference simulation time is set in the past of t0 so that the
+    // simulation starts in the future when the reference wall clock time is
+    // close to the wall clock time when the simulation in initialized.
+    let simulation_ref_offset = 0.3; // [s] must be greater than any `instant_offset`.
+    let simulation_ref = t0 - Duration::from_secs_f64(simulation_ref_offset);
+
+    // Test reference wall clock times in the near past and near future.
+    for wall_clock_offset in [-0.1, 0.1] {
+        // The clock reference is the current time offset by `instant_offset`.
+        let wall_clock_init = SystemTime::now();
+        let wall_clock_ref = if wall_clock_offset >= 0.0 {
+            wall_clock_init + Duration::from_secs_f64(wall_clock_offset)
+        } else {
+            wall_clock_init - Duration::from_secs_f64(-wall_clock_offset)
+        };
+
+        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();
+
+        // Queue a single event at t0 + 0.1s.
+        scheduler
+            .schedule_event(
+                Duration::from_secs_f64(0.1),
+                TimestampModel::trigger,
+                (),
+                &addr,
+            )
+            .unwrap();
+
+        // Check the stamps.
+        for expected_time in [
+            simulation_ref_offset + wall_clock_offset,
+            simulation_ref_offset + wall_clock_offset + 0.1,
+        ] {
+            let measured_time = stamp
+                .next()
+                .unwrap()
+                .1
+                .duration_since(wall_clock_init)
+                .unwrap()
+                .as_secs_f64();
+            assert!(
+                (expected_time - measured_time).abs() <= TOLERANCE,
+                "Expected t = {:.6}s +/- {:.6}s, measured t = {:.6}s",
+                expected_time,
+                TOLERANCE,
+                measured_time,
+            );
+
+            simu.step().unwrap();
+        }
+    }
+}
+
+#[cfg(not(miri))]
+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 instant_t0 = Instant::now();
+
+    let scheduler = simu.scheduler();
+
+    // Queue a periodic event at t0 + 0.2s + k*0.2s.
+    scheduler
+        .schedule_periodic_event(
+            Duration::from_secs_f64(0.2),
+            Duration::from_secs_f64(0.2),
+            TimestampModel::trigger,
+            (),
+            &addr,
+        )
+        .unwrap();
+
+    // Queue a single event at t0 + 0.3s.
+    scheduler
+        .schedule_event(
+            Duration::from_secs_f64(0.3),
+            TimestampModel::trigger,
+            (),
+            &addr,
+        )
+        .unwrap();
+
+    // Check the stamps.
+    for expected_time in [0.0, 0.2, 0.3, 0.4, 0.6] {
+        let measured_time = (stamp.next().unwrap().0 - instant_t0).as_secs_f64();
+        assert!(
+            (expected_time - measured_time).abs() <= TOLERANCE,
+            "Expected t = {:.6}s +/- {:.6}s, measured t = {:.6}s",
+            expected_time,
+            TOLERANCE,
+            measured_time,
+        );
+
+        simu.step().unwrap();
+    }
+}
+
+#[cfg(not(miri))]
+#[test]
+fn system_clock_from_instant_st() {
+    system_clock_from_instant(1);
+}
+
+#[cfg(not(miri))]
+#[test]
+fn system_clock_from_instant_mt() {
+    system_clock_from_instant(MT_NUM_THREADS);
+}
+
+#[cfg(not(miri))]
+#[test]
+fn system_clock_from_system_time_st() {
+    system_clock_from_system_time(1);
+}
+
+#[cfg(not(miri))]
+#[test]
+fn system_clock_from_system_time_mt() {
+    system_clock_from_system_time(MT_NUM_THREADS);
+}
+
+#[cfg(not(miri))]
+#[test]
+fn auto_system_clock_st() {
+    auto_system_clock(1);
+}
+
+#[cfg(not(miri))]
+#[test]
+fn auto_system_clock_mt() {
+    auto_system_clock(MT_NUM_THREADS);
+}

asynchronix/tests/integration/simulation_timeout.rs

@@ -0,0 +1,103 @@
+//! Timeout during step execution.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+use asynchronix::model::Model;
+use asynchronix::ports::Output;
+use asynchronix::simulation::{ExecutionError, Mailbox, SimInit};
+use asynchronix::time::MonotonicTime;
+
+const MT_NUM_THREADS: usize = 4;
+
+#[derive(Default)]
+struct TestModel {
+    output: Output<()>,
+    // A liveliness flag that is cleared when the model is dropped.
+    is_alive: Arc<AtomicBool>,
+}
+impl TestModel {
+    fn new() -> (Self, Arc<AtomicBool>) {
+        let is_alive = Arc::new(AtomicBool::new(true));
+
+        (
+            Self {
+                output: Output::default(),
+                is_alive: is_alive.clone(),
+            },
+            is_alive,
+        )
+    }
+
+    async fn input(&mut self) {
+        self.output.send(()).await;
+    }
+}
+impl Drop for TestModel {
+    fn drop(&mut self) {
+        self.is_alive.store(false, Ordering::Relaxed);
+    }
+}
+impl Model for TestModel {}
+
+fn timeout_untriggered(num_threads: usize) {
+    let (model, _model_is_alive) = TestModel::new();
+    let mbox = Mailbox::new();
+    let addr = mbox.address();
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "test")
+        .set_timeout(Duration::from_secs(1))
+        .init(t0)
+        .unwrap();
+
+    assert!(simu.process_event(TestModel::input, (), addr).is_ok());
+}
+
+fn timeout_triggered(num_threads: usize) {
+    let (mut model, model_is_alive) = TestModel::new();
+    let mbox = Mailbox::new();
+    let addr = mbox.address();
+
+    // Make a loopback connection.
+    model.output.connect(TestModel::input, addr.clone());
+
+    let t0 = MonotonicTime::EPOCH;
+    let mut simu = SimInit::with_num_threads(num_threads)
+        .add_model(model, mbox, "test")
+        .set_timeout(Duration::from_secs(1))
+        .init(t0)
+        .unwrap();
+
+    assert!(matches!(
+        simu.process_event(TestModel::input, (), addr),
+        Err(ExecutionError::Timeout)
+    ));
+
+    // Make sure the request to stop the simulation has succeeded.
+    thread::sleep(Duration::from_millis(10));
+    assert!(!model_is_alive.load(Ordering::Relaxed));
+}
+
+#[test]
+fn timeout_untriggered_st() {
+    timeout_untriggered(1);
+}
+
+#[test]
+fn timeout_untriggered_mt() {
+    timeout_untriggered(MT_NUM_THREADS);
+}
+
+#[test]
+fn timeout_triggered_st() {
+    timeout_triggered(1);
+}
+
+#[test]
+fn timeout_triggered_mt() {
+    timeout_triggered(MT_NUM_THREADS);
+}