diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 1 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 51 |
2 files changed, 37 insertions, 15 deletions
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 196b8b6..c426f99 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -223,6 +223,7 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport let became_ready = remote.ready && peer_state.reports_ready != Some(true); peer_state.reports_ready = Some(remote.ready); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); + peer_state.last_heads = Some((remote.epoch, remote.heads.clone())); drop(peer_state); if left { info!(peer = %peer, "Peer said farewell; dormant until it is heard from again"); diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index a4a4034..26719cd 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -13,7 +13,9 @@ //! 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 STATUS per neighbour (plus the pulls and their replies), whatever the -//! write rate; the delay is the price per hop. Without coalescing, every applied sync +//! write rate; the delay is the price per hop. A neighbour that announced the same heads +//! to us first is skipped, so between two neighbours with the same news usually only one +//! STATUS goes. 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). //! @@ -39,9 +41,13 @@ 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 the transport. +/// How long an announcement waits for further changes to ride along, and a random part +/// on top (`ANNOUNCE_JITTER`), so that of two neighbours with the same news one usually +/// announces first and the other, seeing the first's heads are its own, skips it +/// (`unreached`). Added once per hop to propagation; well below a round trip over the +/// transport. const ANNOUNCE_DELAY: Duration = Duration::from_millis(200); +const ANNOUNCE_JITTER: Duration = Duration::from_millis(200); /// After changes from `from` were applied and our heads changed: tell our other /// neighbours, in the next announcement round. @@ -67,7 +73,11 @@ pub(crate) async fn announce(state: &AppState, from: &str) { } let state = state.clone(); tokio::spawn(async move { - tokio::time::sleep(ANNOUNCE_DELAY).await; + let jitter = crate::transport::paths::random_nonce() + % u64::try_from(ANNOUNCE_JITTER.as_millis()) + .unwrap_or(1) + .max(1); + tokio::time::sleep(ANNOUNCE_DELAY + Duration::from_millis(jitter)).await; let senders = state .announce_pending .lock_or_recover() @@ -95,22 +105,33 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) { /// The neighbours to tell of the changes `senders` sent us. The peer whose changes we /// announce has them already, so with a single sender it is skipped; with several, each -/// lacks the others' changes. +/// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an +/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have. async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> { let skip = match senders.len() { 1 => senders.iter().next().cloned(), _ => None, }; - state - .peer_states - .read() - .await - .keys() - .filter(|peer| { - skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer) - }) - .cloned() - .collect() + let ours = { + let stamp = state.stamp.borrow(); + (stamp.epoch, HeadsSummary::of(stamp.heads.to_vec())) + }; + let mut unreached = Vec::new(); + for (peer, peer_state) in state.peers_snapshot().await { + if skip.as_deref() == Some(peer.as_str()) || !topology::is_neighbour(state, &peer) { + continue; + } + let has_ours = peer_state + .lock() + .await + .last_heads + .as_ref() + .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1); + if !has_ours { + unreached.push(peer); + } + } + unreached } /// Pulls in progress: one neighbour at a time, so each change arrives about once. The |