diff options
Diffstat (limited to 'src/swarm')
| -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 |
4 files changed, 46 insertions, 4 deletions
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 |