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