benchmark delivered and twice

3 files changed, 58 insertions, 0 deletions

diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 2640381..6bf3bd6 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -229,6 +229,10 @@ impl Totals {
             relay.pulls += stats.relay.pulls;
             relay.pulls_queued += stats.relay.pulls_queued;
             relay.tree_forwards += stats.relay.tree_forwards;
+            relay.delivered += stats.relay.delivered;
+            relay.delivered_bytes += stats.relay.delivered_bytes;
+            relay.nothing_new += stats.relay.nothing_new;
+            relay.nothing_new_bytes += stats.relay.nothing_new_bytes;
             for (_, peer_state) in node.peers_snapshot().await {
                 totals.incidents.add(&peer_state.lock().await.incidents);
             }
@@ -475,6 +479,30 @@ fn report(
         (r1.pulls_queued - r0.pulls_queued) as f64 / writes,
         (r1.tree_forwards - r0.tree_forwards) as f64 / writes,
     );
+    let per = |count: u64, bytes: u64| {
+        let count = count as f64;
+        (
+            count / writes,
+            if count > 0.0 {
+                bytes as f64 / count
+            } else {
+                0.0
+            },
+        )
+    };
+    let delivered = per(
+        r1.delivered - r0.delivered,
+        r1.delivered_bytes - r0.delivered_bytes,
+    );
+    let nothing_new = per(
+        r1.nothing_new - r0.nothing_new,
+        r1.nothing_new_bytes - r0.nothing_new_bytes,
+    );
+    let _ = writeln!(
+        out,
+        "    sync per write: {:.1} taken as delivered ({:.0} bytes each), {:.1} with only changes the receiver had ({:.0} bytes each)",
+        delivered.0, delivered.1, nothing_new.0, nothing_new.1,
+    );
     let (i0, i1) = (during.0.incidents, during.1.incidents);
     let _ = writeln!(
         out,
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 {