diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/simtest/bench.rs | 28 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 22 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 8 |
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 { |