subetha_cxc/
cached_clock.rs

1//! Process-global cached wall clock.
2//!
3//! A single background thread refreshes a cached microsecond timestamp
4//! every `REFRESH_INTERVAL`. Readers load one relaxed atomic (~1 ns)
5//! instead of calling `clock_gettime` (~20 ns), trading at most
6//! `REFRESH_INTERVAL` of staleness for the cheaper read.
7//!
8//! This suits primitives whose physical-clock component tolerates coarse
9//! resolution because a logical counter orders sub-interval events - e.g.
10//! a same-host Hybrid Logical Clock, where every process reads the same
11//! hardware clock (zero inter-process skew) and the only thing the cache
12//! changes is the granularity at which the physical timestamp advances.
13//!
14//! Compared with `CLOCK_REALTIME_COARSE` (~1 ms granularity, ~5 ns read)
15//! this is both finer (250 us) and faster (a plain atomic load); the cost
16//! is one background thread per process, spawned lazily on first use.
17
18use std::sync::Once;
19use std::sync::atomic::{AtomicU64, Ordering};
20use std::time::{Duration, SystemTime, UNIX_EPOCH};
21
22static CACHED_US: AtomicU64 = AtomicU64::new(0);
23static INIT: Once = Once::new();
24
25/// Maximum staleness of the cached clock: the cached value lags real wall
26/// time by at most this much. 250 us keeps the updater's wake rate modest
27/// (~4000/s) while staying far finer than `CLOCK_REALTIME_COARSE`.
28const REFRESH_INTERVAL: Duration = Duration::from_micros(250);
29
30#[inline]
31fn real_now_us() -> u64 {
32    SystemTime::now()
33        .duration_since(UNIX_EPOCH)
34        .unwrap_or_default()
35        .as_micros() as u64
36}
37
38/// Start the background updater thread (once per process). Idempotent;
39/// call from a consumer's `create` / `open`. The cache is seeded
40/// synchronously here so the very first [`now_us`] is valid even before
41/// the thread's first refresh.
42pub fn start() {
43    INIT.call_once(|| {
44        CACHED_US.store(real_now_us(), Ordering::Relaxed);
45        std::thread::Builder::new()
46            .name("subetha-cached-clock".into())
47            .spawn(|| {
48                loop {
49                    CACHED_US.store(real_now_us(), Ordering::Relaxed);
50                    std::thread::sleep(REFRESH_INTERVAL);
51                }
52            })
53            .ok(); // detached; the JoinHandle is intentionally dropped
54    });
55}
56
57/// Cached wall-clock microseconds - one relaxed atomic load. Callers must
58/// have invoked [`start`] (e.g. at handle create) so the updater is
59/// running; before the first refresh this returns the seed taken in
60/// `start`. Monotonic to the precision of the underlying clock; a brief
61/// backward NTP step is absorbed by HLC-style `max(prev, now)` callers.
62#[inline]
63pub fn now_us() -> u64 {
64    CACHED_US.load(Ordering::Relaxed)
65}
66
67#[cfg(test)]
68mod tests {
69    use super::*;
70
71    #[test]
72    fn cached_clock_tracks_wall_within_interval() {
73        start();
74        // Give the updater a couple of refresh cycles to populate.
75        std::thread::sleep(REFRESH_INTERVAL * 4);
76        let cached = now_us();
77        let real = real_now_us();
78        assert!(cached > 0, "cache must be seeded");
79        // Within a few refresh intervals of real time (generous for CI).
80        let skew = real.abs_diff(cached);
81        assert!(
82            skew < 50_000,
83            "cached clock {cached} should track real {real} (skew {skew} us)"
84        );
85    }
86
87    #[test]
88    fn now_us_is_monotonic_nondecreasing() {
89        start();
90        let mut prev = now_us();
91        for _ in 0..1000 {
92            let cur = now_us();
93            assert!(cur >= prev, "cached clock must not go backward");
94            prev = cur;
95        }
96    }
97}
subetha_cxc/cached_clock.rs

subetha_cxc/
cached_clock.rs
