From d9c61bd526fd83165d583cb5d75be66385611be4 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 4 Nov 2026 23:32:40 +0100 Subject: time blob messages --- src/api/dashboard/dashboard.html | 2 +- src/swarm/blob.rs | 2 ++ src/swarm/hop.rs | 25 +++++++++++++++++++++++-- src/swarm/peer/inbound.rs | 19 +++++++++++++++++++ src/swarm/throughput.rs | 4 ++-- src/transport/mod.rs | 3 +++ src/transport/mycelium/mod.rs | 2 ++ src/transport/wire.rs | 4 +++- 8 files changed, 55 insertions(+), 6 deletions(-) (limited to 'src') 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 { + 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 { 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(_))); -- cgit v1.3.1