reifydb_runtime/context/clock/
host.rs1#![allow(clippy::disallowed_types)]
5
6use std::{
7 cmp, fmt, ops,
8 sync::{
9 Arc,
10 atomic::{AtomicU64, Ordering},
11 },
12 time,
13 time::{Duration, SystemTime, UNIX_EPOCH},
14};
15
16use reifydb_value::value::{datetime::DateTime, duration::Duration as RvDuration};
17
18use crate::{
19 context::clock::{TimerId, TimerWake},
20 sync::mutex::Mutex,
21};
22
23#[allow(clippy::disallowed_methods)]
24#[inline(always)]
25fn platform_now_nanos() -> u64 {
26 SystemTime::now().duration_since(UNIX_EPOCH).expect("System time is before Unix epoch").as_nanos() as u64
27}
28
29#[derive(Clone, Debug)]
30pub enum Clock {
31 Real,
32
33 Mock(MockClock),
34}
35
36impl Clock {
37 pub fn now(&self) -> DateTime {
38 match self {
39 Clock::Real => DateTime::from_nanos(
40 i64::try_from(platform_now_nanos()).expect("system clock past 2262-04-11"),
41 ),
42 Clock::Mock(mock) => mock.now(),
43 }
44 }
45
46 #[allow(clippy::disallowed_methods)]
47 pub fn instant(&self) -> Instant {
48 match self {
49 Clock::Real => Instant {
50 inner: InstantInner::Real(time::Instant::now()),
51 },
52 Clock::Mock(mock) => Instant {
53 inner: InstantInner::Mock {
54 captured_nanos: u64::try_from(mock.now().to_nanos())
55 .expect("mock clock is never before the Unix epoch"),
56 clock: mock.clone(),
57 },
58 },
59 }
60 }
61
62 pub fn testing() -> Self {
63 #[cfg(reifydb_dst)]
64 return Clock::Mock(MockClock::from_millis(0));
65 #[cfg(not(reifydb_dst))]
66 return Clock::Real;
67 }
68
69 pub fn is_mock(&self) -> bool {
70 matches!(self, Clock::Mock(_))
71 }
72
73 pub fn as_mock(&self) -> Option<&MockClock> {
74 match self {
75 Clock::Mock(mock) => Some(mock),
76 Clock::Real => None,
77 }
78 }
79}
80
81#[derive(Clone, Debug)]
82pub struct MockClock {
83 inner: Arc<MockClockInner>,
84}
85
86#[derive(Debug)]
87struct MockClockInner {
88 time_nanos: AtomicU64,
89 timers: Mutex<Vec<Timer>>,
90 next_timer_id: AtomicU64,
91}
92
93#[derive(Debug)]
94struct Timer {
95 id: u64,
96 deadline_nanos: u64,
97 wake: Arc<dyn TimerWake>,
98}
99
100impl MockClock {
101 pub fn new(initial_nanos: u64) -> Self {
102 Self {
103 inner: Arc::new(MockClockInner {
104 time_nanos: AtomicU64::new(initial_nanos),
105 timers: Mutex::new(Vec::new()),
106 next_timer_id: AtomicU64::new(0),
107 }),
108 }
109 }
110
111 pub fn from_millis(millis: u64) -> Self {
112 Self::new(millis * 1_000_000)
113 }
114
115 pub fn now(&self) -> DateTime {
116 DateTime::from_nanos(
117 i64::try_from(self.inner.time_nanos.load(Ordering::Acquire))
118 .expect("mock clock past 2262-04-11"),
119 )
120 }
121
122 pub fn set_nanos(&self, nanos: u64) {
123 self.inner.time_nanos.store(nanos, Ordering::Release);
124 self.fire_due(nanos);
125 }
126
127 pub fn register_timer(&self, deadline_nanos: u64, wake: Arc<dyn TimerWake>) -> TimerId {
128 let id = self.inner.next_timer_id.fetch_add(1, Ordering::Relaxed);
129 self.inner.timers.lock().push(Timer {
130 id,
131 deadline_nanos,
132 wake,
133 });
134 TimerId(id)
135 }
136
137 pub fn cancel_timer(&self, timer: TimerId) {
138 self.inner.timers.lock().retain(|entry| entry.id != timer.0);
139 }
140
141 fn fire_due(&self, now_nanos: u64) {
142 let due = {
143 let mut timers = self.inner.timers.lock();
144 let mut due: Vec<Arc<dyn TimerWake>> = Vec::new();
145 timers.retain(|entry| {
146 if entry.deadline_nanos <= now_nanos {
147 due.push(entry.wake.clone());
148 return false;
149 }
150 true
151 });
152 due
153 };
154
155 for wake in due {
156 wake.wake();
157 }
158 }
159
160 pub fn set_micros(&self, micros: u64) {
161 self.set_nanos(micros * 1_000);
162 }
163
164 pub fn set_millis(&self, millis: u64) {
165 self.set_nanos(millis * 1_000_000);
166 }
167
168 pub fn advance_nanos(&self, nanos: u64) {
169 let current = u64::try_from(self.now().to_nanos()).expect("mock clock is never before the Unix epoch");
170 self.set_nanos(current.saturating_add(nanos));
171 }
172
173 pub fn advance_micros(&self, micros: u64) {
174 self.advance_nanos(micros * 1_000);
175 }
176
177 pub fn advance_millis(&self, millis: u64) {
178 self.advance_nanos(millis * 1_000_000);
179 }
180
181 pub fn advance_secs(&self, secs: u64) {
182 self.advance_nanos(secs * 1_000_000_000);
183 }
184
185 pub fn advance_minutes(&self, minutes: u64) {
186 self.advance_secs(minutes * 60);
187 }
188
189 pub fn advance_hours(&self, hours: u64) {
190 self.advance_secs(hours * 3600);
191 }
192
193 pub fn advance_days(&self, days: u64) {
194 self.advance_secs(days * 86400);
195 }
196}
197
198#[derive(Clone)]
199enum InstantInner {
200 Real(time::Instant),
201 Mock {
202 captured_nanos: u64,
203 clock: MockClock,
204 },
205}
206
207#[derive(Clone)]
208pub struct Instant {
209 inner: InstantInner,
210}
211
212impl Instant {
213 #[inline]
214 pub fn elapsed(&self) -> Duration {
215 match &self.inner {
216 InstantInner::Real(instant) => instant.elapsed(),
217 InstantInner::Mock {
218 captured_nanos,
219 clock,
220 } => {
221 let now = u64::try_from(clock.now().to_nanos())
222 .expect("mock clock is never before the Unix epoch");
223 let elapsed_nanos = now.saturating_sub(*captured_nanos);
224 Duration::from_nanos(elapsed_nanos)
225 }
226 }
227 }
228
229 #[inline]
230 pub fn duration_since(&self, earlier: &Instant) -> Duration {
231 match (&self.inner, &earlier.inner) {
232 (InstantInner::Real(this), InstantInner::Real(other)) => this.duration_since(*other),
233 (
234 InstantInner::Mock {
235 captured_nanos: this_nanos,
236 ..
237 },
238 InstantInner::Mock {
239 captured_nanos: other_nanos,
240 ..
241 },
242 ) => {
243 let elapsed = this_nanos.saturating_sub(*other_nanos);
244 Duration::from_nanos(elapsed)
245 }
246 _ => panic!("Cannot compare instants from different clock types"),
247 }
248 }
249}
250
251impl PartialEq for Instant {
252 fn eq(&self, other: &Self) -> bool {
253 match (&self.inner, &other.inner) {
254 (InstantInner::Real(a), InstantInner::Real(b)) => a == b,
255 (
256 InstantInner::Mock {
257 captured_nanos: a,
258 ..
259 },
260 InstantInner::Mock {
261 captured_nanos: b,
262 ..
263 },
264 ) => a == b,
265 _ => panic!("Cannot compare instants from different clock types"),
266 }
267 }
268}
269
270impl Eq for Instant {}
271
272impl PartialOrd for Instant {
273 fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
274 Some(self.cmp(other))
275 }
276}
277
278impl Ord for Instant {
279 fn cmp(&self, other: &Self) -> cmp::Ordering {
280 match (&self.inner, &other.inner) {
281 (InstantInner::Real(a), InstantInner::Real(b)) => a.cmp(b),
282 (
283 InstantInner::Mock {
284 captured_nanos: a,
285 ..
286 },
287 InstantInner::Mock {
288 captured_nanos: b,
289 ..
290 },
291 ) => a.cmp(b),
292 _ => panic!("Cannot compare instants from different clock types"),
293 }
294 }
295}
296
297impl ops::Add<Duration> for Instant {
298 type Output = Instant;
299
300 fn add(self, duration: Duration) -> Instant {
301 match self.inner {
302 InstantInner::Real(instant) => Instant {
303 inner: InstantInner::Real(instant + duration),
304 },
305 InstantInner::Mock {
306 captured_nanos,
307 clock,
308 } => Instant {
309 inner: InstantInner::Mock {
310 captured_nanos: captured_nanos.saturating_add(duration.as_nanos() as u64),
311 clock,
312 },
313 },
314 }
315 }
316}
317
318impl ops::Add<RvDuration> for Instant {
319 type Output = Instant;
320
321 fn add(self, duration: RvDuration) -> Instant {
322 self + duration.to_std()
323 }
324}
325
326impl ops::Sub for &Instant {
327 type Output = Duration;
328
329 fn sub(self, other: &Instant) -> Duration {
330 self.duration_since(other)
331 }
332}
333
334impl fmt::Debug for Instant {
335 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
336 match &self.inner {
337 InstantInner::Real(instant) => f.debug_tuple("Instant::Real").field(instant).finish(),
338 InstantInner::Mock {
339 captured_nanos,
340 ..
341 } => f.debug_tuple("Instant::Mock").field(captured_nanos).finish(),
342 }
343 }
344}