From f137a43b44684959a5e1de1f1e54566d8995926f Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 4 Nov 2026 22:15:34 +0100 Subject: throughput from delivered message --- docs/measurements.md | 2 +- src/swarm/sync.rs | 9 +++++++++ src/swarm/throughput.rs | 52 +++++++++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/docs/measurements.md b/docs/measurements.md index 72d2a98..3bd6451 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -271,7 +271,7 @@ 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. 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), 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. | | `epoch`, `sameEpoch`, `heads` | — | What it last advertised; `sameEpoch: false` means we don't sync with it. | diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index a9d98b0..de2d86d 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -440,6 +440,15 @@ pub(crate) async fn handle_sync( throughput .lock_or_recover() .exchange(&peer, wait.started, arrived, rtt); + } else if let Some(delay_ms) = hop_delay.filter(|ms| *ms > 0) { + // Nothing awaited: as long as it took to come. + throughput.lock_or_recover().message( + &peer, + payload_len as u64, + std::time::Duration::from_millis(delay_ms as u64), + arrived.0, + rtt, + ); } peer_state.sync.reply_settled(); diff --git a/src/swarm/throughput.rs b/src/swarm/throughput.rs index 85a3fc8..479239a 100644 --- a/src/swarm/throughput.rs +++ b/src/swarm/throughput.rs @@ -1,6 +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. No transfers of our own. +//! 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. //! //! 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 @@ -81,7 +83,34 @@ impl ThroughputTable { if transfer.is_zero() { return; } - let sample = bytes as f64 / transfer.as_secs_f64(); + self.sample(peer, bytes as f64 / transfer.as_secs_f64(), arrived); + } + + /// A sync message of `bytes` from `peer` arrived at `arrived`, `took` after it was + /// generated (`swarm::hop`: good to half the round trip, which is taken off here as + /// an exchange's round trip is, never more than half). Of messages we awaited + /// nothing for, else the exchange counts it. + pub(crate) fn message( + &mut self, + peer: &str, + bytes: u64, + took: Duration, + arrived: Instant, + rtt: Option, + ) { + if bytes < MIN_BYTES { + return; + } + let transfer = took + .saturating_sub(rtt.unwrap_or_default() / 2) + .max(took / 2); + if transfer.is_zero() { + return; + } + self.sample(peer, bytes as f64 / transfer.as_secs_f64(), arrived); + } + + fn sample(&mut self, peer: &str, sample: f64, arrived: Instant) { let estimate = self.peers.entry(peer.to_owned()).or_insert(Estimate { bytes_per_sec: sample, samples: 0, @@ -124,6 +153,25 @@ mod tests { assert!(table.get("a").is_none()); } + /// One-way: half the round trip taken off, at most half the time; small ones don't + /// count. + #[test] + fn messages_count_one_way() { + let mut table = ThroughputTable::default(); + let now = Instant::now(); + let second = Duration::from_secs(1); + table.message("a", 10_000, second, now, None); + assert!(table.get("a").is_none()); + // 1 MB in 1.5 s, of which half a 1 s round trip: 1 MB/s. + table.message("a", 1_000_000, second * 3 / 2, now, Some(second)); + let a = table.get("a").unwrap(); + assert!((a.bytes_per_sec - 1_000_000.0).abs() < 1.0); + // A round trip longer than twice the time takes off only half of it. + table.message("b", 1_000_000, second, now, Some(second * 5)); + let b = table.get("b").unwrap(); + assert!((b.bytes_per_sec - 2_000_000.0).abs() < 1.0); + } + #[test] fn round_trip_is_taken_off_but_at_most_half() { let mut table = ThroughputTable::default(); -- cgit v1.3.1