des_sim/execution/phase/
source.rs

1//! The `source` module defines the `SourcePhase` struct, which manages the execution
2//! of sources within a micro-step.
3//!
4//! It handles the firing of sources, their rescheduling, and discarding,
5//! interacting with the simulation model and hooks.
6
7use crate::context::{SourceContext, UserContext};
8use crate::execution::phase::MicroStepHandler;
9use crate::modeling::hook::Hook;
10use crate::modeling::model::Model;
11use crate::source_handler::SourceHandler;
12use crate::source_handler::{SourceReadyEntry, SourceView};
13use std::collections::VecDeque;
14
15/// Manages the source execution phase in the simulation.
16///
17/// This structure holds the queue of sources executable in the current micro-step,
18/// and oversees source firing, scheduling, and discarding processes.
19pub struct SourcePhase<E, M: Model<E>> {
20    context: SourceContext<E, M>,
21    /// ### Internal Design Note
22    /// The `source_handler` is temporarily owned by this structure to manage lifetimes
23    /// between phases. Upon completion of the phase, it is safely returned to
24    /// the `SourceContext` via `complete_source_phase`.
25    pub(crate) source_handler: Option<SourceHandler<E, M>>,
26    ready_sources: VecDeque<SourceReadyEntry>,
27}
28
29impl<E, M: Model<E>> SourcePhase<E, M> {
30    /// Creates a new source execution phase.
31    pub(crate) fn new(
32        context: SourceContext<E, M>,
33        source_handler: SourceHandler<E, M>,
34        ready_sources: VecDeque<SourceReadyEntry>,
35    ) -> Self {
36        SourcePhase {
37            context,
38            source_handler: Some(source_handler),
39            ready_sources,
40        }
41    }
42
43    /// Returns a mutable reference to the source context used in the current phase.
44    pub fn get_context(&mut self) -> &mut SourceContext<E, M> {
45        &mut self.context
46    }
47
48    /// Generates a view from the specified source entry.
49    pub fn get_source_view(&self, ready_entry: &SourceReadyEntry) -> SourceView {
50        SourceView::new(ready_entry.source_id(), ready_entry.clone_name_arc())
51    }
52
53    /// Terminates the source phase, returns the `source_handler` to the context,
54    /// and transitions to the next micro-step handler.
55    ///
56    /// Invokes the `after_source_phase` hook to update the simulation state.
57    pub fn complete_source_phase(self, model: &M) -> MicroStepHandler<SourceContext<E, M>> {
58        self.context.hook().after_source_phase(
59            model,
60            self.context.current_tick(),
61            self.context.current_micro_step(),
62        );
63
64        let mut context = self.context;
65        context.source_handler = self.source_handler;
66        MicroStepHandler::new(context)
67    }
68
69    /// Pops one source from the front of the queue.
70    pub fn take_one(&mut self) -> Option<SourceReadyEntry> {
71        self.ready_sources.pop_front()
72    }
73
74    /// Searches for and pops the first source in the queue that satisfies the given predicate.
75    pub fn take_one_if<F>(&mut self, predicate: F) -> Option<SourceReadyEntry>
76    where
77        F: FnOnce(&SourceReadyEntry) -> bool,
78    {
79        self.ready_sources.pop_front_if(|e| predicate(e))
80    }
81
82    /// Pops a source from the front of the queue only if it satisfies the given predicate.
83    pub fn take_front_if<F>(&mut self, predicate: F) -> Option<SourceReadyEntry>
84    where
85        F: FnOnce(&SourceReadyEntry) -> bool,
86    {
87        if self.ready_sources.front().is_some_and(predicate) {
88            self.ready_sources.pop_front()
89        } else {
90            None
91        }
92    }
93
94    /// Takes all sources currently in the queue.
95    pub fn take_all(&mut self) -> VecDeque<SourceReadyEntry> {
96        std::mem::take(&mut self.ready_sources)
97    }
98
99    /// Extracts and returns all sources from the queue that satisfy the given predicate.
100    ///
101    /// Sources that do not satisfy the predicate remain in the queue.
102    pub fn take_all_if<F>(&mut self, predicate: F) -> VecDeque<SourceReadyEntry>
103    where
104        F: FnMut(&SourceReadyEntry) -> bool,
105    {
106        let all_sources = std::mem::take(&mut self.ready_sources);
107
108        let (taken, remaining): (VecDeque<_>, VecDeque<_>) =
109            all_sources.into_iter().partition(predicate);
110
111        self.ready_sources = remaining;
112
113        taken
114    }
115
116    /// Fires the source and schedules it if a next firing time exists.
117    ///
118    /// Invokes the `before_source` and `after_source` hooks surrounding the fire process.
119    pub fn fire_and_schedule(&mut self, model: &M, entry: SourceReadyEntry) {
120        let now = self.context.current_tick();
121        let current_microstep = self.context.current_micro_step();
122        let view = self.get_source_view(&entry);
123
124        self.context
125            .hook()
126            .before_source(model, now, current_microstep, &view);
127
128        let source_handler = self
129            .source_handler
130            .as_mut()
131            .expect("SourcePhase: source_handler is not properly initialized.");
132
133        let source_id = entry.source_id();
134        let entry = source_handler.get_by_source_id(source_id);
135        let next_fire_delay_optional = entry.source.fire(&mut self.context, model);
136
137        if let Some(next_fire_delay) = next_fire_delay_optional {
138            source_handler.schedule_next(now, next_fire_delay, source_id);
139        }
140
141        let computed_next_scheduled_at = next_fire_delay_optional.map(|d| now + d);
142
143        self.context.hook().after_source(
144            model,
145            now,
146            current_microstep,
147            &view,
148            computed_next_scheduled_at,
149        );
150    }
151
152    /// Discards the specified source.
153    ///
154    /// Invokes the `discard_source` hook during the discard process.
155    pub fn discard(&mut self, model: &M, entry: SourceReadyEntry) {
156        let view = self.get_source_view(&entry);
157
158        self.context.hook().discard_source(
159            model,
160            self.context.current_tick(),
161            self.context.current_micro_step(),
162            &view,
163        );
164    }
165}
166
167#[cfg(test)]
168mod tests {
169    use super::*;
170    use crate::context::{EventContext, UserContext};
171    use crate::event_scheduler::EventScheduler;
172    use crate::modeling::event::Event;
173    use crate::modeling::hook::Hook;
174    use crate::modeling::hook::instance::{HookDelegate, SharedHook};
175    use crate::modeling::model::Model;
176    use crate::primitive::id::SourceId;
177    use crate::primitive::time::{Duration, MicroStep, MicroStepStatus, SimTime, TickStatus};
178    use std::collections::VecDeque;
179    use std::rc::Rc;
180    use std::sync::{Arc, Mutex};
181
182    #[derive(Debug, PartialEq, Eq, Clone, Copy)]
183    enum TestEvent {}
184
185    struct TestModel {
186        handled_events: Vec<TestEvent>,
187    }
188
189    impl Model<TestEvent> for TestModel {
190        fn handle_event(
191            &mut self,
192            _context: &mut EventContext<TestEvent, Self>,
193            event: &Event<TestEvent>,
194        ) {
195            self.handled_events.push(event.payload);
196        }
197    }
198
199    /// A hook implementation that tracks discarded events and sources.
200    struct DiscardHook {
201        discarded_events: Rc<Mutex<Vec<TestEvent>>>,
202        discarded_sources: Rc<Mutex<Vec<SourceId>>>,
203    }
204
205    impl Hook<TestEvent, TestModel> for DiscardHook {
206        fn before_simulation(&self, _model: &TestModel) {
207            // none
208        }
209
210        fn after_simulation(&self, _model: &TestModel, _end_tick: SimTime) {
211            // none
212        }
213        fn before_tick(
214            &self,
215            _model: &TestModel,
216            _current_tick: SimTime,
217            _skipped_duration: Duration,
218        ) {
219            // none
220        }
221        fn after_tick(
222            &self,
223            _model: &TestModel,
224            _current_tick: SimTime,
225            _last_micro_step: MicroStep,
226        ) {
227            // none
228        }
229        fn before_micro_step(
230            &self,
231            _model: &TestModel,
232            _current_tick: SimTime,
233            _current_micro_step: MicroStep,
234        ) {
235            // none
236        }
237        fn after_micro_step(
238            &self,
239            _model: &TestModel,
240            _current_tick: SimTime,
241            _current_micro_step: MicroStep,
242        ) {
243            // none
244        }
245        fn on_discard_remain_micro_step(
246            &self,
247            _model: &TestModel,
248            _current_tick: SimTime,
249            _first_discarded_micro_step: MicroStep,
250            _discarded_sources: &[SourceReadyEntry],
251            _discarded_events: &[Event<TestEvent>],
252        ) {
253            // none
254        }
255        fn before_register_source(&self, _model: &TestModel, _name: &str) {
256            // none
257        }
258        fn after_register_source(&self, _model: &TestModel, _name: &str) {
259            // none
260        }
261        fn before_source_phase(
262            &self,
263            _model: &TestModel,
264            _current_tick: SimTime,
265            _current_micro_step: MicroStep,
266        ) {
267            // none
268        }
269        fn before_source(
270            &self,
271            _model: &TestModel,
272            _current_tick: SimTime,
273            _current_micro_step: MicroStep,
274            _source_view: &SourceView,
275        ) {
276            // none
277        }
278        fn after_source(
279            &self,
280            _model: &TestModel,
281            _current_tick: SimTime,
282            _current_micro_step: MicroStep,
283            _source_view: &SourceView,
284            _computed_next_fire: Option<SimTime>,
285        ) {
286            // none
287        }
288        fn cancel_source(
289            &self,
290            _model: &TestModel,
291            _current_tick: SimTime,
292            _current_micro_step: MicroStep,
293            _scheduled_at: SimTime,
294            _source_view: &SourceView,
295        ) {
296            // none
297        }
298        fn discard_source(
299            &self,
300            _model: &TestModel,
301            _current_tick: SimTime,
302            _current_micro_step: MicroStep,
303            source_view: &SourceView,
304        ) {
305            self.discarded_sources
306                .lock()
307                .unwrap()
308                .push(source_view.source_id());
309        }
310        fn after_source_phase(
311            &self,
312            _model: &TestModel,
313            _current_tick: SimTime,
314            _current_micro_step: MicroStep,
315        ) {
316            // none
317        }
318        fn before_event_phase(
319            &self,
320            _model: &TestModel,
321            _current_tick: SimTime,
322            _current_micro_step: MicroStep,
323        ) {
324            // none
325        }
326        fn before_event(
327            &self,
328            _model: &TestModel,
329            _current_tick: SimTime,
330            _current_micro_step: MicroStep,
331            _event: &Event<TestEvent>,
332        ) {
333            // none
334        }
335        fn after_event(
336            &self,
337            _model: &TestModel,
338            _current_tick: SimTime,
339            _current_micro_step: MicroStep,
340            _event: &Event<TestEvent>,
341        ) {
342            // none
343        }
344        fn cancel_event(
345            &self,
346            _model: &TestModel,
347            _current_tick: SimTime,
348            _current_micro_step: MicroStep,
349            _scheduled_at: SimTime,
350            _event: &Event<TestEvent>,
351        ) {
352            // none
353        }
354        fn discard_event(
355            &self,
356            _model: &TestModel,
357            _current_tick: SimTime,
358            _current_micro_step: MicroStep,
359            event: &Event<TestEvent>,
360        ) {
361            self.discarded_events.lock().unwrap().push(event.payload);
362        }
363        fn after_event_phase(
364            &self,
365            _model: &TestModel,
366            _current_tick: SimTime,
367            _current_micro_step: MicroStep,
368        ) {
369            // none
370        }
371    }
372
373    /// Sets up the initial state for a source phase test.
374    fn setup_source_phase() -> (SourcePhase<TestEvent, TestModel>, TestModel) {
375        let model = TestModel {
376            handled_events: Vec::new(),
377        };
378
379        // Note: source_handler is None while residing in SourceContext
380        let source_context = SourceContext {
381            current_tick_status: TickStatus::initialize(),
382            current_micro_step_status: MicroStepStatus::initialize(),
383            hook_delegate: HookDelegate::new(),
384            source_handler: None,
385            event_scheduler: EventScheduler::new(),
386        };
387
388        let source_handler = SourceHandler::new();
389
390        let mut ready_sources = VecDeque::new();
391        ready_sources.push_back(SourceReadyEntry::new(
392            SourceId::new(1),
393            Arc::from("SourceA"),
394        ));
395        ready_sources.push_back(SourceReadyEntry::new(
396            SourceId::new(2),
397            Arc::from("SourceB"),
398        ));
399        ready_sources.push_back(SourceReadyEntry::new(
400            SourceId::new(3),
401            Arc::from("SourceC"),
402        ));
403
404        let source_phase = SourcePhase::new(source_context, source_handler, ready_sources);
405
406        (source_phase, model)
407    }
408
409    #[test]
410    fn test_new() {
411        let (source_phase, _) = setup_source_phase();
412        assert_eq!(source_phase.ready_sources.len(), 3);
413        assert!(source_phase.source_handler.is_some());
414    }
415
416    #[test]
417    fn test_get_context() {
418        let (mut source_phase, _) = setup_source_phase();
419        let context = source_phase.get_context();
420        assert_eq!(context.current_tick(), SimTime::zero());
421    }
422
423    #[test]
424    fn test_get_source_view() {
425        let (source_phase, _) = setup_source_phase();
426        let entry = SourceReadyEntry::new(SourceId::new(1), Arc::from("SourceA"));
427        let view = source_phase.get_source_view(&entry);
428        assert_eq!(view.source_id(), SourceId::new(1));
429        assert_eq!(view.name(), "SourceA");
430    }
431
432    #[test]
433    fn test_take_one() {
434        let (mut source_phase, _) = setup_source_phase();
435        let entry = source_phase.take_one().unwrap();
436        assert_eq!(entry.source_id(), SourceId::new(1));
437        assert_eq!(source_phase.ready_sources.len(), 2);
438    }
439
440    #[test]
441    fn test_take_front_if() {
442        let (mut source_phase, _) = setup_source_phase();
443        let entry_a = source_phase
444            .take_front_if(|e| e.source_id() == SourceId::new(1))
445            .unwrap();
446        assert_eq!(entry_a.source_id(), SourceId::new(1));
447        assert_eq!(source_phase.ready_sources.len(), 2);
448
449        let entry_c = source_phase.take_front_if(|e| e.source_id() == SourceId::new(3));
450        assert!(entry_c.is_none());
451        assert_eq!(source_phase.ready_sources.len(), 2);
452    }
453
454    #[test]
455    fn test_take_all() {
456        let (mut source_phase, _) = setup_source_phase();
457        let all_entries = source_phase.take_all();
458        assert_eq!(all_entries.len(), 3);
459        assert_eq!(source_phase.ready_sources.len(), 0);
460    }
461
462    #[test]
463    fn test_take_all_if() {
464        let (mut source_phase, _) = setup_source_phase();
465        source_phase.ready_sources.push_back(SourceReadyEntry::new(
466            SourceId::new(1),
467            Arc::from("SourceA_again"),
468        ));
469
470        let taken_entries = source_phase.take_all_if(|e| e.source_id() == SourceId::new(1));
471        assert_eq!(taken_entries.len(), 2);
472        assert_eq!(taken_entries.front().unwrap().source_id(), SourceId::new(1));
473        assert_eq!(taken_entries.get(1).unwrap().source_id(), SourceId::new(1));
474
475        assert_eq!(source_phase.ready_sources.len(), 2);
476        assert_eq!(
477            source_phase.ready_sources.front().unwrap().source_id(),
478            SourceId::new(2)
479        );
480        assert_eq!(
481            source_phase.ready_sources.get(1).unwrap().source_id(),
482            SourceId::new(3)
483        );
484    }
485
486    #[test]
487    fn test_fire_and_schedule() {
488        let (source_phase, _) = setup_source_phase();
489        assert!(source_phase.source_handler.is_some());
490    }
491
492    #[test]
493    fn test_discard() {
494        let (mut source_phase, model) = setup_source_phase();
495
496        let hook = SharedHook::new(DiscardHook {
497            discarded_events: Rc::new(Mutex::new(Vec::new())),
498            discarded_sources: Rc::new(Mutex::new(Vec::new())),
499        });
500
501        source_phase
502            .get_context()
503            .hook_delegate
504            .add_shared_hook(hook.clone());
505
506        let source_a_entry = source_phase.take_one().unwrap();
507        source_phase.discard(&model, source_a_entry);
508
509        assert_eq!(hook.get_ref().discarded_sources.lock().unwrap().len(), 1);
510        assert_eq!(
511            hook.get_ref().discarded_sources.lock().unwrap()[0],
512            SourceId::new(1)
513        );
514    }
515
516    #[test]
517    fn test_complete_source_phase() {
518        let (source_phase, model) = setup_source_phase();
519        let micro_step_handler = source_phase.complete_source_phase(&model);
520
521        assert_eq!(
522            micro_step_handler.ref_context().current_tick(),
523            SimTime::zero()
524        );
525        assert!(micro_step_handler.ref_context().source_handler.is_some());
526    }
527}
des_sim/execution/phase/source.rs

des_sim/execution/phase/
source.rs
