#![allow(clippy::disallowed_types)]
use std::{
cmp, fmt, ops,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use reifydb_value::value::{datetime::DateTime, duration::Duration as RvDuration};
use crate::{
context::clock::{TimerId, TimerWake},
sync::mutex::Mutex,
};
#[allow(clippy::disallowed_methods)]
#[inline(always)]
fn platform_now_nanos() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).expect("System time is before Unix epoch").as_nanos() as u64
}
#[derive(Clone)]
pub enum Clock {
Real,
Mock(MockClock),
}
impl Clock {
pub fn now(&self) -> DateTime {
match self {
Clock::Real => DateTime::from_nanos(platform_now_nanos()),
Clock::Mock(mock) => mock.now(),
}
}
#[allow(clippy::disallowed_methods)]
pub fn instant(&self) -> Instant {
match self {
Clock::Real => Instant {
inner: InstantInner::Real(time::Instant::now()),
},
Clock::Mock(mock) => Instant {
inner: InstantInner::Mock {
captured_nanos: mock.now().to_nanos(),
clock: mock.clone(),
},
},
}
}
pub fn testing() -> Self {
#[cfg(reifydb_dst)]
return Clock::Mock(MockClock::from_millis(0));
#[cfg(not(reifydb_dst))]
return Clock::Real;
}
pub fn is_mock(&self) -> bool {
matches!(self, Clock::Mock(_))
}
pub fn as_mock(&self) -> Option<&MockClock> {
match self {
Clock::Mock(mock) => Some(mock),
Clock::Real => None,
}
}
}
#[derive(Clone)]
pub struct MockClock {
inner: Arc<MockClockInner>,
}
struct MockClockInner {
time_nanos: AtomicU64,
timers: Mutex<Vec<Timer>>,
next_timer_id: AtomicU64,
}
struct Timer {
id: u64,
deadline_nanos: u64,
wake: Arc<dyn TimerWake>,
}
impl MockClock {
pub fn new(initial_nanos: u64) -> Self {
Self {
inner: Arc::new(MockClockInner {
time_nanos: AtomicU64::new(initial_nanos),
timers: Mutex::new(Vec::new()),
next_timer_id: AtomicU64::new(0),
}),
}
}
pub fn from_millis(millis: u64) -> Self {
Self::new(millis * 1_000_000)
}
pub fn now(&self) -> DateTime {
DateTime::from_nanos(self.inner.time_nanos.load(Ordering::Acquire))
}
pub fn set_nanos(&self, nanos: u64) {
self.inner.time_nanos.store(nanos, Ordering::Release);
self.fire_due(nanos);
}
pub fn register_timer(&self, deadline_nanos: u64, wake: Arc<dyn TimerWake>) -> TimerId {
let id = self.inner.next_timer_id.fetch_add(1, Ordering::Relaxed);
self.inner.timers.lock().push(Timer {
id,
deadline_nanos,
wake,
});
TimerId(id)
}
pub fn cancel_timer(&self, timer: TimerId) {
self.inner.timers.lock().retain(|entry| entry.id != timer.0);
}
fn fire_due(&self, now_nanos: u64) {
let due = {
let mut timers = self.inner.timers.lock();
let mut due: Vec<Arc<dyn TimerWake>> = Vec::new();
timers.retain(|entry| {
if entry.deadline_nanos <= now_nanos {
due.push(entry.wake.clone());
return false;
}
true
});
due
};
for wake in due {
wake.wake();
}
}
pub fn set_micros(&self, micros: u64) {
self.set_nanos(micros * 1_000);
}
pub fn set_millis(&self, millis: u64) {
self.set_nanos(millis * 1_000_000);
}
pub fn advance_nanos(&self, nanos: u64) {
self.set_nanos(self.now().to_nanos().saturating_add(nanos));
}
pub fn advance_micros(&self, micros: u64) {
self.advance_nanos(micros * 1_000);
}
pub fn advance_millis(&self, millis: u64) {
self.advance_nanos(millis * 1_000_000);
}
pub fn advance_secs(&self, secs: u64) {
self.advance_nanos(secs * 1_000_000_000);
}
pub fn advance_minutes(&self, minutes: u64) {
self.advance_secs(minutes * 60);
}
pub fn advance_hours(&self, hours: u64) {
self.advance_secs(hours * 3600);
}
pub fn advance_days(&self, days: u64) {
self.advance_secs(days * 86400);
}
}
#[derive(Clone)]
enum InstantInner {
Real(time::Instant),
Mock {
captured_nanos: u64,
clock: MockClock,
},
}
#[derive(Clone)]
pub struct Instant {
inner: InstantInner,
}
impl Instant {
#[inline]
pub fn elapsed(&self) -> Duration {
match &self.inner {
InstantInner::Real(instant) => instant.elapsed(),
InstantInner::Mock {
captured_nanos,
clock,
} => {
let now = clock.now().to_nanos();
let elapsed_nanos = now.saturating_sub(*captured_nanos);
Duration::from_nanos(elapsed_nanos)
}
}
}
#[inline]
pub fn duration_since(&self, earlier: &Instant) -> Duration {
match (&self.inner, &earlier.inner) {
(InstantInner::Real(this), InstantInner::Real(other)) => this.duration_since(*other),
(
InstantInner::Mock {
captured_nanos: this_nanos,
..
},
InstantInner::Mock {
captured_nanos: other_nanos,
..
},
) => {
let elapsed = this_nanos.saturating_sub(*other_nanos);
Duration::from_nanos(elapsed)
}
_ => panic!("Cannot compare instants from different clock types"),
}
}
}
impl PartialEq for Instant {
fn eq(&self, other: &Self) -> bool {
match (&self.inner, &other.inner) {
(InstantInner::Real(a), InstantInner::Real(b)) => a == b,
(
InstantInner::Mock {
captured_nanos: a,
..
},
InstantInner::Mock {
captured_nanos: b,
..
},
) => a == b,
_ => panic!("Cannot compare instants from different clock types"),
}
}
}
impl Eq for Instant {}
impl PartialOrd for Instant {
fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for Instant {
fn cmp(&self, other: &Self) -> cmp::Ordering {
match (&self.inner, &other.inner) {
(InstantInner::Real(a), InstantInner::Real(b)) => a.cmp(b),
(
InstantInner::Mock {
captured_nanos: a,
..
},
InstantInner::Mock {
captured_nanos: b,
..
},
) => a.cmp(b),
_ => panic!("Cannot compare instants from different clock types"),
}
}
}
impl ops::Add<Duration> for Instant {
type Output = Instant;
fn add(self, duration: Duration) -> Instant {
match self.inner {
InstantInner::Real(instant) => Instant {
inner: InstantInner::Real(instant + duration),
},
InstantInner::Mock {
captured_nanos,
clock,
} => Instant {
inner: InstantInner::Mock {
captured_nanos: captured_nanos.saturating_add(duration.as_nanos() as u64),
clock,
},
},
}
}
}
impl ops::Add<RvDuration> for Instant {
type Output = Instant;
fn add(self, duration: RvDuration) -> Instant {
self + duration.to_std()
}
}
impl ops::Sub for &Instant {
type Output = Duration;
fn sub(self, other: &Instant) -> Duration {
self.duration_since(other)
}
}
impl fmt::Debug for Instant {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match &self.inner {
InstantInner::Real(instant) => f.debug_tuple("Instant::Real").field(instant).finish(),
InstantInner::Mock {
captured_nanos,
..
} => f.debug_tuple("Instant::Mock").field(captured_nanos).finish(),
}
}
}