From 9aad4bee0fbcac76e96b3ff4e6c57a28b162b861 Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 5 Nov 2026 20:07:47 +0100 Subject: chunk fetch counters --- src/simtest/bench.rs | 9 ++++++++- src/swarm/blob.rs | 13 ++++++++++++- src/swarm/stats.rs | 27 +++++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 2 deletions(-) (limited to 'src') diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs index 1498317..4d34e4b 100644 --- a/src/simtest/bench.rs +++ b/src/simtest/bench.rs @@ -931,6 +931,7 @@ async fn bench_bridge() { let (z0, b0) = (received(&z), b.traffic.total().sent); let p0 = z.stats.snapshot().propagation; let served0 = b.stats.snapshot().relay; + let fetched0 = z.stats.snapshot().relay; let mut rng = crate::testrand::Rng::new(72); let started = Instant::now(); for i in 0..writes { @@ -950,6 +951,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 fetched1 = z.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; @@ -977,7 +979,8 @@ async fn bench_bridge() { 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 {}\n \ - b's blob chunks: {} served, {} of them deflated; {:.1} MiB as {:.1} MiB", + b's blob chunks: {} served, {} of them deflated; {:.1} MiB as {:.1} MiB\n \ + z's chunks: {} timed out, {} asked of a second source, {} came twice, {} from a source dropped", BRIDGE_VALUE / 1024, 1000 / BRIDGE_EVERY.as_millis(), mib(total), @@ -1000,5 +1003,9 @@ async fn bench_bridge() { served1.chunks_deflated - served0.chunks_deflated, mib(served1.chunk_bytes - served0.chunk_bytes), mib(served1.chunk_bytes_sent - served0.chunk_bytes_sent), + fetched1.chunks_timed_out - fetched0.chunks_timed_out, + fetched1.chunks_doubled - fetched0.chunks_doubled, + fetched1.chunks_twice - fetched0.chunks_twice, + fetched1.chunks_from_dropped - fetched0.chunks_from_dropped, ); } diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 82586b8..f3494af 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -642,6 +642,7 @@ pub(crate) async fn fetch( doubled.insert(*index); sources[at].in_flight.insert(*index, now); } + state.stats.chunks_doubled(batch.len()); request(state, &sources[at].peer, id, batch, delta).await; } } @@ -665,7 +666,11 @@ pub(crate) async fn fetch( let at = sources.iter().position(|source| source.peer == from); let sent_at = at.and_then(|at| sources[at].in_flight.remove(&index)); let slot = index as usize; - if slot >= total || chunks[slot].is_some() { + if slot >= total { + continue; + } + if chunks[slot].is_some() { + state.stats.chunk_twice(); continue; } let expected = expected_len(&manifest, slot); @@ -687,6 +692,9 @@ pub(crate) async fn fetch( // A peer that delivers may have one more in flight. source.window = (source.window + 1).min(MAX_WINDOW); } + if at.is_none() { + state.stats.chunk_from_dropped(); + } chunks[slot] = Some(data); done += 1; last_progress = Instant::now(); @@ -724,6 +732,9 @@ pub(crate) async fn fetch( .in_flight .retain(|_, sent| now.saturating_duration_since(*sent) < timeout); if source.in_flight.len() < before { + state + .stats + .chunks_timed_out(before - source.in_flight.len()); source.window = 1; debug!(peer = %source.peer, blob = %id, "Blob chunks timed out; asking others"); } diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index 218d5e8..5cac16a 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -141,6 +141,13 @@ pub(crate) struct Relay { pub(crate) chunks_deflated: u64, pub(crate) chunk_bytes: u64, pub(crate) chunk_bytes_sent: u64, + /// Blob chunks fetched (`blob::fetch`): asked of another source as late, asked of a + /// second source at the end, come when held already, and come from a peer dropped + /// as a source meanwhile (kept, its hash being good). + pub(crate) chunks_timed_out: u64, + pub(crate) chunks_doubled: u64, + pub(crate) chunks_twice: u64, + pub(crate) chunks_from_dropped: 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 @@ -277,6 +284,26 @@ impl NodeStats { inner.relay.chunk_bytes_sent += sent as u64; } + /// `count` blob chunks we fetch went to another source, as late. + pub(crate) fn chunks_timed_out(&self, count: usize) { + self.lock().relay.chunks_timed_out += count as u64; + } + + /// `count` blob chunks we fetch were asked of a second source, at the end. + pub(crate) fn chunks_doubled(&self, count: usize) { + self.lock().relay.chunks_doubled += count as u64; + } + + /// A blob chunk we fetch came when we held it already. + pub(crate) fn chunk_twice(&self) { + self.lock().relay.chunks_twice += 1; + } + + /// A blob chunk we fetch came from a peer no longer a source. + pub(crate) fn chunk_from_dropped(&self) { + self.lock().relay.chunks_from_dropped += 1; + } + /// 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(); -- cgit v1.3.1