diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/stats.rs | 22 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 8 |
2 files changed, 30 insertions, 0 deletions
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index 9d32baa..da7544f 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -110,6 +110,14 @@ pub(crate) struct Relay { /// Changes passed on along writers' trees (`swarm::tree`): a sync message asked for /// per child. pub(crate) tree_forwards: u64, + /// Sync messages sent taken as delivered (`SyncSession::assume_delivered`), and their + /// payload's bytes. + pub(crate) delivered: u64, + pub(crate) delivered_bytes: u64, + /// Sync messages received that carried changes, all of which we had already, and + /// their payload's bytes: changes that came twice. + pub(crate) nothing_new: u64, + pub(crate) nothing_new_bytes: u64, } /// When the initial sync finished, and what it took. @@ -204,6 +212,20 @@ impl NodeStats { self.lock().relay.tree_forwards += 1; } + /// A sync message taken as delivered went out, of `bytes` payload. + pub(crate) fn delivered_sent(&self, bytes: usize) { + let mut inner = self.lock(); + inner.relay.delivered += 1; + inner.relay.delivered_bytes += bytes as u64; + } + + /// A sync message of `bytes` payload carried changes, all of which we had already. + pub(crate) fn nothing_new(&self, bytes: usize) { + let mut inner = self.lock(); + inner.relay.nothing_new += 1; + inner.relay.nothing_new_bytes += bytes as u64; + } + /// A sync message of `bytes` arrived. pub(crate) fn sync_received(&self, bytes: usize) { let mut inner = self.lock(); diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 5be0754..700764f 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -139,6 +139,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let failed = peer_state_mutex.lock().await.failed_whole(Instant::now()); let delta_threshold = state.blobs.delta_threshold_for(rate, failed); + let stats = Arc::clone(&state.stats); // The session the message comes from, for its send's outcome. let (session, generated) = tokio::task::spawn_blocking( move || -> (u64, Result<Option<transport::EncodedMessage>>) { @@ -180,6 +181,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let delivered = level && matches!(body, Some(transport::SyncBody::Message(_))); if delivered { peer_state.sync.assume_delivered(replica.heads()); + if let Some(transport::SyncBody::Message(payload)) = &body { + stats.delivered_sent(payload.0.len()); + } } ( peer_state.sync.id(), @@ -399,6 +403,7 @@ pub(crate) async fn handle_sync( type Applied = (bool, bool, bool, (Vec<String>, bool)); tokio::task::spawn_blocking(move || -> Result<Option<Applied>> { // Before the lock: decoding a large message takes a while. + let payload_len = message.0.len(); let message = crate::document::sync::decode(&message)?; let carried = message.carries_changes(); let mut replica = doc.blocking_write(); @@ -435,6 +440,9 @@ pub(crate) async fn handle_sync( } stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; + if carried && !changed { + stats.nothing_new(payload_len); + } // Whose they are, to pass them on along their writers' trees, however they // came (along a tree, with an answer, pulled). let authors = if changed { |