nautilus_common/clock/
virtual.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Deterministic clock for controlled time advancement.
17
18use std::{
19    collections::{BTreeMap, BinaryHeap},
20    ops::Deref,
21    sync::Arc,
22};
23
24use nautilus_core::{
25    AtomicTime, DurationNanos, UUID4, UnixNanos, correctness::check_predicate_true,
26    string::formatting::Separable,
27};
28use ustr::Ustr;
29
30use super::{
31    CallbackRegistry, Clock, replace_existing_timer, validate_and_prepare_schedule,
32    validate_and_prepare_time_alert, validate_and_prepare_timer,
33};
34use crate::timer::{
35    ScheduledTimeEvent, TimeEvent, TimeEventCallback, TimeEventHandler, TimerInterval,
36    TimerSchedule, VirtualTimer, create_valid_interval,
37};
38
39/// A deterministic clock for controlled time advancement.
40///
41/// The clock stores a manual timestamp, schedules [`VirtualTimer`] instances, and returns due events
42/// without waiting for wall-clock time.
43///
44/// # Threading
45///
46/// This clock is thread-affine; use it only from the thread that created it.
47#[derive(Debug)]
48pub struct VirtualClock {
49    time: AtomicTime,
50    timers: BTreeMap<Ustr, VirtualTimer>,
51    pub(super) timer_queue: BinaryHeap<ScheduledTimeEvent>,
52    callbacks: CallbackRegistry,
53}
54
55impl VirtualClock {
56    /// Creates a virtual clock at the UNIX epoch with no timers or callbacks.
57    #[must_use]
58    pub fn new() -> Self {
59        Self {
60            time: AtomicTime::new(false, UnixNanos::default()),
61            timers: BTreeMap::new(),
62            timer_queue: BinaryHeap::new(),
63            callbacks: CallbackRegistry::new(),
64        }
65    }
66
67    /// Advances active timers through `to_time_ns` and returns their due events.
68    ///
69    /// Events at `to_time_ns` are included and returned in ascending order by event timestamp and
70    /// timer name. If `set_time` is `true`, the stored clock timestamp is also set to `to_time_ns`;
71    /// otherwise the stored timestamp remains unchanged.
72    ///
73    /// # Warnings
74    ///
75    /// Logs a warning if at least 1,000,000 time events are allocated during advancement.
76    ///
77    /// # Panics
78    ///
79    /// Panics if `to_time_ns` is less than the current internal clock time.
80    pub fn advance_time(&mut self, to_time_ns: UnixNanos, set_time: bool) -> Vec<TimeEvent> {
81        const WARN_TIME_EVENTS_THRESHOLD: usize = 1_000_000;
82
83        let from_time_ns = self.time.get_time_ns();
84
85        assert!(
86            to_time_ns >= from_time_ns,
87            "Invariant: time must be non-decreasing, `to_time_ns` {to_time_ns} < `from_time_ns` {from_time_ns}"
88        );
89
90        if set_time {
91            self.time.set_time(to_time_ns);
92        }
93
94        let mut events: Vec<TimeEvent> = Vec::new();
95
96        while self
97            .timer_queue
98            .peek()
99            .is_some_and(|entry| entry.0.ts_event <= to_time_ns)
100        {
101            let entry = self
102                .timer_queue
103                .pop()
104                .expect("timer queue peeked Some but pop returned None");
105
106            let Some((event, next_event)) = self.advance_timer_from_entry(&entry.0) else {
107                continue;
108            };
109
110            events.push(event);
111
112            if let Some(next_event) = next_event {
113                self.timer_queue.push(next_event);
114            }
115        }
116
117        self.compact_timer_queue_if_needed();
118
119        if events.len() >= WARN_TIME_EVENTS_THRESHOLD {
120            log::warn!(
121                "Allocated {} time events during clock advancement from {} to {}, \
122                 consider stopping the timer between large time ranges with no data points",
123                events.len().separate_with_commas(),
124                from_time_ns,
125                to_time_ns
126            );
127        }
128
129        events.sort_by(|a, b| {
130            a.ts_event
131                .cmp(&b.ts_event)
132                .then_with(|| a.name.cmp(&b.name))
133        });
134
135        events
136    }
137
138    /// Matches time events with their registered callbacks, preserving input order.
139    ///
140    /// A named callback takes precedence over the default callback for each event.
141    ///
142    /// # Panics
143    ///
144    /// Panics if an event has neither a named nor default callback.
145    #[must_use]
146    pub fn match_handlers(&self, events: Vec<TimeEvent>) -> Vec<TimeEventHandler> {
147        events
148            .into_iter()
149            .map(|event| self.callbacks.get_handler(event))
150            .collect()
151    }
152
153    fn replace_existing_timer_if_needed(&mut self, name: &Ustr) {
154        replace_existing_timer(&mut self.timers, name);
155        self.compact_timer_queue_if_needed();
156    }
157
158    fn insert_timer(&mut self, timer: VirtualTimer) {
159        self.timer_queue.push(Self::scheduled_event(&timer));
160        self.timers.insert(timer.name, timer);
161        self.compact_timer_queue_if_needed();
162    }
163
164    fn advance_timer_from_entry(
165        &mut self,
166        entry: &TimeEvent,
167    ) -> Option<(TimeEvent, Option<ScheduledTimeEvent>)> {
168        let timer = self.timers.get_mut(&entry.name)?;
169        if timer.next_time_ns() != entry.ts_event {
170            return None;
171        }
172
173        let Some((event, _)) = timer.next() else {
174            self.timers.remove(&entry.name);
175            return None;
176        };
177
178        let next_entry = if timer.is_expired() {
179            self.timers.remove(&entry.name);
180            None
181        } else {
182            Some(Self::scheduled_event(timer))
183        };
184
185        Some((event, next_entry))
186    }
187
188    fn compact_timer_queue_if_needed(&mut self) {
189        if self.timer_queue.len() > self.timers.len().saturating_mul(2) {
190            self.compact_timer_queue();
191        }
192    }
193
194    fn compact_timer_queue(&mut self) {
195        self.timer_queue = self.timers.values().map(Self::scheduled_event).collect();
196    }
197
198    fn scheduled_event(timer: &VirtualTimer) -> ScheduledTimeEvent {
199        ScheduledTimeEvent::new(TimeEvent::new(
200            timer.name,
201            UUID4::new(),
202            timer.next_time_ns(),
203            timer.next_time_ns(),
204        ))
205    }
206    fn register_timer(
207        &mut self,
208        name: Ustr,
209        interval: TimerInterval,
210        start_time_ns: UnixNanos,
211        stop_time_ns: Option<UnixNanos>,
212        callback: Option<TimeEventCallback>,
213        fire_immediately: bool,
214    ) -> anyhow::Result<()> {
215        check_predicate_true(
216            callback.is_some() | self.callbacks.has_any_callback(&name),
217            "No callbacks provided",
218        )?;
219
220        self.replace_existing_timer_if_needed(&name);
221        if let Some(callback) = callback {
222            self.callbacks.register_callback(name, callback);
223        }
224
225        let timer = VirtualTimer::new(
226            name,
227            interval,
228            start_time_ns,
229            stop_time_ns,
230            fire_immediately,
231        );
232        self.insert_timer(timer);
233
234        Ok(())
235    }
236}
237
238impl Default for VirtualClock {
239    /// Creates a new default [`VirtualClock`] instance.
240    fn default() -> Self {
241        Self::new()
242    }
243}
244
245impl Deref for VirtualClock {
246    type Target = AtomicTime;
247
248    fn deref(&self) -> &Self::Target {
249        &self.time
250    }
251}
252
253impl Clock for VirtualClock {
254    fn timestamp_ns(&self) -> UnixNanos {
255        self.time.get_time_ns()
256    }
257
258    fn timestamp_us(&self) -> u64 {
259        self.time.get_time_us()
260    }
261
262    fn timestamp_ms(&self) -> u64 {
263        self.time.get_time_ms()
264    }
265
266    fn timestamp(&self) -> f64 {
267        self.time.get_time()
268    }
269
270    fn timer_names(&self) -> Vec<&str> {
271        self.timers
272            .iter()
273            .filter(|(_, timer)| !timer.is_expired())
274            .map(|(k, _)| k.as_str())
275            .collect()
276    }
277
278    fn timer_count(&self) -> usize {
279        self.timers
280            .iter()
281            .filter(|(_, timer)| !timer.is_expired())
282            .count()
283    }
284
285    fn timer_exists(&self, name: &Ustr) -> bool {
286        self.timers
287            .get(name)
288            .is_some_and(|timer| !timer.is_expired())
289    }
290
291    fn register_default_handler(&mut self, callback: TimeEventCallback) {
292        self.callbacks.register_default_handler(callback);
293    }
294
295    fn cancel_default_handler(&mut self) {
296        self.callbacks.cancel_default_handler();
297    }
298
299    fn cancel_callbacks(&mut self) {
300        self.callbacks.clear();
301    }
302
303    fn set_time_alert_ns(
304        &mut self,
305        name: &str,
306        alert_time_ns: UnixNanos,
307        callback: Option<TimeEventCallback>,
308        allow_past: Option<bool>,
309    ) -> anyhow::Result<()> {
310        let ts_now = self.get_time_ns();
311        let (name, alert_time_ns) =
312            validate_and_prepare_time_alert(name, alert_time_ns, allow_past, ts_now)?;
313
314        check_predicate_true(
315            callback.is_some() | self.callbacks.has_any_callback(&name),
316            "No callbacks provided",
317        )?;
318
319        self.replace_existing_timer_if_needed(&name);
320
321        if let Some(callback) = callback {
322            self.callbacks.register_callback(name, callback);
323        }
324
325        // Safe to calculate interval now that we've ensured alert_time_ns >= ts_now
326        let interval_ns = create_valid_interval(alert_time_ns - ts_now);
327        let fire_immediately = alert_time_ns == ts_now;
328
329        let timer = VirtualTimer::new(
330            name,
331            interval_ns,
332            ts_now,
333            Some(alert_time_ns),
334            fire_immediately,
335        );
336        self.insert_timer(timer);
337
338        Ok(())
339    }
340
341    fn set_timer_ns(
342        &mut self,
343        name: &str,
344        interval_ns: DurationNanos,
345        start_time_ns: Option<UnixNanos>,
346        stop_time_ns: Option<UnixNanos>,
347        callback: Option<TimeEventCallback>,
348        allow_past: Option<bool>,
349        fire_immediately: Option<bool>,
350    ) -> anyhow::Result<()> {
351        let (name, start_time_ns, stop_time_ns, _, fire_immediately) = validate_and_prepare_timer(
352            name,
353            interval_ns,
354            start_time_ns,
355            stop_time_ns,
356            allow_past,
357            fire_immediately,
358            self.get_time_ns(),
359        )?;
360        self.register_timer(
361            name,
362            TimerInterval::Fixed(create_valid_interval(interval_ns)),
363            start_time_ns,
364            stop_time_ns,
365            callback,
366            fire_immediately,
367        )
368    }
369
370    fn set_timer_schedule(
371        &mut self,
372        name: &str,
373        schedule: Arc<dyn TimerSchedule>,
374        stop_time_ns: Option<UnixNanos>,
375        callback: Option<TimeEventCallback>,
376        allow_past: Option<bool>,
377        fire_immediately: Option<bool>,
378    ) -> anyhow::Result<()> {
379        let start_time_ns = schedule.start_time_ns();
380        let interval = TimerInterval::Schedule(schedule);
381        let (name, start_time_ns, stop_time_ns, _, fire_immediately) =
382            validate_and_prepare_schedule(
383                name,
384                &interval,
385                Some(start_time_ns),
386                stop_time_ns,
387                allow_past,
388                fire_immediately,
389                self.get_time_ns(),
390            )?;
391        self.register_timer(
392            name,
393            interval,
394            start_time_ns,
395            stop_time_ns,
396            callback,
397            fire_immediately,
398        )
399    }
400
401    fn next_time_ns(&self, name: &str) -> Option<UnixNanos> {
402        self.timers
403            .get(&Ustr::from(name))
404            .filter(|timer| !timer.is_expired())
405            .map(VirtualTimer::next_time_ns)
406    }
407
408    fn cancel_timer(&mut self, name: &str) {
409        let timer = self.timers.remove(&Ustr::from(name));
410        if let Some(mut timer) = timer {
411            timer.cancel();
412        }
413
414        self.compact_timer_queue_if_needed();
415    }
416
417    fn cancel_timers(&mut self) {
418        for timer in &mut self.timers.values_mut() {
419            timer.cancel();
420        }
421
422        self.timers.clear();
423        self.timer_queue.clear();
424    }
425
426    fn reset(&mut self) {
427        self.time = AtomicTime::new(false, UnixNanos::default());
428        self.timers = BTreeMap::new();
429        self.timer_queue = BinaryHeap::new();
430        self.callbacks.clear();
431    }
432}
nautilus_common/clock/virtual.rs

nautilus_common/clock/
virtual.rs
