diff options
Diffstat (limited to 'src/swarm/topology/mod.rs')
| -rw-r--r-- | src/swarm/topology/mod.rs | 161 |
1 files changed, 152 insertions, 9 deletions
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index d9f361a..14e9de6 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -33,6 +33,7 @@ use tracing::info; use crate::locks::RwLockExt; use crate::state::{AppState, PeerState}; use crate::swarm::members; +use crate::transport::ParsedMessage; mod graph; @@ -91,12 +92,149 @@ pub(crate) trait Topology: Send + Sync + 'static { let _ = (state, inputs); } + /// A message of the topology's own from `peer` (HyParView's NEIGHBOUR, + /// NEIGHBOUR_REPLY, DISCONNECT). By default ignored. + async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) { + let _ = (state, peer, message); + } + /// Changes whenever the topology's own state changed what `neighbours` gives: the /// neighbours are recomputed then, even though nothing they are chosen from did. By /// default always the same. fn version(&self) -> u64 { 0 } + + /// Taking over from another topology, whose neighbours were `neighbours` (the swarm + /// switched, `swarm_ways`): a topology with state of its own starts from them. By + /// default nothing. + fn seed(&self, neighbours: &BTreeSet<String>) { + let _ = neighbours; + } + + /// The neighbours each agreed to, for a topology that keeps such (HyParView's active + /// view), for the tests and the benchmark to look at. By default none. + #[cfg(test)] + fn active(&self) -> Option<BTreeSet<String>> { + None + } +} + +/// The topology and the dissemination in use (`AppState::ways`), and which they are. +pub(crate) struct Ways { + pub(crate) kinds: (Kind, crate::swarm::dissemination::Kind), + pub(crate) topology: Arc<dyn Topology>, + pub(crate) dissemination: Arc<dyn crate::swarm::dissemination::Dissemination>, +} + +impl Ways { + pub(crate) fn new(topology: Kind, dissemination: crate::swarm::dissemination::Kind) -> Self { + Self { + kinds: (topology, dissemination), + topology: topology.make(), + dissemination: dissemination.make(), + } + } +} + +/// Where the swarm records its topology and its dissemination (`swarm_ways`); the +/// genesis node writes its `--topology` and `--dissemination` there +/// (`swarm::bootstrap`). +pub(crate) fn topology_path() -> Vec<String> { + crate::document::meta::path(&[crate::document::meta::TOPOLOGY]) +} + +pub(crate) fn dissemination_path() -> Vec<String> { + crate::document::meta::path(&[crate::document::meta::DISSEMINATION]) +} + +/// The topology and the dissemination the swarm records: its genesis node writes its +/// own (`topology_path`, `dissemination_path`), a protocol writer may change them (every +/// node then swaps to them, `refresh`). Where the document has none (a swarm from +/// before they were recorded), or one this node doesn't know, this node's own (`own`). +pub(crate) fn swarm_ways( + doc: &crate::document::Document, + own: (Kind, crate::swarm::dissemination::Kind), +) -> (Kind, crate::swarm::dissemination::Kind) { + fn recorded<T: serde::de::DeserializeOwned>( + doc: &crate::document::Document, + path: &[String], + what: &str, + own: T, + ) -> T { + match crate::document::store::read_path(doc, path) { + Ok(Some(value)) => match serde_json::from_value(value.clone()) { + Ok(recorded) => recorded, + Err(_) => { + warn_once(&format!( + "The swarm's {what} ({value}) isn't one this node knows; using --{what}" + )); + own + } + }, + _ => own, + } + } + ( + recorded(doc, &topology_path(), "topology", own.0), + recorded(doc, &dissemination_path(), "dissemination", own.1), + ) +} + +/// Swaps to the topology and the dissemination `kinds` if they aren't the ones in use: +/// fresh, the new topology starting from our neighbours (`Topology::seed`). +fn switch_ways(state: &AppState, kinds: (Kind, crate::swarm::dissemination::Kind)) { + let neighbours: BTreeSet<String> = our_neighbours(state).into_iter().collect(); + let mut ways = state.ways.write_or_recover(); + if ways.kinds == kinds { + return; + } + let before = (ways.topology.name(), ways.dissemination.name()); + if ways.kinds.0 != kinds.0 { + ways.topology = kinds.0.make(); + ways.topology.seed(&neighbours); + } + if ways.kinds.1 != kinds.1 { + ways.dissemination = kinds.1.make(); + } + ways.kinds = kinds; + info!( + from_topology = before.0, + from_dissemination = before.1, + topology = ways.topology.name(), + dissemination = ways.dissemination.name(), + "The swarm switched how neighbours are chosen, or how changes pass on" + ); + if ways + .topology + .graph(&[], 2, &|_| false, &|_| false) + .is_none() + && kinds.1 == crate::swarm::dissemination::Kind::WritersTrees + { + warn_once( + "Writers' trees need a graph every node shares, which this topology has none of: changes pass on by announcements alone (a protocol writer may record dissemination \"plumtree\")", + ); + } +} + +/// The topologies to choose from (`--topology`); one per swarm. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum Kind { + /// The member list's graph (`RingAndPairs`). + #[default] + RingAndPairs, + /// HyParView's active view, slimmed down (`swarm::hyparview`). + Hyparview, +} + +impl Kind { + pub(crate) fn make(self) -> Arc<dyn Topology> { + match self { + Kind::RingAndPairs => Arc::new(RingAndPairs), + Kind::Hyparview => Arc::new(crate::swarm::hyparview::HyParView::default()), + } + } } /// What a node's neighbours are chosen from (`Topology::neighbours`): the members, by @@ -128,8 +266,8 @@ pub(crate) struct NeighbourView { /// they say they sync with us (`refresh`). outbound: BTreeSet<String>, neighbours: BTreeSet<String>, - /// The topology's version they were computed with (`Topology::version`). - version: u64, + /// The topology they were computed with, and its version (`Topology::version`). + version: (&'static str, u64), /// Every member not judged unreachable is a neighbour (the tests wait for it, /// `is_full_mesh`). Not the swarm being a full mesh: members on a transport we are /// not on, or that we never reached, don't count against it. @@ -173,7 +311,7 @@ pub(crate) fn member_list_graph( .map(|(member, _)| member.clone()) .collect(); let graph = state - .topology + .topology() .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m))?; for (member, neighbours) in graph { whole.entry(member).or_default().extend(neighbours); @@ -362,6 +500,7 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool { #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> { member_list_graph(state)?; + let topology = state.topology(); let view = state.neighbours.read_or_recover(); let view = view.as_ref()?; let graph = view @@ -370,7 +509,7 @@ pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeS .map(|member| { ( member.clone(), - view.neighbours_of(&*state.topology, state.local_id(), member), + view.neighbours_of(&*topology, state.local_id(), member), ) }) .collect(); @@ -453,8 +592,8 @@ pub(crate) fn change_peer<T>( /// of notifications coalesces into one recomputation. pub(crate) async fn run_worker(state: AppState) { info!( - topology = state.topology.name(), - dissemination = state.dissemination.name(), + topology = state.topology().name(), + dissemination = state.dissemination().name(), "Choosing neighbours" ); loop { @@ -471,6 +610,8 @@ pub(crate) fn affects_neighbours(path: &[String]) -> bool { path.is_empty() || path.starts_with(&members::members_dir()) || path.starts_with(&neighbours_path()) + || path.starts_with(&topology_path()) + || path.starts_with(&dissemination_path()) } /// The most neighbours a swarm may ask each member to keep (`swarm_neighbours`). @@ -543,6 +684,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { let replica = state.doc.read().await; members::apply(state, &replica.doc); let k = swarm_neighbours(&replica.doc, state.neighbours_k); + switch_ways(state, swarm_ways(&replica.doc, state.configured_ways)); let current = members::members(&replica.doc); // Those that said they are alive within the electorate window. let alive: BTreeSet<String> = members::electorate( @@ -646,8 +788,9 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { unreachable: &unreachable_of, stable: &stable_of, }; - state.topology.maintain(state, &inputs).await; - let version = state.topology.version(); + let topology = state.topology(); + topology.maintain(state, &inputs).await; + let version = (topology.name(), topology.version()); let previous = { let view = state.neighbours.read_or_recover(); if view.as_ref().is_some_and(|view| { @@ -694,7 +837,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { Some(previous) => previous, None => state.peer_states.read().await.keys().cloned().collect(), }; - let mut computed = state.topology.neighbours(state.local_id(), &inputs); + let mut computed = topology.neighbours(state.local_id(), &inputs); computed.extend(outbound.iter().cloned()); #[cfg(test)] let listed_neighbours = computed.len(); |
