des_sim/context/
event.rs

1//! The `event` module provides the `EventContext`, which is used by models to interact with the simulation environment
2//! during event processing.
3//!
4//! It allows models to schedule new events, add new sources, and cancel existing scheduled events or sources.
5
6use crate::context::UserContext;
7use crate::event_scheduler::EventScheduler;
8use crate::modeling::event::{Event, EventPriority};
9use crate::modeling::hook::Hook;
10use crate::modeling::hook::instance::HookDelegate;
11use crate::modeling::model::Model;
12use crate::modeling::source::Source;
13use crate::primitive::time::{Duration, MicroStep, MicroStepStatus, SimTime, TickStatus};
14use crate::source_handler::{SourceHandler, SourceReadyEntry, SourceView};
15
16/// Holds and manages context information during event processing.
17///
18/// This provides access to the current simulation time, micro-step status,
19/// hook management, source handling, and the event scheduler during model execution.
20pub struct EventContext<E, M: Model<E>> {
21    pub(crate) current_tick_status: TickStatus,
22    pub(crate) current_micro_step_status: MicroStepStatus,
23    pub(crate) hook_delegate: HookDelegate<E, M>,
24    pub(crate) source_handler: SourceHandler<E, M>,
25    pub(crate) event_scheduler: EventScheduler<E>,
26}
27
28impl<E, M: Model<E>> UserContext<E, M> for EventContext<E, M> {
29    fn current_tick(&self) -> SimTime {
30        self.current_tick_status.current()
31    }
32
33    fn current_micro_step(&self) -> MicroStep {
34        self.current_micro_step_status.current()
35    }
36
37    fn schedule_event(&mut self, delay: Duration, priority: EventPriority, event_payload: E) {
38        self.event_scheduler
39            .schedule(self.current_tick(), delay, priority, event_payload);
40    }
41}
42
43impl<E, M: Model<E>> EventContext<E, M> {
44    /// Retrieves the hooks associated with the current event processing.
45    pub(crate) fn hook(&self) -> &impl Hook<E, M> {
46        &self.hook_delegate
47    }
48
49    /// Registers a new source and adds it to the event processing loop.
50    ///
51    /// # Arguments
52    /// * `model` - The model associated with the source.
53    /// * `name` - A unique identifier for the source.
54    /// * `source` - The source implementation to register.
55    pub fn add_source<S>(&mut self, model: &M, name: &'static str, mut source: S)
56    where
57        S: Source<E, M> + 'static,
58    {
59        self.hook().before_register_source(model, name);
60        let first_fire_delay = source.on_registered(self, model);
61        self.source_handler.add_source_after_registered_action(
62            name,
63            self.current_tick(),
64            first_fire_delay,
65            source,
66        );
67        self.hook().after_register_source(model, name);
68    }
69
70    /// Cancels scheduled sources that satisfy the provided condition.
71    ///
72    /// # Arguments
73    /// * `model` - The associated model.
74    /// * `pred` - A predicate function to determine if a source should be canceled.
75    ///
76    /// # Returns
77    /// A `Vec` containing information about the canceled sources.
78    pub fn cancel_scheduled_sources<S, F>(
79        &mut self,
80        model: &M,
81        pred: F,
82    ) -> Vec<(SimTime, SourceReadyEntry)>
83    where
84        S: Source<E, M> + 'static,
85        F: FnMut(SimTime, &SourceReadyEntry) -> bool,
86    {
87        let mut result = Vec::new();
88        let now = self.current_tick();
89        let micro_step = self.current_micro_step();
90        let canceled = self.source_handler.drain_cancel_scheduled(pred);
91
92        canceled.into_iter().for_each(|(scheduled_at, entry)| {
93            self.hook().cancel_source(
94                model,
95                now,
96                micro_step,
97                scheduled_at,
98                &SourceView::new(entry.source_id(), entry.clone_name_arc()),
99            );
100            result.push((scheduled_at, entry));
101        });
102
103        result
104    }
105
106    /// Cancels scheduled events that satisfy the provided condition.
107    ///
108    /// # Arguments
109    /// * `model` - The associated model.
110    /// * `pred` - A predicate function to determine if an event should be canceled.
111    ///
112    /// # Returns
113    /// A `Vec` containing information about the canceled events.
114    pub fn cancel_scheduled_events<F>(&mut self, model: &M, pred: F) -> Vec<(SimTime, Event<E>)>
115    where
116        F: FnMut(SimTime, &Event<E>) -> bool,
117    {
118        let mut result = Vec::new();
119        let now = self.current_tick();
120        let micro_step = self.current_micro_step();
121        let canceled = self.event_scheduler.drain_cancel_scheduled(pred);
122
123        canceled.into_iter().for_each(|(scheduled_at, event)| {
124            self.hook()
125                .cancel_event(model, now, micro_step, scheduled_at, &event);
126            result.push((scheduled_at, event));
127        });
128
129        result
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136    use crate::context::SourceContext;
137    use crate::modeling::hook::instance::SharedHook;
138    use std::cell::RefCell;
139    use std::fmt::Debug;
140    use std::rc::Rc;
141
142    /// Event structure for testing purposes.
143    #[derive(Debug, Clone, PartialEq)]
144    struct TestEvent;
145
146    /// Model definition for testing purposes.
147    struct TestModel;
148
149    impl Model<TestEvent> for TestModel {
150        fn handle_event(
151            &mut self,
152            _event_context: &mut EventContext<TestEvent, Self>,
153            _event: &Event<TestEvent>,
154        ) {
155            // No-op for test purposes.
156        }
157    }
158
159    /// Source implementation for testing purposes.
160    struct TestSource {
161        initial_delay: Duration,
162    }
163
164    impl Source<TestEvent, TestModel> for TestSource {
165        fn on_registered(
166            &mut self,
167            context: &mut dyn UserContext<TestEvent, TestModel>,
168            _model: &TestModel,
169        ) -> Option<Duration> {
170            context.schedule_event(Duration::ticks(5), EventPriority::minimum(), TestEvent);
171            Some(self.initial_delay)
172        }
173
174        fn fire(
175            &mut self,
176            context: &mut SourceContext<TestEvent, TestModel>,
177            _model: &TestModel,
178        ) -> Option<Duration> {
179            context.schedule_event(Duration::ticks(5), EventPriority::minimum(), TestEvent);
180            Some(Duration::ticks(5))
181        }
182    }
183
184    /// Mock implementation for tracking hook invocation history.
185    ///
186    /// Used to verify that hook methods are called as expected during tests.
187    #[derive(Default)]
188    struct MockHook {
189        before_register_source_called: Rc<RefCell<Vec<String>>>,
190        after_register_source_called: Rc<RefCell<Vec<String>>>,
191        cancel_source_called: Rc<RefCell<Vec<String>>>,
192        cancel_event_called: Rc<RefCell<Vec<String>>>,
193    }
194
195    impl MockHook {
196        /// Creates a new mock hook.
197        fn new() -> Self {
198            Default::default()
199        }
200    }
201
202    impl<E: Debug, M: Model<E>> Hook<E, M> for MockHook {
203        fn before_simulation(&self, _model: &M) {
204            unreachable!();
205        }
206        fn after_simulation(&self, _model: &M, _end_tick: SimTime) {
207            unreachable!();
208        }
209        fn before_tick(&self, _model: &M, _current_tick: SimTime, _skipped_duration: Duration) {
210            unreachable!();
211        }
212        fn after_tick(&self, _model: &M, _current_tick: SimTime, _last_micro_step: MicroStep) {
213            unreachable!();
214        }
215        fn before_micro_step(
216            &self,
217            _model: &M,
218            _current_tick: SimTime,
219            _current_micro_step: MicroStep,
220        ) {
221            unreachable!();
222        }
223        fn after_micro_step(
224            &self,
225            _model: &M,
226            _current_tick: SimTime,
227            _current_micro_step: MicroStep,
228        ) {
229            unreachable!();
230        }
231        fn on_discard_remain_micro_step(
232            &self,
233            _model: &M,
234            _current_tick: SimTime,
235            _first_discarded_micro_step: MicroStep,
236            _discarded_sources: &[SourceReadyEntry],
237            _discarded_events: &[Event<E>],
238        ) {
239            unreachable!();
240        }
241
242        fn before_register_source(&self, _model: &M, name: &str) {
243            self.before_register_source_called
244                .borrow_mut()
245                .push(name.to_string());
246        }
247
248        fn after_register_source(&self, _model: &M, name: &str) {
249            self.after_register_source_called
250                .borrow_mut()
251                .push(name.to_string());
252        }
253
254        fn before_source_phase(
255            &self,
256            _model: &M,
257            _current_tick: SimTime,
258            _current_micro_step: MicroStep,
259        ) {
260            unreachable!();
261        }
262
263        fn before_source(
264            &self,
265            _model: &M,
266            _current_tick: SimTime,
267            _current_micro_step: MicroStep,
268            _source_view: &SourceView,
269        ) {
270            unreachable!();
271        }
272        fn after_source(
273            &self,
274            _model: &M,
275            _current_tick: SimTime,
276            _current_micro_step: MicroStep,
277            _source_view: &SourceView,
278            _computed_next_fire: Option<SimTime>,
279        ) {
280            unreachable!();
281        }
282
283        fn cancel_source(
284            &self,
285            _model: &M,
286            _now: SimTime,
287            _micro_step: MicroStep,
288            _scheduled_at: SimTime,
289            source_view: &SourceView,
290        ) {
291            self.cancel_source_called
292                .borrow_mut()
293                .push(source_view.name().to_string());
294        }
295
296        fn discard_source(
297            &self,
298            _model: &M,
299            _current_tick: SimTime,
300            _current_micro_step: MicroStep,
301            _source_view: &SourceView,
302        ) {
303            unreachable!();
304        }
305
306        fn after_source_phase(
307            &self,
308            _model: &M,
309            _current_tick: SimTime,
310            _current_micro_step: MicroStep,
311        ) {
312            unreachable!();
313        }
314        fn before_event_phase(
315            &self,
316            _model: &M,
317            _current_tick: SimTime,
318            _current_micro_step: MicroStep,
319        ) {
320            unreachable!();
321        }
322        fn before_event(
323            &self,
324            _model: &M,
325            _current_tick: SimTime,
326            _current_micro_step: MicroStep,
327            _event: &Event<E>,
328        ) {
329            unreachable!();
330        }
331        fn after_event(
332            &self,
333            _model: &M,
334            _current_tick: SimTime,
335            _current_micro_step: MicroStep,
336            _event: &Event<E>,
337        ) {
338            unreachable!();
339        }
340
341        fn cancel_event(
342            &self,
343            _model: &M,
344            _now: SimTime,
345            _micro_step: MicroStep,
346            _scheduled_at: SimTime,
347            event: &Event<E>,
348        ) {
349            self.cancel_event_called
350                .borrow_mut()
351                .push(format!("{:?}", event.payload));
352        }
353
354        fn discard_event(
355            &self,
356            _model: &M,
357            _current_tick: SimTime,
358            _current_micro_step: MicroStep,
359            _event: &Event<E>,
360        ) {
361            unreachable!();
362        }
363        fn after_event_phase(
364            &self,
365            _model: &M,
366            _current_tick: SimTime,
367            _current_micro_step: MicroStep,
368        ) {
369            unreachable!();
370        }
371    }
372
373    /// Sets up the test environment, returning an EventContext and a SharedHook.
374    fn setup() -> (
375        EventContext<TestEvent, TestModel>,
376        SharedHook<TestEvent, TestModel, MockHook>,
377    ) {
378        let test_hook = MockHook::new();
379        let shared_hook = SharedHook::new(test_hook);
380        let mut hook_delegate = HookDelegate::new();
381        hook_delegate.add_shared_hook(shared_hook.clone());
382
383        let context = EventContext {
384            current_tick_status: TickStatus::initialize(),
385            current_micro_step_status: MicroStepStatus::new(MicroStep::zero()),
386            hook_delegate,
387            source_handler: SourceHandler::new(),
388            event_scheduler: EventScheduler::new(),
389        };
390
391        (context, shared_hook)
392    }
393
394    #[test]
395    fn test_current_tick() {
396        let (context, _) = setup();
397        assert_eq!(context.current_tick(), SimTime::from_ticks(0));
398    }
399
400    #[test]
401    fn test_current_micro_step() {
402        let (context, _) = setup();
403        assert_eq!(context.current_micro_step(), MicroStep::zero());
404    }
405
406    #[test]
407    fn test_add_source_after() {
408        let model = TestModel;
409        let (mut context, shared_hook) = setup();
410        let initial_sources_count = context.source_handler.ready_queue_len();
411        let delay = Duration::ticks(10);
412
413        context.add_source(
414            &model,
415            "test_source",
416            TestSource {
417                initial_delay: delay,
418            },
419        );
420
421        context.source_handler.flush_pending();
422        context.event_scheduler.flush_pending(); // Ensure event scheduler is also flushed if it was used by source initialization
423
424        assert_eq!(
425            context.source_handler.ready_queue_len(),
426            initial_sources_count + 1
427        );
428
429        let (scheduled_at, scheduled_source) = context.source_handler.peek().unwrap();
430        assert_eq!(scheduled_at, SimTime::from_ticks(0) + delay);
431        assert_eq!(scheduled_source.source_id.value(), 0); // Assuming it's the first source added
432
433        assert_eq!(
434            shared_hook
435                .get_ref()
436                .before_register_source_called
437                .borrow()
438                .len(),
439            1
440        );
441        assert_eq!(
442            shared_hook.get_ref().before_register_source_called.borrow()[0],
443            "test_source"
444        );
445        assert_eq!(
446            shared_hook
447                .get_ref()
448                .after_register_source_called
449                .borrow()
450                .len(),
451            1
452        );
453        assert_eq!(
454            shared_hook.get_ref().after_register_source_called.borrow()[0],
455            "test_source"
456        );
457    }
458
459    #[test]
460    fn test_add_source_at_now() {
461        let model = TestModel;
462        let (mut context, shared_hook) = setup();
463        let initial_sources_count = context.source_handler.ready_queue_len();
464
465        context.add_source(
466            &model,
467            "test_source_now",
468            TestSource {
469                initial_delay: Duration::zero(),
470            },
471        );
472
473        context.source_handler.flush_pending();
474        context.event_scheduler.flush_pending(); // Ensure event scheduler is also flushed if it was used by source initialization
475
476        assert_eq!(
477            context.source_handler.ready_queue_len(),
478            initial_sources_count + 1
479        );
480
481        let (scheduled_at, scheduled_source) = context.source_handler.peek().unwrap();
482        assert_eq!(scheduled_at, SimTime::from_ticks(0));
483        assert_eq!(scheduled_source.source_id.value(), 0); // Assuming it's the first source added
484
485        assert_eq!(
486            shared_hook
487                .get_ref()
488                .before_register_source_called
489                .borrow()
490                .len(),
491            1
492        );
493        assert_eq!(
494            shared_hook.get_ref().before_register_source_called.borrow()[0],
495            "test_source_now"
496        );
497        assert_eq!(
498            shared_hook
499                .get_ref()
500                .after_register_source_called
501                .borrow()
502                .len(),
503            1
504        );
505        assert_eq!(
506            shared_hook.get_ref().after_register_source_called.borrow()[0],
507            "test_source_now"
508        );
509    }
510
511    #[test]
512    fn test_schedule_event() {
513        let (mut context, _) = setup();
514
515        // Populate existing events
516        context.event_scheduler.schedule(
517            SimTime::from_ticks(0),
518            Duration::one(),
519            EventPriority::minimum(),
520            TestEvent,
521        );
522        context.event_scheduler.schedule(
523            SimTime::from_ticks(0),
524            Duration::one(),
525            EventPriority::minimum(),
526            TestEvent,
527        );
528        context.event_scheduler.schedule(
529            SimTime::from_ticks(0),
530            Duration::one(),
531            EventPriority::minimum(),
532            TestEvent,
533        );
534        context.event_scheduler.flush_pending();
535
536        let initial_scheduled_events_count = context.event_scheduler.ready_queue_len();
537        let delay = Duration::ticks(5);
538        let priority = EventPriority::new(10);
539        let event_payload = TestEvent;
540
541        context.schedule_event(delay, priority, event_payload.clone());
542        context.source_handler.flush_pending(); // Ensure source handler is also flushed
543        context.event_scheduler.flush_pending();
544
545        assert_eq!(
546            context.event_scheduler.ready_queue_len(),
547            initial_scheduled_events_count + 1
548        );
549
550        let (scheduled_at, event) = context.event_scheduler.peek().unwrap();
551        assert_eq!(scheduled_at, SimTime::from_ticks(1));
552        assert_eq!(event.priority, EventPriority::minimum());
553        assert_eq!(event.payload, TestEvent);
554    }
555
556    #[test]
557    fn test_cancel_scheduled_events() {
558        let (mut context, shared_hook) = setup();
559        let model = TestModel;
560
561        context.schedule_event(Duration::ticks(5), EventPriority::minimum(), TestEvent);
562        context.schedule_event(Duration::ticks(10), EventPriority::minimum(), TestEvent);
563        context.source_handler.flush_pending();
564        context.event_scheduler.flush_pending();
565
566        let canceled_events = context.cancel_scheduled_events(&model, |_, _| true);
567        assert_eq!(canceled_events.len(), 2);
568        assert_eq!(context.event_scheduler.ready_queue_len(), 0);
569        assert_eq!(shared_hook.get_ref().cancel_event_called.borrow().len(), 2);
570        assert_eq!(
571            shared_hook.get_ref().cancel_event_called.borrow()[0],
572            "TestEvent"
573        );
574        assert_eq!(
575            shared_hook.get_ref().cancel_event_called.borrow()[1],
576            "TestEvent"
577        );
578
579        // Re-schedule
580        context.schedule_event(Duration::ticks(5), EventPriority::minimum(), TestEvent);
581        context.schedule_event(Duration::ticks(10), EventPriority::minimum(), TestEvent);
582
583        context.source_handler.flush_pending();
584        context.event_scheduler.flush_pending();
585
586        let canceled_events_filtered = context
587            .cancel_scheduled_events(&model, |scheduled_at, _| {
588                scheduled_at == SimTime::from_ticks(5)
589            });
590        assert_eq!(canceled_events_filtered.len(), 1);
591        assert_eq!(canceled_events_filtered[0].0, SimTime::from_ticks(5));
592        assert_eq!(context.event_scheduler.ready_queue_len(), 1);
593        let (remaining_scheduled_at, _) = context.event_scheduler.peek().unwrap();
594        assert_eq!(remaining_scheduled_at, SimTime::from_ticks(10));
595        assert_eq!(shared_hook.get_ref().cancel_event_called.borrow().len(), 3);
596        assert_eq!(
597            shared_hook.get_ref().cancel_event_called.borrow()[2],
598            "TestEvent"
599        );
600    }
601
602    #[test]
603    fn test_cancel_scheduled_sources() {
604        let (mut context, shared_hook) = setup();
605        let model = TestModel;
606
607        context.add_source(
608            &model,
609            "source1",
610            TestSource {
611                initial_delay: Duration::ticks(5),
612            },
613        );
614        context.add_source(
615            &model,
616            "source2",
617            TestSource {
618                initial_delay: Duration::ticks(10),
619            },
620        );
621        context.source_handler.flush_pending();
622        context.event_scheduler.flush_pending();
623
624        let canceled_sources =
625            context.cancel_scheduled_sources::<TestSource, _>(&model, |_, _| true);
626        assert_eq!(canceled_sources.len(), 2);
627        assert_eq!(context.source_handler.ready_queue_len(), 0);
628        assert_eq!(shared_hook.get_ref().cancel_source_called.borrow().len(), 2);
629        assert_eq!(
630            shared_hook.get_ref().cancel_source_called.borrow()[0],
631            "source1"
632        );
633        assert_eq!(
634            shared_hook.get_ref().cancel_source_called.borrow()[1],
635            "source2"
636        );
637
638        // Re-schedule
639        context.add_source(
640            &model,
641            "source3",
642            TestSource {
643                initial_delay: Duration::ticks(5),
644            },
645        );
646        context.add_source(
647            &model,
648            "source4",
649            TestSource {
650                initial_delay: Duration::ticks(10),
651            },
652        );
653        context.source_handler.flush_pending();
654        context.event_scheduler.flush_pending();
655
656        let canceled_sources_filtered = context
657            .cancel_scheduled_sources::<TestSource, _>(&model, |scheduled_at, _| {
658                scheduled_at == SimTime::from_ticks(5)
659            });
660        assert_eq!(canceled_sources_filtered.len(), 1);
661        assert_eq!(canceled_sources_filtered[0].0, SimTime::from_ticks(5));
662        assert_eq!(context.source_handler.ready_queue_len(), 1);
663        let (remaining_scheduled_at, _) = context.source_handler.peek().unwrap();
664        assert_eq!(remaining_scheduled_at, SimTime::from_ticks(10));
665        assert_eq!(shared_hook.get_ref().cancel_source_called.borrow().len(), 3);
666        assert_eq!(
667            shared_hook.get_ref().cancel_source_called.borrow()[2],
668            "source3"
669        );
670    }
671}
des_sim/context/event.rs

des_sim/context/
event.rs
