diff options
| -rw-r--r-- | src/node.rs | 1 | ||||
| -rw-r--r-- | src/state/mod.rs | 2 | ||||
| -rw-r--r-- | src/swarm/dissemination.rs | 51 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 6 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 9 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 6 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 36 |
8 files changed, 107 insertions, 5 deletions
diff --git a/src/node.rs b/src/node.rs index de72e6f..b03f0a2 100644 --- a/src/node.rs +++ b/src/node.rs @@ -101,6 +101,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), // 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/state/mod.rs b/src/state/mod.rs index 7e87f31..d5c18b3 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -70,6 +70,8 @@ pub(crate) struct AppState { pub(crate) transport: Arc<dyn transport::Transport>, /// How members' neighbours are chosen (`swarm::topology::Topology`). pub(crate) topology: Arc<dyn crate::swarm::topology::Topology>, + /// How a change reaches every member (`swarm::dissemination::Dissemination`). + pub(crate) dissemination: Arc<dyn crate::swarm::dissemination::Dissemination>, pub(crate) firewall: Arc<Firewall>, /// Messages and bytes exchanged, per peer (`transport::traffic`). pub(crate) traffic: Arc<transport::traffic::Traffic>, diff --git a/src/swarm/dissemination.rs b/src/swarm/dissemination.rs new file mode 100644 index 0000000..1d92b06 --- /dev/null +++ b/src/swarm/dissemination.rs @@ -0,0 +1,51 @@ +//! How a change reaches every member: whom a node sends a writer's changes to at once, +//! and whom it leaves to the announcement rounds (`swarm::relay`, which repair what it +//! misses: announcements, pulls, the slow round). Writers' trees (`swarm::tree`) are +//! one way; others can be tried against them in the benchmark (`docs/future-work.md`, +//! Swappable topology and dissemination). One per swarm: a node's part in passing a +//! change on is worked out by itself and by its neighbours, and must agree. +//! +//! What every way keeps: the base underneath, pairs of neighbours syncing with +//! Automerge, which delivers everything in the end; a way only decides whom to push to +//! first, and so how soon. + +use async_trait::async_trait; + +use crate::document::ChangeHash; +use crate::state::AppState; +use crate::swarm::tree::Plan; + +#[async_trait] +pub(crate) trait Dissemination: Send + Sync + 'static { + /// Its name (in logs, and once chosen by the swarm, in the document). + fn name(&self) -> &'static str; + + /// Our part in passing on `writer`'s changes (by its pair key, `topology::pair_key`): + /// whom we send them to at once, and how the announcement rounds treat our other + /// neighbours (`Plan`). None: we have no part, and announcements do it all. + async fn plan(&self, state: &AppState, writer: u64) -> Option<Plan>; + + /// `writer`'s `changes` came from `from` (none: we had them already, and they came + /// as the writer's now): passed on at once to those we send them to, the others + /// counted as having them where they get them elsewhere. Returns our part, as + /// `plan`. + async fn forward( + &self, + state: &AppState, + writer: u64, + from: &str, + changes: &[ChangeHash], + ) -> Option<Plan>; + + /// What a sync message from `peer` says about how well this works (`/admin/stats`): + /// whether it came as `writer`'s and `carried` changes, and the new changes it + /// brought, by author. Not while catching up. + async fn note_arrival( + &self, + state: &AppState, + peer: &str, + writer: Option<u64>, + carried: bool, + authors: &[(String, Vec<ChangeHash>)], + ); +} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 7014438..011807c 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -5,6 +5,7 @@ pub(crate) mod auth; pub(crate) mod blob; pub(crate) mod bootstrap; pub(crate) mod changes; +pub(crate) mod dissemination; pub(crate) mod document_file; pub(crate) mod hop; pub(crate) mod links; diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 0f8a697..a8e3d57 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -157,7 +157,11 @@ pub(crate) async fn announce( let mut plans = Vec::new(); let mut treeless = unauthored || writers.is_empty(); for (writer, changes) in writers { - match crate::swarm::tree::forward(state, *writer, from, changes).await { + match state + .dissemination + .forward(state, *writer, from, changes) + .await + { Some(plan) => plans.push(plan), None => treeless = true, } diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 66fa4a5..45d3d75 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -687,7 +687,10 @@ pub(crate) async fn handle_sync( crate::swarm::relay::pulled(state, peer); if state.initial_sync_done.load(Ordering::Acquire) && !just_caught_up { - crate::swarm::tree::note_arrival(state, peer, writer, carried, &authors).await; + state + .dissemination + .note_arrival(state, peer, writer, carried, &authors) + .await; } // Caught up: our other neighbours may lack what the source had. if changed || just_caught_up { @@ -703,7 +706,7 @@ pub(crate) async fn handle_sync( // A writer's changes along its tree that we had already: they came another way // first, not saying whose (with an answer to one of our messages, say), so we // passed nothing on then. Our children in its tree may lack them still. - crate::swarm::tree::forward(state, writer, peer, &[]).await; + state.dissemination.forward(state, writer, peer, &[]).await; } state @@ -941,7 +944,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece let own = crate::swarm::topology::pair_key(state.local_id()); let ours = own_changes(&state, &mut seen_heads).await; let covered = match &ours { - Some(ours) => match crate::swarm::tree::plan(&state, own).await { + Some(ours) => match state.dissemination.plan(&state, own).await { Some(plan) => { for neighbour in &plan.covered { if let Some(peer_state) = state.get_peer_state(neighbour).await { diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index 647d3a3..b5da4d8 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -396,7 +396,11 @@ pub(crate) fn change_peer<T>( /// a peer's state (`change_peer`), and every 5 s as a fallback (`swarm::peer`). A burst /// of notifications coalesces into one recomputation. pub(crate) async fn run_worker(state: AppState) { - info!(topology = state.topology.name(), "Choosing neighbours"); + info!( + topology = state.topology.name(), + dissemination = state.dissemination.name(), + "Choosing neighbours" + ); loop { state.topology_changed.notified().await; if let Some(change) = refresh(&state).await { diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs index 064f5b9..de865de 100644 --- a/src/swarm/tree.rs +++ b/src/swarm/tree.rs @@ -707,6 +707,42 @@ pub(crate) async fn forward( Some(plan) } +/// Writers' trees as the way changes reach every member +/// (`swarm::dissemination::Dissemination`): `plan`, `forward` and `note_arrival` above. +pub(crate) struct WritersTrees; + +#[async_trait::async_trait] +impl crate::swarm::dissemination::Dissemination for WritersTrees { + fn name(&self) -> &'static str { + "writers-trees" + } + + async fn plan(&self, state: &AppState, writer: u64) -> Option<Plan> { + plan(state, writer).await + } + + async fn forward( + &self, + state: &AppState, + writer: u64, + from: &str, + changes: &[crate::document::ChangeHash], + ) -> Option<Plan> { + forward(state, writer, from, changes).await + } + + async fn note_arrival( + &self, + state: &AppState, + peer: &str, + writer: Option<u64>, + carried: bool, + authors: &[(String, Vec<crate::document::ChangeHash>)], + ) { + note_arrival(state, peer, writer, carried, authors).await; + } +} + #[cfg(test)] mod tests { use super::*; |