diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 5 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 19 | ||||
| -rw-r--r-- | src/swarm/throughput.rs | 168 |
4 files changed, 192 insertions, 1 deletions
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 83e2d2d..6f194dc 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -12,4 +12,5 @@ pub(crate) mod source; pub(crate) mod stability; pub(crate) mod stats; pub(crate) mod sync; +pub(crate) mod throughput; pub(crate) mod topology; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 7a3eed8..3b85539 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -496,6 +496,11 @@ pub(crate) async fn run_peer_worker(state: AppState) { { let known = state.peer_states.read().await; state.traffic.retain(|peer| known.contains_key(peer)); + state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .retain(|peer| known.contains_key(peer)); } if let Some(change) = topology::refresh(&state).await { apply_neighbour_change(&state, change).await; diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 09e10f4..3284015 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -14,6 +14,7 @@ use tracing::{debug, info, trace, warn}; use crate::state::{AppState, Liveness, Sending, Standing}; use crate::swarm::peer::{peer_swarm, send_hello_once}; use crate::swarm::source; +use crate::swarm::throughput::SyncBytes; use crate::transport::{self, ParsedMessage}; #[derive(Debug)] @@ -127,6 +128,8 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let transport = Arc::clone(&state.transport); let peer_owned = peer.to_owned(); let ready = state.is_ready().await; + // Before this message is counted. + let bytes_before = SyncBytes::of(&state.traffic, peer); let generated = tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> { @@ -157,7 +160,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { if in_step { peer_state.sync.reply_settled(); } else if message.is_some() { - peer_state.sync.expect_reply(sent.clone(), Instant::now()); + peer_state + .sync + .expect_reply(sent.clone(), Instant::now(), bytes_before); } if message.is_some() { // Marked under the same lock as the generation, so no second @@ -255,6 +260,8 @@ pub(crate) async fn handle_sync( envelope: transport::SyncEnvelope, ) -> Result<()> { debug!(peer = %peer, "Processing sync envelope"); + // This message is counted already (`transport::traffic`). + let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer)); let peer_state_mutex = state.get_or_create_peer_state(peer).await; @@ -306,6 +313,10 @@ pub(crate) async fn handle_sync( let doc = Arc::clone(&state.doc); let peer_state_mutex = Arc::clone(&peer_state_mutex); let stats = Arc::clone(&state.stats); + let throughput = Arc::clone(&state.throughput); + let rtt = state + .lock_rtt() + .effective(peer, Instant::now(), state.timing.rtt_fresh_for()); let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire); // While catching up, everything arrives late; that says nothing about propagation. let measure_propagation = !check_caught_up; @@ -334,6 +345,12 @@ pub(crate) async fn handle_sync( .receive_sync_message(&mut peer_state.sync.state, message) .context("Automerge synchronization failed")?; peer_state.last_sync_rx = Some(Instant::now()); + if let Some(wait) = &peer_state.sync.reply { + throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .exchange(&peer, wait.started, arrived, rtt); + } peer_state.sync.reply_settled(); resend_if_needed(&peer, &mut peer_state.sync.state); diff --git a/src/swarm/throughput.rs b/src/swarm/throughput.rs new file mode 100644 index 0000000..f1b9b14 --- /dev/null +++ b/src/swarm/throughput.rs @@ -0,0 +1,168 @@ +//! 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. +//! +//! 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 +//! never more than half of it, so a stale round trip can't inflate a sample. What +//! remains also includes the time the peer took to apply and answer: this is the rate a +//! peer delivers at in practice, not the raw capacity of the path. +//! +//! Samples are smoothed (exponential moving average, weight 1/4). Large exchanges are +//! rare (catching up, big writes), so estimates are kept however old; they come with +//! their age. + +use std::collections::HashMap; +use std::time::Duration; + +use tokio::time::Instant; + +use crate::transport::traffic::Traffic; + +/// Smallest exchange that counts: some dozens of Mycelium chunks. +const MIN_BYTES: u64 = 32 * 1024; +const WEIGHT: f64 = 0.25; + +/// Sync bytes exchanged with a peer so far (from the traffic counters). +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct SyncBytes { + sent: u64, + received: u64, +} + +impl SyncBytes { + pub(crate) fn of(traffic: &Traffic, peer: &str) -> Self { + traffic + .peer(peer) + .map(|t| Self { + sent: t.sent.sync.bytes, + received: t.received.sync.bytes, + }) + .unwrap_or_default() + } + + /// Bytes either way since `earlier` (0 if the counters were dropped meanwhile). + fn since(&self, earlier: &SyncBytes) -> u64 { + self.sent.saturating_sub(earlier.sent) + self.received.saturating_sub(earlier.received) + } +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct Estimate { + pub(crate) bytes_per_sec: f64, + pub(crate) samples: u32, + /// When the last sample was taken. + pub(crate) at: Instant, +} + +#[derive(Debug, Default)] +pub(crate) struct ThroughputTable { + peers: HashMap<String, Estimate>, +} + +impl ThroughputTable { + /// A sync exchange with `peer` finished: it started at `started` with the counters + /// at `before`, and its answer arrived at `arrived` with the counters at `after`. + pub(crate) fn exchange( + &mut self, + peer: &str, + (started, before): (Instant, SyncBytes), + (arrived, after): (Instant, SyncBytes), + rtt: Option<Duration>, + ) { + let bytes = after.since(&before); + if bytes < MIN_BYTES { + return; + } + let elapsed = arrived.saturating_duration_since(started); + let transfer = elapsed + .saturating_sub(rtt.unwrap_or_default()) + .max(elapsed / 2); + if transfer.is_zero() { + return; + } + let sample = bytes as f64 / transfer.as_secs_f64(); + let estimate = self.peers.entry(peer.to_owned()).or_insert(Estimate { + bytes_per_sec: sample, + samples: 0, + at: arrived, + }); + if estimate.samples > 0 { + estimate.bytes_per_sec += WEIGHT * (sample - estimate.bytes_per_sec); + } + estimate.samples += 1; + estimate.at = arrived; + } + + pub(crate) fn get(&self, peer: &str) -> Option<Estimate> { + self.peers.get(peer).copied() + } + + pub(crate) fn retain(&mut self, mut keep: impl FnMut(&str) -> bool) { + self.peers.retain(|peer, _| keep(peer)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn bytes(sent: u64, received: u64) -> SyncBytes { + SyncBytes { sent, received } + } + + #[test] + fn small_exchanges_dont_count() { + let mut table = ThroughputTable::default(); + let t0 = Instant::now(); + table.exchange( + "a", + (t0, bytes(0, 0)), + (t0 + Duration::from_millis(200), bytes(500, 10_000)), + None, + ); + assert!(table.get("a").is_none()); + } + + #[test] + fn round_trip_is_taken_off_but_at_most_half() { + let mut table = ThroughputTable::default(); + let t0 = Instant::now(); + // 1 MB in 2 s, of which 1 s round trip: 1 MB/s. + table.exchange( + "a", + (t0, bytes(100, 0)), + (t0 + Duration::from_secs(2), bytes(200, 999_900)), + Some(Duration::from_secs(1)), + ); + let a = table.get("a").unwrap(); + assert!((a.bytes_per_sec - 1_000_000.0).abs() < 1.0); + + // A round trip longer than the exchange takes off only half. + table.exchange( + "b", + (t0, bytes(0, 0)), + (t0 + Duration::from_secs(2), bytes(0, 1_000_000)), + Some(Duration::from_secs(5)), + ); + assert!((table.get("b").unwrap().bytes_per_sec - 1_000_000.0).abs() < 1.0); + } + + #[test] + fn smoothed() { + let mut table = ThroughputTable::default(); + let t0 = Instant::now(); + for received in [500_000u64, 1_000_000] { + table.exchange( + "a", + (t0, bytes(0, 0)), + (t0 + Duration::from_secs(1), bytes(0, received)), + None, + ); + } + // Samples of 500 kB/s, then 1 MB/s: 500 kB/s + (1 MB/s - 500 kB/s) / 4. + let a = table.get("a").unwrap(); + assert_eq!(a.samples, 2); + assert!((a.bytes_per_sec - 625_000.0).abs() < 1.0); + } +} |