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

nautilus_common/clock/
virtual.rs
