diff options
| -rw-r--r-- | docs/dashboard.md | 2 | ||||
| -rw-r--r-- | docs/measurements.md | 4 | ||||
| -rw-r--r-- | docs/openapi.yaml | 2 | ||||
| -rw-r--r-- | src/api/dashboard/dashboard.html | 2 | ||||
| -rw-r--r-- | src/swarm/blob.rs | 2 | ||||
| -rw-r--r-- | src/swarm/hop.rs | 25 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 19 | ||||
| -rw-r--r-- | src/swarm/throughput.rs | 4 | ||||
| -rw-r--r-- | src/transport/mod.rs | 3 | ||||
| -rw-r--r-- | src/transport/mycelium/mod.rs | 2 | ||||
| -rw-r--r-- | src/transport/wire.rs | 4 |
11 files changed, 59 insertions, 10 deletions
diff --git a/docs/dashboard.md b/docs/dashboard.md index 98f4077..47badc4 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -213,7 +213,7 @@ Every peer this node knows, with: | seen | How long ago this node last heard from it. | | via | The address this node sends to it at, and whether `direct` or `relayed`. | | rtt | The round trip on that path. | -| hop | How long its sync messages take to come, by its clock and this node's, its clock offset taken off (`hopDelay` in [`measurements.md`](measurements.md)): this minute's mean, then the last 15 minutes' means as bars, oldest first (a minute without messages blank). Good to half the round trip, but the trend exactly: bars growing mean the link, or a queue on it, gets slower. All figures in the tooltip. | +| hop | How long its sync and blob messages take to come, by its clock and this node's, its clock offset taken off (`hopDelay` in [`measurements.md`](measurements.md)): this minute's mean, then the last 15 minutes' means as bars, oldest first (a minute without messages blank). Good to half the round trip, but the trend exactly: bars growing mean the link, or a queue on it, gets slower. All figures in the tooltip. | | clock | How far its clock is ahead of this node's (behind: `−`), give or take (in the tooltip); in orange beyond `--clock-skew-warn`, for sure ([`measurements.md`](measurements.md), Clocks). | | neighbour | Whether this node keeps contact with it (HELLOs, keepalives, sync); others it only answers. | | reaches | How many members besides this node it reaches, as it said ([`peers.md`](peers.md)); `(+n)`: of those, how many this node doesn't reach itself, which it is a way to. By name in the tooltip, those marked `+`. Blank until it said so in this session. | diff --git a/docs/measurements.md b/docs/measurements.md index 3bd6451..a8c2730 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -271,9 +271,9 @@ In `/peers`, and as `peer_…` series with a `peer` label: | `destination.paths[].rttMs`, `.down`, `.connection` | `peer_path_rtt_seconds{transport}`, `peer_path_down{transport}` | Every path to it, best first: its round-trip time (smoothed, from PINGs), whether it is down, and the transport's connection there (`direct` or `relayed`, where it has connections). The first is the one in use: the peer's round-trip time. | | `lastSyncSecsAgo` | `peer_last_sync_seconds` | Since a sync message from it was applied (or the epoch changed). | | `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; and over messages of at least 32 KiB we awaited nothing for (taken as delivered: no answer to time), from how long they took to come (`hopDelay`), half the round trip taken off. 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. | +| `syncThroughput` | `peer_sync_throughput_bytes_per_second` | Sync bytes per second over exchanges of at least 32 KiB, round trip taken off, smoothed; and over messages of at least 32 KiB we awaited nothing for (taken as delivered: no answer to time), and blob chunks (deltas), from how long they took to come (`hopDelay`), half the round trip taken off. 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. | +| `hopDelay` | — | How long its sync and blob 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 0509709..384e517 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1771,7 +1771,7 @@ components: 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. + description: How long a peer's sync and blob 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: diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index 876c386..8ee1ee5 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -431,7 +431,7 @@ function renderPeers() { const table = document.getElementById("peers"); const headers = { reaches: "How many members besides this node it reaches (an active " + "session with each), as it said; +: those of them this node doesn't reach itself (it " + - "is a way to them)", hop: "How long its sync messages take to come, by its clock and " + + "is a way to them)", hop: "How long its sync and blob messages take to come, by its clock and " + "this node's (its offset taken off): this minute's mean, and the last 15 minutes' as " + "a line; good to half the round trip, but the trend exactly" }; const rows = [el("tr", {}, ...["peer", "status", "seen", "via", "rtt", "hop", "clock", diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index ce9a2bd..98dee66 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -278,6 +278,7 @@ async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) { epoch_seq, id, body, + sent: state.clock.now_millis(), }); if let Err(error) = state.transport.send_message(peer, &message).await { debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob answer failed"); @@ -347,6 +348,7 @@ async fn request( chunks, delta: delta.cloned(), }, + sent: state.clock.now_millis(), }); if let Err(error) = state.transport.send_message(peer, &message).await { debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob request failed"); diff --git a/src/swarm/hop.rs b/src/swarm/hop.rs index 4cd413f..1084ddd 100644 --- a/src/swarm/hop.rs +++ b/src/swarm/hop.rs @@ -1,5 +1,6 @@ -//! 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 +//! 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). @@ -61,6 +62,26 @@ impl HopDelays { } } +/// 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<i64> { + 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( diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 53f1aa3..f43a45a 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -15,6 +15,7 @@ use super::keepalive::{ heal_if_diverged, heard_from, local_status, note_alive, note_status, status_received, }; use super::lanes::SyncLanes; +use crate::locks::MutexExt; use crate::state::AppState; use crate::swarm::sync::SyncEvent; use crate::transport::{self, ParsedMessage}; @@ -151,6 +152,24 @@ pub(crate) async fn handle_message( lanes.enqueue(&peer, envelope); } transport::ParsedMessage::Blob(blob) => { + // How long it took to come, and of a chunk, the throughput (it was asked + // for: the request's round trip is no exchange of sync bytes to time). + if let Some(delay_ms) = crate::swarm::hop::note(state, &peer, blob.sent).await + && delay_ms > 0 + && let transport::BlobBody::Chunk { data, .. } = &blob.body + { + let rtt = + state + .lock_rtt() + .effective(&peer, Instant::now(), state.timing.rtt_fresh_for()); + state.throughput.lock_or_recover().message( + &peer, + data.len() as u64, + Duration::from_millis(delay_ms as u64), + Instant::now(), + rtt, + ); + } crate::swarm::blob::handle(state, &peer, blob).await; } transport::ParsedMessage::Status(remote) => { diff --git a/src/swarm/throughput.rs b/src/swarm/throughput.rs index 479239a..ec50693 100644 --- a/src/swarm/throughput.rs +++ b/src/swarm/throughput.rs @@ -1,8 +1,8 @@ //! Sync throughput with peers, measured from sync exchanges we have anyway: from sending //! a sync message until the peer's answer arrives, the sync bytes that went either way //! with that peer; and from a large message we awaited nothing for (one taken as -//! delivered, which gets no answer to time), how long it took to come (`swarm::hop`). No -//! transfers of our own. +//! delivered, which gets no answer to time), and from a blob chunk (`swarm::blob`), how +//! long it took to come (`swarm::hop`). No transfers of our own. //! //! Only exchanges of at least `MIN_BYTES` count; smaller ones measure the round trip, //! not the throughput. The round trip (`swarm::rtt`) is taken off the elapsed time, but diff --git a/src/transport/mod.rs b/src/transport/mod.rs index ae6b593..4ddfe32 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -295,6 +295,9 @@ pub(crate) struct BlobMessage { pub(crate) epoch_seq: u32, pub(crate) id: BlobId, pub(crate) body: BlobBody, + /// When the sender sent it, by its clock (as `SyncEnvelope::sent`): the receiver + /// takes how long it took to come (`swarm::hop`), and of a chunk, the throughput. + pub(crate) sent: crate::clock::UnixMillis, } #[derive(Clone, Serialize, Deserialize)] diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 3e33a9f..3552edf 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -752,6 +752,7 @@ mod tests { index: 5, data: vec![1u8; 64 * 1024].into(), }, + sent: crate::clock::UnixMillis(6), }); let bytes = wire::encode(&chunk).unwrap(); // A 64 KiB chunk costs only a few bytes of framing. @@ -773,6 +774,7 @@ mod tests { chunk_size: 64, hashes: vec![[1; 32], [2; 32]], }, + sent: crate::clock::UnixMillis(6), }); let decoded = decode_blob(&wire::encode(&manifest).unwrap()); assert!(matches!( diff --git a/src/transport/wire.rs b/src/transport/wire.rs index d1b3353..b3a8ee9 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 = 26; +pub(crate) const VERSION: u8 = 27; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -420,6 +420,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { epoch_seq: 1, id: BlobId::Content([5; 32]), body: BlobBody::Missing, + sent: crate::clock::UnixMillis(9), }), ParsedMessage::Who(proof.clone()), ParsedMessage::Me(proof), @@ -603,6 +604,7 @@ mod tests { epoch_seq: 0, id: BlobId::Content([1; 32]), body: BlobBody::Missing, + sent: crate::clock::UnixMillis(2), }); let payload = encode(&blob).unwrap(); assert!(matches!(decode(&payload).unwrap(), ParsedMessage::Blob(_))); |