diff options
| -rw-r--r-- | docs/measurements.md | 1 | ||||
| -rw-r--r-- | docs/openapi.yaml | 32 | ||||
| -rw-r--r-- | src/api/peers.rs | 42 | ||||
| -rw-r--r-- | src/state/peer.rs | 3 | ||||
| -rw-r--r-- | src/swarm/hop.rs | 102 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 16 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 11 | ||||
| -rw-r--r-- | src/transport/mod.rs | 3 | ||||
| -rw-r--r-- | src/transport/wire.rs | 3 |
10 files changed, 211 insertions, 3 deletions
diff --git a/docs/measurements.md b/docs/measurements.md index 483e185..72d2a98 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -273,6 +273,7 @@ In `/peers`, and as `peer_…` series with a `peer` label: | `divergedForSecs` | `peer_diverged_seconds` | How long its heads have differed from ours with no sync message from it in between, nor both moving on; absent while they match. Long values precede a healing re-handshake ([`peers.md`](peers.md)). | | `syncThroughput` | `peer_sync_throughput_bytes_per_second` | Sync bytes per second over exchanges of at least 32 KiB, round trip taken off, smoothed. Absent before the first large exchange; comes with its age (`secsAgo`) since large exchanges are rare. This is the rate the peer delivers at in practice (its apply time included), not the path's capacity. | | `clock.offsetMs`, `.withinMs` | `peer_clock_offset_seconds`, `peer_clock_offset_within_seconds` | How far its clock is ahead of ours (behind if negative), give or take. Each PONG says when the peer answered, by its clock, which lies between our PING going and the PONG coming: within half the round trip of its middle, however unevenly the two ways took. Of its paths and the last 5 minutes, the narrowest bound is kept; a sample it contradicts (a clock set anew) replaces it at once. Absent before a PONG from it. | +| `hopDelay` | — | How long its sync messages take to come: each says when the peer generated it, by its clock; we take how long until we handle it, its clock offset (`clock`) taken off. So the transport, the network and the queues at both ends, which a PING's round trip leaves out (PINGs wait behind nothing); `halfRttMs` to compare. Good to the offset's error, half the round trip, but that error is the same for every message while the offset stands: the means of the last 15 minutes (`byMinuteMs`) show a link getting worse exactly. All since start in buckets (`halfWithinMs`, `p99WithinMs`), as propagation's. Absent before a message came while we knew its clock. | | `epoch`, `sameEpoch`, `heads` | — | What it last advertised; `sameEpoch: false` means we don't sync with it. | | `handshakeComplete`, `hasPendingDelivery` | — | Session state, for debugging. | diff --git a/docs/openapi.yaml b/docs/openapi.yaml index fd0b855..0509709 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1686,6 +1686,10 @@ components: oneOf: - $ref: '#/components/schemas/ClockOffset' - type: 'null' + hopDelay: + oneOf: + - $ref: '#/components/schemas/HopDelay' + - type: 'null' Counter: type: object @@ -1765,6 +1769,34 @@ components: type: integer description: Give or take this much (half the round trip). + HopDelay: + type: object + description: How long a peer's sync messages take to come, from its clock when it generated one to ours when we handle it, its clock offset taken off (`measurements.md`, Per peer). Good to half the round trip, the offset's error; that error is the same for every message while the offset stands, so the minutes' means show a trend exactly. + required: [count, meanMs, halfWithinMs, p99WithinMs, belowZero, byMinuteMs, halfRttMs] + properties: + count: + type: integer + description: Messages measured since this node started. + meanMs: + type: integer + halfWithinMs: + type: [integer, 'null'] + description: The bucket bound by which half came (the propagation buckets of `/admin/stats`); null if beyond the last. + p99WithinMs: + type: [integer, 'null'] + description: The same for 99%. + belowZero: + type: integer + description: Messages that took less than nothing by the clocks (the offset's error was larger), counted as 0 above. + byMinuteMs: + type: array + items: + type: [integer, 'null'] + description: The mean of each of the last 15 minutes, oldest first, the last the current one; null for a minute without messages. May be negative. + halfRttMs: + type: [integer, 'null'] + description: Half the round trip to it now, from PINGs, which wait behind nothing; what the delay takes beyond it waits in queues (or is the offset's error). Null without a recent round trip. + Throughput: type: object required: [bytesPerSec, samples, secsAgo] diff --git a/src/api/peers.rs b/src/api/peers.rs index 6662a55..b00593f 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -74,6 +74,34 @@ pub(crate) struct PeerStatus { /// How far its clock is ahead of ours, give or take (`Transport::clock_offset`); /// absent before a PONG from it. pub(crate) clock: Option<ClockOffset>, + /// How long its sync messages take to come (`swarm::hop`); absent before one came + /// while we knew its clock. + pub(crate) hop_delay: Option<HopDelayInfo>, +} + +/// How long a peer's sync messages take to come, by its clock and ours, its offset +/// taken off: good to half the round trip (the offset's error), the same for every +/// message while the offset stands, so the minutes' means show a trend exactly. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct HopDelayInfo { + /// Messages measured since we started. + pub(crate) count: u64, + pub(crate) mean_ms: u64, + /// The bucket bounds by which half, and 99%, came (`/admin/stats`' propagation + /// buckets); absent if beyond the last. + pub(crate) half_within_ms: Option<u64>, + pub(crate) p99_within_ms: Option<u64>, + /// Messages that took less than nothing by the clocks: the offset's error was larger + /// (counted as 0 above). + pub(crate) below_zero: u64, + /// The mean of each of the last 15 minutes, oldest first (the last is the current + /// one); null for a minute without messages. May be negative, as `belowZero`. + pub(crate) by_minute_ms: Vec<Option<i64>>, + /// Half the round trip to it now, from PINGs (which wait behind nothing): what a + /// message would take with nothing waiting anywhere; what the delay takes beyond it + /// waits in queues (and is the offset's error). Absent without a recent round trip. + pub(crate) half_rtt_ms: Option<u64>, } #[derive(Debug, Serialize)] @@ -163,6 +191,19 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { let addresses = state.transport.addresses_of(&peer); let destination = state.transport.destination(&peer); let clock = state.transport.clock_offset(&peer); + let hops = &ps.hop_delays; + let hop_delay = (hops.all.count > 0).then(|| HopDelayInfo { + count: hops.all.count, + mean_ms: hops.all.sum_ms / hops.all.count, + half_within_ms: hops.all.within(0.5), + p99_within_ms: hops.all.within(0.99), + below_zero: hops.all.ahead, + by_minute_ms: hops.by_minute(now), + half_rtt_ms: state + .lock_rtt() + .effective(&peer, now, state.timing.rtt_fresh_for()) + .map(|rtt| rtt.as_millis() as u64 / 2), + }); peers.insert( peer, PeerStatus { @@ -192,6 +233,7 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { incidents: ps.incidents, sync_throughput, clock, + hop_delay, }, ); } diff --git a/src/state/peer.rs b/src/state/peer.rs index 8f79886..787a772 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -153,6 +153,8 @@ pub(crate) struct PeerState { /// with its epoch: once they are ours, it has every change we would announce to it /// (`swarm::relay`). pub(crate) last_heads: Option<(transport::Epoch, transport::HeadsSummary)>, + /// How long its sync messages took to come (`swarm::hop`). + pub(crate) hop_delays: crate::swarm::hop::HopDelays, /// When we last told it our heads in a slow round (`swarm::relay::SlowRounds`). pub(crate) slow_announced_at: Option<Instant>, /// Whom it said it reaches (`swarm::reach`), by pair key, and when. @@ -288,6 +290,7 @@ impl PeerState { syncs_with_us: None, reported_heads: None, last_heads: None, + hop_delays: Default::default(), slow_announced_at: None, reaches: None, tree_writer: None, diff --git a/src/swarm/hop.rs b/src/swarm/hop.rs new file mode 100644 index 0000000..4cd413f --- /dev/null +++ b/src/swarm/hop.rs @@ -0,0 +1,102 @@ +//! How long sync messages take from a peer to us, per hop: from the peer's clock when it +//! generated one (`SyncEnvelope::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<Option<i64>> { + 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 + } +} + +/// 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"); + } +} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 985e968..071f00e 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -6,6 +6,7 @@ pub(crate) mod blob; pub(crate) mod bootstrap; pub(crate) mod changes; pub(crate) mod document_file; +pub(crate) mod hop; pub(crate) mod links; pub(crate) mod members; pub(crate) mod pair; diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index 74076a4..de63962 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -59,7 +59,7 @@ pub(crate) struct Propagation { } impl Propagation { - fn record(&mut self, delay_ms: i64) { + pub(crate) fn record(&mut self, delay_ms: i64) { if delay_ms < 0 { self.ahead += 1; } @@ -73,6 +73,20 @@ impl Propagation { self.sum_ms += delay_ms; } + /// The bound of the bucket by which `quantile` of the delays came: None if beyond + /// the last bound, or none came. + pub(crate) fn within(&self, quantile: f64) -> Option<u64> { + let wanted = (self.count as f64 * quantile).ceil() as u64; + let mut seen = 0; + for (bound, count) in self.buckets() { + seen += count; + if seen >= wanted && seen > 0 { + return bound; + } + } + None + } + /// Each bucket's upper bound (None for the last) and count. pub(crate) fn buckets(&self) -> impl Iterator<Item = (Option<u64>, u64)> + '_ { PROPAGATION_BUCKETS_MS diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 4bb6ba5..a9d98b0 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -140,6 +140,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let delta_threshold = state.blobs.delta_threshold_for(rate, failed); let stats = Arc::clone(&state.stats); + let clock = Arc::clone(&state.clock); // The session the message comes from, for its send's outcome. let (session, generated) = tokio::task::spawn_blocking( move || -> (u64, Result<Option<transport::EncodedMessage>>) { @@ -205,6 +206,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { ready, writer, delivered, + sent: clock.now_millis(), }; let encoded = transport .encode(&peer_owned, &ParsedMessage::Sync(envelope)) @@ -318,11 +320,18 @@ pub(crate) async fn handle_sync( debug!(peer = %peer, "Processing sync envelope"); // This message is counted already (`transport::traffic`). let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer)); + // How long it took to come, where we know the peer's clock (`swarm::hop`). + let hop_delay = state.transport.clock_offset(peer).map(|offset| { + crate::swarm::hop::delay_ms(envelope.sent, state.clock.now_millis(), offset.offset_ms) + }); let peer_state_mutex = state.get_or_create_peer_state(peer).await; let needs_hello = { - let peer_state = peer_state_mutex.lock().await; + let mut peer_state = peer_state_mutex.lock().await; + if let Some(delay_ms) = hop_delay { + peer_state.hop_delays.record(delay_ms, arrived.0); + } match &peer_state.standing { Standing::Stranger | Standing::Admitted => { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 426ef72..ae6b593 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -671,6 +671,9 @@ pub(crate) struct SyncEnvelope { /// heads, and the sender awaits no answer. A receiver it leaves level with the /// sender answers nothing; one that has changes the sender lacks answers with them. pub(crate) delivered: bool, + /// When the sender generated it, by its clock: the receiver takes how long it took + /// to come (`swarm::hop`). + pub(crate) sent: crate::clock::UnixMillis, } /// What a sync envelope carries. diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 4199927..d1b3353 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 25; +pub(crate) const VERSION: u8 = 26; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -411,6 +411,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { ready: true, writer: Some(7), delivered: true, + sent: crate::clock::UnixMillis(8), }), ParsedMessage::Status(status.clone()), ParsedMessage::StatusReply(status), |