measure deflated chunks

4 files changed, 38 insertions, 9 deletions

diff --git a/Cargo.toml b/Cargo.toml
index 888c4b9..e6cd6c6 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -46,6 +46,8 @@ blake3 = "1.8"
 bytes = "1.12"
 clap = "4.6"
 dashmap = "6.2"
+# Deflate (pure Rust), for large sync messages; Automerge uses it already.
+flate2 = "1.1"
 futures = { version = "0.3", optional = true }
 libp2p = { version = "0.57", optional = true, features = ["tokio", "tcp", "quic", "dns", "noise", "yamux", "ed25519", "identify", "ping", "macros", "relay", "dcutr", "upnp", "autonat", "mdns", "kad"] }
 libp2p-stream = { version = "0.5.0-alpha", optional = true }
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index ae22fd1..19603d2 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -892,6 +892,7 @@ async fn bench_bridge() {
     let received = |node: &AppState| node.traffic.total().received;
     let (z0, b0) = (received(&z), b.traffic.total().sent);
     let p0 = z.stats.snapshot().propagation;
+    let served0 = b.stats.snapshot().relay;
     let mut rng = crate::testrand::Rng::new(72);
     let started = Instant::now();
     for i in 0..writes {
@@ -910,6 +911,7 @@ async fn bench_bridge() {
 
     let (z1, b1) = (received(&z), b.traffic.total().sent);
     let p1 = z.stats.snapshot().propagation;
+    let served1 = b.stats.snapshot().relay;
     let diff = |c1: Counter, c0: Counter| (c1.messages - c0.messages, c1.bytes - c0.bytes);
     let mib = |bytes: u64| bytes as f64 / f64::from(1 << 20);
     let total = (writes * BRIDGE_VALUE) as u64;
@@ -936,7 +938,8 @@ async fn bench_bridge() {
          written: {:.1} MiB in {:.1} s; z converged {} ms after the last write\n  \
          b sent (to a and z): SYNC {} messages, {:.1} MiB; BLOB {} messages, {:.1} MiB\n  \
          z received: SYNC {} messages, {:.1} MiB; BLOB {} messages, {:.1} MiB: {:.2} times what was written\n  \
-         z's delays: {} changes, mean {} ms, half within {}, 99% within {}",
+         z's delays: {} changes, mean {} ms, half within {}, 99% within {}\n  \
+         b's blob chunks: {} served, {} of them deflated; {:.1} MiB as {:.1} MiB",
         BRIDGE_VALUE / 1024,
         1000 / BRIDGE_EVERY.as_millis(),
         mib(total),
@@ -955,5 +958,9 @@ async fn bench_bridge() {
         propagation.sum_ms / propagation.count.max(1),
         within(0.5),
         within(0.99),
+        served1.chunks_served - served0.chunks_served,
+        served1.chunks_deflated - served0.chunks_deflated,
+        mib(served1.chunk_bytes - served0.chunk_bytes),
+        mib(served1.chunk_bytes_sent - served0.chunk_bytes_sent),
     );
 }
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 0bd2d37..f06b251 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -339,14 +339,20 @@ async fn serve(
         let chunk = &prepared.bytes[start..end];
         // Deflated where it shrinks (`compress`): changes are stored uncompressed.
         let body = match crate::compress::deflate(chunk) {
-            Some(deflated) => BlobBody::DeflatedChunk {
-                index,
-                data: deflated.into(),
-            },
-            None => BlobBody::Chunk {
-                index,
-                data: bytes::Bytes::copy_from_slice(chunk),
-            },
+            Some(deflated) => {
+                state.stats.chunk_served(chunk.len(), deflated.len());
+                BlobBody::DeflatedChunk {
+                    index,
+                    data: deflated.into(),
+                }
+            }
+            None => {
+                state.stats.chunk_served(chunk.len(), chunk.len());
+                BlobBody::Chunk {
+                    index,
+                    data: bytes::Bytes::copy_from_slice(chunk),
+                }
+            }
         };
         reply(state, peer, id, body).await;
     }
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index 6332901..31b6a3b 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -137,6 +137,11 @@ pub(crate) struct Relay {
     pub(crate) deflated: u64,
     pub(crate) deflated_from: u64,
     pub(crate) deflated_to: u64,
+    /// Blob chunks served, those of them deflated, and their bytes before and as sent.
+    pub(crate) chunks_served: u64,
+    pub(crate) chunks_deflated: u64,
+    pub(crate) chunk_bytes: u64,
+    pub(crate) chunk_bytes_sent: u64,
     /// Sync messages received that carried changes, all of which we had already, and
     /// their payload's bytes: changes that came twice. Of those, how many came along a
     /// writer's tree (`SyncEnvelope::writer`), and how many were taken as delivered by
@@ -258,6 +263,15 @@ impl NodeStats {
         self.lock().relay.tree_forwards += 1;
     }
 
+    /// A blob chunk of `bytes` was served as `sent` bytes (fewer if deflated).
+    pub(crate) fn chunk_served(&self, bytes: usize, sent: usize) {
+        let mut inner = self.lock();
+        inner.relay.chunks_served += 1;
+        inner.relay.chunks_deflated += u64::from(sent < bytes);
+        inner.relay.chunk_bytes += bytes as u64;
+        inner.relay.chunk_bytes_sent += sent as u64;
+    }
+
     /// A sync message of `from` bytes went out deflated to `to`.
     pub(crate) fn deflated(&self, from: usize, to: usize) {
         let mut inner = self.lock();