aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/swarm/topology/graph.rs17
-rw-r--r--src/swarm/topology/mod.rs161
2 files changed, 120 insertions, 58 deletions
diff --git a/src/swarm/topology/graph.rs b/src/swarm/topology/graph.rs
index 92240e9..d8b2921 100644
--- a/src/swarm/topology/graph.rs
+++ b/src/swarm/topology/graph.rs
@@ -34,8 +34,8 @@ pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> {
/// The neighbours of `local` among `members` (by the transports each is on), on each of
/// `local_transports`: on each, its neighbours among the members on it
/// (`neighbours`), all together.
-pub(crate) fn neighbours_across(
- topology: &dyn Topology,
+pub(crate) fn neighbours_across<T: Topology + ?Sized>(
+ topology: &T,
members: &BTreeMap<String, BTreeSet<String>>,
local: &str,
local_transports: &BTreeSet<String>,
@@ -56,9 +56,10 @@ pub(crate) fn neighbours_across(
}
/// The neighbours of `local` among `members` (which may or may not contain `local`),
-/// skipping members for which `unreachable` holds.
-pub(crate) fn neighbours(
- topology: &dyn Topology,
+/// skipping members for which `unreachable` holds; none by a topology without a shared
+/// graph.
+pub(crate) fn neighbours<T: Topology + ?Sized>(
+ topology: &T,
members: &[String],
local: &str,
k: usize,
@@ -69,7 +70,7 @@ pub(crate) fn neighbours(
all.push(local.to_owned());
topology
.graph(&all, k, &|m: &str| m != local && unreachable(m), stable)
- .remove(local)
+ .and_then(|mut graph| graph.remove(local))
.unwrap_or_default()
}
@@ -88,8 +89,8 @@ impl Topology for RingAndPairs {
k: usize,
unreachable: &dyn Fn(&str) -> bool,
stable: &dyn Fn(&str) -> bool,
- ) -> BTreeMap<String, BTreeSet<String>> {
- graph(members, k, unreachable, stable)
+ ) -> Option<BTreeMap<String, BTreeSet<String>>> {
+ Some(graph(members, k, unreachable, stable))
}
}
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index b5da4d8..d9f361a 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -41,26 +41,73 @@ pub(crate) use graph::{RingAndPairs, pair_key, pair_score};
/// How members' neighbours are chosen: the graph above (`RingAndPairs`) is one way;
/// others can be tried against it in the benchmark (`docs/future-work.md`, Swappable
-/// topology and dissemination). One per swarm, as every node must compute the same
-/// graph: the writers' trees are built on it (`member_list_graph`, `swarm::tree`).
+/// topology and dissemination). One per swarm: a node's neighbours must count it as
+/// theirs.
///
-/// A topology each node keeps from its own messages instead (HyParView's views) has no
-/// graph every node shares; it takes more than this (its state, its messages, and a
-/// dissemination that needs no shared graph), and comes with one.
+/// A topology either computes a graph every node shares (`graph`: the same inputs give
+/// every node the same), or keeps its own state from its own messages (HyParView's
+/// views), and has none. What needs the shared graph does nothing without one: writers'
+/// trees (`member_list_graph`, `swarm::tree`), the links members record where it is
+/// wrong (`working_graph`, `swarm::working_links`), indirect probes through a member's
+/// neighbours, the dashboard's map of the intended links (`intended_graph`). A
+/// dissemination that needs none goes with such a topology (PlumTree).
+#[async_trait::async_trait]
pub(crate) trait Topology: Send + Sync + 'static {
/// Its name (in logs, and once chosen by the swarm, in the document).
fn name(&self) -> &'static str;
/// Every reachable member's neighbours among `members` (those on one transport),
/// each keeping about `k`, skipping those `unreachable`, preferring those `stable`
- /// as it does. Pure: the same inputs give every node the same graph.
+ /// as it does. Pure: the same inputs give every node the same graph. None for a
+ /// topology without a shared graph.
fn graph(
&self,
members: &[String],
k: usize,
unreachable: &dyn Fn(&str) -> bool,
stable: &dyn Fn(&str) -> bool,
- ) -> BTreeMap<String, BTreeSet<String>>;
+ ) -> Option<BTreeMap<String, BTreeSet<String>>>;
+
+ /// Our neighbours, from what they are chosen from (`Inputs`), each time those may
+ /// have changed (`refresh`). By default our part of the shared graph, on each of our
+ /// transports (`neighbours_across`); a topology with state of its own keeps them
+ /// there.
+ fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
+ neighbours_across(
+ self,
+ inputs.transports,
+ local,
+ inputs.local_transports,
+ inputs.k,
+ inputs.unreachable,
+ inputs.stable,
+ )
+ }
+
+ /// With every recomputation (`refresh`, at least every 5 s), before `neighbours`:
+ /// what a topology with state of its own does about it (asking members to be
+ /// neighbours, giving up on those that don't answer). By default nothing.
+ async fn maintain(&self, state: &AppState, inputs: &Inputs<'_>) {
+ let _ = (state, inputs);
+ }
+
+ /// 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
+ }
+}
+
+/// What a node's neighbours are chosen from (`Topology::neighbours`): the members, by
+/// the transports each is on (as the member list gives their addresses), ours, the
+/// neighbours each keeps, whom we judge unreachable, who is stable.
+pub(crate) struct Inputs<'a> {
+ pub(crate) transports: &'a BTreeMap<String, BTreeSet<String>>,
+ pub(crate) local_transports: &'a BTreeSet<String>,
+ pub(crate) k: usize,
+ pub(crate) unreachable: &'a (dyn Fn(&str) -> bool + Sync),
+ pub(crate) stable: &'a (dyn Fn(&str) -> bool + Sync),
}
/// This node's current neighbours, and the member list they were computed from.
@@ -81,6 +128,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,
/// 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.
@@ -88,66 +137,71 @@ pub(crate) struct NeighbourView {
full_mesh: bool,
/// The graph the member list gives, as if every member reached every other
/// (`member_list_graph`): computed when first asked for, kept while the member list
- /// stays the same.
- whole: std::sync::OnceLock<Arc<BTreeMap<String, BTreeSet<String>>>>,
+ /// stays the same. None for a topology without a shared graph.
+ whole: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
/// Where members recorded their links differ from that graph (`swarm::working_links`),
/// those with a record.
recorded: BTreeMap<String, members::RecordedLinks>,
/// That graph with the links members recorded (`working_graph`): computed when first
/// asked for, kept while the member list and the records stay the same.
- working: std::sync::OnceLock<Arc<BTreeMap<String, BTreeSet<String>>>>,
+ working: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
}
/// Every member's neighbours as the member list gives them, as if every member reached
/// every other: the same on every node that knows the same member list, whatever each
/// reaches (`swarm::tree` builds writers' trees on it). On each transport among the
-/// members on it, all together. None until the view is known.
+/// members on it, all together. None until the view is known, and for a topology
+/// without a shared graph (`Topology::graph`).
pub(crate) fn member_list_graph(
state: &AppState,
) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
let view = state.neighbours.read_or_recover();
let view = view.as_ref()?;
- Some(Arc::clone(view.whole.get_or_init(|| {
- let mut whole: BTreeMap<String, BTreeSet<String>> = view
- .members
- .iter()
- .map(|member| (member.clone(), BTreeSet::new()))
- .collect();
- let transports: BTreeSet<&String> = view.transports.values().flatten().collect();
- for transport in transports {
- let on_it: Vec<String> = view
- .transports
+ view.whole
+ .get_or_init(|| {
+ let mut whole: BTreeMap<String, BTreeSet<String>> = view
+ .members
.iter()
- .filter(|(_, on)| on.contains(transport))
- .map(|(member, _)| member.clone())
+ .map(|member| (member.clone(), BTreeSet::new()))
.collect();
- let graph = state
- .topology
- .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m));
- for (member, neighbours) in graph {
- whole.entry(member).or_default().extend(neighbours);
+ let transports: BTreeSet<&String> = view.transports.values().flatten().collect();
+ for transport in transports {
+ let on_it: Vec<String> = view
+ .transports
+ .iter()
+ .filter(|(_, on)| on.contains(transport))
+ .map(|(member, _)| member.clone())
+ .collect();
+ let graph = state
+ .topology
+ .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m))?;
+ for (member, neighbours) in graph {
+ whole.entry(member).or_default().extend(neighbours);
+ }
}
- }
- Arc::new(whole)
- })))
+ Some(Arc::new(whole))
+ })
+ .clone()
}
/// The member list's graph (`member_list_graph`) with the links members recorded
/// (`swarm::working_links`): without those either end recorded as unreached, with those
/// one recorded as reached (`with_recorded`). The same on every node that has the same
/// document; what the writers' trees are built on (`swarm::tree`). None until the view
-/// is known.
+/// is known, and for a topology without a shared graph.
pub(crate) fn working_graph(state: &AppState) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
let whole = member_list_graph(state)?;
let view = state.neighbours.read_or_recover();
let view = view.as_ref()?;
- Some(Arc::clone(view.working.get_or_init(|| {
- if view.recorded.is_empty() {
- whole
- } else {
- Arc::new(with_recorded(&whole, &view.recorded))
- }
- })))
+ view.working
+ .get_or_init(|| {
+ Some(if view.recorded.is_empty() {
+ whole
+ } else {
+ Arc::new(with_recorded(&whole, &view.recorded))
+ })
+ })
+ .clone()
}
/// `graph` with the links members recorded (`recorded`): those one recorded as reached
@@ -303,9 +357,11 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
/// they are (with members that only connect out, or aren't listed yet). For a map of
/// the swarm (the dashboard); None until the view is known. A link exists if either end
/// chose it: draw one if either lists the other. About `members²` work: compute it when
-/// the member list may have changed, not on every look.
+/// the member list may have changed, not on every look. None too for a topology without
+/// a shared graph.
#[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 view = state.neighbours.read_or_recover();
let view = view.as_ref()?;
let graph = view
@@ -581,10 +637,22 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
}
}
}
+ let unreachable_of = |m: &str| unreachable.contains(m);
+ let stable_of = |m: &str| stable.contains(m);
+ let inputs = Inputs {
+ transports: &transports,
+ local_transports: &local_transports,
+ k,
+ unreachable: &unreachable_of,
+ stable: &stable_of,
+ };
+ state.topology.maintain(state, &inputs).await;
+ let version = state.topology.version();
let previous = {
let view = state.neighbours.read_or_recover();
if view.as_ref().is_some_and(|view| {
- view.members == current
+ view.version == version
+ && view.members == current
&& view.unlisted == unlisted
&& view.outbound == outbound
&& view.k == k
@@ -626,15 +694,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 = neighbours_across(
- &*state.topology,
- &transports,
- state.local_id(),
- &local_transports,
- k,
- &|m| unreachable.contains(m),
- &|m| stable.contains(m),
- );
+ let mut computed = state.topology.neighbours(state.local_id(), &inputs);
computed.extend(outbound.iter().cloned());
#[cfg(test)]
let listed_neighbours = computed.len();
@@ -680,6 +740,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
unlisted,
outbound,
neighbours: computed,
+ version,
whole,
recorded,
working,