ferroui_reactive/concurrency/
virtual_time_scheduler.rs

1//! `VirtualTimeSchedulerBase<TAbsolute, TRelative>` and `VirtualTimeScheduler<TAbsolute, TRelative>`
2//! (`VirtualTimeScheduler.cs`).
3
4use super::{IScheduler, IStopwatch, IStopwatchProvider, ScheduledAction, ScheduledItem};
5use crate::internal::PriorityQueue;
6use ferroui_system::IDisposable;
7use std::cell::{Cell, RefCell};
8use std::rc::{Rc, Weak};
9use std::time::{Duration, SystemTime};
10
11/// The time of a virtual-time scheduler: what the abstract members of upstream's
12/// `VirtualTimeSchedulerBase` state (`Add`, `ToDateTimeOffset`, `ToRelative`), and how an absolute
13/// schedule treats a due time (`ScheduleAbsolute`, which a derived scheduler may override).
14pub trait VirtualTimeClock: 'static {
15    /// The absolute time (`TAbsolute`).
16    type Absolute: Ord + Copy + 'static;
17    /// The relative time (`TRelative`).
18    type Relative: Copy + 'static;
19
20    /// `absolute` moved by `relative` (`Add`).
21    fn add(absolute: Self::Absolute, relative: Self::Relative) -> Self::Absolute;
22
23    /// The absolute time as a time of the system (`ToDateTimeOffset`).
24    fn to_system_time(absolute: Self::Absolute) -> SystemTime;
25
26    /// A duration as a relative time (`ToRelative`).
27    fn to_relative(duration: Duration) -> Self::Relative;
28
29    /// `absolute` moved back by `duration`: what upstream's `Add` does with the negative relative
30    /// time of a due time in the past. The default keeps `absolute` (the past is due now).
31    fn subtract(absolute: Self::Absolute, _duration: Duration) -> Self::Absolute {
32        absolute
33    }
34
35    /// The due time an absolute schedule uses for `due_time` when the clock is at `clock`
36    /// (`ScheduleAbsolute`): `due_time` itself. The test scheduler moves work due now or earlier to
37    /// one tick later.
38    fn adjust_due_time(due_time: Self::Absolute, _clock: Self::Absolute) -> Self::Absolute {
39        due_time
40    }
41}
42
43/// A scheduler of virtual time (`VirtualTimeScheduler<TAbsolute, TRelative>`): work runs when the
44/// clock is advanced to its due time (`start`, `advance_to`, `advance_by`), in the order of due
45/// times and, for equal ones, in the order it was scheduled.
46pub struct VirtualTimeScheduler<C: VirtualTimeClock> {
47    this: Weak<VirtualTimeScheduler<C>>,
48    clock: Cell<C::Absolute>,
49    is_enabled: Cell<bool>,
50    queue: RefCell<PriorityQueue<C::Absolute, Rc<ScheduledItem<C::Absolute>>>>,
51}
52
53impl<C: VirtualTimeClock> VirtualTimeScheduler<C> {
54    /// A scheduler with its clock at `initial_clock`.
55    pub fn new(initial_clock: C::Absolute) -> Rc<Self> {
56        Rc::new_cyclic(|this| Self {
57            this: this.clone(),
58            clock: Cell::new(initial_clock),
59            is_enabled: Cell::new(false),
60            queue: RefCell::new(PriorityQueue::new()),
61        })
62    }
63
64    fn this(&self) -> Rc<dyn IScheduler> {
65        self.this.upgrade().expect("the scheduler is alive while it runs work")
66    }
67
68    /// The current time of the clock (`Clock`).
69    pub fn clock(&self) -> C::Absolute {
70        self.clock.get()
71    }
72
73    /// Sets the clock (`Clock`'s protected setter), for derived schedulers.
74    pub fn set_clock(&self, clock: C::Absolute) {
75        self.clock.set(clock);
76    }
77
78    /// Whether the scheduler is running work (`IsEnabled`).
79    pub fn is_enabled(&self) -> bool {
80        self.is_enabled.get()
81    }
82
83    /// Schedules work at an absolute virtual time (`ScheduleAbsolute`).
84    pub fn schedule_absolute(&self, due_time: C::Absolute, action: ScheduledAction) -> Rc<dyn IDisposable> {
85        let due_time = C::adjust_due_time(due_time, self.clock.get());
86        let item = ScheduledItem::new(due_time, action);
87        self.queue.borrow_mut().push(due_time, item.clone());
88        item
89    }
90
91    /// Schedules work at a virtual time relative to the clock (`ScheduleRelative`).
92    pub fn schedule_relative(&self, due_time: C::Relative, action: ScheduledAction) -> Rc<dyn IDisposable> {
93        self.schedule_absolute(C::add(self.clock.get(), due_time), action)
94    }
95
96    /// The next work that has not been cancelled (`GetNext`), left in the queue.
97    fn peek_next(&self) -> Option<(C::Absolute, Rc<ScheduledItem<C::Absolute>>)> {
98        let mut queue = self.queue.borrow_mut();
99        loop {
100            let (due, item) = queue.peek().map(|(due, item)| (*due, item.clone()))?;
101            if item.is_canceled() {
102                queue.pop();
103                continue;
104            }
105            return Some((due, item));
106        }
107    }
108
109    fn run_next(&self, due: C::Absolute, item: Rc<ScheduledItem<C::Absolute>>) {
110        // The work leaves the queue as it starts (upstream removes it in the wrapper of the
111        // action): what it schedules for the same time runs after it.
112        self.queue.borrow_mut().remove_where(|queued| Rc::ptr_eq(queued, &item));
113        if due > self.clock.get() {
114            self.clock.set(due);
115        }
116        item.invoke(&self.this());
117    }
118
119    /// Runs the queued work, moving the clock to the due time of each, until there is none or the
120    /// scheduler is stopped (`Start`).
121    pub fn start(&self) {
122        if self.is_enabled.replace(true) {
123            return;
124        }
125        while self.is_enabled.get() {
126            match self.peek_next() {
127                Some((due, item)) => self.run_next(due, item),
128                None => self.is_enabled.set(false),
129            }
130        }
131    }
132
133    /// Stops the scheduler after the work that runs (`Stop`).
134    pub fn stop(&self) {
135        self.is_enabled.set(false);
136    }
137
138    /// Runs the work due up to `time` and moves the clock to it (`AdvanceTo`). Panics when `time`
139    /// is before the clock or the scheduler is running, where upstream throws.
140    pub fn advance_to(&self, time: C::Absolute) {
141        if time < self.clock.get() {
142            panic!("Specified argument was out of the range of valid values. (Parameter 'time')");
143        }
144        if time == self.clock.get() {
145            return;
146        }
147        if self.is_enabled.replace(true) {
148            panic!("Cannot advance while running: AdvanceTo.");
149        }
150        while self.is_enabled.get() {
151            match self.peek_next() {
152                Some((due, item)) if due <= time => self.run_next(due, item),
153                _ => self.is_enabled.set(false),
154            }
155        }
156        self.clock.set(time);
157    }
158
159    /// Runs the work due within `time` from the clock and moves the clock by it (`AdvanceBy`).
160    pub fn advance_by(&self, time: C::Relative) {
161        let target = C::add(self.clock.get(), time);
162        if target < self.clock.get() {
163            panic!("Specified argument was out of the range of valid values. (Parameter 'time')");
164        }
165        if target == self.clock.get() {
166            return;
167        }
168        if self.is_enabled.get() {
169            panic!("Cannot advance while running: AdvanceBy.");
170        }
171        self.advance_to(target);
172    }
173
174    /// Moves the clock by `time` without running work (`Sleep`).
175    pub fn sleep(&self, time: C::Relative) {
176        let target = C::add(self.clock.get(), time);
177        if target < self.clock.get() {
178            panic!("Specified argument was out of the range of valid values. (Parameter 'time')");
179        }
180        self.clock.set(target);
181    }
182}
183
184impl<C: VirtualTimeClock> IScheduler for VirtualTimeScheduler<C> {
185    fn now(&self) -> SystemTime {
186        C::to_system_time(self.clock.get())
187    }
188
189    fn schedule(&self, action: ScheduledAction) -> Rc<dyn IDisposable> {
190        self.schedule_absolute(self.clock.get(), action)
191    }
192
193    fn schedule_after(&self, due_time: Duration, action: ScheduledAction) -> Rc<dyn IDisposable> {
194        self.schedule_relative(C::to_relative(due_time), action)
195    }
196
197    fn schedule_at(&self, due_time: SystemTime, action: ScheduledAction) -> Rc<dyn IDisposable> {
198        match due_time.duration_since(self.now()) {
199            Ok(relative) => self.schedule_relative(C::to_relative(relative), action),
200            // A due time in the past keeps its place before the work due now
201            // (`ScheduleRelative(ToRelative(dueTime - Now))` with a negative time).
202            Err(past) => self.schedule_absolute(C::subtract(self.clock.get(), past.duration()), action),
203        }
204    }
205
206    fn as_stopwatch_provider(&self) -> Option<&dyn IStopwatchProvider> {
207        Some(self)
208    }
209}
210
211impl<C: VirtualTimeClock> IStopwatchProvider for VirtualTimeScheduler<C> {
212    fn start_stopwatch(&self) -> Box<dyn IStopwatch> {
213        Box::new(VirtualTimeStopwatch::<C> { scheduler: self.this.clone(), start: self.now() })
214    }
215}
216
217/// A stopwatch on the virtual clock (`VirtualTimeStopwatch`).
218struct VirtualTimeStopwatch<C: VirtualTimeClock> {
219    scheduler: Weak<VirtualTimeScheduler<C>>,
220    start: SystemTime,
221}
222
223impl<C: VirtualTimeClock> IStopwatch for VirtualTimeStopwatch<C> {
224    fn elapsed(&self) -> Duration {
225        match self.scheduler.upgrade() {
226            Some(scheduler) => scheduler.now().duration_since(self.start).unwrap_or(Duration::ZERO),
227            None => Duration::ZERO,
228        }
229    }
230}
ferroui_reactive/concurrency/virtual_time_scheduler.rs

ferroui_reactive/concurrency/
virtual_time_scheduler.rs
