//! How long sync messages (and BLOB messages) take from a peer to us, per hop: from the //! peer's clock when it generated one (`SyncEnvelope::sent`, `BlobMessage::sent`) to //! ours when we start handling it, the peer's //! clock offset taken off (`Transport::clock_offset`, from PING and PONG). So the //! transport, the network and the queues at both ends, which a PING's round trip leaves //! out (PINGs don't wait behind sync messages). //! //! The offset is known to within half a round trip only, so a single figure is too; //! but that error is the same for every message from a peer while the offset stands, so //! how the delay changes over time is exact: a link getting worse shows whatever the //! offset. Hence the means of the last `MINUTES` minutes besides all since start, and //! half the round trip to compare with (`api::peers`). use std::collections::VecDeque; use tokio::time::{Duration, Instant}; use crate::swarm::stats::Propagation; /// The minutes whose means are kept. pub(crate) const MINUTES: usize = 15; const MINUTE: Duration = Duration::from_secs(60); /// A peer's hop delays (`PeerState::hop_delays`). #[derive(Debug, Default, Clone)] pub(crate) struct HopDelays { /// All since we started, in buckets. pub(crate) all: Propagation, /// The minutes with messages, oldest first: when each began, how many, their sum. minutes: VecDeque<(Instant, u64, i64)>, } impl HopDelays { /// A message that took `delay_ms` (negative where the offset's error exceeds it). pub(crate) fn record(&mut self, delay_ms: i64, now: Instant) { self.all.record(delay_ms); match self.minutes.back_mut() { Some((began, count, sum)) if now.saturating_duration_since(*began) < MINUTE => { *count += 1; *sum += delay_ms; } _ => self.minutes.push_back((now, 1, delay_ms)), } while self.minutes.front().is_some_and(|(began, _, _)| { now.saturating_duration_since(*began) >= MINUTE * MINUTES as u32 }) { self.minutes.pop_front(); } } /// The mean of each of the last `MINUTES` minutes, oldest first, None for one without /// messages; the last is the current one. pub(crate) fn by_minute(&self, now: Instant) -> Vec> { let mut means = vec![None; MINUTES]; for (began, count, sum) in &self.minutes { let ago = (now.saturating_duration_since(*began).as_secs() / 60) as usize; if ago < MINUTES { means[MINUTES - 1 - ago] = Some(sum / (*count).max(1) as i64); } } means } } /// A message from `peer` that it `sent` (its clock) is here: how long it took, recorded /// with the peer's hop delays (`PeerState::hop_delays`); None while its clock isn't /// known. pub(crate) async fn note( state: &crate::state::AppState, peer: &str, sent: crate::clock::UnixMillis, ) -> Option { let offset = state.transport.clock_offset(peer)?; let delay = delay_ms(sent, state.clock.now_millis(), offset.offset_ms); if let Some(peer_state) = state.get_peer_state(peer).await { peer_state .lock() .await .hop_delays .record(delay, Instant::now()); } Some(delay) } /// How long a sync message `sent` by a peer (its clock) took to come, as `arrived` (ours), /// with the peer's clock `offset_ms` ahead of ours. pub(crate) fn delay_ms( sent: crate::clock::UnixMillis, arrived: crate::clock::UnixMillis, offset_ms: i64, ) -> i64 { arrived.as_u64() as i64 - sent.as_u64() as i64 + offset_ms } #[cfg(test)] mod tests { use super::*; use crate::clock::UnixMillis; /// Sent at 1000 by a clock 300 ahead (700 by ours), arrived at 750: 50 ms. #[test] fn the_offset_is_taken_off() { assert_eq!(delay_ms(UnixMillis(1000), UnixMillis(750), 300), 50); } /// Each minute's mean, oldest first; minutes past `MINUTES` forgotten. #[test] fn means_by_minute() { let start = Instant::now(); let mut delays = HopDelays::default(); delays.record(10, start); delays.record(30, start + Duration::from_secs(10)); delays.record(100, start + MINUTE * 2); let now = start + MINUTE * 2 + Duration::from_secs(1); let means = delays.by_minute(now); assert_eq!(means[MINUTES - 1], Some(100)); assert_eq!(means[MINUTES - 2], None); assert_eq!(means[MINUTES - 3], Some(20)); assert_eq!(delays.all.count, 3); delays.record(5, start + MINUTE * (MINUTES as u32 + 1)); assert_eq!(delays.minutes.len(), 2, "the first minute is forgotten"); } }