coalesce announcements

3 files changed, 60 insertions, 2 deletions

diff --git a/src/main.rs b/src/main.rs
index 98abf7a..217a9a4 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -144,6 +144,7 @@ async fn main() -> Result<()> {
         config: Arc::new(config.report()),
         neighbours: Default::default(),
         pulling: Default::default(),
+        announce_pending: Default::default(),
         rtt: Default::default(),
         throughput: Default::default(),
         stability: Default::default(),
diff --git a/src/state.rs b/src/state.rs
index 0dd3a81..9c5552f 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -613,6 +613,9 @@ pub(crate) struct AppState {
     pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>,
     /// The neighbour we are pulling changes from, and since when (`swarm::relay`).
     pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>,
+    /// An announcement round is scheduled, for changes from these peers
+    /// (`swarm::relay`).
+    pub(crate) announce_pending: Arc<std::sync::Mutex<Option<std::collections::HashSet<String>>>>,
     /// Our neighbours, once known (`swarm::topology`).
     pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,
 
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 492b59d..703bc3f 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -9,8 +9,16 @@
 //! receives a change about once; the rest is announcements. Regular keepalives work the
 //! same way, so any gap is noticed within a keepalive interval.
 //!
+//! Announcements are coalesced: the first applied change schedules one announcement
+//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
+//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the
+//! write rate; the delay is the price per hop. Without coalescing, every applied sync
+//! message would announce to every neighbour, and announcements would outnumber the
+//! data they announce (about 15 to 1 at 250 nodes).
+//!
 //! In a full mesh the writer reaches everyone itself, so nothing is announced.
 
+use std::collections::HashSet;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 
@@ -26,18 +34,64 @@ use crate::transport::{self, HeadsSummary};
 /// A pull that brought nothing for this long no longer blocks pulling from another peer.
 const PULL_TIMEOUT: Duration = Duration::from_secs(15);
 
+/// How long an announcement waits for further changes to ride along. Added once per
+/// hop to propagation; well below a round trip over Mycelium.
+const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
+
 /// After changes from `from` were applied and our heads changed: tell our other
-/// neighbours.
+/// neighbours, in the next announcement round.
 pub(crate) async fn announce(state: &AppState, from: &str) {
     if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
         return;
     }
+    let schedule = {
+        let mut pending = state
+            .announce_pending
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        match pending.as_mut() {
+            Some(senders) => {
+                senders.insert(from.to_owned());
+                false
+            }
+            None => {
+                *pending = Some(HashSet::from([from.to_owned()]));
+                true
+            }
+        }
+    };
+    if !schedule {
+        return;
+    }
+    let state = state.clone();
+    tokio::spawn(async move {
+        tokio::time::sleep(ANNOUNCE_DELAY).await;
+        let senders = state
+            .announce_pending
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .take()
+            .unwrap_or_default();
+        announce_now(&state, &senders).await;
+    });
+}
+
+/// One announcement round. The peer whose changes we announce has them already, so a
+/// round with a single sender skips it; with several, each lacks the others' changes,
+/// so every neighbour is told.
+async fn announce_now(state: &AppState, senders: &HashSet<String>) {
+    let skip = match senders.len() {
+        1 => senders.iter().next().cloned(),
+        _ => None,
+    };
     let neighbours: Vec<String> = state
         .peer_states
         .read()
         .await
         .keys()
-        .filter(|peer| peer.as_str() != from && topology::is_neighbour(state, peer))
+        .filter(|peer| {
+            skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer)
+        })
         .cloned()
         .collect();
     for peer in neighbours {