asks by asker

3 files changed, 77 insertions, 2 deletions

diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 08aa365..8368f7d 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -12,6 +12,7 @@
 //! long: not taken off, since writes make keepalives unneeded (a pair that exchanges
 //! anything needs none), so the quiet phase's count isn't part of a workload's.
 
+use std::collections::HashMap;
 use std::fmt::Write;
 use std::time::Duration;
 
@@ -300,6 +301,9 @@ struct Totals {
     /// Per node, by name: its arrivals, their delays summed, and those later than
     /// `LATE_MS`.
     by_node: Vec<(String, u64, u64, u64)>,
+    /// Per node, by name: the changes it asked for that the asked had counted as its
+    /// (`PeerState::asked_known`), summed over those it asked.
+    asked_by: HashMap<String, u64>,
 }
 
 /// Arrivals later than this are late (`report`): the propagation buckets above it.
@@ -314,7 +318,15 @@ impl Totals {
             incidents: Incidents::default(),
             propagation: Propagation::default(),
             by_node: Vec::new(),
+            asked_by: HashMap::new(),
         };
+        let names: HashMap<String, String> = nodes
+            .iter()
+            .map(|node| {
+                let id = node.local_id().to_owned();
+                (id.clone(), node.alias.clone().unwrap_or(id))
+            })
+            .collect();
         for node in nodes {
             let sent = node.traffic.total().sent;
             for (total, theirs) in counters_mut(&mut totals.sent)
@@ -353,8 +365,11 @@ impl Totals {
             relay.waiting += stats.relay.waiting;
             relay.off_tree += stats.relay.off_tree;
             relay.off_tree_ms += stats.relay.off_tree_ms;
-            for (_, peer_state) in node.peers_snapshot().await {
-                totals.incidents.add(&peer_state.lock().await.incidents);
+            for (peer, peer_state) in node.peers_snapshot().await {
+                let peer_state = peer_state.lock().await;
+                totals.incidents.add(&peer_state.incidents);
+                let name = names.get(&peer).cloned().unwrap_or(peer);
+                *totals.asked_by.entry(name).or_default() += peer_state.asked_known;
             }
             let propagation = &mut totals.propagation;
             for (sum, count) in propagation
@@ -860,6 +875,60 @@ fn report(
             if late.len() > 5 { ", …" } else { "" }
         );
     }
+    // Who asked for what it was counted as having: the most first, with how many of the
+    // writes reached it late; and the share of the nodes most writes reached late.
+    let late_of: HashMap<&str, u64> = late.iter().map(|(name, late, _)| (*name, *late)).collect();
+    let mut askers: Vec<(&str, u64)> = during
+        .1
+        .asked_by
+        .iter()
+        .map(|(name, asked)| {
+            let before = during.0.asked_by.get(name).copied().unwrap_or(0);
+            (name.as_str(), asked - before)
+        })
+        .filter(|(_, asked)| *asked > 0)
+        .collect();
+    askers.sort_by_key(|(name, asked)| (std::cmp::Reverse(*asked), *name));
+    let asked: u64 = askers.iter().map(|(_, asked)| asked).sum();
+    if asked > 0 {
+        let listed: Vec<String> = askers
+            .iter()
+            .take(5)
+            .map(|(name, asked)| {
+                format!(
+                    "{name} {asked} ({} late)",
+                    late_of.get(name).copied().unwrap_or(0)
+                )
+            })
+            .collect();
+        let writes_seen = |name: &str| {
+            during
+                .0
+                .by_node
+                .iter()
+                .zip(&during.1.by_node)
+                .find(|((n, ..), _)| n == name)
+                .map_or(0, |((_, c0, ..), (_, c1, ..))| c1 - c0)
+        };
+        let mostly_late: Vec<&str> = late
+            .iter()
+            .filter(|(name, late, _)| 2 * late > writes_seen(name))
+            .map(|(name, ..)| *name)
+            .collect();
+        let by_mostly_late: u64 = askers
+            .iter()
+            .filter(|(name, _)| mostly_late.contains(name))
+            .map(|(_, asked)| asked)
+            .sum();
+        let _ = writeln!(
+            out,
+            "    asked for, though counted as theirs, by: {}{}; {:.0}% by the {} nodes most writes reached later than {LATE_MS} ms",
+            listed.join(", "),
+            if askers.len() > 5 { ", …" } else { "" },
+            100.0 * by_mostly_late as f64 / asked as f64,
+            mostly_late.len(),
+        );
+    }
 }
 
 /// Values written in `bench_bridge`, unless `BENCH_BRIDGE_WRITES` says otherwise.
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 4923f0f..e2a47ec 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -182,6 +182,10 @@ pub(crate) struct PeerState {
 
     /// What went wrong with this peer so far (counted only, for /peers and /admin/stats).
     pub(crate) incidents: Incidents,
+    /// Changes it asked for that we had counted as its, as another parent in their
+    /// writer's tree brings them (`SyncSession::resend_if_needed`): per asker, for the
+    /// bench (`simtest::bench`).
+    pub(crate) asked_known: u64,
 
     /// The largest sync message to it that didn't make it (its send failed, or it went
     /// unanswered), in bytes, and when: messages about that large go as deltas for a
@@ -307,6 +311,7 @@ impl PeerState {
             last_hello_at: None,
             unreachable_since: None,
             incidents: Incidents::default(),
+            asked_known: 0,
             whole_failed: None,
         }
     }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 62bb27e..594e5de 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -599,6 +599,7 @@ pub(crate) async fn handle_sync(
                 .sync
                 .resend_if_needed(&replica.doc, &peer, Instant::now());
             stats.asked_known_elsewhere(asked);
+            peer_state.asked_known += (asked.0 + asked.1) as u64;
             replica.refresh_heads();
             let heads_after = replica.heads().to_vec();
             if delivered {