t2t_clock/
manual.rs

1//! Clocks set and moved by hand, for tests and replays.
2
3use core::cell::Cell;
4use core::fmt;
5
6#[cfg(atomic_clock)]
7pub use atomic::AtomicManualClock;
8use derive_more::Debug;
9use t2t_core::{TimePoint, Timestamp};
10
11use crate::Clock;
12
13/// A clock set and moved by hand, read on one thread.
14///
15/// Reading it costs a load; [`AtomicManualClock`] is the one to share across threads. It is not
16/// `Clone`, since a copy would fork time: share it by reference.
17///
18/// # Examples
19/// ```
20/// use t2t_clock::{Clock, ManualClock};
21/// use t2t_core::{Timedelta, Timestamp};
22///
23/// let clock = ManualClock::new(Timestamp::UNIX_EPOCH);
24/// clock.advance(Timedelta::SECOND);
25/// assert_eq!(clock.now(), Timestamp::from_secs(1), "moved on by a second");
26/// clock.set(Timestamp::from_secs(60));
27/// assert_eq!(clock.now(), Timestamp::from_secs(60), "moved to a minute");
28/// ```
29#[derive(Debug, Default)]
30#[debug("ManualClock({:?})", reading.get())]
31#[debug(bound(P: TimePoint + fmt::Debug))]
32pub struct ManualClock<P = Timestamp> {
33    /// The point the clock reads.
34    reading: Cell<P>,
35}
36
37impl<P: TimePoint> ManualClock<P> {
38    /// A clock reading `start`.
39    #[inline]
40    #[must_use]
41    pub const fn new(start: P) -> Self {
42        Self { reading: Cell::new(start) }
43    }
44
45    /// Moves the clock to `point`, earlier or later.
46    #[inline]
47    pub fn set(&self, point: P) {
48        self.reading.set(point);
49    }
50
51    /// Moves the clock by `span`, back for a negative one, saturating as the point's addition does.
52    #[inline]
53    #[expect(
54        clippy::arithmetic_side_effects,
55        reason = "the point's own addition, which saturates for t2t's"
56    )]
57    pub fn advance(&self, span: P::Span) {
58        self.reading.set(self.reading.get() + span);
59    }
60}
61
62impl<P: TimePoint> Clock for ManualClock<P> {
63    type Reading = P;
64
65    #[inline]
66    fn now(&self) -> P {
67        self.reading.get()
68    }
69}
70
71/// The clock shared across threads, where the target has a 64-bit atomic.
72#[cfg(atomic_clock)]
73mod atomic {
74    use core::fmt;
75    use core::marker::PhantomData;
76
77    use derive_more::Debug;
78    use t2t_core::{TimePoint, Timestamp};
79
80    use crate::Clock;
81    use crate::sync::{AtomicI64, Ordering};
82
83    /// A clock set and moved by hand, shared across threads.
84    ///
85    /// A thread that reads a point another set or moved the clock to also sees what that thread
86    /// wrote before. It is not `Clone`, since a copy would fork time: share it by reference or
87    /// `Arc`.
88    ///
89    /// # Examples
90    /// ```
91    /// use std::thread;
92    ///
93    /// use t2t_clock::{AtomicManualClock, Clock};
94    /// use t2t_core::{Timedelta, Uptime};
95    ///
96    /// let clock = AtomicManualClock::new(Uptime::from_secs(1));
97    /// thread::scope(|scope| {
98    ///     scope.spawn(|| clock.advance(Timedelta::SECOND));
99    /// });
100    /// assert_eq!(clock.now(), Uptime::from_secs(2), "the other thread's move");
101    /// ```
102    #[derive(Debug)]
103    #[debug("AtomicManualClock({:?})", self.now())]
104    #[debug(bound(P: TimePoint + fmt::Debug))]
105    pub struct AtomicManualClock<P = Timestamp> {
106        /// The count of the point the clock reads.
107        reading: AtomicI64,
108        /// The kind of point, held by no value, so the clock is `Send` and `Sync` whatever it is.
109        marker: PhantomData<fn() -> P>,
110    }
111
112    impl<P: TimePoint> AtomicManualClock<P> {
113        /// A clock reading `start`.
114        #[inline]
115        #[must_use]
116        pub fn new(start: P) -> Self {
117            Self { reading: AtomicI64::new(start.count()), marker: PhantomData }
118        }
119
120        /// Moves the clock to `point`, earlier or later.
121        #[inline]
122        pub fn set(&self, point: P) {
123            // ORDERING: Release, pairing with the Acquire load in `now`.
124            self.reading.store(point.count(), Ordering::Release);
125        }
126
127        /// Moves the clock by `span`, back for a negative one, saturating as the point's addition
128        /// does.
129        #[inline]
130        #[expect(
131            clippy::arithmetic_side_effects,
132            reason = "the point's own addition, which saturates for t2t's"
133        )]
134        pub fn advance(&self, span: P::Span) {
135            // By hand: std's `update` is this loop, but loom 0.7's atomics have only
136            // `fetch_update`, which the pinned nightly deprecates for `try_update`.
137            //
138            // ORDERING: Relaxed; the count only seeds the exchange, which orders the move.
139            let mut expected = self.reading.load(Ordering::Relaxed);
140            loop {
141                let target = (P::from_count(expected) + span).count();
142                // ORDERING: AcqRel on success, pairing with the Acquire load in `now` and with the
143                // Release of the last `set` or `advance`; Relaxed on failure, which only reloads.
144                match self.reading.compare_exchange_weak(
145                    expected,
146                    target,
147                    Ordering::AcqRel,
148                    Ordering::Relaxed,
149                ) {
150                    Ok(_) => return,
151                    Err(actual) => expected = actual,
152                }
153            }
154        }
155    }
156
157    // What the docs promise of a clock to share.
158    const _: () = {
159        /// Compiles only for a type that is `Send` and `Sync`.
160        const fn is_send_and_sync<T: Send + Sync>() {}
161        is_send_and_sync::<AtomicManualClock<Timestamp>>();
162    };
163
164    impl<P: TimePoint> Clock for AtomicManualClock<P> {
165        type Reading = P;
166
167        #[inline]
168        fn now(&self) -> P {
169            // ORDERING: Acquire, pairing with the Release of whichever thread set or moved the
170            // clock.
171            P::from_count(self.reading.load(Ordering::Acquire))
172        }
173    }
174
175    /// By hand, since a derived one would start the count at zero, not at the point's default.
176    impl<P: TimePoint + Default> Default for AtomicManualClock<P> {
177        fn default() -> Self {
178            Self::new(P::default())
179        }
180    }
181
182    // The atomic clock under loom, which runs each model once for every interleaving of its two
183    // threads that the memory model allows: `cargo test -p t2t-clock --lib --release --config
184    // 'target."cfg(all())".rustflags=["--cfg","loom"]'`. Loom has two gaps here: it orders a store
185    // and an exchange that did not see it only partly, so a `set` racing an `advance` is not
186    // modelled; and past `advance`'s seed load, its search misses an exchange without Release.
187    #[cfg(test)]
188    #[cfg(loom)]
189    mod model {
190        use loom::sync::Arc;
191        use loom::sync::atomic::{AtomicU64, Ordering};
192        use loom::{model, thread};
193        use t2t_core::{Timedelta, Timestamp};
194
195        use super::AtomicManualClock;
196        use crate::Clock;
197
198        /// A clock at the epoch, and a note for a thread to write before it moves the clock.
199        fn clock_and_note() -> (Arc<AtomicManualClock>, Arc<AtomicU64>) {
200            (Arc::new(AtomicManualClock::new(Timestamp::UNIX_EPOCH)), Arc::new(AtomicU64::new(0)))
201        }
202
203        #[test]
204        fn a_reader_that_sees_a_set_point_sees_what_was_written_before_it() {
205            model(|| {
206                let (clock, note) = clock_and_note();
207                let setter = {
208                    let (clock, note) = (Arc::clone(&clock), Arc::clone(&note));
209                    thread::spawn(move || {
210                        // ORDERING: Relaxed; the Release in `set` publishes it.
211                        note.store(7, Ordering::Relaxed);
212                        clock.set(Timestamp::from_secs(1));
213                    })
214                };
215                if clock.now() == Timestamp::from_secs(1) {
216                    // ORDERING: Relaxed; the Acquire in `now`, which saw the set, orders it after.
217                    assert_eq!(note.load(Ordering::Relaxed), 7, "the note written before the set");
218                }
219                setter.join().expect("the setting thread ends");
220            });
221        }
222
223        #[test]
224        fn a_reader_that_sees_an_advance_sees_what_was_written_before_it() {
225            model(|| {
226                let (clock, note) = clock_and_note();
227                let advancer = {
228                    let (clock, note) = (Arc::clone(&clock), Arc::clone(&note));
229                    thread::spawn(move || {
230                        // ORDERING: Relaxed; the AcqRel exchange in `advance` publishes it.
231                        note.store(7, Ordering::Relaxed);
232                        clock.advance(Timedelta::SECOND);
233                    })
234                };
235                if clock.now() == Timestamp::from_secs(1) {
236                    // ORDERING: Relaxed; the Acquire in `now`, which saw the advance, orders it
237                    // after.
238                    assert_eq!(
239                        note.load(Ordering::Relaxed),
240                        7,
241                        "the note written before the advance"
242                    );
243                }
244                advancer.join().expect("the advancing thread ends");
245            });
246        }
247
248        #[test]
249        fn two_threads_advancing_at_once_both_count() {
250            model(|| {
251                let (clock, _) = clock_and_note();
252                let advancer = {
253                    let clock = Arc::clone(&clock);
254                    thread::spawn(move || clock.advance(Timedelta::SECOND))
255                };
256                clock.advance(Timedelta::SECOND);
257                advancer.join().expect("the advancing thread ends");
258                assert_eq!(clock.now(), Timestamp::from_secs(2), "neither move lost");
259            });
260        }
261    }
262}
263
264#[cfg(test)]
265#[cfg(not(loom))]
266mod tests {
267    use alloc::format;
268    use std::thread;
269
270    use t2t_core::{Tickdelta, Tickstamp, Timedelta, Timestamp};
271
272    use crate::{AtomicManualClock, Clock, ManualClock};
273
274    #[test]
275    fn a_manual_clock_moves_both_ways_and_saturates() {
276        let clock = ManualClock::new(Tickstamp::from_ticks(100));
277        clock.advance(Tickdelta::from_ticks(-40));
278        assert_eq!(clock.now(), Tickstamp::from_ticks(60), "moved back by a negative span");
279        clock.set(Tickstamp::MAX);
280        clock.advance(Tickdelta::from_ticks(1));
281        assert_eq!(clock.now(), Tickstamp::MAX, "and saturating at the end");
282    }
283
284    #[test]
285    fn an_atomic_manual_clock_sums_every_threads_moves() {
286        let clock = AtomicManualClock::new(Timestamp::UNIX_EPOCH);
287        thread::scope(|scope| {
288            for _ in 0..8 {
289                scope.spawn(|| {
290                    for _ in 0..1_000 {
291                        clock.advance(Timedelta::NANOSECOND);
292                    }
293                });
294            }
295        });
296        assert_eq!(clock.now(), Timestamp::from_nanos(8_000), "no move lost");
297    }
298
299    #[test]
300    fn a_clock_is_read_through_a_reference_and_debugged_as_its_reading() {
301        let clock = ManualClock::<Timestamp>::default();
302        let reference: &dyn Clock<Reading = Timestamp> = &clock;
303        assert_eq!(reference.now(), Timestamp::UNIX_EPOCH, "through a reference");
304        let written = format!("{clock:?}");
305        assert_eq!(written, "ManualClock(1970-01-01T00:00:00.000000000Z)", "as its reading");
306        let atomic = format!("{:?}", AtomicManualClock::new(Tickstamp::from_ticks(5)));
307        assert_eq!(atomic, "AtomicManualClock(5 ticks)", "for the shared clock too");
308    }
309}
t2t_clock/manual.rs

t2t_clock/
manual.rs
