des_sim/execution/runner/instance/
parallel.rs

1//! The `parallel` module provides the `ParallelRunner`, an implementation of the `Runner` trait
2//! that enables parallel execution of events.
3//!
4//! This runner processes events concurrently, with an option to synchronize high-priority
5//! events on the main thread to prevent data races. It requires models to implement
6//! the `ParallelModel` trait for safe state management across threads.
7
8use crate::context::{EventContext, ExecutorStatus};
9use crate::execution::SimulationResult;
10use crate::execution::engine::Engine;
11use crate::execution::phase::MicroStepResult;
12use crate::execution::runner::Runner;
13use crate::execution::strategy::{AlwaysContinueStrategy, ContinueStrategy};
14use crate::modeling::event::{Event, EventPriority};
15use crate::modeling::model::Model;
16use crate::primitive::time::TickStatus;
17use std::marker::PhantomData;
18use std::sync::mpsc::Sender;
19
20/// An extension trait for models supporting parallel execution.
21///
22/// This trait extracts model state changes into commands that are aggregated
23/// on the main thread, preventing data races in concurrent environments.
24pub trait ParallelModel<E, Command>: Model<E> {
25    /// An event handler invoked from asynchronous (parallel) threads.
26    /// Sends a `Command` representing a state change request through the provided sender.
27    fn handle_event_parallel(&self, event: Event<E>, sender: Sender<Command>);
28
29    /// Applies the result of parallel computation (`Command`) to the model on the main thread.
30    ///
31    /// Since `&mut self` is provided here, state updates can be performed safely.
32    fn apply_command(&mut self, context: &mut EventContext<E, Self>, command: Command)
33    where
34        Self: Sized;
35}
36
37/// A runner that enables parallel event execution.
38///
39/// Events with a priority exceeding `sync_priority_threshold` are processed
40/// synchronously on the main thread, while others are processed in parallel.
41#[derive(Clone)]
42pub struct ParallelRunner<Command, CS> {
43    skippable: bool,
44    continue_strategy: CS,
45    /// Events with a priority greater than or equal to this threshold are processed synchronously.
46    sync_priority_threshold: EventPriority,
47    _command: PhantomData<Command>,
48}
49
50impl<E, Command, M: ParallelModel<E, Command>, CS: ContinueStrategy<E, M, ()>> Runner<E, M, CS>
51    for ParallelRunner<Command, CS>
52where
53    E: Send + 'static,
54    M: Send + Sync + 'static,
55    Command: Send,
56{
57    type Err = ();
58
59    fn run<F>(
60        &mut self,
61        engine: Engine<E, M>,
62        mut model: M,
63        mut should_stop: F,
64    ) -> SimulationResult<M, CS::Err>
65    where
66        F: FnMut(&M, ExecutorStatus, TickStatus) -> bool,
67    {
68        let mut runner_error: Option<CS::Err> = None;
69        let mut executor = engine.begin_simulation(&model);
70
71        loop {
72            let (executor_status, tick_status) = executor.peek_next_tick();
73            if should_stop(&model, executor_status, tick_status) {
74                break;
75            }
76
77            let mut active_executor = executor.begin_tick(&model);
78
79            loop {
80                // 1. Begin micro-step
81                let micro_step_handler = active_executor.begin_micro_step(&model);
82
83                // 2. Source phase (Sources are typically processed synchronously as they depend on state)
84                let mut source_phase = micro_step_handler.start_source_phase(&model);
85                while let Some(source_ready) = source_phase.take_one() {
86                    source_phase.fire_and_schedule(&model, source_ready);
87                }
88                let micro_step_handler = source_phase.complete_source_phase(&model);
89
90                // 3. Event phase
91                let mut event_phase = micro_step_handler.to_event_phase(&model);
92
93                loop {
94                    // Check if the front event is eligible for synchronous processing
95                    if let Some(event_ready) =
96                        event_phase.take_front_if(|e| e.priority >= self.sync_priority_threshold)
97                    {
98                        // [Synchronous Execution] Process immediately on the main thread
99                        event_phase.handle_event(&mut model, event_ready);
100                    } else {
101                        // Extract remaining events for parallel processing
102                        let parallel_events = event_phase.take_all();
103                        let (sender, receiver) = std::sync::mpsc::channel();
104
105                        // Utilize std::thread::scope to safely share &model with the Rayon thread pool
106                        std::thread::scope(|_scope| {
107                            use rayon::prelude::*;
108
109                            let sender = sender;
110                            parallel_events.into_par_iter().for_each_with(
111                                sender,
112                                |sender_worker, event_ready| {
113                                    let model_ref = &model;
114                                    model_ref
115                                        .handle_event_parallel(event_ready, sender_worker.clone());
116                                },
117                            );
118                        });
119
120                        // Apply accumulated commands to the model on the main thread
121                        while let Ok(command) = receiver.try_recv() {
122                            model.apply_command(event_phase.get_context(), command);
123                        }
124
125                        break;
126                    }
127                }
128
129                let micro_step_handler = event_phase.complete_event_phase(&model);
130
131                // 4. End micro-step
132                match micro_step_handler.end_micro_step(&model) {
133                    MicroStepResult::Continue(unchecked) => {
134                        match self
135                            .continue_strategy
136                            .handle_micro_step_continue(&model, unchecked)
137                        {
138                            Ok(new_active_executor) => {
139                                active_executor = new_active_executor;
140                                continue;
141                            }
142                            Err((new_active_executor, error)) => {
143                                active_executor = new_active_executor;
144                                runner_error = Some(error);
145                                break;
146                            }
147                        }
148                    }
149                    MicroStepResult::Complete(new_active_executor, _) => {
150                        active_executor = new_active_executor;
151                        break;
152                    }
153                }
154            }
155
156            executor = if self.skippable {
157                active_executor.end_tick_with_jump_to_next_tick(&model)
158            } else {
159                active_executor.end_tick_with_increment_tick(&model)
160            };
161
162            if runner_error.is_some() {
163                break;
164            }
165        }
166
167        if let Some(error) = runner_error.take() {
168            executor.end_simulation_as_error(model, error)
169        } else {
170            executor.end_simulation_as_ok(model)
171        }
172    }
173}
174
175impl<Command> ParallelRunner<Command, AlwaysContinueStrategy> {
176    /// Creates a new `ParallelRunner` using the `AlwaysContinueStrategy`.
177    ///
178    /// # Arguments
179    /// * `skippable` - If `true`, the runner allows skipping empty ticks to optimize simulation time.
180    /// * `sync_priority_threshold` - Events with a priority greater than or equal to this
181    ///   threshold will be processed synchronously on the main thread.
182    pub fn new(skippable: bool, sync_priority_threshold: EventPriority) -> Self {
183        ParallelRunner {
184            skippable,
185            continue_strategy: AlwaysContinueStrategy::new(),
186            sync_priority_threshold,
187            _command: PhantomData,
188        }
189    }
190}
191
192impl<Command, CS> ParallelRunner<Command, CS> {
193    /// Creates a new `ParallelRunner` with a custom `ContinueStrategy`.
194    ///
195    /// # Arguments
196    /// * `skippable` - If `true`, the runner allows skipping empty ticks to optimize simulation time.
197    /// * `sync_priority_threshold` - Events with a priority greater than or equal to this
198    ///   threshold will be processed synchronously on the main thread.
199    /// * `continue_strategy` - The specific strategy used to determine how the simulation
200    ///   handles micro-step transitions.
201    pub fn new_with_continue_strategy(
202        skippable: bool,
203        sync_priority_threshold: EventPriority,
204        continue_strategy: CS,
205    ) -> Self {
206        ParallelRunner {
207            skippable,
208            continue_strategy,
209            sync_priority_threshold,
210            _command: PhantomData,
211        }
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218    use crate::context::{EventContext, ExecutorStatus, SourceContext, UserContext};
219    use crate::execution::engine::Engine;
220    use crate::execution::strategy::{AlwaysContinueStrategy, LimitAbortStrategy};
221    use crate::modeling::event::{Event, EventPriority};
222    use crate::modeling::hook::Hook;
223    use crate::modeling::model::Model;
224    use crate::modeling::source::Source;
225    use crate::primitive::time::{Duration, MicroStep, SimTime, TickStatus};
226    use crate::source_handler::{SourceReadyEntry, SourceView};
227    use std::sync::atomic::{AtomicUsize, Ordering};
228    use std::sync::mpsc::Sender;
229    use std::sync::{Arc, Mutex};
230
231    /// Common commands for test simulation.
232    #[allow(unused)]
233    #[derive(Debug)]
234    enum TestCommand {
235        IncrementSync,
236        IncrementParallel,
237    }
238
239    #[derive(Debug, PartialEq, Eq, Clone, Copy)]
240    enum TestEvent {
241        SyncTarget,
242        ParallelTarget,
243    }
244
245    /// Model used for testing parallel and synchronous execution paths.
246    struct TestParallelModel {
247        sync_event_count: Arc<AtomicUsize>,
248        parallel_command_count: Arc<AtomicUsize>,
249    }
250
251    impl Model<TestEvent> for TestParallelModel {
252        fn handle_event(
253            &mut self,
254            _context: &mut EventContext<TestEvent, Self>,
255            event: &Event<TestEvent>,
256        ) {
257            // Synchronous execution path
258            if let TestEvent::SyncTarget = event.payload {
259                self.sync_event_count.fetch_add(1, Ordering::SeqCst);
260            }
261        }
262    }
263
264    impl ParallelModel<TestEvent, TestCommand> for TestParallelModel {
265        fn handle_event_parallel(&self, event: Event<TestEvent>, sender: Sender<TestCommand>) {
266            // Asynchronous execution path; forwards request to main thread
267            if let TestEvent::ParallelTarget = event.payload {
268                let _ = sender.send(TestCommand::IncrementParallel);
269            }
270        }
271
272        fn apply_command(
273            &mut self,
274            _context: &mut EventContext<TestEvent, Self>,
275            command: TestCommand,
276        ) {
277            // Apply result on main thread
278            if let TestCommand::IncrementParallel = command {
279                self.parallel_command_count.fetch_add(1, Ordering::SeqCst);
280            }
281        }
282    }
283
284    #[test]
285    fn test_parallel_runner_new_and_args() {
286        let runner = ParallelRunner::<TestCommand, AlwaysContinueStrategy>::new(
287            true,
288            EventPriority::new(100),
289        );
290        assert!(runner.skippable);
291        assert_eq!(runner.sync_priority_threshold, EventPriority::new(100));
292    }
293
294    #[test]
295    fn test_parallel_runner_sync_and_parallel_execution() {
296        let sync_counter = Arc::new(AtomicUsize::new(0));
297        let parallel_counter = Arc::new(AtomicUsize::new(0));
298
299        let model = TestParallelModel {
300            sync_event_count: sync_counter.clone(),
301            parallel_command_count: parallel_counter.clone(),
302        };
303
304        let mut engine = Engine::new();
305
306        // 1. High-priority event (synchronous)
307        engine.schedule_event_at(
308            SimTime::zero(),
309            EventPriority::new(10),
310            TestEvent::SyncTarget,
311        );
312        // 2. Low-priority event (parallel)
313        engine.schedule_event_at(
314            SimTime::zero(),
315            EventPriority::new(0),
316            TestEvent::ParallelTarget,
317        );
318
319        // Set the threshold to "5". 1 should be assigned to synchronous, 2 should be assigned to parallel
320        let mut runner = ParallelRunner::new(true, EventPriority::new(5));
321
322        let mut tick_counter = 0;
323        let should_stop = |_m: &TestParallelModel, _status: ExecutorStatus, _tick: TickStatus| {
324            if tick_counter >= 1 {
325                true
326            } else {
327                tick_counter += 1;
328                false
329            }
330        };
331
332        let result = runner.run(engine, model, should_stop);
333        assert!(result.is_ok());
334
335        // Verify whether the count-up ran according to the expected route.
336        assert_eq!(sync_counter.load(Ordering::SeqCst), 1);
337        assert_eq!(parallel_counter.load(Ordering::SeqCst), 1);
338    }
339
340    #[test]
341    fn test_parallel_runner_boundary_priority() {
342        let sync_counter = Arc::new(AtomicUsize::new(0));
343        let parallel_counter = Arc::new(AtomicUsize::new(0));
344
345        let model = TestParallelModel {
346            sync_event_count: sync_counter.clone(),
347            parallel_command_count: parallel_counter.clone(),
348        };
349
350        let mut engine = Engine::new();
351
352        // Events with “exactly the same” priority as the threshold
353        engine.schedule_event_at(
354            SimTime::zero(),
355            EventPriority::new(5),
356            TestEvent::SyncTarget,
357        );
358
359        // Check the specifications for events that are higher than the threshold (5) to be processed synchronously (>= judged)
360        let mut runner = ParallelRunner::new(true, EventPriority::new(5));
361        let mut tick_counter = 0;
362        let should_stop = |_m: &TestParallelModel, _status: ExecutorStatus, _tick: TickStatus| {
363            if tick_counter >= 1 {
364                true
365            } else {
366                tick_counter += 1;
367                false
368            }
369        };
370
371        let result = runner.run(engine, model, should_stop);
372        assert!(result.is_ok());
373
374        // Boundary (>= threshold) is processed synchronously
375        assert_eq!(sync_counter.load(Ordering::SeqCst), 1);
376        assert_eq!(parallel_counter.load(Ordering::SeqCst), 0);
377    }
378
379    #[test]
380    fn test_parallel_runner_massive_parallel_events() {
381        let sync_counter = Arc::new(AtomicUsize::new(0));
382        let parallel_counter = Arc::new(AtomicUsize::new(0));
383        let model = TestParallelModel {
384            sync_event_count: sync_counter.clone(),
385            parallel_command_count: parallel_counter.clone(),
386        };
387
388        let mut engine = Engine::new();
389
390        // Input a large number of parallel events at once
391        let event_count = 100;
392        for _ in 0..event_count {
393            engine.schedule_event_at(
394                SimTime::zero(),
395                EventPriority::new(0), //Below Threshold
396                TestEvent::ParallelTarget,
397            );
398        }
399
400        let mut runner = ParallelRunner::new(true, EventPriority::new(10));
401        let mut tick_counter = 0;
402        let should_stop = |_m: &TestParallelModel, _status: ExecutorStatus, _tick: TickStatus| {
403            if tick_counter >= 1 {
404                true
405            } else {
406                tick_counter += 1;
407                false
408            }
409        };
410
411        let result = runner.run(engine, model, should_stop);
412        assert!(result.is_ok());
413        assert_eq!(parallel_counter.load(Ordering::SeqCst), event_count);
414    }
415
416    #[test]
417    fn test_parallel_runner_aborts_on_strategy_error() {
418        let mut engine = Engine::new();
419
420        // Set the first trigger event at time 0
421        engine.schedule_event_at(
422            SimTime::zero(),
423            // Since the threshold is less than 5, it will be processed in parallel.
424            EventPriority::new(0),
425            TestEvent::ParallelTarget,
426        );
427
428        struct TestAbortModel;
429        impl Model<TestEvent> for TestAbortModel {
430            fn handle_event(
431                &mut self,
432                _context: &mut EventContext<TestEvent, Self>,
433                _event: &Event<TestEvent>,
434            ) {
435                // none
436            }
437        }
438        impl ParallelModel<TestEvent, TestCommand> for TestAbortModel {
439            fn handle_event_parallel(&self, event: Event<TestEvent>, sender: Sender<TestCommand>) {
440                if let TestEvent::ParallelTarget = event.payload {
441                    let _ = sender.send(TestCommand::IncrementParallel);
442                }
443            }
444            fn apply_command(
445                &mut self,
446                context: &mut EventContext<TestEvent, Self>,
447                _command: TestCommand,
448            ) {
449                // Re-schedule in the same tick to trigger Continue strategy
450                context.schedule_event(
451                    Duration::zero(),
452                    EventPriority::new(0),
453                    TestEvent::ParallelTarget,
454                );
455            }
456        }
457
458        let abort_model = TestAbortModel;
459
460        // Combine ParallelRunner with a continuation strategy with a 0 limit setting that immediately fails.
461        let strategy = LimitAbortStrategy::new(0, 0);
462        let mut runner =
463            ParallelRunner::new_with_continue_strategy(true, EventPriority::new(5), strategy);
464
465        let mut loop_count = 0;
466        let should_stop = |_m: &TestAbortModel, _status: ExecutorStatus, _tick: TickStatus| {
467            loop_count += 1;
468            loop_count > 5
469        };
470
471        // When executed, immediately after the first event is processed, the event is re-registered to the same Tick in apply_command,
472        // and MicroStepResult::Continue occurs at the first microstep end determination (end_micro_step).
473        // Immediately after, LimitAbortStrategy detects that the limit has been exceeded and returns Err.
474        let result = runner.run(engine, abort_model, should_stop);
475
476        // Verify that the process did not terminate normally and detected an error (Err) in the continuation strategy to safely terminate the process.
477        assert!(result.is_err());
478    }
479
480    #[test]
481    fn test_parallel_runner_without_aborts_on_strategy_error() {
482        let model = TestParallelModel {
483            sync_event_count: Arc::new(AtomicUsize::new(0)),
484            parallel_command_count: Arc::new(AtomicUsize::new(0)),
485        };
486
487        let mut engine = Engine::new();
488
489        // An event at time 0 registered in the engine is registered to be processed as an event at 0 tick when the simulation starts,
490        // so even if the microstep upper limit is 0, it will not be affected by the upper limit.
491        engine.schedule_event_at(
492            SimTime::zero(),
493            EventPriority::new(0),
494            TestEvent::ParallelTarget,
495        );
496
497        // Combine ParallelRunner with a continuation strategy with a 0 limit setting that immediately fails.
498        let strategy = LimitAbortStrategy::new(0, 0);
499        let mut runner =
500            ParallelRunner::new_with_continue_strategy(true, EventPriority::new(5), strategy);
501
502        let mut loop_count = 0;
503        let should_stop = |_m: &TestParallelModel, _status: ExecutorStatus, _tick: TickStatus| {
504            loop_count += 1;
505            loop_count > 5
506        };
507
508        let result = runner.run(engine, model, should_stop);
509
510        // Verify that the process completed without any errors in the continuation strategy.
511        assert!(result.is_ok());
512    }
513
514    // Lifecycle event definition to track call order
515    #[derive(Debug, PartialEq, Eq, Clone)]
516    enum LifecycleEvent {
517        BeforeSimulation,
518        BeforeTick(SimTime),
519        BeforeFireSource(SimTime),
520        BeforeScheduleEvent,
521        AfterScheduleEvent,
522        AfterFireSource(SimTime),
523        AfterTick(SimTime),
524        AfterSimulation,
525    }
526
527    // A dummy source that shares and records trace logs for each event for testing purposes.
528    struct TraceSource {
529        trace: Arc<Mutex<Vec<LifecycleEvent>>>,
530        initial_delay: Duration,
531        interval_delay: Option<Duration>,
532    }
533
534    impl Source<TestEvent, TestParallelModel> for TraceSource {
535        fn on_registered(
536            &mut self,
537            _context: &mut dyn UserContext<TestEvent, TestParallelModel>,
538            _model: &TestParallelModel,
539        ) -> Option<Duration> {
540            self.trace
541                .lock()
542                .unwrap()
543                .push(LifecycleEvent::BeforeSimulation);
544            Some(self.initial_delay)
545        }
546
547        fn fire(
548            &mut self,
549            context: &mut SourceContext<TestEvent, TestParallelModel>,
550            _model: &TestParallelModel,
551        ) -> Option<Duration> {
552            let mut t = self.trace.lock().unwrap();
553            t.push(LifecycleEvent::BeforeFireSource(context.current_tick()));
554            t.push(LifecycleEvent::BeforeScheduleEvent);
555
556            // Fire an event
557            context.schedule_event(
558                Duration::zero(),
559                EventPriority::new(0),
560                TestEvent::ParallelTarget,
561            );
562
563            t.push(LifecycleEvent::AfterScheduleEvent);
564            t.push(LifecycleEvent::AfterFireSource(context.current_tick()));
565            self.interval_delay
566        }
567    }
568
569    #[test]
570    fn test_parallel_runner_lifecycle_execution_order_scenario() {
571        let trace = Arc::new(Mutex::new(Vec::new()));
572        let model = TestParallelModel {
573            sync_event_count: Arc::new(AtomicUsize::new(0)),
574            parallel_command_count: Arc::new(AtomicUsize::new(0)),
575        };
576        let mut engine = Engine::new();
577
578        //Register the source that fires once at 1 tick
579        engine.add_source(
580            "trace_source",
581            TraceSource {
582                trace: Arc::clone(&trace),
583                initial_delay: Duration::ticks(1),
584                interval_delay: None,
585            },
586        );
587
588        let mut runner = ParallelRunner::new(true, EventPriority::new(5));
589
590        // Stop condition: Stop when 2 tick is reached
591        let should_stop =
592            move |_m: &TestParallelModel, _status: ExecutorStatus, tick: TickStatus| {
593                tick.is_done_ticks(false, 2)
594            };
595
596        let result = runner.run(engine, model, should_stop);
597        assert!(result.is_ok());
598
599        // After exiting the loop, the simulation is safely finished,
600        // so we manually record it at the end.
601        trace.lock().unwrap().push(LifecycleEvent::AfterSimulation);
602
603        let final_trace = trace.lock().unwrap();
604
605        let expected = vec![
606            LifecycleEvent::BeforeSimulation,
607            // iteration at 1 tick
608            LifecycleEvent::BeforeFireSource(SimTime::from_ticks(1)),
609            LifecycleEvent::BeforeScheduleEvent,
610            LifecycleEvent::AfterScheduleEvent,
611            LifecycleEvent::AfterFireSource(SimTime::from_ticks(1)),
612            // Termination processing after exiting the loop
613            LifecycleEvent::AfterSimulation,
614        ];
615
616        assert_eq!(*final_trace, expected);
617    }
618
619    #[test]
620    fn test_parallel_runner_lifecycle_interruption_on_strategy_error() {
621        let trace = Arc::new(Mutex::new(Vec::new()));
622        let model = TestParallelModel {
623            sync_event_count: Arc::new(AtomicUsize::new(0)),
624            parallel_command_count: Arc::new(AtomicUsize::new(0)),
625        };
626        let mut engine = Engine::new();
627
628        // Prepare a source that will definitely fire infinitely at the first tick (time 0)
629        engine.add_source(
630            "loop_source",
631            TraceSource {
632                trace: Arc::clone(&trace),
633                initial_delay: Duration::zero(),
634                interval_delay: Some(Duration::zero()),
635            },
636        );
637
638        // Strategies to instantly generate upper bound errors for microsteps
639        let strategy = LimitAbortStrategy::new(0, 0);
640        let mut runner =
641            ParallelRunner::new_with_continue_strategy(true, EventPriority::new(5), strategy);
642
643        let trace_for_stop = Arc::clone(&trace);
644        let should_stop =
645            move |_m: &TestParallelModel, _status: ExecutorStatus, tick: TickStatus| {
646                trace_for_stop
647                    .lock()
648                    .unwrap()
649                    .push(LifecycleEvent::BeforeTick(tick.current()));
650                false
651            };
652
653        let result = runner.run(engine, model, should_stop);
654
655        // Confirmed abnormal termination due to strategy error
656        assert!(result.is_err());
657
658        let final_trace = trace.lock().unwrap();
659
660        // Verify that even when interrupted due to an error, the process that is paired with the started hook detects an abnormality and is interrupted.
661        assert!(final_trace.contains(&LifecycleEvent::BeforeSimulation));
662        assert!(final_trace.contains(&LifecycleEvent::BeforeTick(SimTime::zero())));
663        assert!(!final_trace.contains(&LifecycleEvent::BeforeTick(SimTime::from_ticks(1))));
664
665        // Verify that the normal lifecycle event (such as AfterTick) after the microstep where the error occurred is not executed
666        // and the loop is safely exited.
667        let last_event = final_trace.last().unwrap();
668        assert_ne!(last_event, &LifecycleEvent::AfterTick(SimTime::zero()));
669    }
670
671    // An enum that records that the hook was called, along with detailed parameters.
672    #[derive(Debug, PartialEq, Eq, Clone)]
673    enum HookCall {
674        BeforeSimulation,
675        AfterSimulation(SimTime),
676        BeforeTick {
677            current: SimTime,
678            skipped: Duration,
679        },
680        AfterTick {
681            current: SimTime,
682            last_micro: MicroStep,
683        },
684        BeforeMicroStep {
685            current: SimTime,
686            micro: MicroStep,
687        },
688        AfterMicroStep {
689            current: SimTime,
690            micro: MicroStep,
691        },
692        BeforeSourcePhase {
693            current: SimTime,
694            micro: MicroStep,
695        },
696        AfterSourcePhase {
697            current: SimTime,
698            micro: MicroStep,
699        },
700        BeforeEventPhase {
701            current: SimTime,
702            micro: MicroStep,
703        },
704        AfterEventPhase {
705            current: SimTime,
706            micro: MicroStep,
707        },
708    }
709
710    // テスト用の Hook 実装体
711    struct MockHook {
712        calls: Arc<Mutex<Vec<HookCall>>>,
713    }
714
715    impl<E, M: Model<E>> Hook<E, M> for MockHook {
716        fn before_simulation(&self, _model: &M) {
717            self.calls.lock().unwrap().push(HookCall::BeforeSimulation);
718        }
719        fn after_simulation(&self, _model: &M, end_tick: SimTime) {
720            self.calls
721                .lock()
722                .unwrap()
723                .push(HookCall::AfterSimulation(end_tick));
724        }
725        fn before_tick(&self, _model: &M, current_tick: SimTime, skipped_duration: Duration) {
726            self.calls.lock().unwrap().push(HookCall::BeforeTick {
727                current: current_tick,
728                skipped: skipped_duration,
729            });
730        }
731        fn after_tick(&self, _model: &M, current_tick: SimTime, last_micro_step: MicroStep) {
732            self.calls.lock().unwrap().push(HookCall::AfterTick {
733                current: current_tick,
734                last_micro: last_micro_step,
735            });
736        }
737        fn before_micro_step(
738            &self,
739            _model: &M,
740            current_tick: SimTime,
741            current_micro_step: MicroStep,
742        ) {
743            self.calls.lock().unwrap().push(HookCall::BeforeMicroStep {
744                current: current_tick,
745                micro: current_micro_step,
746            });
747        }
748        fn after_micro_step(
749            &self,
750            _model: &M,
751            current_tick: SimTime,
752            current_micro_step: MicroStep,
753        ) {
754            self.calls.lock().unwrap().push(HookCall::AfterMicroStep {
755                current: current_tick,
756                micro: current_micro_step,
757            });
758        }
759        fn on_discard_remain_micro_step(
760            &self,
761            _model: &M,
762            _current_tick: SimTime,
763            _first_discarded_micro_step: MicroStep,
764            _discarded_sources: &[SourceReadyEntry],
765            _discarded_events: &[Event<E>],
766        ) {
767        }
768        fn before_register_source(&self, _model: &M, _name: &str) {}
769        fn after_register_source(&self, _model: &M, _name: &str) {}
770        fn before_source_phase(
771            &self,
772            _model: &M,
773            current_tick: SimTime,
774            current_micro_step: MicroStep,
775        ) {
776            self.calls
777                .lock()
778                .unwrap()
779                .push(HookCall::BeforeSourcePhase {
780                    current: current_tick,
781                    micro: current_micro_step,
782                });
783        }
784        fn before_source(
785            &self,
786            _model: &M,
787            _current_tick: SimTime,
788            _current_micro_step: MicroStep,
789            _source_view: &SourceView,
790        ) {
791        }
792        fn after_source(
793            &self,
794            _model: &M,
795            _current_tick: SimTime,
796            _current_micro_step: MicroStep,
797            _source_view: &SourceView,
798            _computed_next_fire: Option<SimTime>,
799        ) {
800        }
801        fn cancel_source(
802            &self,
803            _model: &M,
804            _current_tick: SimTime,
805            _current_micro_step: MicroStep,
806            _scheduled_at: SimTime,
807            _source_view: &SourceView,
808        ) {
809        }
810        fn discard_source(
811            &self,
812            _model: &M,
813            _current_tick: SimTime,
814            _current_micro_step: MicroStep,
815            _source_view: &SourceView,
816        ) {
817        }
818        fn after_source_phase(
819            &self,
820            _model: &M,
821            current_tick: SimTime,
822            current_micro_step: MicroStep,
823        ) {
824            self.calls.lock().unwrap().push(HookCall::AfterSourcePhase {
825                current: current_tick,
826                micro: current_micro_step,
827            });
828        }
829        fn before_event_phase(
830            &self,
831            _model: &M,
832            current_tick: SimTime,
833            current_micro_step: MicroStep,
834        ) {
835            self.calls.lock().unwrap().push(HookCall::BeforeEventPhase {
836                current: current_tick,
837                micro: current_micro_step,
838            });
839        }
840        fn before_event(
841            &self,
842            _model: &M,
843            _current_tick: SimTime,
844            _current_micro_step: MicroStep,
845            _event: &Event<E>,
846        ) {
847        }
848        fn after_event(
849            &self,
850            _model: &M,
851            _current_tick: SimTime,
852            _current_micro_step: MicroStep,
853            _event: &Event<E>,
854        ) {
855        }
856        fn cancel_event(
857            &self,
858            _model: &M,
859            _current_tick: SimTime,
860            _current_micro_step: MicroStep,
861            _scheduled_at: SimTime,
862            _event: &Event<E>,
863        ) {
864        }
865        fn discard_event(
866            &self,
867            _model: &M,
868            _current_tick: SimTime,
869            _current_micro_step: MicroStep,
870            _event: &Event<E>,
871        ) {
872        }
873        fn after_event_phase(
874            &self,
875            _model: &M,
876            current_tick: SimTime,
877            current_micro_step: MicroStep,
878        ) {
879            self.calls.lock().unwrap().push(HookCall::AfterEventPhase {
880                current: current_tick,
881                micro: current_micro_step,
882            });
883        }
884    }
885
886    #[test]
887    fn test_parallel_runner_hook_lifecycle_flow_with_include_zero_tick() {
888        use crate::modeling::hook::instance::SharedHook;
889
890        let hook = MockHook {
891            calls: Arc::new(Mutex::new(Vec::new())),
892        };
893        let shared_hook = SharedHook::new(hook);
894
895        let model = TestParallelModel {
896            sync_event_count: Arc::new(AtomicUsize::new(0)),
897            parallel_command_count: Arc::new(AtomicUsize::new(0)),
898        };
899        let mut engine = Engine::new();
900        engine.add_shared_hook(shared_hook.clone());
901
902        // Place only one parallel event (priority 0) at 1 tick
903        engine.schedule_event_at(
904            SimTime::from_ticks(1),
905            EventPriority::new(0),
906            TestEvent::ParallelTarget,
907        );
908
909        let mut runner = ParallelRunner::new(true, EventPriority::new(5));
910
911        // Stop condition: End when processing for two times has been completed.
912        let should_stop = |_m: &TestParallelModel, _status: ExecutorStatus, tick: TickStatus| {
913            tick.is_done_ticks(true, 2)
914        };
915
916        let _result = runner.run(engine, model, should_stop);
917
918        let final_calls = shared_hook.get_ref().calls.lock().unwrap();
919
920        let expected = vec![
921            HookCall::BeforeSimulation,
922            // --- process of 0 tick ---
923            HookCall::BeforeTick {
924                current: SimTime::from_ticks(0),
925                skipped: Duration::zero(),
926            },
927            HookCall::BeforeMicroStep {
928                current: SimTime::from_ticks(0),
929                micro: MicroStep::zero(),
930            },
931            HookCall::BeforeSourcePhase {
932                current: SimTime::from_ticks(0),
933                micro: MicroStep::zero(),
934            },
935            HookCall::AfterSourcePhase {
936                current: SimTime::from_ticks(0),
937                micro: MicroStep::zero(),
938            },
939            HookCall::BeforeEventPhase {
940                current: SimTime::from_ticks(0),
941                micro: MicroStep::zero(),
942            },
943            HookCall::AfterEventPhase {
944                current: SimTime::from_ticks(0),
945                micro: MicroStep::zero(),
946            },
947            HookCall::AfterMicroStep {
948                current: SimTime::from_ticks(0),
949                micro: MicroStep::zero(),
950            },
951            HookCall::AfterTick {
952                current: SimTime::from_ticks(0),
953                last_micro: MicroStep::zero(),
954            },
955            // --- process of 1 tick ---
956            HookCall::BeforeTick {
957                current: SimTime::from_ticks(1),
958                skipped: Duration::ticks(0),
959            },
960            HookCall::BeforeMicroStep {
961                current: SimTime::from_ticks(1),
962                micro: MicroStep::zero(),
963            },
964            HookCall::BeforeSourcePhase {
965                current: SimTime::from_ticks(1),
966                micro: MicroStep::zero(),
967            },
968            HookCall::AfterSourcePhase {
969                current: SimTime::from_ticks(1),
970                micro: MicroStep::zero(),
971            },
972            HookCall::BeforeEventPhase {
973                current: SimTime::from_ticks(1),
974                micro: MicroStep::zero(),
975            },
976            HookCall::AfterEventPhase {
977                current: SimTime::from_ticks(1),
978                micro: MicroStep::zero(),
979            },
980            HookCall::AfterMicroStep {
981                current: SimTime::from_ticks(1),
982                micro: MicroStep::zero(),
983            },
984            HookCall::AfterTick {
985                current: SimTime::from_ticks(1),
986                last_micro: MicroStep::zero(),
987            },
988            HookCall::AfterSimulation(SimTime::from_ticks(1)),
989        ];
990
991        assert_eq!(*final_calls, expected);
992    }
993
994    #[test]
995    fn test_parallel_runner_hook_lifecycle_flow_without_include_zero_tick() {
996        use crate::modeling::hook::instance::SharedHook;
997
998        let hook = MockHook {
999            calls: Arc::new(Mutex::new(Vec::new())),
1000        };
1001        let shared_hook = SharedHook::new(hook);
1002
1003        let model = TestParallelModel {
1004            sync_event_count: Arc::new(AtomicUsize::new(0)),
1005            parallel_command_count: Arc::new(AtomicUsize::new(0)),
1006        };
1007        let mut engine = Engine::new();
1008        engine.add_shared_hook(shared_hook.clone());
1009
1010        // Place only one parallel event (priority 0) at 1 tick
1011        engine.schedule_event_at(
1012            SimTime::from_ticks(1),
1013            EventPriority::new(0),
1014            TestEvent::ParallelTarget,
1015        );
1016
1017        let mut runner = ParallelRunner::new(true, EventPriority::new(5));
1018
1019        // Stop condition: End when processing for 2 tick has been completed.
1020        let should_stop = |_m: &TestParallelModel, _status: ExecutorStatus, tick: TickStatus| {
1021            tick.is_done_ticks(false, 2)
1022        };
1023
1024        let _result = runner.run(engine, model, should_stop);
1025
1026        let final_calls = shared_hook.get_ref().calls.lock().unwrap();
1027
1028        let expected = vec![
1029            HookCall::BeforeSimulation,
1030            // --- process of 0 tick ---
1031            HookCall::BeforeTick {
1032                current: SimTime::from_ticks(0),
1033                skipped: Duration::zero(),
1034            },
1035            HookCall::BeforeMicroStep {
1036                current: SimTime::from_ticks(0),
1037                micro: MicroStep::zero(),
1038            },
1039            HookCall::BeforeSourcePhase {
1040                current: SimTime::from_ticks(0),
1041                micro: MicroStep::zero(),
1042            },
1043            HookCall::AfterSourcePhase {
1044                current: SimTime::from_ticks(0),
1045                micro: MicroStep::zero(),
1046            },
1047            HookCall::BeforeEventPhase {
1048                current: SimTime::from_ticks(0),
1049                micro: MicroStep::zero(),
1050            },
1051            HookCall::AfterEventPhase {
1052                current: SimTime::from_ticks(0),
1053                micro: MicroStep::zero(),
1054            },
1055            HookCall::AfterMicroStep {
1056                current: SimTime::from_ticks(0),
1057                micro: MicroStep::zero(),
1058            },
1059            HookCall::AfterTick {
1060                current: SimTime::from_ticks(0),
1061                last_micro: MicroStep::zero(),
1062            },
1063            // --- process of 1 tick ---
1064            HookCall::BeforeTick {
1065                current: SimTime::from_ticks(1),
1066                skipped: Duration::ticks(0),
1067            },
1068            HookCall::BeforeMicroStep {
1069                current: SimTime::from_ticks(1),
1070                micro: MicroStep::zero(),
1071            },
1072            HookCall::BeforeSourcePhase {
1073                current: SimTime::from_ticks(1),
1074                micro: MicroStep::zero(),
1075            },
1076            HookCall::AfterSourcePhase {
1077                current: SimTime::from_ticks(1),
1078                micro: MicroStep::zero(),
1079            },
1080            HookCall::BeforeEventPhase {
1081                current: SimTime::from_ticks(1),
1082                micro: MicroStep::zero(),
1083            },
1084            HookCall::AfterEventPhase {
1085                current: SimTime::from_ticks(1),
1086                micro: MicroStep::zero(),
1087            },
1088            HookCall::AfterMicroStep {
1089                current: SimTime::from_ticks(1),
1090                micro: MicroStep::zero(),
1091            },
1092            HookCall::AfterTick {
1093                current: SimTime::from_ticks(1),
1094                last_micro: MicroStep::zero(),
1095            },
1096            // --- process of 2 tick ---
1097            HookCall::BeforeTick {
1098                current: SimTime::from_ticks(2),
1099                skipped: Duration::zero(),
1100            },
1101            HookCall::BeforeMicroStep {
1102                current: SimTime::from_ticks(2),
1103                micro: MicroStep::zero(),
1104            },
1105            HookCall::BeforeSourcePhase {
1106                current: SimTime::from_ticks(2),
1107                micro: MicroStep::zero(),
1108            },
1109            HookCall::AfterSourcePhase {
1110                current: SimTime::from_ticks(2),
1111                micro: MicroStep::zero(),
1112            },
1113            HookCall::BeforeEventPhase {
1114                current: SimTime::from_ticks(2),
1115                micro: MicroStep::zero(),
1116            },
1117            HookCall::AfterEventPhase {
1118                current: SimTime::from_ticks(2),
1119                micro: MicroStep::zero(),
1120            },
1121            HookCall::AfterMicroStep {
1122                current: SimTime::from_ticks(2),
1123                micro: MicroStep::zero(),
1124            },
1125            HookCall::AfterTick {
1126                current: SimTime::from_ticks(2),
1127                last_micro: MicroStep::zero(),
1128            },
1129            HookCall::AfterSimulation(SimTime::from_ticks(2)),
1130        ];
1131
1132        assert_eq!(*final_calls, expected);
1133    }
1134}
des_sim/execution/runner/instance/parallel.rs

des_sim/execution/runner/instance/
parallel.rs
