robot_bus/runtime/
timers.rs

1//! Periodic timers for [`super::Executor`] (ROS 2–style `create_timer`).
2
3use std::sync::Arc;
4use std::time::{Duration, Instant};
5
6use crate::runtime::callback_group::CallbackGroup;
7use crate::runtime::worker_pool::WorkerPool;
8
9pub type TimerCallback = Arc<dyn Fn() + Send + Sync>;
10
11/// Opaque id returned by [`super::Executor::create_timer`].
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
13pub struct TimerHandle {
14    pub(crate) id: u64,
15}
16
17/// Opaque id returned by [`super::Executor::subscribe`] / Node create_subscription.
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
19pub struct SubscriptionHandle {
20    pub(crate) id: u64,
21}
22
23impl SubscriptionHandle {
24    pub fn id(self) -> u64 {
25        self.id
26    }
27}
28
29pub(crate) struct Timer {
30    pub id: u64,
31    pub period: Duration,
32    pub next_deadline: Instant,
33    pub callback: TimerCallback,
34    pub group: CallbackGroup,
35    pub cancelled: bool,
36}
37
38impl Timer {
39    pub fn new(id: u64, period: Duration, callback: TimerCallback, group: CallbackGroup) -> Self {
40        Self {
41            id,
42            period,
43            next_deadline: Instant::now() + period,
44            callback,
45            group,
46            cancelled: false,
47        }
48    }
49}
50
51/// Fire every due timer once, then reschedule from `now + period`.
52///
53/// Returns `true` if at least one callback was scheduled/ran.
54pub(crate) fn tick_timers(
55    timers: &mut [Timer],
56    now: Instant,
57    worker_pool: Option<&WorkerPool>,
58) -> bool {
59    let mut fired = false;
60    for timer in timers.iter_mut() {
61        if timer.cancelled || timer.next_deadline > now {
62            continue;
63        }
64        let callback = Arc::clone(&timer.callback);
65        let group = timer.group.clone();
66        group.run(worker_pool, move || callback());
67        timer.next_deadline = now + timer.period;
68        fired = true;
69    }
70    fired
71}
72
73/// Milliseconds until the soonest active timer, or `None` if none.
74pub(crate) fn ms_until_next_timer(timers: &[Timer], now: Instant) -> Option<i64> {
75    timers
76        .iter()
77        .filter(|t| !t.cancelled)
78        .map(|t| {
79            if t.next_deadline <= now {
80                0
81            } else {
82                t.next_deadline
83                    .duration_since(now)
84                    .as_millis()
85                    .min(i64::MAX as u128) as i64
86            }
87        })
88        .min()
89}
90
91pub(crate) fn effective_poll_timeout_ms(timers: &[Timer], requested_ms: i64, now: Instant) -> i64 {
92    match ms_until_next_timer(timers, now) {
93        Some(until) => until.min(requested_ms.max(0)),
94        None => requested_ms.max(0),
95    }
96}
robot_bus/runtime/timers.rs

robot_bus/runtime/
timers.rs
