diff options
| -rw-r--r-- | src/swarm/topology.rs | 132 | ||||
| -rw-r--r-- | src/transport/firewall.rs | 4 | ||||
| -rw-r--r-- | src/transport/identity.rs | 8 | ||||
| -rw-r--r-- | src/transport/mod.rs | 8 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 4 |
5 files changed, 152 insertions, 4 deletions
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index 5150e38..2c6b867 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -16,6 +16,14 @@ //! `k` neighbours. With at most `k + 1` reachable members the graph is the full mesh: //! small swarms behave exactly as before. A member joining, leaving or going offline //! changes only links near it. +//! +//! Members on several transports (Mycelium, libp2p) get such a graph per transport, +//! among the members on it (as their addresses in the member list say), and a node's +//! neighbours are the union of its neighbours on each transport it is on +//! (`neighbours_across`). Each of those graphs is connected (its ring), and a member on +//! two transports is in both: as long as members on several transports link the +//! transports, the swarm stays connected, and such a member has about `k` neighbours per +//! transport. With one transport, it is the one graph as before. use std::collections::{BTreeMap, BTreeSet}; @@ -33,6 +41,9 @@ pub(crate) struct NeighbourView { k: usize, unreachable: BTreeSet<String>, stable: BTreeSet<String>, + /// The transports (address schemes) of each member, and ours. + transports: BTreeMap<String, BTreeSet<String>>, + local_transports: BTreeSet<String>, neighbours: BTreeSet<String>, /// Every reachable member is a neighbour. full_mesh: bool, @@ -132,7 +143,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { if !state.is_ready().await { return None; } - let (current, stable) = { + let (current, stable, transports) = { let replica = state.doc.read().await; members::learn_addresses(state, &replica.doc); let current = members::members(&replica.doc); @@ -141,8 +152,13 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable) .cloned() .collect(); - (current, stable) + let transports: BTreeMap<String, BTreeSet<String>> = current + .iter() + .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m)))) + .collect(); + (current, stable, transports) }; + let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect(); let k = state.neighbours_k; // Those that don't answer, and those on none of our transports. let mut unreachable = BTreeSet::new(); @@ -165,6 +181,8 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { && view.k == k && view.unreachable == unreachable && view.stable == stable + && view.transports == transports + && view.local_transports == local_transports }) { return None; } @@ -176,9 +194,10 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { Some(previous) => previous, None => state.peer_states.read().await.keys().cloned().collect(), }; - let computed = neighbours( - ¤t, + let computed = neighbours_across( + &transports, state.local_id(), + &local_transports, k, &|m| unreachable.contains(m), &|m| stable.contains(m), @@ -215,6 +234,8 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { k, unreachable, stable, + transports, + local_transports, neighbours: computed, }); Some(change) @@ -230,6 +251,43 @@ const PAIR_LABEL: &[u8] = b"continuum/topology-pair/2"; /// graph stays connected and symmetric. const STABLE_FANIN_FACTOR: usize = 4; +/// The transports `addresses` are on: their schemes (`mycelium` of `mycelium:<key>`); +/// the empty one for an address without (a transport whose identities are addresses). +pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> { + addresses + .iter() + .map(|address| { + address + .split_once(':') + .map_or("", |(scheme, _)| scheme) + .to_owned() + }) + .collect() +} + +/// 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( + members: &BTreeMap<String, BTreeSet<String>>, + local: &str, + local_transports: &BTreeSet<String>, + k: usize, + unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, +) -> BTreeSet<String> { + let mut all = BTreeSet::new(); + for transport in local_transports { + let on_it: Vec<String> = members + .iter() + .filter(|(member, transports)| *member != local && transports.contains(transport)) + .map(|(member, _)| member.clone()) + .collect(); + all.extend(neighbours(&on_it, local, k, unreachable, stable)); + } + all +} + /// The neighbours of `local` among `members` (which may or may not contain `local`), /// skipping members for which `unreachable` holds. pub(crate) fn neighbours( @@ -642,6 +700,72 @@ mod tests { ); } + /// Two transports with members on one or the other, and two members on both: every + /// link is over a shared transport, and the two members on both connect the rest. + #[test] + fn members_on_two_transports_bridge_the_others() { + let k = 8; + let ms = members(201); + let on = |names: &[&str]| -> BTreeSet<String> { + names.iter().map(|name| (*name).to_owned()).collect() + }; + let transports: BTreeMap<String, BTreeSet<String>> = ms + .iter() + .enumerate() + .map(|(i, m)| { + let transports = match i { + 0 | 1 => on(&["a", "b"]), + i if i % 2 == 0 => on(&["a"]), + _ => on(&["b"]), + }; + (m.clone(), transports) + }) + .collect(); + 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) { + assert!( + !transports[&n].is_disjoint(own), + "{m} and {n} share no transport" + ); + graph.entry(m.clone()).or_default().insert(n.clone()); + graph.entry(n).or_default().insert(m.clone()); + } + } + assert_eq!(graph.len(), ms.len()); + assert!(connected(&graph)); + let bridge = &ms[0]; + assert!(graph[bridge].iter().any(|n| !transports[n].contains("b"))); + assert!(graph[bridge].iter().any(|n| !transports[n].contains("a"))); + + // One transport: the one graph, as without transports. + let one: BTreeMap<String, BTreeSet<String>> = + 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) + ); + } + } + + #[test] + fn schemes_of_addresses() { + let addresses = + |all: &[&str]| -> Vec<String> { all.iter().map(|a| (*a).to_owned()).collect() }; + assert_eq!( + schemes(&addresses(&[ + "mycelium:ab", + "libp2p:/ip4/1.2.3.4/tcp/1", + "libp2p:/ip6/::1/tcp/1" + ])), + BTreeSet::from(["libp2p".to_owned(), "mycelium".to_owned()]) + ); + assert_eq!(schemes(&addresses(&["a"])), BTreeSet::from([String::new()])); + assert!(schemes(&[]).is_empty()); + } + #[test] fn hub_degree_is_bounded_however_few_stable_members() { for (n, stable_count) in [(250, 10), (400, 5)] { diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs index e6816d6..78d00fc 100644 --- a/src/transport/firewall.rs +++ b/src/transport/firewall.rs @@ -92,6 +92,10 @@ impl Transport for FirewallTransport { self.inner.discovered_seeds() } + fn transports(&self) -> Vec<String> { + self.inner.transports() + } + async fn resolve_seed(&self, seed: &str) -> Result<String> { self.inner.resolve_seed(seed).await } diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 04375f6..af0fe74 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -504,6 +504,14 @@ impl Transport for IdentityTransport { self.shared.destination(peer).is_some() } + fn transports(&self) -> Vec<String> { + self.shared + .links + .iter() + .map(|link| link.scheme.to_owned()) + .collect() + } + /// Each transport's, with its scheme. fn discovered_seeds(&self) -> Vec<String> { self.shared diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 6ae3814..3f77306 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -555,6 +555,14 @@ pub(crate) trait Transport: Send + Sync { vec![peer.to_owned()] } + /// The transports this node is on, as the schemes of their addresses + /// (`swarm::topology`), whether or not it has an address on each (one that only + /// connects out has none). By default the empty scheme, for a transport whose + /// identities are addresses. + fn transports(&self) -> Vec<String> { + vec![String::new()] + } + /// Seeds the transport found by itself since last asked (libp2p: nodes on the LAN), /// for a node that is joining. By default none. fn discovered_seeds(&self) -> Vec<String> { diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 83b0cff..b332d3d 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -169,6 +169,10 @@ impl Transport for CountingTransport { self.inner.discovered_seeds() } + fn transports(&self) -> Vec<String> { + self.inner.transports() + } + async fn resolve_seed(&self, seed: &str) -> Result<String> { self.inner.resolve_seed(seed).await } |