benchmark delivered and twice

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 {