diff options
| -rw-r--r-- | src/api/mod.rs | 1 | ||||
| -rw-r--r-- | src/api/peers.rs | 23 | ||||
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/state.rs | 15 | ||||
| -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 |
8 files changed, 231 insertions, 2 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index 95d63aa..85427b0 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -159,6 +159,7 @@ fn response_for_value( crate::transport::traffic::ByKind, crate::transport::traffic::Counter, crate::state::Incidents, + peers::ThroughputInfo, ) ), tags( diff --git a/src/api/peers.rs b/src/api/peers.rs index 5b7d576..0305647 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -70,6 +70,18 @@ pub(crate) struct PeerStatus { /// since it was last listed). traffic: PeerTraffic, incidents: Incidents, + /// Smoothed sync throughput (`swarm::throughput`); absent before the first large + /// exchange. + sync_throughput: Option<ThroughputInfo>, +} + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ThroughputInfo { + bytes_per_sec: f64, + samples: u32, + /// Age of the last sample. + secs_ago: u64, } #[derive(Debug, Serialize, ToSchema)] @@ -122,6 +134,16 @@ pub(crate) async fn list_peers( .effective(&public_key, now, state.timing.rtt_fresh_for()) .map(|rtt| rtt.as_secs_f64() * 1000.0); let traffic = state.traffic.peer(&public_key).unwrap_or_default(); + let sync_throughput = state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&public_key) + .map(|estimate| ThroughputInfo { + bytes_per_sec: estimate.bytes_per_sec.round(), + samples: estimate.samples, + secs_ago: now.saturating_duration_since(estimate.at).as_secs(), + }); let ps = peer_state_mutex.lock().await; let (status, dormant_for_secs) = match ps.liveness { @@ -149,6 +171,7 @@ pub(crate) async fn list_peers( rtt_ms, traffic, incidents: ps.incidents, + sync_throughput, }); } diff --git a/src/main.rs b/src/main.rs index 622f755..7261d6a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -204,6 +204,7 @@ async fn main() -> Result<()> { neighbours: Default::default(), pulling: Default::default(), rtt: Default::default(), + throughput: Default::default(), stability: Default::default(), stale: Default::default(), catch_up_source: Default::default(), diff --git a/src/state.rs b/src/state.rs index b8944ab..d8f0bbf 100644 --- a/src/state.rs +++ b/src/state.rs @@ -16,6 +16,7 @@ use crate::backoff::Backoff; use crate::document::replica::{Epoch, Replica}; use crate::remake::{Host, PeerSummary, RemakeState}; use crate::swarm::sync::{ChangeSet, SyncEvent}; +use crate::swarm::throughput::{SyncBytes, ThroughputTable}; use crate::transport; use crate::transport::firewall::Firewall; @@ -204,6 +205,9 @@ pub(crate) struct SyncSession { #[derive(Clone, Debug)] pub(crate) struct SyncReplyWait { pub(crate) deadline: Instant, + /// When the first of the unanswered messages was generated, and the sync bytes + /// exchanged with the peer by then (`swarm::throughput`). + pub(crate) started: (Instant, SyncBytes), /// Changes first sent in the unanswered messages. pub(crate) sent: Vec<automerge::ChangeHash>, } @@ -266,13 +270,20 @@ impl SyncSession { } /// A sync message went out; `sent` are the changes it carried for the first time. - pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) { + /// `bytes` are the sync bytes exchanged with the peer before it. + pub(crate) fn expect_reply( + &mut self, + sent: Vec<automerge::ChangeHash>, + now: Instant, + bytes: SyncBytes, + ) { match &mut self.reply { // Still waiting for an earlier one: keep its deadline, remember these too. Some(wait) => wait.sent.extend(sent), None => { self.reply = Some(SyncReplyWait { deadline: now + self.reply_backoff.current(), + started: (now, bytes), sent, }); } @@ -586,6 +597,8 @@ pub(crate) struct AppState { pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>, /// Round-trip times to peers (`swarm::rtt`). pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>, + /// Sync throughput with peers (`swarm::throughput`). + pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>, /// The neighbour we are pulling changes from, and since when (`swarm::relay`). pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>, /// Our neighbours, once known (`swarm::topology`). 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); + } +} |