chunk fetch counters

3 files changed, 47 insertions, 2 deletions

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();