diff options
| -rw-r--r-- | docs/architecture.md | 3 | ||||
| -rw-r--r-- | docs/peers.md | 12 | ||||
| -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 |
7 files changed, 102 insertions, 10 deletions
diff --git a/docs/architecture.md b/docs/architecture.md index 2ad7638..a63e36c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -229,7 +229,8 @@ the write lock, stamped with the clock) → `publish_change` (bumps the version, neighbour applies it and sends it at once to its children in the writer's tree (the writer known from the changes' author), which do the same; the members the trees miss are told of the new heads by announcement and pull what they lack, at once where no -tree reaches them, else in the slow round 2 s later. +tree reaches them, else in the slow round 2 s later (up to 16 s while nobody it tells +lacks anything). **An inbound message.** Consumer → `handle_message`: drop our own and stale-run replies, admit (swarm key), mark the peer seen, record its epoch (ignoring late ones), diff --git a/docs/peers.md b/docs/peers.md index 051db4e..d22b7c5 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -342,10 +342,14 @@ be lost, the child's other neighbours tell it in their slow rounds (below), and pulls; a child that reaches no other member is told in its parent's. The announcement round of a node that passed changes on goes only to the neighbours the tree leaves out (no member they reach comes before them, or the member list's graph doesn't have -them); all others are told in a slow round 2 s later, which -catches what a tree missed (a parent that went away, nodes whose views of the tree -differed) and covers everything that came meanwhile: while changes keep coming, each -neighbour gets an announcement about every 2 s, not with every change. +them); all others are told in a slow round 2 s later, which catches what a tree missed +(a parent that went away, nodes whose views of the tree differed, the last of a run of +messages along a tree lost) and covers everything that came meanwhile: while changes +keep coming, each neighbour gets an announcement about every 2 s, not with every +change. As a neighbour answers an announcement only when it lacks something, an +answered slow round means a tree missed; while none is answered, the trees work, and +each slow round waits twice as long as the one before, up to 16 s, and 2 s again after +the first answer. A member without a parent it reaches, or whose parent missed the change, gets it by announcement and pull as before, then passes it on to its own children. With every link working, each member gets each change once (`bench_writer_trees`: 1999 messages per 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); + } +} |