From b18c75f20f937229c9b12bf3f07901090bd35317 Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 16:34:47 +0100 Subject: topology trait --- src/node.rs | 1 + src/state/mod.rs | 2 ++ src/swarm/topology/graph.rs | 27 ++++++++++++++++++++++++-- src/swarm/topology/mod.rs | 47 +++++++++++++++++++++++++++++++++++++++++---- src/swarm/topology/tests.rs | 33 ++++++++++++++++++++++++++----- 5 files changed, 99 insertions(+), 11 deletions(-) diff --git a/src/node.rs b/src/node.rs index 9322b75..de72e6f 100644 --- a/src/node.rs +++ b/src/node.rs @@ -100,6 +100,7 @@ impl Node { discovered_swarm_id: None, })), peer_states: Arc::new(RwLock::new(HashMap::new())), + topology: Arc::new(crate::swarm::topology::RingAndPairs), // 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 d1a32df..7e87f31 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -68,6 +68,8 @@ pub(crate) struct AppState { pub(crate) peer_states: Arc>>>>, pub(crate) transport: Arc, + /// How members' neighbours are chosen (`swarm::topology::Topology`). + pub(crate) topology: Arc, pub(crate) firewall: Arc, /// Messages and bytes exchanged, per peer (`transport::traffic`). pub(crate) traffic: Arc, diff --git a/src/swarm/topology/graph.rs b/src/swarm/topology/graph.rs index 5401506..92240e9 100644 --- a/src/swarm/topology/graph.rs +++ b/src/swarm/topology/graph.rs @@ -35,6 +35,7 @@ pub(crate) fn schemes(addresses: &[String]) -> BTreeSet { /// `local_transports`: on each, its neighbours among the members on it /// (`neighbours`), all together. pub(crate) fn neighbours_across( + topology: &dyn Topology, members: &BTreeMap>, local: &str, local_transports: &BTreeSet, @@ -49,7 +50,7 @@ pub(crate) fn neighbours_across( .filter(|(member, transports)| *member != local && transports.contains(transport)) .map(|(member, _)| member.clone()) .collect(); - all.extend(neighbours(&on_it, local, k, unreachable, stable)); + all.extend(neighbours(topology, &on_it, local, k, unreachable, stable)); } all } @@ -57,6 +58,7 @@ 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, members: &[String], local: &str, k: usize, @@ -65,11 +67,32 @@ pub(crate) fn neighbours( ) -> BTreeSet { let mut all: Vec = members.to_vec(); all.push(local.to_owned()); - graph(&all, k, &|m: &str| m != local && unreachable(m), stable) + topology + .graph(&all, k, &|m: &str| m != local && unreachable(m), stable) .remove(local) .unwrap_or_default() } +/// The graph described above (`graph`): a ring by key hash, plus each member's +/// lowest-scoring pairs, stable members preferred. +pub(crate) struct RingAndPairs; + +impl Topology for RingAndPairs { + fn name(&self) -> &'static str { + "ring-and-pairs" + } + + fn graph( + &self, + members: &[String], + k: usize, + unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, + ) -> BTreeMap> { + graph(members, k, unreachable, stable) + } +} + /// Every reachable member's neighbours. Hashes each member once and scores pairs by /// mixing the two hashes (`pair_score`), so the N² scores are a few nanoseconds each: /// tens of milliseconds for thousands of members. `stable` members (`swarm::stability`) diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index 2c41f0b..647d3a3 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -37,7 +37,31 @@ use crate::swarm::members; mod graph; use graph::*; -pub(crate) use graph::{pair_key, pair_score}; +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`). +/// +/// 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. +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. + fn graph( + &self, + members: &[String], + k: usize, + unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, + ) -> BTreeMap>; +} /// This node's current neighbours, and the member list they were computed from. #[derive(Debug, Default)] @@ -97,7 +121,9 @@ pub(crate) fn member_list_graph( .filter(|(_, on)| on.contains(transport)) .map(|(member, _)| member.clone()) .collect(); - let graph = graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m)); + 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); } @@ -285,7 +311,12 @@ pub(crate) fn intended_graph(state: &AppState) -> Option BTreeSet } impl NeighbourView { - fn neighbours_of(&self, local: &str, member: &str) -> BTreeSet { + fn neighbours_of( + &self, + topology: &dyn Topology, + local: &str, + member: &str, + ) -> BTreeSet { if member == local { return self.neighbours.clone(); } let own = self.transports.get(member).cloned().unwrap_or_default(); neighbours_across( + topology, &self.transports, member, &own, @@ -359,6 +396,7 @@ pub(crate) fn change_peer( /// 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"); loop { state.topology_changed.notified().await; if let Some(change) = refresh(&state).await { @@ -585,6 +623,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option { None => state.peer_states.read().await.keys().cloned().collect(), }; let mut computed = neighbours_across( + &*state.topology, &transports, state.local_id(), &local_transports, diff --git a/src/swarm/topology/tests.rs b/src/swarm/topology/tests.rs index ede3b6b..8ee83c2 100644 --- a/src/swarm/topology/tests.rs +++ b/src/swarm/topology/tests.rs @@ -102,7 +102,14 @@ fn random_swarms() { if let Some(local) = live.iter().nth(rng.below(live.len())) { let others: Vec = all.iter().filter(|m| *m != local).cloned().collect(); assert_eq!( - neighbours(&others, local, k, &is_unreachable, &is_stable), + neighbours( + &RingAndPairs, + &others, + local, + k, + &is_unreachable, + &is_stable + ), g[local] ); } @@ -114,7 +121,7 @@ fn small_swarms_are_a_full_mesh() { let ms = members(13); for m in &ms { assert_eq!( - neighbours(&ms, m, 12, &all_reachable, &none_stable).len(), + neighbours(&RingAndPairs, &ms, m, 12, &all_reachable, &none_stable).len(), 12 ); } @@ -254,7 +261,15 @@ fn members_on_two_transports_bridge_the_others() { let no_stable = |_: &str| false; let mut graph: BTreeMap> = BTreeMap::new(); for (m, own) in &transports { - for n in neighbours_across(&transports, m, own, k, &all_reachable, &no_stable) { + for n in neighbours_across( + &RingAndPairs, + &transports, + m, + own, + k, + &all_reachable, + &no_stable, + ) { assert!( !transports[&n].is_disjoint(own), "{m} and {n} share no transport" @@ -274,8 +289,16 @@ fn members_on_two_transports_bridge_the_others() { ms.iter().map(|m| (m.clone(), on(&[""]))).collect(); for m in &ms[..5] { assert_eq!( - neighbours_across(&one, m, &on(&[""]), k, &all_reachable, &no_stable), - neighbours(&ms, m, k, &all_reachable, &no_stable) + neighbours_across( + &RingAndPairs, + &one, + m, + &on(&[""]), + k, + &all_reachable, + &no_stable + ), + neighbours(&RingAndPairs, &ms, m, k, &all_reachable, &no_stable) ); } } -- cgit v1.3.1