diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 9 | ||||
| -rw-r--r-- | src/swarm/reach.rs | 61 | ||||
| -rw-r--r-- | src/swarm/rtt.rs | 63 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 183 |
4 files changed, 251 insertions, 65 deletions
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index aa34e70..44f2879 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -41,6 +41,7 @@ pub(crate) async fn announce_heads(state: &AppState, peer: &str, with_reaches: b let status = transport::Status { announcing: true, reaches: local.reaches.filter(|_| with_reaches), + round_trips: local.round_trips.filter(|_| with_reaches), ..local }; state @@ -189,6 +190,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta Some(peer_state) => peer_state.lock().await.syncs_on(epoch), None => false, }; + let (reaches, round_trips) = crate::swarm::reach::published_with_round_trips(state); transport::Status { heads, epoch, @@ -197,7 +199,8 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta syncs_with_you, leaving: state.is_closing(), announcing: false, - reaches: Some(crate::swarm::reach::published(state)), + reaches: Some(reaches), + round_trips: Some(round_trips), } } @@ -236,6 +239,10 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport peer_state.last_heads = Some((remote.epoch, remote.heads.clone())); if let Some(reaches) = &remote.reaches { peer_state.reaches = Some((Instant::now(), reaches.clone())); + peer_state.round_trips = remote + .round_trips + .clone() + .filter(|round_trips| round_trips.len() == reaches.len()); } drop(peer_state); if left { diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs index 8c22ef6..e6cee2f 100644 --- a/src/swarm/reach.rs +++ b/src/swarm/reach.rs @@ -7,7 +7,8 @@ //! `Timing::unreached_after`, asked again halfway, with nothing else heard from it; a //! member gone shows within seconds, not after `dormant_after`). It tells its //! neighbours so in its STATUS (`Status::reaches`, each member by its pair key, -//! `topology::pair_key`, 8 bytes instead of a node ID): in every keepalive STATUS and +//! `topology::pair_key`, 8 bytes instead of a node ID, with its round trip to each as a +//! bucket: `Status::round_trips`, `rtt::bucket`): in every keepalive STATUS and //! STATUS_REPLY, and soon to every neighbour when it changed (the peer worker looks //! every 5 s, and at once when a member we reach stops answering), but not in //! announcements of heads, which go often. So every node knows whom its neighbours @@ -32,17 +33,20 @@ use tracing::debug; use crate::locks::MutexExt; use crate::state::{AppState, HelloState, Standing}; use crate::swarm::peer::announce_heads; -use crate::swarm::{members, topology}; +use crate::swarm::{members, rtt, topology}; -/// Whom we reach, as last told our neighbours. +/// Whom we reach, as last told our neighbours, and our round trip to each. #[derive(Default)] pub(crate) struct Reach { published: Option<Vec<u64>>, + /// One per member in `published`, in its order (`rtt::bucket`). + round_trips: Vec<u8>, } /// The pair keys of the members we reach now, sorted: those with an active session -/// that answer. -async fn measure(state: &AppState) -> Vec<u64> { +/// that answer; and our round trip to each, as a bucket, held near a bound to the one +/// told before (`before`, by pair key: `rtt::bucket`). +async fn measure(state: &AppState, before: &HashMap<u64, u8>) -> (Vec<u64>, Vec<u8>) { let now = Instant::now(); let mut reached = Vec::new(); for (peer, peer_state) in state.peers_snapshot().await { @@ -53,21 +57,37 @@ async fn measure(state: &AppState) -> Vec<u64> { && !peer_state.answer_overdue(now) && matches!(peer_state.standing, Standing::Member { .. }) { - reached.push(topology::pair_key(&peer)); + reached.push((topology::pair_key(&peer), peer)); } } reached.sort_unstable(); - reached + let rtts = state.lock_rtt(); + let round_trips = reached + .iter() + .map(|(key, peer)| { + let rtt = rtts.effective(peer, now, state.timing.rtt_fresh_for()); + rtt::bucket(rtt, before.get(key).copied()) + }) + .collect(); + ( + reached.into_iter().map(|(key, _)| key).collect(), + round_trips, + ) } /// Whom we reach, as we tell it (`Status::reaches`); empty until first measured. pub(crate) fn published(state: &AppState) -> Vec<u64> { - state - .reach - .lock_or_recover() - .published - .clone() - .unwrap_or_default() + published_with_round_trips(state).0 +} + +/// Whom we reach and our round trip to each, as we tell it (`Status::reaches`, +/// `Status::round_trips`); empty until first measured. +pub(crate) fn published_with_round_trips(state: &AppState) -> (Vec<u64>, Vec<u8>) { + let reach = state.reach.lock_or_recover(); + match &reach.published { + Some(keys) => (keys.clone(), reach.round_trips.clone()), + None => (Vec::new(), Vec::new()), + } } /// With the peer worker's periodic pass (every 5 s), and at once when a member we said @@ -75,13 +95,24 @@ pub(crate) fn published(state: &AppState) -> Vec<u64> { /// our neighbours are told at once (a STATUS each, unanswered unless they lack some of /// our heads). A change that is undone within a pass isn't told. pub(crate) async fn check(state: &AppState) { - let now = measure(state).await; + let before: HashMap<u64, u8> = { + let reach = state.reach.lock_or_recover(); + reach + .published + .iter() + .flatten() + .copied() + .zip(reach.round_trips.iter().copied()) + .collect() + }; + let (now, round_trips) = measure(state, &before).await; { let mut reach = state.reach.lock_or_recover(); - if reach.published.as_ref() == Some(&now) { + if reach.published.as_ref() == Some(&now) && reach.round_trips == round_trips { return; } reach.published = Some(now.clone()); + reach.round_trips = round_trips; } debug!( reached = now.len(), diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs index cf91ac2..370754c 100644 --- a/src/swarm/rtt.rs +++ b/src/swarm/rtt.rs @@ -108,6 +108,51 @@ impl RttTable { } } +/// Upper bounds of the round-trip buckets in which members tell how far those they +/// reach are (`swarm::reach`), in ms; the last bucket is above them all. Doubling: a +/// writer's tree only needs to tell which of a member's possible parents is clearly the +/// faster (`swarm::tree`), and a coarse value changes seldom, and with it what is told. +const BUCKET_BOUNDS_MS: [u64; 6] = [25, 50, 100, 200, 400, 800]; + +/// The bucket of a link without a round trip measured: in the middle, neither fast +/// nor slow. +pub(crate) const UNKNOWN_BUCKET: u8 = 3; + +/// How far past a bound a round trip goes before its bucket changes: a round trip close +/// to a bound would otherwise move between two buckets, and every move is told to every +/// neighbour and may change who passes changes to whom. +const BUCKET_HYSTERESIS: f64 = 0.2; + +/// The round-trip bucket of a link (`BUCKET_BOUNDS_MS`): `rtt`'s, unless it is still +/// within `BUCKET_HYSTERESIS` of the bucket `before` (as last told). Without a round +/// trip, as before; never measured, `UNKNOWN_BUCKET`. +pub(crate) fn bucket(rtt: Option<Duration>, before: Option<u8>) -> u8 { + let Some(rtt) = rtt else { + return before.unwrap_or(UNKNOWN_BUCKET); + }; + let ms = rtt.as_secs_f64() * 1000.0; + let plain = BUCKET_BOUNDS_MS + .iter() + .take_while(|bound| ms >= **bound as f64) + .count() as u8; + let Some(before) = before.filter(|before| *before != plain) else { + return plain; + }; + let index = usize::from(before); + let lower = index + .checked_sub(1) + .and_then(|below| BUCKET_BOUNDS_MS.get(below)) + .map_or(0.0, |bound| *bound as f64 * (1.0 - BUCKET_HYSTERESIS)); + let upper = BUCKET_BOUNDS_MS.get(index).map_or(f64::INFINITY, |bound| { + *bound as f64 * (1.0 + BUCKET_HYSTERESIS) + }); + if (lower..upper).contains(&ms) { + before + } else { + plain + } +} + #[cfg(test)] mod tests { use super::*; @@ -169,4 +214,22 @@ mod tests { let stale = start + FRESH + Duration::from_secs(1); assert_eq!(table.effective("a", stale, FRESH), None); } + + #[test] + fn buckets_double_and_hold_near_their_bounds() { + let ms = |ms| Some(Duration::from_millis(ms)); + assert_eq!(bucket(ms(10), None), 0); + assert_eq!(bucket(ms(25), None), 1); + assert_eq!(bucket(ms(150), None), 3); + assert_eq!(bucket(ms(5000), None), 6); + assert_eq!(bucket(None, None), UNKNOWN_BUCKET); + assert_eq!(bucket(None, Some(1)), 1, "unmeasured now: as before"); + // Just past a bound: still the bucket told before. + assert_eq!(bucket(ms(110), Some(2)), 2); + assert_eq!(bucket(ms(45), Some(2)), 2); + // Well past it: the new one. + assert_eq!(bucket(ms(130), Some(2)), 3); + assert_eq!(bucket(ms(35), Some(2)), 1); + assert_eq!(bucket(ms(2000), Some(6)), 6); + } } diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs index c84056c..f04e442 100644 --- a/src/swarm/tree.rs +++ b/src/swarm/tree.rs @@ -7,7 +7,8 @@ //! (`swarm::reach`). Every member's distance from the writer over that graph orders the //! members: by distance, then whether it reaches a member a hop closer, then pair key; //! a member's parent is the neighbour it reaches that comes before it, the closest to -//! the writer first, then one that reaches closer, then by pair score +//! the writer first, then one that reaches closer, then the one it has the shortest +//! round trip to (as it said, a bucket: `rtt::bucket`), then by pair score //! (`topology::pair_score`). Coming before rules out cycles, and taking only a neighbour //! it reaches routes around links that don't work: where a writer behind NAT reaches a //! single member, that one comes first and the others take it. @@ -32,7 +33,7 @@ use tracing::trace; use crate::locks::MutexExt; use crate::state::AppState; -use crate::swarm::{reach, topology}; +use crate::swarm::{reach, rtt, topology}; /// The writers whose distances are kept at most. const MAX_WRITERS: usize = 256; @@ -94,7 +95,7 @@ fn distances(state: &AppState, writer: u64) -> Option<(Arc<Graph>, Arc<HashMap<S /// Members are ordered by distance from the writer, then whether they reach a member a /// hop closer (as they say: else, while it hasn't the change from its own side, it is /// as good as farther), then pair key; a member's parent is the neighbour it reaches -/// that comes before it, by the same, then pair score. So where the writer reaches few +/// that comes before it, by the same, then round trip, then pair score. So where the writer reaches few /// (behind NAT), those it reaches come first among the members as far from it, and the /// others take them as parents. What a member reaches we know of ourselves and our /// neighbours, as they said; of others, the graph is taken as right (if we're wrong, @@ -127,12 +128,15 @@ fn rank( } /// `member`'s parent in the writer's tree (`rank`): the neighbour it reaches that comes -/// before it, the closest to the writer first, then one that reaches closer, then by -/// pair score; None if none does (or it is the writer). +/// before it, the closest to the writer first, then one that reaches closer, then the +/// one it has the shortest round trip to, as it said (`round_trip(member, candidate)`, +/// a bucket: `rtt::bucket`), then by pair score; None if none does (or it is the +/// writer). fn parent<'g>( graph: &'g Graph, distance: &HashMap<String, u32>, reached: &dyn Fn(&str, &str) -> bool, + round_trip: &dyn Fn(&str, &str) -> u8, member: &str, ) -> Option<&'g str> { let its_rank = rank(graph, distance, reached, member)?; @@ -147,19 +151,25 @@ fn parent<'g>( }) .min_by_key(|candidate| { let (d, far, _) = rank(graph, distance, reached, candidate).expect("ranked"); - (d, far, topology::pair_score(candidate, member)) + ( + d, + far, + round_trip(member, candidate), + topology::pair_score(candidate, member), + ) }) } -/// `writer`'s whole tree, as far as `reached` knows whom each member reaches (the -/// dashboard's map, from the links members report): every member's parent, None for -/// those without one (they get the writer's changes by announcement). None while the -/// member list isn't known, or the writer isn't in it. +/// `writer`'s whole tree, as far as `reached` knows whom each member reaches and +/// `round_trip` how far (the dashboard's map, from the links members report): every +/// member's parent, None for those without one (they get the writer's changes by +/// announcement). None while the member list isn't known, or the writer isn't in it. #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) fn whole( state: &AppState, writer: &str, reached: &dyn Fn(&str, &str) -> bool, + round_trip: &dyn Fn(&str, &str) -> u8, ) -> Option<BTreeMap<String, Option<String>>> { let (graph, distance) = distances(state, topology::pair_key(writer))?; Some( @@ -167,29 +177,70 @@ pub(crate) fn whole( .keys() .filter(|member| distance.get(member.as_str()).is_some_and(|d| *d > 0)) .map(|member| { - let parent = parent(&graph, &distance, reached, member).map(str::to_owned); + let parent = + parent(&graph, &distance, reached, round_trip, member).map(str::to_owned); (member.clone(), parent) }) .collect(), ) } -/// Whom we and our neighbours in `graph` reach, by pair key, as far as we know (our -/// own, and what our neighbours said). -async fn known_reaches<'g>(state: &AppState, graph: &'g Graph) -> HashMap<&'g str, Vec<u64>> { +/// Whom a member reaches, by pair key, and its round trip to each (`rtt::bucket`), as +/// it said. +struct Said { + keys: Vec<u64>, + round_trips: Option<Vec<u8>>, +} + +/// Whom we and our neighbours in `graph` reach, and how far, as far as we know (our +/// own, and what our neighbours said). Of other members, nothing: the graph is taken as +/// right, and their round trips as alike. +struct Known<'g>(HashMap<&'g str, Said>); + +impl Known<'_> { + /// Whether `by` reaches `member`: as it said, or by the graph. + fn reached(&self, by: &str, member: &str) -> bool { + self.0 + .get(by) + .is_none_or(|said| said.keys.contains(&topology::pair_key(member))) + } + + /// `by`'s round trip to `member`, as a bucket; `rtt::UNKNOWN_BUCKET` if not said. + fn round_trip(&self, by: &str, member: &str) -> u8 { + let key = topology::pair_key(member); + self.0 + .get(by) + .and_then(|said| { + let index = said.keys.iter().position(|said| *said == key)?; + said.round_trips.as_ref()?.get(index).copied() + }) + .unwrap_or(rtt::UNKNOWN_BUCKET) + } +} + +async fn known_reaches<'g>(state: &AppState, graph: &'g Graph) -> Known<'g> { let me = state.local_id(); - let mut reaches: HashMap<&str, Vec<u64>> = HashMap::new(); + let mut reaches: HashMap<&str, Said> = HashMap::new(); if let Some((me, _)) = graph.get_key_value(me) { - reaches.insert(me.as_str(), reach::published(state)); + let (keys, round_trips) = reach::published_with_round_trips(state); + reaches.insert( + me.as_str(), + Said { + keys, + round_trips: Some(round_trips), + }, + ); } for member in graph.get(me).into_iter().flatten() { - if let Some(peer_state) = state.get_peer_state(member).await - && let Some(keys) = reach::of(peer_state.lock().await.reaches.as_ref()) - { - reaches.insert(member.as_str(), keys); + if let Some(peer_state) = state.get_peer_state(member).await { + let peer_state = peer_state.lock().await; + if let Some(keys) = reach::of(peer_state.reaches.as_ref()) { + let round_trips = peer_state.round_trips.clone(); + reaches.insert(member.as_str(), Said { keys, round_trips }); + } } } - reaches + Known(reaches) } /// Our own tree, as our writes go along it (`Plan`, as every member computes its part). @@ -213,13 +264,10 @@ pub(crate) struct Shape { pub(crate) async fn own_shape(state: &AppState) -> Option<Shape> { let me = state.local_id(); let (graph, distance) = distances(state, topology::pair_key(me))?; - let reaches = known_reaches(state, &graph).await; - let reached = |by: &str, member: &str| { - reaches - .get(by) - .is_none_or(|keys| keys.contains(&topology::pair_key(member))) - }; - Some(shape(&graph, &distance, &reached, me)) + let known = known_reaches(state, &graph).await; + let reached = |by: &str, member: &str| known.reached(by, member); + let round_trip = |by: &str, member: &str| known.round_trip(by, member); + Some(shape(&graph, &distance, &reached, &round_trip, me)) } /// `own_shape` over `graph`, from `me` at distance 0. @@ -227,12 +275,16 @@ fn shape( graph: &Graph, distance: &HashMap<String, u32>, reached: &dyn Fn(&str, &str) -> bool, + round_trip: &dyn Fn(&str, &str) -> u8, me: &str, ) -> Shape { let parents: HashMap<&str, Option<&str>> = graph .keys() .filter(|member| distance.get(member.as_str()).is_some_and(|d| *d > 0)) - .map(|member| (member.as_str(), parent(graph, distance, reached, member))) + .map(|member| { + let parent = parent(graph, distance, reached, round_trip, member); + (member.as_str(), parent) + }) .collect(); // Hops from us along parents; None where the chain ends without us. Parents come // before their children, so chains end. @@ -275,9 +327,9 @@ fn shape( /// How long our writes take to reach the members along our own tree, by link latency /// alone (the bench's, which knows each link's): the mean over the members it reaches -/// (and the others do), against two other trees of ours. In one, each member's parent is chosen as now, but -/// among those as close to us and as well placed (`rank`) by latency before pair -/// score; in the other, every member takes the fastest path over the links members +/// (and the others do), against two other trees of ours. In one, each member's parent is +/// chosen as now, but among those as close to us and as well placed (`rank`) by the +/// exact latency, not the round-trip bucket it said; in the other, every member takes the fastest path over the links members /// reach (no tree computed alike everywhere gets below it). In milliseconds, one way. #[cfg(test)] #[derive(Debug, Clone, Copy)] @@ -297,12 +349,9 @@ pub(crate) async fn own_latencies( ) -> Option<Latencies> { let me = state.local_id(); let (graph, distance) = distances(state, topology::pair_key(me))?; - let reaches = known_reaches(state, &graph).await; - let reached = |by: &str, member: &str| { - reaches - .get(by) - .is_none_or(|keys| keys.contains(&topology::pair_key(member))) - }; + let known = known_reaches(state, &graph).await; + let reached = |by: &str, member: &str| known.reached(by, member); + let round_trip = |by: &str, member: &str| known.round_trip(by, member); let members: Vec<&str> = graph .keys() .map(String::as_str) @@ -310,7 +359,12 @@ pub(crate) async fn own_latencies( .collect(); let now: HashMap<&str, Option<&str>> = members .iter() - .map(|member| (*member, parent(&graph, &distance, &reached, member))) + .map(|member| { + ( + *member, + parent(&graph, &distance, &reached, &round_trip, member), + ) + }) .collect(); let faster: HashMap<&str, Option<&str>> = members .iter() @@ -428,12 +482,9 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> { if !distance.contains_key(me) { return None; } - let reaches = known_reaches(state, &graph).await; - let reached = |by: &str, member: &str| { - reaches - .get(by) - .is_none_or(|keys| keys.contains(&topology::pair_key(member))) - }; + let known = known_reaches(state, &graph).await; + let reached = |by: &str, member: &str| known.reached(by, member); + let round_trip = |by: &str, member: &str| known.round_trip(by, member); let my_rank = rank(&graph, &distance, &reached, me)?; let my_key = topology::pair_key(me); let mut children = Vec::new(); @@ -448,14 +499,15 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> { covered.push(member.clone()); continue; } - match parent(&graph, &distance, &reached, member) { + match parent(&graph, &distance, &reached, &round_trip, member) { None => uncovered.push(member.clone()), Some(parent) if parent == me && its_rank > my_rank && topology::reaches(state, member).await => { - if reaches + if known + .0 .get(member.as_str()) - .is_some_and(|keys| keys.iter().all(|key| *key == my_key)) + .is_some_and(|said| said.keys.iter().all(|key| *key == my_key)) { alone.push(member.clone()); } @@ -549,6 +601,33 @@ pub(crate) async fn forward( mod tests { use super::*; + /// Of the members as close to the writer and as well placed, the one a member has + /// the shorter round trip to is its parent, whatever the pair scores. + #[test] + fn the_nearer_parent_wins() { + let mut graph = Graph::new(); + for (a, b) in [("w", "a"), ("w", "b"), ("a", "c"), ("b", "c")] { + graph.entry(a.into()).or_default().insert(b.into()); + graph.entry(b.into()).or_default().insert(a.into()); + } + let distance: HashMap<String, u32> = [("w", 0), ("a", 1), ("b", 1), ("c", 2)] + .into_iter() + .map(|(member, d)| (member.to_owned(), d)) + .collect(); + let reached = |_: &str, _: &str| true; + for (near, far) in [("a", "b"), ("b", "a")] { + let round_trip = |by: &str, member: &str| match (by, member) { + ("c", m) if m == near => 1, + ("c", m) if m == far => 5, + _ => rtt::UNKNOWN_BUCKET, + }; + assert_eq!( + parent(&graph, &distance, &reached, &round_trip, "c"), + Some(near) + ); + } + } + /// Hops along parents from us; members without a parent, below one without, or /// not connected count as off the tree. #[test] @@ -570,7 +649,13 @@ mod tests { // c doesn't reach us: no parent, nor e below it. let reached = |by: &str, member: &str| !(by == "c" && member == "me"); assert_eq!( - shape(&graph, &distance, &reached, "me"), + shape( + &graph, + &distance, + &reached, + &|_, _| rtt::UNKNOWN_BUCKET, + "me" + ), Shape { members: 2, depth: 2, |