diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/node.rs | 1 | ||||
| -rw-r--r-- | src/state/mod.rs | 2 | ||||
| -rw-r--r-- | src/state/peer.rs | 3 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 5 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 86 |
5 files changed, 92 insertions, 5 deletions
diff --git a/src/node.rs b/src/node.rs index b376baa..5bfce31 100644 --- a/src/node.rs +++ b/src/node.rs @@ -142,6 +142,7 @@ impl Node { pulling: Default::default(), announce_pending: Default::default(), announce_slow: Default::default(), + slow_rounds: Default::default(), reach: Default::default(), trees: Default::default(), rtt: Default::default(), diff --git a/src/state/mod.rs b/src/state/mod.rs index 53d8178..e127877 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -154,6 +154,8 @@ pub(crate) struct AppState { /// The slow round after changes passed on along writers' trees is scheduled /// (`swarm::relay`). pub(crate) announce_slow: Arc<std::sync::Mutex<Option<crate::swarm::relay::Pending>>>, + /// How long the next slow round waits. + pub(crate) slow_rounds: Arc<std::sync::Mutex<crate::swarm::relay::SlowRounds>>, /// Distances in writers' trees (`swarm::tree`). pub(crate) trees: Arc<std::sync::Mutex<crate::swarm::tree::Trees>>, /// Whom we reach, as told our neighbours (`swarm::reach`). diff --git a/src/state/peer.rs b/src/state/peer.rs index 32b35fc..8f79886 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -153,6 +153,8 @@ pub(crate) struct PeerState { /// with its epoch: once they are ours, it has every change we would announce to it /// (`swarm::relay`). pub(crate) last_heads: Option<(transport::Epoch, transport::HeadsSummary)>, + /// When we last told it our heads in a slow round (`swarm::relay::SlowRounds`). + pub(crate) slow_announced_at: Option<Instant>, /// Whom it said it reaches (`swarm::reach`), by pair key, and when. pub(crate) reaches: Option<(Instant, Vec<u64>)>, /// Whose change the next sync message to it passes on along the writer's tree @@ -286,6 +288,7 @@ impl PeerState { syncs_with_us: None, reported_heads: None, last_heads: None, + slow_announced_at: None, reaches: None, tree_writer: None, left: false, diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index cfafbd3..53f1aa3 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -190,6 +190,11 @@ pub(crate) async fn handle_message( } } transport::ParsedMessage::StatusReply(remote) => { + // An answer to our slow round's announcement: a tree missed. + let announced = peer_state.lock().await.slow_announced_at.take(); + if let Some(at) = announced { + crate::swarm::relay::slow_round_answered(state, at.elapsed()); + } on_status(state, &peer, &remote, reactivated).await; } transport::ParsedMessage::Links(request) => { diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 98d4ca0..58196d3 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -26,9 +26,9 @@ //! go at once to the receiver's children in the writer's tree (`swarm::tree`). Then the //! tree carries them, and the round after `ANNOUNCE_DELAY` goes only to the neighbours //! the tree leaves out (`Plan::uncovered`: no member they reach comes before them). The -//! others are told in a slow round (`SLOW_ANNOUNCE_DELAY`): in case their parent missed -//! the change (it went away, or the nodes' views of the tree differed), coalescing all -//! that came meanwhile. Changes that don't say whose (pulled, a delta, from a node that +//! others are told in a slow round (`SLOW_ANNOUNCE_DELAY`, longer while nobody lacks +//! anything when told: `SlowRounds`): in case their parent missed the change (it went +//! away, or the nodes' views of the tree differed), coalescing all that came meanwhile. Changes that don't say whose (pulled, a delta, from a node that //! has no part in the writer's tree) are announced to all, as before. //! //! In a full mesh too: the member list makes every member the writer's neighbour, but @@ -69,8 +69,63 @@ const ANNOUNCE_JITTER: Duration = Duration::from_millis(200); /// How long the slow round after changes passed on along writers' trees waits: what a /// tree misses arrives this much later, and while changes keep coming, every neighbour -/// is told at most this often. +/// is told at most this often. Doubles with every slow round that nobody lacked +/// anything of (`SlowRounds`), up to `SLOW_ANNOUNCE_MAX`. const SLOW_ANNOUNCE_DELAY: Duration = Duration::from_secs(2); +const SLOW_ANNOUNCE_MAX: Duration = Duration::from_secs(16); + +/// A STATUS_REPLY this soon after our slow round's announcement to the peer answers it. +const SLOW_ANSWER_WITHIN: Duration = Duration::from_secs(5); + +/// How long the slow round waits. Since a neighbour answers an announcement only if it +/// lacks something (`swarm::peer::inbound`), an answer to the slow round means a tree +/// missed: then it is back to `SLOW_ANNOUNCE_DELAY`. While nobody answers, the trees +/// work, and the round waits twice as long each time, up to `SLOW_ANNOUNCE_MAX`: it is +/// a safety net, and most of its announcements go to neighbours that have everything. +/// (A lost message along a tree is noticed sooner by the next one, which its receiver +/// can't apply and asks after; the slow round is for the last of a run of writes, a +/// parent gone away, views of the tree that differ.) +pub(crate) struct SlowRounds { + delay: Duration, + /// An answer came since the last slow round was scheduled. + answered: bool, +} + +impl Default for SlowRounds { + fn default() -> Self { + // The first round is the short one. + Self { + delay: SLOW_ANNOUNCE_DELAY, + answered: true, + } + } +} + +impl SlowRounds { + /// The wait for the slow round being scheduled now: as long as the last if it was + /// answered (or the first), else twice that (up to the most). + fn next(&mut self) -> Duration { + self.delay = if std::mem::take(&mut self.answered) { + SLOW_ANNOUNCE_DELAY + } else { + (self.delay * 2).min(SLOW_ANNOUNCE_MAX) + }; + self.delay + } +} + +/// A STATUS_REPLY from `peer`, which we told of our heads in a slow round +/// `announced_ago`: it lacked something, so a tree missed; the next slow round waits +/// `SLOW_ANNOUNCE_DELAY` again. +pub(crate) fn slow_round_answered(state: &AppState, announced_ago: Duration) { + if announced_ago < SLOW_ANSWER_WITHIN { + let mut rounds = state.slow_rounds.lock_or_recover(); + if !rounds.answered { + debug!("A slow round's announcement was answered; next one in 2 s"); + } + rounds.answered = true; + } +} /// An announcement round scheduled: for changes from these peers, and the neighbours /// that were sent them along a writer's tree meanwhile (`swarm::tree`). @@ -139,7 +194,8 @@ pub(crate) async fn announce(state: &AppState, from: &str, writers: &[u64], unau schedule }; if schedule { - round(state, &state.announce_slow, SLOW_ANNOUNCE_DELAY, true); + let delay = state.slow_rounds.lock_or_recover().next(); + round(state, &state.announce_slow, delay, true); } } @@ -177,6 +233,9 @@ async fn announce_now(state: &AppState, pending: &Pending, slow: bool) { } state.stats.announce_round(pending.all, slow); for peer in neighbours { + if slow && let Some(peer_state) = state.get_peer_state(&peer).await { + peer_state.lock().await.slow_announced_at = Some(Instant::now()); + } if let Err(error) = announce_heads(state, &peer, false).await { debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); } @@ -383,3 +442,20 @@ async fn pull_next(state: &AppState) { } } } + +#[cfg(test)] +mod tests { + use super::*; + + /// The slow round waits twice as long after each one nobody answered, up to the + /// most, and the short time again after an answer. + #[test] + fn slow_rounds_back_off_while_unanswered() { + let mut rounds = SlowRounds::default(); + let waits: Vec<u64> = (0..6).map(|_| rounds.next().as_secs()).collect(); + assert_eq!(waits, [2, 4, 8, 16, 16, 16]); + rounds.answered = true; + assert_eq!(rounds.next(), SLOW_ANNOUNCE_DELAY); + assert_eq!(rounds.next(), SLOW_ANNOUNCE_DELAY * 2); + } +} |