topology trait

1 files changed, 43 insertions, 4 deletions

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,