From ab8f5447680da56ce658abbc33c222219c3204ef Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 16:47:32 +0100 Subject: plumtree --- src/swarm/plumtree.rs | 324 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 324 insertions(+) create mode 100644 src/swarm/plumtree.rs (limited to 'src/swarm/plumtree.rs') diff --git a/src/swarm/plumtree.rs b/src/swarm/plumtree.rs new file mode 100644 index 0000000..9560b9f --- /dev/null +++ b/src/swarm/plumtree.rs @@ -0,0 +1,324 @@ +//! PlumTree (Leitão, Pereira and Rodrigues, "Epidemic Broadcast Trees", 2007) as the way +//! changes reach every member (`swarm::dissemination`, `--dissemination plumtree`), over +//! the same neighbours as writers' trees (`swarm::topology`). +//! +//! Each node keeps which of its neighbours it pushes changes to at once (eager) and +//! which it leaves to announcements (lazy), one set for all writers, as in the paper: +//! at first all are eager. A change that comes along a writer's push with nothing new +//! (a duplicate: it came another way first) makes the receiver ask the sender to leave +//! changes to announcements from now on (PRUNE), and both hold the link lazy. A change +//! that comes another way (pulled after an announcement, PlumTree's IHAVE: the eager +//! links didn't bring it) makes the receiver ask the one it came from to push at once +//! (GRAFT), and both hold the link eager. So the eager links settle into a tree made of +//! the links that delivered first: one that follows latency by itself, where writers' +//! trees are computed alike on every node from the member list (`swarm::tree`). One +//! tree for all writers settles over every writer's changes; trees per writer (tried +//! first) never did with writers in turn, each writing too rarely, every new one +//! flooding until pruned. PRUNE and GRAFT name the writer, for such a variant; this one +//! doesn't use it. +//! +//! PlumTree's IHAVE is the announcement (`swarm::relay`): lazy neighbours, a few of them +//! at random (`LAZY_FANOUT`), are told the new heads in the quick round, coalesced, all +//! of them in the slow round, and count as having the changes +//! (`Plan::covered`), as their eager links bring them. One that still lacks them a round +//! trip later pulls (`relay::pull_after_grace`: the graft timer), and the pull, bringing +//! a writer's changes another way, grafts the link. Told only in the slow round instead +//! (tried second), repair took seconds; every lazy link told per round is PlumTree's +//! cost (one STATUS each; a message of its own would only be smaller), so telling all +//! in the quick round tripled the messages (tried third). +//! +//! The paper's optimisation: a lazy neighbour that told us of changes before an eager +//! one pushed them has the faster link (an announcement waits `relay::ANNOUNCE_DELAY`, +//! a push doesn't), so it is grafted, once it was so for a few pushes in a row +//! (`AHEAD_TIMES`); the old parent's pushes then come as duplicates, +//! and are pruned. A link we just grafted isn't pruned for a duplicate while what it +//! pushed meanwhile may still be on its way (`settling`). Else a tree stays as it first +//! formed: a pull beats a push only when +//! the push is slower by an announcement's delay and two round trips. +//! Nothing is written to the document. + +use std::collections::{HashMap, HashSet}; +use std::sync::Mutex; +use std::time::Duration; + +use async_trait::async_trait; +use tokio::time::Instant; +use tracing::debug; + +use crate::document::ChangeHash; +use crate::locks::MutexExt; +use crate::state::AppState; +use crate::swarm::dissemination::{Dissemination, pass_on}; +use crate::swarm::topology; +use crate::swarm::tree::Plan; +use crate::transport::{ParsedMessage, TreeLink}; + +/// Lazy links kept at most; past it, all are forgotten (every link eager again, and +/// pruned again where it delivers twice). More than a node has neighbours. +const MAX_LAZY: usize = 4096; + +/// Lazy neighbours told in the quick round per change passed on, picked at random each +/// time (`some_of`); the others hear in the slow round. Told all (patch 997), every node +/// announced to all its lazy neighbours at every hop: about 240 STATUS per write at 50 +/// nodes, three times the messages of PlumTree told only in the slow round. +const LAZY_FANOUT: usize = 3; + +/// Up to `count` of `peers`, picked at random. +fn some_of(peers: &[String], count: usize) -> Vec { + let mut peers = peers.to_vec(); + let count = count.min(peers.len()); + for picked in 0..count { + let rest = (peers.len() - picked) as u64; + let other = picked + (crate::transport::paths::random_nonce() % rest) as usize; + peers.swap(picked, other); + } + peers.truncate(count); + peers +} + +/// Heads we lacked when told of them, kept at most; past it, all are forgotten (a +/// better link is noticed with the next change). +const MAX_ANNOUNCED: usize = 4096; + +/// Who first told us of a head we lacked is noticed for this long. +const ANNOUNCED_FOR: Duration = Duration::from_secs(10); + +/// Pushes in a row a lazy neighbour must have told of first to be grafted (`ahead`). +/// Grafted at once (patch 1001), trees shifted with every lucky announcement: on Tor, +/// with one writer, 24 PRUNE and GRAFT per write against 6, and changes along the +/// eager links took 987 ms against 471. +const AHEAD_TIMES: u32 = 3; + +/// How long after we grafted a link its duplicates don't prune it, in round trips to +/// the peer (`settling`), and when we know none. +const SETTLE_ROUND_TRIPS: u32 = 2; +const SETTLE_UNKNOWN: Duration = Duration::from_secs(2); + +/// The neighbours whose links are lazy, who first told us of heads we lacked, how many +/// pushes in a row a lazy one told of first, and when we grafted each. +#[derive(Default)] +pub(crate) struct PlumTree { + lazy: Mutex>, + announced: Mutex>, + ahead_times: Mutex>, + /// When we last grafted each link (`settling`). + grafted: Mutex>, +} + +impl PlumTree { + /// Holds the link to `peer` lazy, or eager; whether that changed it. + fn set(&self, peer: &str, lazy: bool) -> bool { + let mut links = self.lazy.lock_or_recover(); + if lazy { + if links.len() >= MAX_LAZY { + links.clear(); + } + links.insert(peer.to_owned()) + } else { + links.remove(peer) + } + } + + /// A lazy neighbour, not `peer`, that told us of any of the changes `peer` just + /// pushed before they came (since `ANNOUNCED_FOR`), for the last `AHEAD_TIMES` + /// pushes in a row: its link is the faster, by at least an announcement's delay + /// (`relay::ANNOUNCE_DELAY`), which a push over it doesn't wait. The changes are + /// forgotten as announced. + fn ahead(&self, peer: &str, authors: &[(String, Vec)]) -> Option { + let now = Instant::now(); + let told: Vec<(String, Instant)> = { + let mut announced = self.announced.lock_or_recover(); + authors + .iter() + .flat_map(|(_, changes)| changes) + .filter_map(|change| announced.remove(change)) + .collect() + }; + let first = { + let lazy = self.lazy.lock_or_recover(); + told.into_iter() + .filter(|(by, at)| { + by != peer && lazy.contains(by) && now.duration_since(*at) < ANNOUNCED_FOR + }) + .min_by_key(|(_, at)| *at) + .map(|(by, _)| by) + }; + let mut times = self.ahead_times.lock_or_recover(); + let Some(first) = first else { + times.clear(); + return None; + }; + times.retain(|by, _| *by == first); + let count = times.entry(first.clone()).or_default(); + *count += 1; + if *count < AHEAD_TIMES { + return None; + } + times.clear(); + Some(first) + } + + /// We grafted the link to `peer` now. + fn grafted(&self, peer: &str) { + let mut grafted = self.grafted.lock_or_recover(); + if grafted.len() >= MAX_LAZY { + grafted.clear(); + } + grafted.insert(peer.to_owned(), Instant::now()); + } + + /// Whether we grafted the link to `peer` so lately (`SETTLE_ROUND_TRIPS` round trips + /// to it) that a duplicate over it doesn't prune it: what it pushed was on its way + /// as we pulled the same from it, or from another. Pruned at once, a node no + /// neighbour pushed to pulled every change, grafted the one it pulled from, and + /// pruned it at its next push, a duplicate of the pull, over and over (patch 1007: + /// on Tor, such a node got every change late). + fn settling(&self, state: &AppState, peer: &str) -> bool { + let Some(at) = self.grafted.lock_or_recover().get(peer).copied() else { + return false; + }; + let now = Instant::now(); + let settle = state + .lock_rtt() + .effective(peer, now, state.timing.rtt_fresh_for()) + .map_or(SETTLE_UNKNOWN, |rtt| rtt * SETTLE_ROUND_TRIPS); + now.duration_since(at) < settle + } + + /// The changes, come another way, are forgotten as announced. + fn forget_announced(&self, authors: &[(String, Vec)]) { + let mut announced = self.announced.lock_or_recover(); + for change in authors.iter().flat_map(|(_, changes)| changes) { + announced.remove(change); + } + } + + /// Tells `peer` to hold the link for `writer` as we now do: a GRAFT (`eager`) or a + /// PRUNE. + async fn tell(state: &AppState, peer: &str, writer: u64, eager: bool) { + let (epoch, epoch_seq) = state.epoch_stamp(); + let link = TreeLink { + epoch, + epoch_seq, + writer, + }; + let message = if eager { + ParsedMessage::Graft(link) + } else { + ParsedMessage::Prune(link) + }; + if let Err(error) = state.transport.send_message(peer, &message).await { + debug!(peer = %peer, error = %format!("{error:#}"), "PRUNE or GRAFT failed"); + } + } +} + +#[async_trait] +impl Dissemination for PlumTree { + fn name(&self) -> &'static str { + "plumtree" + } + + /// Eager neighbours get the writer's changes at once (`Plan::children`); lazy ones + /// count as having them (`Plan::covered`), `LAZY_FANOUT` of them, picked at random, + /// told in the quick round (`Plan::uncovered`: the IHAVE), all in the slow round. + async fn plan(&self, state: &AppState, writer: u64) -> Option { + let _ = writer; + let neighbours = topology::our_neighbours(state); + let lazy = self.lazy.lock_or_recover().clone(); + let (lazy, eager): (Vec, Vec) = + neighbours.into_iter().partition(|peer| lazy.contains(peer)); + Some(Plan { + children: eager, + alone: Vec::new(), + uncovered: some_of(&lazy, LAZY_FANOUT), + covered: lazy, + }) + } + + async fn forward( + &self, + state: &AppState, + writer: u64, + from: &str, + changes: &[ChangeHash], + ) -> Option { + let plan = self.plan(state, writer).await?; + Some(pass_on(state, writer, from, changes, plan).await) + } + + /// A duplicate along a writer's push prunes the link; changes of a writer that + /// came another way graft the link to whoever brought them. Told every time, not + /// only when our side of the link changed: a duplicate pushed over a link we hold + /// lazy, or changes pulled over one we hold eager, say the other end holds it the + /// other way (a PRUNE or GRAFT lost, or two crossing), and told only on a change, + /// it stayed so: a node every neighbour held lazy while it held them eager pulled + /// every change (patch 999: one such on Tor, a few links eager at one end). Counted as the trees' + /// counters count (`/admin/stats`): a duplicate as a message from another than the + /// parent, changes another way as ones the trees didn't bring. + async fn note_arrival( + &self, + state: &AppState, + peer: &str, + writer: Option, + carried: bool, + authors: &[(String, Vec)], + ) { + match writer { + Some(writer) if carried && authors.is_empty() => { + state.stats.tree_from_other(); + if self.settling(state, peer) { + return; + } + self.set(peer, true); + Self::tell(state, peer, writer, false).await; + } + Some(writer) if !authors.is_empty() => { + self.set(peer, false); + if let Some(ahead) = self.ahead(peer, authors) { + debug!(peer = %ahead, "A lazy neighbour told of changes before they were pushed, every time lately; grafting it"); + self.set(&ahead, false); + self.grafted(&ahead); + Self::tell(state, &ahead, writer, true).await; + } + } + None => { + self.forget_announced(authors); + for (author, changes) in authors { + if author == state.local_id() { + continue; + } + state.stats.tree_missed(changes.len()); + self.set(peer, false); + self.grafted(peer); + Self::tell(state, peer, topology::pair_key(author), true).await; + } + } + Some(_) => {} + } + } + + async fn linked(&self, state: &AppState, peer: &str, writer: u64, eager: bool) { + let _ = (state, writer); + self.set(peer, !eager); + } + + /// The first to tell us of each head is the one noticed. + fn announced(&self, peer: &str, heads: &[ChangeHash]) { + let mut announced = self.announced.lock_or_recover(); + if announced.len() + heads.len() > MAX_ANNOUNCED { + announced.clear(); + } + let now = Instant::now(); + for head in heads { + announced + .entry(*head) + .or_insert_with(|| (peer.to_owned(), now)); + } + } + + #[cfg(test)] + fn lazy(&self) -> Option> { + Some(self.lazy.lock_or_recover().clone()) + } +} -- cgit v1.3.1