node: Do not mix monotonic and system time
The `struct Epoch` introduced in e404f1038f
inadvertently made `radicle-node` suffer from
https://github.com/rust-lang/rust/issues/87906
That is, updating of the "current time" was skewed *way more*
than expected by a slowly ticking monotonic clock. Such slow
ticking can be caused by suspending the system.
This commit is contained in:
parent
58305cda32
commit
4a5a51e6e5
|
|
@ -211,7 +211,7 @@ pub trait ReactionHandler: Send + Iterator<Item = Action<Self::Listener, Self::T
|
||||||
type Transport: EventHandler + Source + Send + Debug + WriteAtomic;
|
type Transport: EventHandler + Source + Send + Debug + WriteAtomic;
|
||||||
|
|
||||||
/// Method called by the reactor on the start of each event loop once the poll has returned.
|
/// Method called by the reactor on the start of each event loop once the poll has returned.
|
||||||
fn tick(&mut self, instant: Instant);
|
fn tick(&mut self);
|
||||||
|
|
||||||
/// Method called by the reactor when a previously set timeout is fired.
|
/// Method called by the reactor when a previously set timeout is fired.
|
||||||
///
|
///
|
||||||
|
|
@ -387,8 +387,8 @@ impl<H: ReactionHandler> Runtime<H> {
|
||||||
// Blocking
|
// Blocking
|
||||||
let res = self.poll.poll(&mut events, Some(timeout));
|
let res = self.poll.poll(&mut events, Some(timeout));
|
||||||
|
|
||||||
|
self.service.tick();
|
||||||
let tick = Instant::now();
|
let tick = Instant::now();
|
||||||
self.service.tick(tick);
|
|
||||||
|
|
||||||
// The way this is currently used basically ignores which keys have
|
// The way this is currently used basically ignores which keys have
|
||||||
// timed out. So as long as *something* timed out, we wake the service.
|
// timed out. So as long as *something* timed out, we wake the service.
|
||||||
|
|
|
||||||
|
|
@ -291,35 +291,6 @@ impl Peers {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The epoch time of when the node started.
|
|
||||||
struct Epoch {
|
|
||||||
/// The system time when the node started.
|
|
||||||
started_time: SystemTime,
|
|
||||||
/// The instant when the node started.
|
|
||||||
started_at: Instant,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl Epoch {
|
|
||||||
/// Construct a new [`Epoch`].
|
|
||||||
fn new(started_time: SystemTime, started_at: Instant) -> Self {
|
|
||||||
Self {
|
|
||||||
started_time,
|
|
||||||
started_at,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Construct an [`Epoch`] where both values are recorded using their
|
|
||||||
/// equivalent `now` constructors.
|
|
||||||
fn now() -> Self {
|
|
||||||
Self::new(SystemTime::now(), Instant::now())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Get the elapsed [`SystemTime`] given a later [`Instant`].
|
|
||||||
fn elapsed_time(&self, later: Instant) -> SystemTime {
|
|
||||||
self.started_time + (later - self.started_at)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Wire protocol implementation for a set of peers.
|
/// Wire protocol implementation for a set of peers.
|
||||||
pub(crate) struct Wire<D, S, G: crypto::signature::Signer<crypto::Signature> + Ecdh> {
|
pub(crate) struct Wire<D, S, G: crypto::signature::Signer<crypto::Signature> + Ecdh> {
|
||||||
/// Backing service instance.
|
/// Backing service instance.
|
||||||
|
|
@ -342,8 +313,6 @@ pub(crate) struct Wire<D, S, G: crypto::signature::Signer<crypto::Signature> + E
|
||||||
peers: Peers,
|
peers: Peers,
|
||||||
/// A (practically) infinite source of tokens to identify transports and listeners.
|
/// A (practically) infinite source of tokens to identify transports and listeners.
|
||||||
tokens: Tokens,
|
tokens: Tokens,
|
||||||
/// Record of system time and instant when the node started.
|
|
||||||
epoch: Epoch,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<D, S, G> Wire<D, S, G>
|
impl<D, S, G> Wire<D, S, G>
|
||||||
|
|
@ -366,14 +335,9 @@ where
|
||||||
listening: RandomMap::default(),
|
listening: RandomMap::default(),
|
||||||
peers: Peers(RandomMap::default()),
|
peers: Peers(RandomMap::default()),
|
||||||
tokens: Tokens::default(),
|
tokens: Tokens::default(),
|
||||||
epoch: Epoch::now(),
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn time(&self, instant: Instant) -> SystemTime {
|
|
||||||
self.epoch.elapsed_time(instant)
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn listen(&mut self, socket: Listener) {
|
pub fn listen(&mut self, socket: Listener) {
|
||||||
let token = self.tokens.advance();
|
let token = self.tokens.advance();
|
||||||
self.listening.insert(token, socket.local_addr());
|
self.listening.insert(token, socket.local_addr());
|
||||||
|
|
@ -532,7 +496,7 @@ where
|
||||||
type Listener = Listener;
|
type Listener = Listener;
|
||||||
type Transport = Transport<WireSession<G>>;
|
type Transport = Transport<WireSession<G>>;
|
||||||
|
|
||||||
fn tick(&mut self, time: Instant) {
|
fn tick(&mut self) {
|
||||||
self.metrics.open_channels = self
|
self.metrics.open_channels = self
|
||||||
.peers
|
.peers
|
||||||
.iter()
|
.iter()
|
||||||
|
|
@ -546,7 +510,7 @@ where
|
||||||
.sum();
|
.sum();
|
||||||
self.metrics.worker_queue_size = self.worker.len();
|
self.metrics.worker_queue_size = self.worker.len();
|
||||||
|
|
||||||
self.service.tick(self.time(time).into(), &self.metrics);
|
self.service.tick(SystemTime::now().into(), &self.metrics);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn timer_reacted(&mut self) {
|
fn timer_reacted(&mut self) {
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue