diff options
| -rw-r--r-- | src/swarm/topology.rs | 99 |
1 files changed, 89 insertions, 10 deletions
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index dddf81b..8e25a2a 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -32,6 +32,7 @@ pub(crate) struct NeighbourView { members: Vec<String>, k: usize, unreachable: BTreeSet<String>, + stable: BTreeSet<String>, neighbours: BTreeSet<String>, /// Every reachable member is a neighbour. full_mesh: bool, @@ -74,7 +75,16 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { if !state.is_ready().await { return None; } - let current = members::members(&state.doc.read().await.doc); + let (current, stable) = { + let replica = state.doc.read().await; + let current = members::members(&replica.doc); + let stable: BTreeSet<String> = current + .iter() + .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable) + .cloned() + .collect(); + (current, stable) + }; let k = state.neighbours_k; let now = Instant::now(); let mut unreachable = BTreeSet::new(); @@ -91,7 +101,10 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { .read() .unwrap_or_else(|poisoned| poisoned.into_inner()); if view.as_ref().is_some_and(|view| { - view.members == current && view.k == k && view.unreachable == unreachable + view.members == current + && view.k == k + && view.unreachable == unreachable + && view.stable == stable }) { return None; } @@ -103,7 +116,13 @@ 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, state.local_id(), k, &|m| unreachable.contains(m)); + let computed = neighbours( + ¤t, + state.local_id(), + k, + &|m| unreachable.contains(m), + &|m| stable.contains(m), + ); let change = NeighbourChange { added: computed .iter() @@ -119,6 +138,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { info!( members = current.len(), unreachable = unreachable.len(), + stable = stable.len(), k, neighbours = computed.len(), added = change.added.len(), @@ -137,6 +157,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { members: current, k, unreachable, + stable, neighbours: computed, }); Some(change) @@ -152,16 +173,18 @@ pub(crate) fn neighbours( local: &str, k: usize, unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, ) -> BTreeSet<String> { let mut all: Vec<String> = members.to_vec(); all.push(local.to_owned()); - graph(&all, k, &|m: &str| m != local && unreachable(m)) + graph(&all, k, &|m: &str| m != local && unreachable(m), stable) .remove(local) .unwrap_or_default() } /// Every reachable member's neighbours. Costs about N² hashes (a few milliseconds for -/// hundreds of members). +/// hundreds of members). `stable` members (`swarm::stability`) are preferred for half of +/// each member's scored slots. /// /// Each reachable member walks the ring (by key hash) both ways to the first reachable /// member, which gives a ring over the reachable members, and adds the `k - 2` @@ -174,6 +197,7 @@ pub(crate) fn graph( members: &[String], k: usize, unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, ) -> BTreeMap<String, BTreeSet<String>> { let k = k.max(2); let all: BTreeSet<&str> = members.iter().map(String::as_str).collect(); @@ -226,7 +250,22 @@ pub(crate) fn graph( .map(|other| (pair_score(member, other), other)) .collect(); candidates.sort_unstable(); - for (_, partner) in candidates.into_iter().take(k - 2) { + // Half the slots (rounded up) prefer stable members, so long-running nodes form + // a backbone; the rest go by pair score alone, keeping paths short and the load + // on stable members bounded. + let stable_slots = (k - 2).div_ceil(2); + let preferred: Vec<&str> = candidates + .iter() + .map(|(_, other)| *other) + .filter(|other| stable(other)) + .take(stable_slots) + .collect(); + let rest = candidates + .iter() + .map(|(_, other)| *other) + .filter(|other| !preferred.contains(other)) + .take(k - 2 - preferred.len()); + for partner in preferred.iter().copied().chain(rest) { link(member, partner); } } @@ -264,8 +303,12 @@ mod tests { false } + fn none_stable(_: &str) -> bool { + false + } + fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> { - super::graph(members, k, &all_reachable) + super::graph(members, k, &all_reachable, &none_stable) } fn connected(g: &BTreeMap<String, BTreeSet<String>>) -> bool { @@ -303,7 +346,10 @@ mod tests { fn small_swarms_are_a_full_mesh() { let ms = members(13); for m in &ms { - assert_eq!(neighbours(&ms, m, 12, &all_reachable).len(), 12); + assert_eq!( + neighbours(&ms, m, 12, &all_reachable, &none_stable).len(), + 12 + ); } } @@ -344,14 +390,14 @@ mod tests { // Every tenth member is online. let online: BTreeSet<String> = ms.iter().step_by(10).cloned().collect(); let offline = |m: &str| !online.contains(m); - let g = super::graph(&ms, 12, &offline); + let g = super::graph(&ms, 12, &offline, &none_stable); assert_eq!(g.len(), 10); // Ten reachable members and k = 12: a full mesh among them. assert!(g.values().all(|ns| ns.len() == 9)); let online: BTreeSet<String> = ms.iter().step_by(4).cloned().collect(); let offline = |m: &str| !online.contains(m); - let g = super::graph(&ms, 8, &offline); + let g = super::graph(&ms, 8, &offline, &none_stable); assert_eq!(g.len(), 25); assert!(connected(&g)); assert!( @@ -359,4 +405,37 @@ mod tests { .all(|ns| ns.len() >= 8 && ns.iter().all(|n| online.contains(n))) ); } + + #[test] + fn stable_members_become_hubs_with_bounded_degree() { + let ms = members(250); + let stable: BTreeSet<String> = ms.iter().step_by(12).take(20).cloned().collect(); + let is_stable = |m: &str| stable.contains(m); + let g = super::graph(&ms, 12, &all_reachable, &is_stable); + assert!(connected(&g)); + + let degree = |pred: &dyn Fn(&String) -> bool| -> (usize, f64) { + let degrees: Vec<usize> = g + .iter() + .filter(|(m, _)| pred(m)) + .map(|(_, ns)| ns.len()) + .collect(); + let max = *degrees.iter().max().unwrap(); + ( + max, + degrees.iter().sum::<usize>() as f64 / degrees.len() as f64, + ) + }; + let (stable_max, stable_mean) = degree(&|m| stable.contains(m)); + let (_, transient_mean) = degree(&|m| !stable.contains(m)); + // 230 transient members put about 5 of their 10 scored slots on 20 stable ones. + assert!( + stable_mean > 3.0 * transient_mean, + "{stable_mean} vs {transient_mean}" + ); + assert!( + stable_max <= 100, + "a stable member has {stable_max} neighbours" + ); + } } |