topology trait

3 files changed, 96 insertions, 11 deletions

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<String> {
 /// `local_transports`: on each, its neighbours among the members on it
 /// (`neighbours`), all together.
 pub(crate) fn neighbours_across(
+    topology: &dyn Topology,
     members: &BTreeMap<String, BTreeSet<String>>,
     local: &str,
     local_transports: &BTreeSet<String>,
@@ -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<String> {
     let mut all: Vec<String> = 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<String, BTreeSet<String>> {
+        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<String, BTreeSet<String>>;
+}
 
 /// 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<BTreeMap<String, BTreeS
     let graph = view
         .members
         .iter()
-        .map(|member| (member.clone(), view.neighbours_of(state.local_id(), member)))
+        .map(|member| {
+            (
+                member.clone(),
+                view.neighbours_of(&*state.topology, state.local_id(), member),
+            )
+        })
         .collect();
     Some(graph)
 }
@@ -316,12 +347,18 @@ pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String>
 }
 
 impl NeighbourView {
-    fn neighbours_of(&self, local: &str, member: &str) -> BTreeSet<String> {
+    fn neighbours_of(
+        &self,
+        topology: &dyn Topology,
+        local: &str,
+        member: &str,
+    ) -> BTreeSet<String> {
         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<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");
     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<NeighbourChange> {
         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<String> = 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<String, BTreeSet<String>> = 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)
         );
     }
 }