diff options
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/state.rs | 2 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 2 | ||||
| -rw-r--r-- | src/swarm/source.rs | 1 | ||||
| -rw-r--r-- | src/swarm/stability.rs | 7 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 179 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 8 | ||||
| -rw-r--r-- | src/swarm/topology.rs | 38 |
9 files changed, 238 insertions, 1 deletions
diff --git a/src/main.rs b/src/main.rs index 8b6f823..622f755 100644 --- a/src/main.rs +++ b/src/main.rs @@ -191,6 +191,7 @@ async fn main() -> Result<()> { )), firewall, traffic, + stats: Default::default(), sync_events: sync_tx, change_tx, initial_sync_done: Arc::new(AtomicBool::new(false)), diff --git a/src/state.rs b/src/state.rs index c3b23de..aae33f5 100644 --- a/src/state.rs +++ b/src/state.rs @@ -519,6 +519,8 @@ pub(crate) struct AppState { pub(crate) firewall: Arc<Firewall>, /// Messages and bytes exchanged, per peer (`transport::traffic`). pub(crate) traffic: Arc<transport::traffic::Traffic>, + /// Node-wide figures for /admin/stats (`swarm::stats`). + pub(crate) stats: Arc<crate::swarm::stats::NodeStats>, pub(crate) sync_events: mpsc::UnboundedSender<SyncEvent>, diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index ee0de20..83e2d2d 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -10,5 +10,6 @@ pub(crate) mod relay; pub(crate) mod rtt; pub(crate) mod source; pub(crate) mod stability; +pub(crate) mod stats; pub(crate) mod sync; pub(crate) mod topology; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 3ceea3c..7a3eed8 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -492,6 +492,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { apply_forget_records(&state).await; crate::swarm::members::ensure_registered(&state).await; crate::swarm::stability::tick(&state).await; + crate::swarm::stats::measure_saved_size(&state); { let known = state.peer_states.read().await; state.traffic.retain(|peer| known.contains_key(peer)); @@ -725,6 +726,7 @@ pub(crate) async fn handle_message( // we were down) says nothing about now; the peer answers our current HELLO. if let transport::ParsedMessage::HelloReply(reply) = &message { if reply.run_id != state.run_id { + state.stats.stale_hello_reply(); debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run"); return Ok(()); } diff --git a/src/swarm/source.rs b/src/swarm/source.rs index 36a333e..4025f0e 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -111,6 +111,7 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt ), None => debug!(peer = %peer, rtt_ms, "Catching up from this peer first"), } + state.stats.catch_up_source_chosen(); *source = Some(CatchUpSource { peer: peer.to_owned(), last_heard: now, diff --git a/src/swarm/stability.rs b/src/swarm/stability.rs index 31ba4b8..9d79b17 100644 --- a/src/swarm/stability.rs +++ b/src/swarm/stability.rs @@ -135,6 +135,13 @@ fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Stability> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } +/// The fraction of the observed time (up to a day) we had a live peer, and how much +/// time was observed. +pub(crate) fn measured(state: &AppState) -> (f64, Duration) { + let stability = lock(state); + (stability.fraction(), stability.observed()) +} + /// A HELLO_REPLY told us how long ago that peer last heard from us before this run. pub(crate) fn note_away(state: &AppState, away: Option<Duration>) { if let Some(away) = away { diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs new file mode 100644 index 0000000..79e5c83 --- /dev/null +++ b/src/swarm/stats.rs @@ -0,0 +1,179 @@ +//! Node-wide figures for /admin/stats that no other part keeps. Recorded where things +//! happen anyway; the one exception is the document's saved size (see +//! `measure_saved_size`). + +use std::sync::{Mutex, MutexGuard}; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::{debug, warn}; + +use crate::state::AppState; + +/// Saving takes time proportional to the document; we spend at most about 1% of it on +/// measuring, and measure at least this long apart. +const SAVED_SIZE_MIN_INTERVAL: Duration = Duration::from_secs(60); +const SAVED_SIZE_DUTY: u32 = 100; + +#[derive(Debug)] +pub(crate) struct NodeStats { + started: Instant, + inner: Mutex<Inner>, +} + +#[derive(Debug, Default)] +struct Inner { + caught_up: Option<CaughtUp>, + catch_up_sources: u32, + max_heads: usize, + stale_hello_replies: u64, + saved_size: Option<SavedSize>, + measuring: bool, +} + +/// The document's size as saved (compressed), when it was measured and how long that +/// took (copy and save). +#[derive(Debug, Clone, Copy)] +pub(crate) struct SavedSize { + pub(crate) bytes: u64, + pub(crate) at: Instant, + pub(crate) took: Duration, +} + +/// When the initial sync finished, and what it took. +#[derive(Debug, Clone, Copy)] +pub(crate) struct CaughtUp { + /// Since the process started (handshakes included). + pub(crate) after: Duration, + /// Sync payload received until then, from all peers. + pub(crate) sync_bytes: u64, +} + +/// A copy of the figures. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Snapshot { + pub(crate) uptime: Duration, + pub(crate) caught_up: Option<CaughtUp>, + pub(crate) catch_up_sources: u32, + pub(crate) max_heads: usize, + pub(crate) stale_hello_replies: u64, +} + +impl Default for NodeStats { + fn default() -> Self { + Self { + started: Instant::now(), + inner: Mutex::default(), + } + } +} + +impl NodeStats { + fn lock(&self) -> MutexGuard<'_, Inner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + pub(crate) fn caught_up(&self, sync_bytes: u64) { + let after = self.started.elapsed(); + self.lock() + .caught_up + .get_or_insert(CaughtUp { after, sync_bytes }); + } + + /// A catch-up source was chosen (the first, or a replacement). + pub(crate) fn catch_up_source_chosen(&self) { + self.lock().catch_up_sources += 1; + } + + /// Our document has `count` heads (after applying a sync message). + pub(crate) fn heads_seen(&self, count: usize) { + let mut inner = self.lock(); + inner.max_heads = inner.max_heads.max(count); + } + + pub(crate) fn stale_hello_reply(&self) { + self.lock().stale_hello_replies += 1; + } + + /// The last measurement of the saved size, if any. + pub(crate) fn saved_size(&self) -> Option<SavedSize> { + self.lock().saved_size + } + + /// Whether a measurement is due; if so, it counts as started. + fn start_measuring(&self, now: Instant) -> bool { + let mut inner = self.lock(); + let due = !inner.measuring + && inner.saved_size.is_none_or(|last| { + now >= last.at + SAVED_SIZE_MIN_INTERVAL.max(last.took * SAVED_SIZE_DUTY) + }); + if due { + inner.measuring = true; + } + due + } + + fn finish_measuring(&self, measured: Option<SavedSize>) { + let mut inner = self.lock(); + inner.measuring = false; + if let Some(measured) = measured { + inner.saved_size = Some(measured); + } + } + + pub(crate) fn snapshot(&self) -> Snapshot { + let inner = self.lock(); + Snapshot { + uptime: self.started.elapsed(), + caught_up: inner.caught_up, + catch_up_sources: inner.catch_up_sources, + max_heads: inner.max_heads, + stale_hello_replies: inner.stale_hello_replies, + } + } +} + +/// Measures the document's saved size in the background when due: the first time, then +/// at least a minute apart and spending no more than about 1% of the time on it (a +/// document that takes 2 s to copy and save is measured every 200 s). Called by the +/// peer worker every few seconds. +/// +/// The copy holds the read lock (writes wait for it); the save runs without any lock. +pub(crate) fn measure_saved_size(state: &AppState) { + if !state.stats.start_measuring(Instant::now()) { + return; + } + let state = state.clone(); + tokio::spawn(async move { + let start = Instant::now(); + let doc = std::sync::Arc::clone(&state.doc); + // Both steps take time proportional to the document: off the async threads. + let measured = match tokio::task::spawn_blocking(move || { + let mut copy = doc.blocking_read().doc.clone(); + copy.save().len() as u64 + }) + .await + { + Ok(bytes) => { + let took = start.elapsed(); + debug!( + bytes, + took_ms = took.as_millis() as u64, + "Measured the saved document size" + ); + Some(SavedSize { + bytes, + at: Instant::now(), + took, + }) + } + Err(error) => { + warn!(error = %error, "Measuring the saved document size failed"); + None + } + }; + state.stats.finish_measuring(measured); + }); +} diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 383cdaf..adf8146 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -305,6 +305,7 @@ pub(crate) async fn handle_sync( let applied = { let doc = Arc::clone(&state.doc); let peer_state_mutex = Arc::clone(&peer_state_mutex); + let stats = Arc::clone(&state.stats); let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire); let epoch = envelope.epoch; let message = envelope.message; @@ -334,7 +335,9 @@ pub(crate) async fn handle_sync( peer_state.sync.reply_settled(); resend_if_needed(&peer, &mut peer_state.sync.state); - let changed = replica.doc.get_heads() != heads_before; + let heads_after = replica.doc.get_heads(); + stats.heads_seen(heads_after.len()); + let changed = heads_after != heads_before; Ok(Some(( // Only from a ready peer: a peer still catching up has empty heads, which @@ -367,6 +370,9 @@ pub(crate) async fn handle_sync( if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) { info!(peer = %peer, "Initial sync complete"); + state + .stats + .caught_up(state.traffic.total().received.sync.bytes); source::finished(state, peer); } diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index 8e25a2a..c5a02f1 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -48,6 +48,44 @@ pub(crate) fn is_full_mesh(state: &AppState) -> bool { .is_none_or(|view| view.full_mesh) } +/// The current view in numbers, once known. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Summary { + /// Registered members, us included. + pub(crate) members: usize, + /// Members other than us not currently judged unreachable. + pub(crate) reachable: usize, + pub(crate) stable: usize, + pub(crate) neighbours: usize, + pub(crate) stable_neighbours: usize, + pub(crate) full_mesh: bool, +} + +pub(crate) fn summary(state: &AppState) -> Option<Summary> { + let local = state.local_id(); + state + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .map(|view| Summary { + members: view.members.len(), + reachable: view + .members + .iter() + .filter(|m| m.as_str() != local && !view.unreachable.contains(*m)) + .count(), + stable: view.stable.len(), + neighbours: view.neighbours.len(), + stable_neighbours: view + .neighbours + .iter() + .filter(|m| view.stable.contains(*m)) + .count(), + full_mesh: view.full_mesh, + }) +} + /// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync /// openings. Everyone is a neighbour until the view is known (while joining, or before /// any member registered), so joining works as in a full mesh. We always *answer* |