aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/stats/prometheus.rs1
-rw-r--r--src/config/mod.rs9
-rw-r--r--src/main.rs1
-rw-r--r--src/node.rs4
-rw-r--r--src/simtest/bench.rs75
-rw-r--r--src/simtest/mod.rs5
-rw-r--r--src/swarm/dissemination.rs100
-rw-r--r--src/swarm/mod.rs1
-rw-r--r--src/swarm/peer/inbound.rs12
-rw-r--r--src/swarm/plumtree.rs324
-rw-r--r--src/swarm/relay.rs21
-rw-r--r--src/swarm/tree.rs46
-rw-r--r--src/transport/mod.rs23
-rw-r--r--src/transport/traffic.rs7
-rw-r--r--src/transport/wire.rs23
15 files changed, 592 insertions, 60 deletions
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 963dd0c..11fde59 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -192,6 +192,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
(direction, "paths", by.paths),
(direction, "links", by.links),
(direction, "probe", by.probe),
+ (direction, "plumtree", by.plumtree),
]
};
let mut series = Vec::new();
diff --git a/src/config/mod.rs b/src/config/mod.rs
index 93ae7fd..5f6634f 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -285,6 +285,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
"Members that renewed their alive stamp within this time vote on remakes (a majority of them must agree). At least 7h. Members renew their stamps every seventh of the shortest window any of them publishes, hourly at most, so nodes may differ. Matters on the node that coordinates a remake.",
),
option(
+ "dissemination",
+ Kind::Text,
+ "writers-trees|plumtree",
+ Some("writers-trees"),
+ "How a change reaches every member: along writers' trees computed alike on every node from the member list, or by PlumTree (eager links pruned where they deliver twice, grafted where announcements had to bring a change). For experiments: every node of a swarm must use the same one.",
+ ),
+ option(
"superseded-writes",
Kind::Text,
"drop|quarantine",
@@ -576,6 +583,8 @@ pub(crate) struct Settings {
)]
pub(crate) electorate_window: Duration,
#[serde(default)]
+ pub(crate) dissemination: crate::swarm::dissemination::Kind,
+ #[serde(default)]
pub(crate) superseded_writes: crate::remake::Superseded,
#[serde(
default = "default_clock_skew_warn",
diff --git a/src/main.rs b/src/main.rs
index 128e5a0..002ce7f 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -225,6 +225,7 @@ async fn main() -> Result<()> {
clock_skew_warn: args.clock_skew_warn,
stream_heartbeat: args.stream_heartbeat,
superseded_writes: args.superseded_writes,
+ dissemination: args.dissemination,
// At most 4MiB (`Settings::check`).
blob_chunk_bytes: args.blob_chunk as u32,
swarm_key,
diff --git a/src/node.rs b/src/node.rs
index b03f0a2..a9d99b2 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -53,6 +53,8 @@ pub(crate) struct NodeConfig {
pub(crate) stream_heartbeat: Duration,
/// `--superseded-writes`.
pub(crate) superseded_writes: crate::remake::Superseded,
+ /// `--dissemination`.
+ pub(crate) dissemination: crate::swarm::dissemination::Kind,
pub(crate) blob_chunk_bytes: u32,
pub(crate) swarm_key: Option<Arc<SwarmKey>>,
pub(crate) protocol_writers: HashSet<String>,
@@ -101,7 +103,7 @@ impl Node {
})),
peer_states: Arc::new(RwLock::new(HashMap::new())),
topology: Arc::new(crate::swarm::topology::RingAndPairs),
- dissemination: Arc::new(crate::swarm::tree::WritersTrees),
+ dissemination: config.dissemination.make(),
// Counted above the firewall (in the paths layer, under the identity layer):
// what it drops outbound counts as sent (it plays a lossy network), what it
// drops inbound never arrived. Without the identity layer (tests) there is
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index c3996a1..d797115 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -424,7 +424,7 @@ impl Totals {
}
/// The kinds, in the order of the report.
-const KINDS: [&str; 10] = [
+const KINDS: [&str; 11] = [
"SYNC",
"STATUS",
"STATUS_REPLY",
@@ -435,9 +435,10 @@ const KINDS: [&str; 10] = [
"WHO/ME",
"LINKS",
"PROBE",
+ "PRUNE/GRAFT",
];
-fn counters(by: &ByKind) -> [Counter; 10] {
+fn counters(by: &ByKind) -> [Counter; 11] {
[
by.sync,
by.status,
@@ -449,10 +450,11 @@ fn counters(by: &ByKind) -> [Counter; 10] {
by.identity,
by.links,
by.probe,
+ by.plumtree,
]
}
-fn counters_mut(by: &mut ByKind) -> [&mut Counter; 10] {
+fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] {
[
&mut by.sync,
&mut by.status,
@@ -464,6 +466,7 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 10] {
&mut by.identity,
&mut by.links,
&mut by.probe,
+ &mut by.plumtree,
]
}
@@ -545,10 +548,11 @@ async fn run(seed: u64, scenario: &'static Scenario) -> String {
let _ = writeln!(
out,
- "\n{} ({} writes at {} a second each)",
+ "\n{} ({} writes at {} a second each; {})",
scenario.name,
writes(),
- 1000 / WRITE_EVERY.as_millis()
+ 1000 / WRITE_EVERY.as_millis(),
+ all[0].dissemination.name()
);
let mut joined = initial;
for (w, (workload, writer)) in [
@@ -611,6 +615,7 @@ async fn run(seed: u64, scenario: &'static Scenario) -> String {
);
trees(&mut out, &nodes, &sim.net).await;
sent_kept(&mut out, &nodes).await;
+ eager_links(&mut out, &nodes);
if let Some(writer) = writer {
slowest(&mut out, &nodes, &nodes[writer], &sim.net).await;
}
@@ -727,6 +732,66 @@ pub(super) async fn sent_kept(out: &mut String, nodes: &[AppState]) {
);
}
+/// PlumTree's links after the workload (`swarm::plumtree`): how many neighbours each
+/// node pushes to at once (eager), how many push to it, the nodes none pushes to (they
+/// get every change by announcement and pull), and the links eager at one end only (a
+/// PRUNE or GRAFT lost, or crossing). Nothing for a way that keeps no such links.
+fn eager_links(out: &mut String, nodes: &[AppState]) {
+ let mut pushes: HashMap<String, Vec<String>> = HashMap::new();
+ for node in nodes {
+ let Some(lazy) = node.dissemination.lazy() else {
+ return;
+ };
+ let eager = crate::swarm::topology::our_neighbours(node)
+ .into_iter()
+ .filter(|peer| !lazy.contains(peer))
+ .collect();
+ pushes.insert(node.local_id().to_owned(), eager);
+ }
+ let name = |id: &str| {
+ nodes
+ .iter()
+ .find(|node| node.local_id() == id)
+ .and_then(|node| node.alias.clone())
+ .unwrap_or_else(|| id.to_owned())
+ };
+ let mut pushed_by: HashMap<&str, usize> = HashMap::new();
+ let mut one_ended = 0;
+ for (node, eager) in &pushes {
+ for peer in eager {
+ *pushed_by.entry(peer.as_str()).or_default() += 1;
+ if !pushes.get(peer).is_some_and(|theirs| theirs.contains(node)) {
+ one_ended += 1;
+ }
+ }
+ }
+ let count = pushes.len().max(1) as f64;
+ let out_sum: usize = pushes.values().map(Vec::len).sum();
+ let in_least = pushes
+ .keys()
+ .map(|node| pushed_by.get(node.as_str()).copied().unwrap_or(0))
+ .min()
+ .unwrap_or(0);
+ let mut unpushed: Vec<String> = pushes
+ .keys()
+ .filter(|node| !pushed_by.contains_key(node.as_str()))
+ .map(|node| name(node))
+ .collect();
+ unpushed.sort();
+ let _ = writeln!(
+ out,
+ " PlumTree's links at the end: each node pushes to {:.1} neighbours on average; \
+ pushed to by {in_least} at least; {} pushed to by none{}; {one_ended} links eager at one end only",
+ out_sum as f64 / count,
+ unpushed.len(),
+ if unpushed.is_empty() {
+ String::new()
+ } else {
+ format!(" ({})", unpushed.join(", "))
+ },
+ );
+}
+
/// Whether the nodes' views of the writers' trees agree: each member's parent as the
/// member chose it (`swarm::tree::own_parent`) against what that parent and the
/// others plan (`swarm::tree::plan`: their children, the neighbours they count as
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index e9d8c15..c3b2368 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -193,6 +193,11 @@ impl Sim {
clock_skew_warn: Duration::from_secs(2),
stream_heartbeat: Duration::from_secs(30),
superseded_writes: Default::default(),
+ // `SIM_DISSEMINATION=plumtree`: every simulated node passes changes on so.
+ dissemination: match std::env::var("SIM_DISSEMINATION").as_deref() {
+ Ok("plumtree") => crate::swarm::dissemination::Kind::Plumtree,
+ _ => Default::default(),
+ },
blob_chunk_bytes: 64 * 1024,
swarm_key: None,
protocol_writers: HashSet::new(),
diff --git a/src/swarm/dissemination.rs b/src/swarm/dissemination.rs
index 1d92b06..93ad4e9 100644
--- a/src/swarm/dissemination.rs
+++ b/src/swarm/dissemination.rs
@@ -10,6 +10,7 @@
//! first, and so how soon.
use async_trait::async_trait;
+use tracing::trace;
use crate::document::ChangeHash;
use crate::state::AppState;
@@ -48,4 +49,103 @@ pub(crate) trait Dissemination: Send + Sync + 'static {
carried: bool,
authors: &[(String, Vec<ChangeHash>)],
);
+
+ /// `peer` asks us to send it `writer`'s changes at once (`eager`, a GRAFT) or to
+ /// leave them to announcements (a PRUNE), for a way that keeps such links
+ /// (PlumTree). By default nothing.
+ async fn linked(&self, state: &AppState, peer: &str, writer: u64, eager: bool) {
+ let _ = (state, peer, writer, eager);
+ }
+
+ /// `peer` told us of `heads` we lack (a STATUS: an announcement, or its keepalive's),
+ /// for a way that minds which neighbour has changes first (PlumTree). By default
+ /// nothing.
+ fn announced(&self, peer: &str, heads: &[ChangeHash]) {
+ let _ = (peer, heads);
+ }
+
+ /// The neighbours whose links are lazy, for a way that keeps such links (PlumTree),
+ /// for the benchmark to report. By default none.
+ #[cfg(test)]
+ fn lazy(&self) -> Option<std::collections::HashSet<String>> {
+ None
+ }
+}
+
+/// What every way does with its `plan` for `writer`'s `changes` from `from` (none: we
+/// had them already): those it leaves to others (`Plan::covered`) count as having them
+/// (`SyncSession::known_to_peer`): else, with writers in turn, each change coming along
+/// another way, the next message we pass on to one would carry all we got since our
+/// last to it, mostly what its other parents brought it; should one lack them after
+/// all, it asks, and gets them (`resend_if_needed`). Those it sends to
+/// (`Plan::children`, but `from`) get them at once, with those of them that `changes`
+/// build on, if they were counted so lately (`SyncSession::with_predecessors`).
+pub(crate) async fn pass_on(
+ state: &AppState,
+ writer: u64,
+ from: &str,
+ changes: &[ChangeHash],
+ mut plan: Plan,
+) -> Plan {
+ if !changes.is_empty() {
+ for neighbour in &plan.covered {
+ if let Some(peer_state) = state.get_peer_state(neighbour).await {
+ peer_state
+ .lock()
+ .await
+ .sync
+ .known_to_peer(changes, tokio::time::Instant::now());
+ }
+ }
+ }
+ plan.children.retain(|child| child != from);
+ let children = &plan.children;
+ let replica = if changes.is_empty() {
+ None
+ } else {
+ Some(state.doc.read().await)
+ };
+ for child in children {
+ if let Some(peer_state) = state.get_peer_state(child).await {
+ let mut peer_state = peer_state.lock().await;
+ peer_state.tree_writer = Some(writer);
+ if let Some(replica) = &replica {
+ peer_state.sync.with_predecessors(
+ &replica.doc,
+ changes,
+ tokio::time::Instant::now(),
+ );
+ }
+ }
+ state.stats.tree_forward();
+ state.request_sync(child);
+ }
+ if !children.is_empty() {
+ trace!(
+ writer,
+ children = children.len(),
+ "Passing a change on along the writer's tree"
+ );
+ }
+ plan
+}
+
+/// The ways to choose from (`--dissemination`); one per swarm.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub(crate) enum Kind {
+ /// Writers' trees (`swarm::tree`).
+ #[default]
+ WritersTrees,
+ /// PlumTree (`swarm::plumtree`).
+ Plumtree,
+}
+
+impl Kind {
+ pub(crate) fn make(self) -> std::sync::Arc<dyn Dissemination> {
+ match self {
+ Kind::WritersTrees => std::sync::Arc::new(crate::swarm::tree::WritersTrees),
+ Kind::Plumtree => std::sync::Arc::new(crate::swarm::plumtree::PlumTree::default()),
+ }
+ }
}
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 011807c..e01c6f1 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -12,6 +12,7 @@ pub(crate) mod links;
pub(crate) mod members;
pub(crate) mod pair;
pub(crate) mod peer;
+pub(crate) mod plumtree;
pub(crate) mod reach;
pub(crate) mod relay;
pub(crate) mod rtt;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 7154357..ddcc376 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -254,6 +254,18 @@ pub(crate) async fn handle_message(
transport::ParsedMessage::ProbeReply(reply) => {
crate::swarm::working_links::answered(state, &peer, reply);
}
+ transport::ParsedMessage::Prune(link) => {
+ state
+ .dissemination
+ .linked(state, &peer, link.writer, false)
+ .await;
+ }
+ transport::ParsedMessage::Graft(link) => {
+ state
+ .dissemination
+ .linked(state, &peer, link.writer, true)
+ .await;
+ }
transport::ParsedMessage::Who(_)
| transport::ParsedMessage::Me(_)
| transport::ParsedMessage::Ping(_)
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<String> {
+ 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<HashSet<String>>,
+ announced: Mutex<HashMap<ChangeHash, (String, Instant)>>,
+ ahead_times: Mutex<HashMap<String, u32>>,
+ /// When we last grafted each link (`settling`).
+ grafted: Mutex<HashMap<String, Instant>>,
+}
+
+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<ChangeHash>)]) -> Option<String> {
+ 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<ChangeHash>)]) {
+ 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<Plan> {
+ let _ = writer;
+ let neighbours = topology::our_neighbours(state);
+ let lazy = self.lazy.lock_or_recover().clone();
+ let (lazy, eager): (Vec<String>, Vec<String>) =
+ 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<Plan> {
+ 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<u64>,
+ carried: bool,
+ authors: &[(String, Vec<ChangeHash>)],
+ ) {
+ 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<HashSet<String>> {
+ Some(self.lazy.lock_or_recover().clone())
+ }
+}
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index a8e3d57..bee8687 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -333,14 +333,19 @@ pub(crate) async fn lacks(state: &AppState, heads: &HeadsSummary) -> bool {
/// end (`PeerState::pull_pending`).
pub(crate) async fn pull_after_grace(state: &AppState, peer: &str, remote: &transport::Status) {
if HeadsSummary::of(state.stamp.borrow().heads.to_vec()) != remote.heads {
- let grace = state
- .lock_rtt()
- .effective(peer, Instant::now(), state.timing.rtt_fresh_for())
- .map(|rtt| rtt.min(PULL_GRACE_MAX));
- if let Some(grace) = grace
- && lacks(state, &remote.heads).await
- {
- tokio::time::sleep(grace).await;
+ let missing = {
+ let replica = state.doc.read().await;
+ crate::document::changes::missing(&replica.doc, remote.heads.listed())
+ };
+ if !missing.is_empty() {
+ state.dissemination.announced(peer, &missing);
+ let grace = state
+ .lock_rtt()
+ .effective(peer, Instant::now(), state.timing.rtt_fresh_for())
+ .map(|rtt| rtt.min(PULL_GRACE_MAX));
+ if let Some(grace) = grace {
+ tokio::time::sleep(grace).await;
+ }
}
}
let latest = match state.get_peer_state(peer).await {
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index de865de..db3e920 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -30,8 +30,6 @@
use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
use std::sync::Arc;
-use tracing::trace;
-
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::{reach, rtt, topology};
@@ -663,48 +661,8 @@ pub(crate) async fn forward(
from: &str,
changes: &[crate::document::ChangeHash],
) -> Option<Plan> {
- let mut plan = plan(state, writer).await?;
- if !changes.is_empty() {
- for neighbour in &plan.covered {
- if let Some(peer_state) = state.get_peer_state(neighbour).await {
- peer_state
- .lock()
- .await
- .sync
- .known_to_peer(changes, tokio::time::Instant::now());
- }
- }
- }
- plan.children.retain(|child| child != from);
- let children = &plan.children;
- let replica = if changes.is_empty() {
- None
- } else {
- Some(state.doc.read().await)
- };
- for child in children {
- if let Some(peer_state) = state.get_peer_state(child).await {
- let mut peer_state = peer_state.lock().await;
- peer_state.tree_writer = Some(writer);
- if let Some(replica) = &replica {
- peer_state.sync.with_predecessors(
- &replica.doc,
- changes,
- tokio::time::Instant::now(),
- );
- }
- }
- state.stats.tree_forward();
- state.request_sync(child);
- }
- if !children.is_empty() {
- trace!(
- writer,
- children = children.len(),
- "Passing a change on along the writer's tree"
- );
- }
- Some(plan)
+ let plan = plan(state, writer).await?;
+ Some(crate::swarm::dissemination::pass_on(state, writer, from, changes, plan).await)
}
/// Writers' trees as the way changes reach every member
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 743e713..52b544a 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -247,6 +247,10 @@ pub(crate) enum ParsedMessage {
Probe(ProbeRequest),
/// Answers a PROBE.
ProbeReply(ProbeReply),
+ /// Asks the recipient to leave a writer's changes to announcements (`swarm::plumtree`).
+ Prune(TreeLink),
+ /// Asks the recipient to send a writer's changes at once (`swarm::plumtree`).
+ Graft(TreeLink),
}
/// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -391,6 +395,8 @@ pub(crate) enum MessageKind {
LinksReply,
Probe,
ProbeReply,
+ Prune,
+ Graft,
}
impl ParsedMessage {
@@ -410,6 +416,8 @@ impl ParsedMessage {
ParsedMessage::LinksReply(_) => MessageKind::LinksReply,
ParsedMessage::Probe(_) => MessageKind::Probe,
ParsedMessage::ProbeReply(_) => MessageKind::ProbeReply,
+ ParsedMessage::Prune(_) => MessageKind::Prune,
+ ParsedMessage::Graft(_) => MessageKind::Graft,
}
}
@@ -430,6 +438,9 @@ impl ParsedMessage {
ParsedMessage::LinksReply(reply) => Some((reply.epoch, reply.epoch_seq)),
ParsedMessage::Probe(request) => Some((request.epoch, request.epoch_seq)),
ParsedMessage::ProbeReply(reply) => Some((reply.epoch, reply.epoch_seq)),
+ ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => {
+ Some((link.epoch, link.epoch_seq))
+ }
ParsedMessage::Who(_)
| ParsedMessage::Me(_)
| ParsedMessage::Ping(_)
@@ -438,6 +449,18 @@ impl ParsedMessage {
}
}
+/// PRUNE and GRAFT: a writer's changes left to announcements, or sent at once, between
+/// the sender and the recipient (`swarm::plumtree`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct TreeLink {
+ #[serde(with = "crate::transport::wire::codec::epoch")]
+ pub(crate) epoch: Epoch,
+ /// See `SyncEnvelope::epoch_seq`.
+ pub(crate) epoch_seq: u32,
+ /// The writer, by its pair key (`topology::pair_key`).
+ pub(crate) writer: u64,
+}
+
/// PROBE: whether the recipient reaches `member` (`swarm::working_links`), as SWIM
/// asks others before it suspects a member. Answered from what the recipient knows,
/// without contacting the member.
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 2e4266c..501c77b 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -55,6 +55,8 @@ pub(crate) struct ByKind {
pub(crate) links: Counter,
/// PROBE and PROBE_REPLY (`swarm::working_links`).
pub(crate) probe: Counter,
+ /// PRUNE and GRAFT (`swarm::plumtree`).
+ pub(crate) plumtree: Counter,
}
impl ByKind {
@@ -70,6 +72,7 @@ impl ByKind {
MessageKind::Ping,
MessageKind::Links,
MessageKind::Probe,
+ MessageKind::Prune,
] {
let theirs = *other.get(kind);
self.get_mut(kind).absorb(theirs);
@@ -88,6 +91,7 @@ impl ByKind {
MessageKind::Ping | MessageKind::Pong => &self.paths,
MessageKind::Links | MessageKind::LinksReply => &self.links,
MessageKind::Probe | MessageKind::ProbeReply => &self.probe,
+ MessageKind::Prune | MessageKind::Graft => &self.plumtree,
}
}
@@ -103,6 +107,7 @@ impl ByKind {
MessageKind::Ping | MessageKind::Pong => &mut self.paths,
MessageKind::Links | MessageKind::LinksReply => &mut self.links,
MessageKind::Probe | MessageKind::ProbeReply => &mut self.probe,
+ MessageKind::Prune | MessageKind::Graft => &mut self.plumtree,
}
}
}
@@ -295,6 +300,8 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) {
MessageKind::LinksReply => ("LINKS_REPLY", "links"),
MessageKind::Probe => ("PROBE", "probe"),
MessageKind::ProbeReply => ("PROBE_REPLY", "probe"),
+ MessageKind::Prune => ("PRUNE", "plumtree"),
+ MessageKind::Graft => ("GRAFT", "plumtree"),
}
}
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index bf46fc9..729acd3 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
/// The wire format's version, the first byte of every payload. Bump it with every change
/// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 36;
+pub(crate) const VERSION: u8 = 37;
/// What the second byte of a payload says.
fn tag(kind: MessageKind) -> u8 {
@@ -33,6 +33,8 @@ fn tag(kind: MessageKind) -> u8 {
MessageKind::LinksReply => 11,
MessageKind::Probe => 12,
MessageKind::ProbeReply => 13,
+ MessageKind::Prune => 14,
+ MessageKind::Graft => 15,
}
}
@@ -89,6 +91,9 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
ParsedMessage::LinksReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
ParsedMessage::Probe(request) => rmp_serde::encode::write(&mut payload, request)?,
ParsedMessage::ProbeReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
+ ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => {
+ rmp_serde::encode::write(&mut payload, link)?
+ }
}
Ok(payload)
}
@@ -162,6 +167,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
ParsedMessage::Probe(rmp_serde::from_slice(body)?)
} else if is(MessageKind::ProbeReply) {
ParsedMessage::ProbeReply(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::Prune) {
+ ParsedMessage::Prune(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::Graft) {
+ ParsedMessage::Graft(rmp_serde::from_slice(body)?)
} else {
return Err(WireError::UnknownKind(kind));
})
@@ -532,6 +541,16 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
member: "q".into(),
reached: None,
}),
+ ParsedMessage::Prune(crate::transport::TreeLink {
+ epoch,
+ epoch_seq: 1,
+ writer: 5,
+ }),
+ ParsedMessage::Graft(crate::transport::TreeLink {
+ epoch,
+ epoch_seq: 1,
+ writer: 5,
+ }),
]
}
@@ -563,7 +582,7 @@ mod tests {
#[test]
fn every_kind_round_trips() {
let samples = samples();
- assert_eq!(samples.len(), 14, "a kind is missing from `samples`");
+ assert_eq!(samples.len(), 16, "a kind is missing from `samples`");
for message in samples {
let bytes = encode(&message).unwrap();
let decoded = decode(&bytes).unwrap();