diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/blob.rs | 1 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 10 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 34 |
3 files changed, 44 insertions, 1 deletions
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index a2feb98..d68a78c 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -124,6 +124,7 @@ pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) { match message.body { BlobBody::Request { chunks } => { let Ok(permit) = Arc::clone(&state.blobs.answering).try_acquire_owned() else { + state.stats.blob_request_dropped(); debug!(peer = %peer, blob = %message.id, "Answering too many blob requests; dropping one"); return; }; diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 087e028..2fa5f5f 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -81,6 +81,7 @@ pub(crate) async fn announce(state: &AppState, from: &str) { /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. async fn announce_now(state: &AppState, senders: &HashSet<String>) { + state.stats.announce_round(); let skip = match senders.len() { 1 => senders.iter().next().cloned(), _ => None, @@ -165,7 +166,13 @@ pub(crate) async fn pull_if_behind( Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT ); if busy { - pulling.waiting.insert(peer.to_owned(), remote.clone()); + if pulling + .waiting + .insert(peer.to_owned(), remote.clone()) + .is_none() + { + state.stats.pull_queued(); + } return false; } pulling.current = Some((peer.to_owned(), now)); @@ -174,6 +181,7 @@ pub(crate) async fn pull_if_behind( // Automerge asks for whatever these heads need that we don't have. peer_state.lock().await.sync.pull_from(theirs); + state.stats.pull_started(); debug!(peer = %peer, "Neighbour has changes we lack; pulling"); let _ = state .sync_events diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index f9a5b7b..8172233 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -29,6 +29,8 @@ struct Inner { catch_up_sources: u32, max_heads: usize, stale_hello_replies: u64, + relay: Relay, + blob_requests_dropped: u64, saved_size: Option<SavedSize>, measuring: bool, /// The document version (`AppState::change_tx`) and epoch the saved size is of. @@ -90,6 +92,17 @@ pub(crate) struct SavedSize { pub(crate) took: Duration, } +/// Passing changes on (`swarm::relay`), since start. +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct Relay { + /// Announcement rounds (each a PING to the neighbours). + pub(crate) announce_rounds: u64, + /// Pulls started. + pub(crate) pulls: u64, + /// Announcements of heads we lacked that had to wait for the pull in progress. + pub(crate) pulls_queued: u64, +} + /// When the initial sync finished, and what it took. #[derive(Debug, Clone, Copy)] pub(crate) struct CaughtUp { @@ -107,6 +120,8 @@ pub(crate) struct Snapshot { pub(crate) catch_up_sources: u32, pub(crate) max_heads: usize, pub(crate) stale_hello_replies: u64, + pub(crate) relay: Relay, + pub(crate) blob_requests_dropped: u64, pub(crate) propagation: Propagation, } @@ -154,6 +169,23 @@ impl NodeStats { self.lock().stale_hello_replies += 1; } + pub(crate) fn announce_round(&self) { + self.lock().relay.announce_rounds += 1; + } + + pub(crate) fn pull_started(&self) { + self.lock().relay.pulls += 1; + } + + pub(crate) fn pull_queued(&self) { + self.lock().relay.pulls_queued += 1; + } + + /// A blob request went unanswered: too many were being answered. + pub(crate) fn blob_request_dropped(&self) { + self.lock().blob_requests_dropped += 1; + } + /// The last measurement of the saved size, if any. pub(crate) fn saved_size(&self) -> Option<SavedSize> { self.lock().saved_size @@ -203,6 +235,8 @@ impl NodeStats { catch_up_sources: inner.catch_up_sources, max_heads: inner.max_heads, stale_hello_replies: inner.stale_hello_replies, + relay: inner.relay, + blob_requests_dropped: inner.blob_requests_dropped, propagation: inner.propagation, } } |