From a8762b78d327e97c3c818842e5a948ba118fefd8 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 4 Nov 2026 21:07:42 +0100 Subject: benchmark what sync bytes are --- src/document/sync.rs | 12 ++++++++++++ src/simtest/bench.rs | 32 ++++++++++++++++++++++++++++++++ src/swarm/stats.rs | 27 +++++++++++++++++++++++++-- src/swarm/sync.rs | 6 +++++- 4 files changed, 74 insertions(+), 3 deletions(-) (limited to 'src') diff --git a/src/document/sync.rs b/src/document/sync.rs index f45a316..29efb5d 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -27,6 +27,18 @@ impl Received { pub(crate) fn carries_changes(&self) -> bool { !self.0.changes.is_empty() } + + /// The bytes of the changes it carries, and of its bloom filters (`have`). + pub(crate) fn sizes(&self) -> (usize, usize) { + let changes = self.0.changes.iter().map(<[u8]>::len).sum(); + let blooms = self + .0 + .have + .iter() + .map(|have| have.bloom.to_bytes().len()) + .sum(); + (changes, blooms) + } } /// `payload` decoded. Before taking the document's lock: decoding a large message takes diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs index ecf44a5..290fd7c 100644 --- a/src/simtest/bench.rs +++ b/src/simtest/bench.rs @@ -235,6 +235,11 @@ impl Totals { relay.nothing_new_bytes += stats.relay.nothing_new_bytes; relay.nothing_new_along_trees += stats.relay.nothing_new_along_trees; relay.nothing_new_delivered += stats.relay.nothing_new_delivered; + relay.carrying += stats.relay.carrying; + relay.carrying_change_bytes += stats.relay.carrying_change_bytes; + relay.carrying_bloom_bytes += stats.relay.carrying_bloom_bytes; + relay.nothing_new_change_bytes += stats.relay.nothing_new_change_bytes; + relay.nothing_new_bloom_bytes += stats.relay.nothing_new_bloom_bytes; for (_, peer_state) in node.peers_snapshot().await { totals.incidents.add(&peer_state.lock().await.incidents); } @@ -510,6 +515,33 @@ fn report( (r1.nothing_new_along_trees - r0.nothing_new_along_trees) as f64 / writes, (r1.nothing_new_delivered - r0.nothing_new_delivered) as f64 / writes, ); + let each = |bytes: u64, count: u64| { + if count > 0 { + bytes as f64 / count as f64 + } else { + 0.0 + } + }; + let carrying = r1.carrying - r0.carrying; + let twice = r1.nothing_new - r0.nothing_new; + let _ = writeln!( + out, + " received with changes: {:.1} per write, each {:.0} bytes of changes and {:.0} of bloom filters; of those with nothing new, {:.0} and {:.0}", + carrying as f64 / writes, + each( + r1.carrying_change_bytes - r0.carrying_change_bytes, + carrying + ), + each(r1.carrying_bloom_bytes - r0.carrying_bloom_bytes, carrying), + each( + r1.nothing_new_change_bytes - r0.nothing_new_change_bytes, + twice + ), + each( + r1.nothing_new_bloom_bytes - r0.nothing_new_bloom_bytes, + twice + ), + ); 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 34542ff..905e668 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -122,6 +122,13 @@ pub(crate) struct Relay { pub(crate) nothing_new_bytes: u64, pub(crate) nothing_new_along_trees: u64, pub(crate) nothing_new_delivered: u64, + /// Sync messages received that carried changes, the bytes of those changes, and of + /// their bloom filters; and the same of those with nothing new. + pub(crate) carrying: u64, + pub(crate) carrying_change_bytes: u64, + pub(crate) carrying_bloom_bytes: u64, + pub(crate) nothing_new_change_bytes: u64, + pub(crate) nothing_new_bloom_bytes: u64, } /// When the initial sync finished, and what it took. @@ -224,13 +231,29 @@ impl NodeStats { } /// A sync message of `bytes` payload carried changes, all of which we had already; - /// it came `along_tree`, or was `delivered`. - pub(crate) fn nothing_new(&self, bytes: usize, along_tree: bool, delivered: bool) { + /// it came `along_tree`, or was `delivered`; `sizes`: of its changes and blooms. + pub(crate) fn nothing_new( + &self, + bytes: usize, + along_tree: bool, + delivered: bool, + sizes: (usize, usize), + ) { let mut inner = self.lock(); inner.relay.nothing_new += 1; inner.relay.nothing_new_bytes += bytes as u64; inner.relay.nothing_new_along_trees += u64::from(along_tree); inner.relay.nothing_new_delivered += u64::from(delivered); + inner.relay.nothing_new_change_bytes += sizes.0 as u64; + inner.relay.nothing_new_bloom_bytes += sizes.1 as u64; + } + + /// A sync message carrying changes arrived; `sizes`: of its changes and blooms. + pub(crate) fn carrying(&self, sizes: (usize, usize)) { + let mut inner = self.lock(); + inner.relay.carrying += 1; + inner.relay.carrying_change_bytes += sizes.0 as u64; + inner.relay.carrying_bloom_bytes += sizes.1 as u64; } /// A sync message of `bytes` arrived. diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 07c990c..04d06de 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -406,6 +406,10 @@ pub(crate) async fn handle_sync( let payload_len = message.0.len(); let message = crate::document::sync::decode(&message)?; let carried = message.carries_changes(); + let sizes = if carried { message.sizes() } else { (0, 0) }; + if carried { + stats.carrying(sizes); + } let mut replica = doc.blocking_write(); // Authoritative check, under the same lock as the apply: the document may @@ -447,7 +451,7 @@ pub(crate) async fn handle_sync( peer_state.sync.came_from_peer(&replica.doc, &heads_before); } if carried && !changed { - stats.nothing_new(payload_len, writer.is_some(), delivered); + stats.nothing_new(payload_len, writer.is_some(), delivered, sizes); } // Whose they are, to pass them on along their writers' trees, however they // came (along a tree, with an answer, pulled). -- cgit v1.3.1