1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
|
//! Wall-clock time as this node sees it, in Unix milliseconds: commit stamps, alive
//! stamps, remake deadlines and votes, archive and quarantine times.
//!
//! Every such reading goes through the node's [`Clock`], so tests can put one node's
//! clock ahead of or behind the others (`docs/trust.md`, "What depends on clocks").
//! Elapsed time within a node (timeouts, keepalives) uses tokio's monotonic clock and
//! isn't affected.
use std::sync::atomic::{AtomicI64, Ordering};
use std::time::Duration;
use serde::{Deserialize, Serialize};
/// A point in time as Unix milliseconds, as stored in the document and reported: not a
/// duration (those are `Duration`, or plain milliseconds named `…_ms`). Comparable
/// across nodes as far as their clocks agree; for elapsed time within a node use
/// tokio's `Instant`.
#[derive(
Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
)]
#[serde(transparent)]
pub(crate) struct UnixMillis(pub(crate) u64);
impl UnixMillis {
pub(crate) fn as_u64(self) -> u64 {
self.0
}
/// `ms` milliseconds later (saturating).
pub(crate) fn plus_ms(self, ms: u64) -> Self {
Self(self.0.saturating_add(ms))
}
/// `duration` later (saturating, to the millisecond).
pub(crate) fn plus(self, duration: Duration) -> Self {
self.plus_ms(u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
}
/// Milliseconds from `earlier` to this; 0 if `earlier` is later.
pub(crate) fn ms_since(self, earlier: Self) -> u64 {
self.0.saturating_sub(earlier.0)
}
}
impl std::fmt::Display for UnixMillis {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.0.fmt(f)
}
}
/// The node's wall clock: the system clock plus an offset (zero outside tests).
#[derive(Debug, Default)]
pub(crate) struct Clock {
offset_ms: AtomicI64,
}
impl Clock {
/// Now.
pub(crate) fn now_millis(&self) -> UnixMillis {
let offset = self.offset_ms.load(Ordering::Relaxed);
UnixMillis(system_millis().saturating_add_signed(offset))
}
/// Puts this clock `offset_ms` ahead of the system clock (behind if negative).
#[cfg(test)]
pub(crate) fn set_offset_ms(&self, offset_ms: i64) {
self.offset_ms.store(offset_ms, Ordering::Relaxed);
}
}
/// The system clock, in Unix milliseconds; 0 if it is before 1970.
fn system_millis() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn offset_moves_the_clock() {
let clock = Clock::default();
let before = clock.now_millis();
clock.set_offset_ms(3_600_000);
let ahead = clock.now_millis();
assert!(ahead >= before.plus_ms(3_600_000));
clock.set_offset_ms(-3_600_000);
assert!(clock.now_millis() < before);
}
}
|
