nautilus_common/live/
timer.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//! Live timer implementation using Tokio for real-time scheduling.
17
18use std::{
19    num::NonZeroU64,
20    sync::{
21        Arc,
22        atomic::{self, AtomicU64},
23    },
24};
25
26use nautilus_core::{
27    UUID4, UnixNanos,
28    correctness::{FAILED, check_valid_string_utf8},
29    datetime::floor_to_nearest_microsecond,
30    time::get_atomic_clock_realtime,
31};
32use tokio::{
33    task::JoinHandle,
34    time::{Duration, Instant},
35};
36use ustr::Ustr;
37
38use super::runtime::get_runtime;
39use crate::{
40    runner::TimeEventSender,
41    timer::{TimeEvent, TimeEventCallback, TimeEventHandler, Timer},
42};
43
44/// A live timer for use with a `LiveClock`.
45///
46/// `LiveTimer` triggers events at specified intervals in a real-time environment,
47/// using Tokio's async runtime to handle scheduling and execution.
48///
49/// # Threading
50///
51/// The timer runs on the runtime thread that created it and dispatches events across threads as needed.
52#[derive(Debug)]
53pub struct LiveTimer {
54    /// The name of the timer.
55    pub name: Ustr,
56    /// The start time of the timer in UNIX nanoseconds.
57    pub interval_ns: NonZeroU64,
58    /// The start time of the timer in UNIX nanoseconds.
59    pub start_time_ns: UnixNanos,
60    /// The optional stop time of the timer in UNIX nanoseconds.
61    pub stop_time_ns: Option<UnixNanos>,
62    /// If the timer should fire immediately at start time.
63    pub fire_immediately: bool,
64    next_time_ns: Arc<AtomicU64>,
65    callback: TimeEventCallback,
66    task_handle: Option<JoinHandle<()>>,
67    canceled: bool,
68    sender: Option<Arc<dyn TimeEventSender>>,
69}
70
71impl LiveTimer {
72    /// Creates a new [`LiveTimer`] instance.
73    ///
74    /// # Panics
75    ///
76    /// Panics if `name` is not a valid string.
77    #[must_use]
78    pub fn new(
79        name: Ustr,
80        interval_ns: NonZeroU64,
81        start_time_ns: UnixNanos,
82        stop_time_ns: Option<UnixNanos>,
83        callback: TimeEventCallback,
84        fire_immediately: bool,
85        sender: Option<Arc<dyn TimeEventSender>>,
86    ) -> Self {
87        check_valid_string_utf8(name, stringify!(name)).expect(FAILED);
88
89        let next_time_ns = if fire_immediately {
90            start_time_ns.as_u64()
91        } else {
92            start_time_ns.as_u64() + interval_ns.get()
93        };
94
95        log::trace!("Creating timer '{name}'");
96
97        Self {
98            name,
99            interval_ns,
100            start_time_ns,
101            stop_time_ns,
102            fire_immediately,
103            next_time_ns: Arc::new(AtomicU64::new(next_time_ns)),
104            callback,
105            task_handle: None,
106            canceled: false,
107            sender,
108        }
109    }
110
111    /// Returns the next time in UNIX nanoseconds when the timer will fire.
112    ///
113    /// Provides the scheduled time for the next event based on the current state of the timer.
114    #[must_use]
115    pub fn next_time_ns(&self) -> UnixNanos {
116        UnixNanos::from(self.next_time_ns.load(atomic::Ordering::SeqCst))
117    }
118
119    /// Returns whether the timer is expired.
120    ///
121    /// An expired timer will not trigger any further events.
122    /// A timer that has not been started is not expired.
123    #[must_use]
124    pub fn is_expired(&self) -> bool {
125        self.canceled
126            || self
127                .task_handle
128                .as_ref()
129                .is_some_and(tokio::task::JoinHandle::is_finished)
130    }
131
132    /// Starts the timer.
133    ///
134    /// Time events will begin triggering at the specified intervals.
135    /// The generated events are handled by the provided callback function.
136    ///
137    /// # Panics
138    ///
139    /// Panics if using a Rust callback (`Rust` or `RustLocal`) without a `TimeEventSender`.
140    #[allow(unused_variables)]
141    pub fn start(&mut self) {
142        let event_name = self.name;
143        let stop_time_ns = self.stop_time_ns;
144        let interval_ns = self.interval_ns.get();
145
146        if self.callback.is_local() {
147            log::debug!(
148                "Timer '{event_name}' uses a RustLocal callback on a live Tokio timer; \
149                 callback registry dispatch is needed to avoid cloning Rc on worker threads"
150            );
151        }
152
153        let callback = self.callback.clone();
154
155        // Get current time
156        let clock = get_atomic_clock_realtime();
157        let now_ns = clock.get_time_ns();
158
159        // Check if the timer's alert time is in the past and adjust if needed
160        let now_raw = now_ns.as_u64();
161        let mut observed_next = self.next_time_ns.load(atomic::Ordering::SeqCst);
162
163        if observed_next <= now_raw {
164            loop {
165                match self.next_time_ns.compare_exchange(
166                    observed_next,
167                    now_raw,
168                    atomic::Ordering::SeqCst,
169                    atomic::Ordering::SeqCst,
170                ) {
171                    Ok(_) => {
172                        if observed_next < now_raw {
173                            let original = UnixNanos::from(observed_next);
174                            log::warn!(
175                                "Timer '{event_name}' alert time {} was in the past, adjusted to current time for immediate fire",
176                                original.to_rfc3339(),
177                            );
178                        }
179                        observed_next = now_raw;
180                        break;
181                    }
182                    Err(actual) => {
183                        observed_next = actual;
184                        if observed_next > now_raw {
185                            break;
186                        }
187                    }
188                }
189            }
190        }
191
192        // Floor the next time to the nearest microsecond which is within the timers accuracy
193        let mut next_time_ns = UnixNanos::from(floor_to_nearest_microsecond(observed_next));
194        let next_time_atomic = self.next_time_ns.clone();
195        next_time_atomic.store(next_time_ns.as_u64(), atomic::Ordering::SeqCst);
196
197        let sender = self.sender.clone();
198
199        let rt = get_runtime();
200        let handle = rt.spawn(async move {
201            let clock = get_atomic_clock_realtime();
202
203            // 1-millisecond delay to account for the overhead of initializing a tokio timer
204            let overhead = Duration::from_millis(1);
205            let delay_ns = next_time_ns.saturating_sub(now_ns.as_u64());
206            let mut delay = Duration::from_nanos(delay_ns);
207
208            // Subtract the estimated startup overhead; saturating to zero for sub-ms delays
209            if delay > overhead {
210                delay -= overhead;
211            } else {
212                delay = Duration::from_nanos(0);
213            }
214
215            let start = Instant::now() + delay;
216
217            let mut timer = tokio::time::interval_at(start, Duration::from_nanos(interval_ns));
218
219            loop {
220                // `timer.tick` is cancellation safe, if the cancel branch completes
221                // first then no tick has been consumed (no event was ready).
222                timer.tick().await;
223                let now_ns = clock.get_time_ns();
224
225                let event = TimeEvent::new(event_name, UUID4::new(), next_time_ns, now_ns);
226
227                if let Some(sender) = sender.as_ref() {
228                    // TODO: `RustLocal` still clones an `Rc` on the timer worker.
229                    // Move callbacks into an event-loop registry and send an id instead.
230                    let handler = TimeEventHandler::new(event, callback.clone());
231                    sender.send(handler);
232                } else {
233                    #[cfg(feature = "python")]
234                    if matches!(&callback, TimeEventCallback::Python(_)) {
235                        callback.call(event);
236                    } else {
237                        panic!("timer event sender was unset for Rust callback system");
238                    }
239
240                    #[cfg(not(feature = "python"))]
241                    {
242                        panic!("timer event sender was unset for Rust callback system");
243                    }
244                }
245
246                // Prepare next time interval
247                next_time_ns += interval_ns;
248                next_time_atomic.store(next_time_ns.as_u64(), atomic::Ordering::SeqCst);
249
250                // Check if expired
251                if let Some(stop_time_ns) = stop_time_ns
252                    && std::cmp::max(next_time_ns, now_ns) >= stop_time_ns
253                {
254                    break; // Timer expired
255                }
256            }
257        });
258
259        self.task_handle = Some(handle);
260        self.canceled = false;
261    }
262
263    /// Cancels the timer.
264    ///
265    /// The timer will not generate a final event.
266    pub fn cancel(&mut self) {
267        log::trace!("Cancel timer '{}'", self.name);
268
269        if let Some(handle) = self.task_handle.take() {
270            handle.abort();
271        }
272        self.canceled = true;
273    }
274}
275
276impl Timer for LiveTimer {
277    fn is_expired(&self) -> bool {
278        Self::is_expired(self)
279    }
280
281    fn cancel(&mut self) {
282        Self::cancel(self);
283    }
284}
285
286impl Drop for LiveTimer {
287    fn drop(&mut self) {
288        if let Some(handle) = self.task_handle.take() {
289            handle.abort();
290        }
291    }
292}
293
294#[cfg(test)]
295mod tests {
296    use std::{num::NonZeroU64, sync::Arc};
297    #[cfg(feature = "python")]
298    use std::{
299        sync::{Mutex, mpsc},
300        time::Duration as StdDuration,
301    };
302
303    use nautilus_core::{
304        UnixNanos, datetime::floor_to_nearest_microsecond, time::get_atomic_clock_realtime,
305    };
306    #[cfg(feature = "python")]
307    use pyo3::{
308        Python,
309        types::{PyAnyMethods, PyList, PyListMethods},
310    };
311    use rstest::*;
312    use ustr::Ustr;
313
314    use super::LiveTimer;
315    use crate::{
316        runner::TimeEventSender,
317        timer::{TimeEventCallback, TimeEventHandler},
318    };
319
320    #[rstest]
321    fn test_live_timer_fire_immediately_field() {
322        let timer = LiveTimer::new(
323            Ustr::from("TEST_TIMER"),
324            NonZeroU64::new(1000).unwrap(),
325            UnixNanos::from(100),
326            None,
327            TimeEventCallback::from(|_| {}),
328            true, // fire_immediately = true
329            None, // time_event_sender
330        );
331
332        // Verify the field is set correctly
333        assert!(timer.fire_immediately);
334
335        // With fire_immediately=true, next_time_ns should be start_time_ns
336        assert_eq!(timer.next_time_ns(), UnixNanos::from(100));
337    }
338
339    #[rstest]
340    fn test_live_timer_fire_immediately_false_field() {
341        let timer = LiveTimer::new(
342            Ustr::from("TEST_TIMER"),
343            NonZeroU64::new(1000).unwrap(),
344            UnixNanos::from(100),
345            None,
346            TimeEventCallback::from(|_| {}),
347            false, // fire_immediately = false
348            None,  // time_event_sender
349        );
350
351        // Verify the field is set correctly
352        assert!(!timer.fire_immediately);
353
354        // With fire_immediately=false, next_time_ns should be start_time_ns + interval
355        assert_eq!(timer.next_time_ns(), UnixNanos::from(1100));
356    }
357
358    #[rstest]
359    fn test_live_timer_adjusts_past_due_start_time() {
360        #[derive(Debug)]
361        struct NoopSender;
362
363        impl TimeEventSender for NoopSender {
364            fn send(&self, _handler: TimeEventHandler) {}
365        }
366
367        let sender = Arc::new(NoopSender);
368        let mut timer = LiveTimer::new(
369            Ustr::from("PAST_TIMER"),
370            NonZeroU64::new(1).unwrap(),
371            UnixNanos::from(0),
372            None,
373            TimeEventCallback::from(|_| {}),
374            true,
375            Some(sender),
376        );
377
378        let before = get_atomic_clock_realtime().get_time_ns();
379
380        timer.start();
381
382        // `next_time_ns` is floored to microsecond precision, so compare against
383        // the same floor applied to the baseline
384        let before_floored = UnixNanos::from(floor_to_nearest_microsecond(before.as_u64()));
385        assert!(timer.next_time_ns() >= before_floored);
386
387        timer.cancel();
388    }
389
390    #[cfg(feature = "python")]
391    #[rstest]
392    fn test_live_timer_with_sender_defers_python_callback_to_handler() {
393        #[derive(Debug)]
394        struct ChannelSender {
395            tx: Mutex<mpsc::Sender<TimeEventHandler>>,
396        }
397
398        impl TimeEventSender for ChannelSender {
399            fn send(&self, handler: TimeEventHandler) {
400                self.tx
401                    .lock()
402                    .expect("sender mutex should lock")
403                    .send(handler)
404                    .expect("handler should send");
405            }
406        }
407
408        Python::initialize();
409
410        Python::attach(|py| {
411            let py_list = PyList::empty(py);
412            let py_append = py_list
413                .getattr("append")
414                .expect("append should exist")
415                .unbind();
416            let callback = TimeEventCallback::from(py_append);
417            let (tx, rx) = mpsc::channel();
418            let sender = Arc::new(ChannelSender { tx: Mutex::new(tx) });
419            let now = get_atomic_clock_realtime().get_time_ns();
420
421            let mut timer = LiveTimer::new(
422                Ustr::from("PY_TIMER"),
423                NonZeroU64::new(1_000_000).unwrap(),
424                now,
425                Some(UnixNanos::from(now.as_u64() + 2_000_000)),
426                callback,
427                true,
428                Some(sender),
429            );
430
431            timer.start();
432            let handler = rx
433                .recv_timeout(StdDuration::from_secs(1))
434                .expect("timer handler should arrive without acquiring the GIL on the worker");
435            timer.cancel();
436
437            assert_eq!(py_list.len(), 0);
438            handler.run();
439            assert_eq!(py_list.len(), 1);
440        });
441    }
442}
nautilus_common/live/timer.rs

nautilus_common/live/
timer.rs
