From 24327be7e09c20a9820eac61ba3ac60fb7bdb424 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 02:18:01 +0100 Subject: measure paths --- src/transport/paths.rs | 568 ++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 517 insertions(+), 51 deletions(-) (limited to 'src/transport/paths.rs') diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 3330b2d..30aab8d 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -11,10 +11,19 @@ //! as configured (`--transports`). //! //! A message to a node with several addresses goes to the first of them that takes it -//! (`send_first`), in the order `rank` gives: addresses whose last send failed, for -//! `DOWN_FOR`, after the others, and among each the configured order. A send that fails -//! is tried at once at the next address, the message encoded for its transport: when -//! one transport fails, the next carries on within the same send. +//! (`send_first`), in the order `rank` gives. A send that fails is tried at once at the +//! next address, the message encoded for its transport: when one transport fails, the +//! next carries on within the same send. +//! +//! Paths measures the addresses of such nodes itself: every `PROBE_EVERY`, each address +//! it had a choice of within `ACTIVE_FOR` gets a PING with a random nonce, which the +//! other side answers with a PONG at once, over the same path. Answers give each +//! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a +//! send that failed within `DOWN_FOR`, make a path down. `rank` puts paths that are up +//! first, the one in use first among them, and the rest by round-trip time; it changes +//! the one in use when another is faster by `SWITCH_GAIN` (a quarter) and at least +//! `SWITCH_MIN` (both measured), so it doesn't flap. Until both are measured, the +//! configured order decides. PING and PONG end here: the identity layer never sees them. use std::collections::HashMap; use std::sync::{Arc, Mutex}; @@ -22,12 +31,12 @@ use std::time::Duration; use anyhow::{Context, Result, anyhow}; use tokio::time::Instant; -use tracing::debug; +use tracing::{debug, info}; use crate::locks::MutexExt; - -use crate::transport::stats::TransportStats; -use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport}; +use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats}; +use crate::transport::traffic::Traffic; +use crate::transport::{Connection, EncodedMessage, MessageKind, ParsedMessage, Transport}; /// Called with the sender's address, the message and its `payload_len`. pub(crate) type PathHandler = dyn Fn(String, ParsedMessage, usize) -> std::pin::Pin> + Send>> @@ -56,27 +65,259 @@ pub(crate) fn scheme_of(address: &str) -> &str { address.split_once(':').map_or("", |(scheme, _)| scheme) } -/// An address whose send failed is ranked after the others for this long (unless a +/// A path whose send failed is down for this long (unless it answers a PING, or a /// send there works again). const DOWN_FOR: Duration = Duration::from_secs(60); +/// How often a path in use is pinged, how long an answer may take, and how many PINGs +/// in a row unanswered make it down. +const PROBE_EVERY: Duration = Duration::from_secs(30); +const PROBE_TIMEOUT: Duration = Duration::from_secs(15); +const UNANSWERED_DOWN: u32 = 3; +/// How often the probe task looks for paths due. +const PROBE_TICK: Duration = Duration::from_secs(5); +/// Paths this node had a choice of within this long are probed; ones it didn't use for +/// twice as long are forgotten. +const ACTIVE_FOR: Duration = Duration::from_secs(300); +/// The path in use is changed for one this much faster (a fraction of its round-trip +/// time), and at least `SWITCH_MIN`. +const SWITCH_GAIN: f64 = 0.25; +const SWITCH_MIN: Duration = Duration::from_millis(5); +/// Paths kept at most: they are of peers this node chose among, but bounded anyway. +const MAX_PATHS: usize = 4096; + +/// What paths knows about one path (an address as paths keep it). +#[derive(Debug, Default)] +struct PathState { + /// When it was last among several to choose from (`rank`). + used: Option, + /// Smoothed round-trip time of its PINGs. + rtt: Option, + /// The PING out, by nonce, and when it went. + probe: Option<(u64, Instant)>, + last_probe: Option, + /// PINGs in a row unanswered. + unanswered: u32, + /// When a send there last failed. + failed_at: Option, +} + +impl PathState { + fn is_down(&self, now: Instant) -> bool { + self.unanswered >= UNANSWERED_DOWN + || self + .failed_at + .is_some_and(|at| now.saturating_duration_since(at) < DOWN_FOR) + } +} + +/// Every path's state, and which one is in use for each set to choose from. +#[derive(Debug, Default)] +struct Table { + paths: HashMap, + /// By the set's addresses (sorted, joined): the one in use. + chosen: HashMap, + counts: PathCounts, +} + +impl Table { + /// `usable`, with each one's place in the configured order, best first (as + /// `Paths::rank` says), noting the choice. + fn rank(&mut self, mut usable: Vec<(String, usize)>, now: Instant) -> Vec { + usable.sort_by_key(|(_, preference)| *preference); + usable.dedup_by(|a, b| a.0 == b.0); + if usable.len() < 2 { + return usable.into_iter().map(|(address, _)| address).collect(); + } + if self.paths.len() >= MAX_PATHS { + self.forget_unused(now); + } + for (address, _) in &usable { + if self.paths.len() < MAX_PATHS || self.paths.contains_key(address) { + self.paths.entry(address.clone()).or_default().used = Some(now); + } + } + let rtt = |table: &Self, address: &str| table.paths.get(address).and_then(|path| path.rtt); + let (up, down): (Vec<_>, Vec<_>) = usable.into_iter().partition(|(address, _)| { + !self + .paths + .get(address) + .is_some_and(|path| path.is_down(now)) + }); + let mut set: Vec<&str> = up + .iter() + .chain(down.iter()) + .map(|(address, _)| address.as_str()) + .collect(); + set.sort_unstable(); + let key = set.join(" "); + let mut ranked: Vec = Vec::new(); + if !up.is_empty() { + let mut current = self + .chosen + .get(&key) + .filter(|chosen| up.iter().any(|(address, _)| address == *chosen)) + .cloned() + .unwrap_or_else(|| up[0].0.clone()); + let fastest = up + .iter() + .filter_map(|(address, _)| Some((address, rtt(self, address)?))) + .min_by_key(|(_, rtt)| *rtt); + if let (Some((fastest, fastest_rtt)), Some(current_rtt)) = + (fastest, rtt(self, ¤t)) + && *fastest != current + && current_rtt.saturating_sub(fastest_rtt) + >= SWITCH_MIN.max(current_rtt.mul_f64(SWITCH_GAIN)) + { + info!( + from = %current, + to = %fastest, + from_rtt = ?current_rtt, + to_rtt = ?fastest_rtt, + "Switching to a faster path" + ); + self.counts.switches += 1; + current = fastest.clone(); + } + if self.chosen.len() < MAX_PATHS || self.chosen.contains_key(&key) { + self.chosen.insert(key, current.clone()); + } + // The rest: measured ones by round-trip time, then unmeasured ones in the + // configured order. + let mut rest: Vec<&(String, usize)> = up + .iter() + .filter(|(address, _)| *address != current) + .collect(); + rest.sort_by_key(|(address, preference)| { + ( + rtt(self, address).is_none(), + rtt(self, address), + *preference, + ) + }); + ranked.push(current); + ranked.extend(rest.into_iter().map(|(address, _)| address.clone())); + } + ranked.extend(down.into_iter().map(|(address, _)| address)); + ranked + } + + /// A send to `address` failed. + fn failed(&mut self, address: &str, now: Instant) { + self.counts.send_failures += 1; + if let Some(path) = self.paths.get_mut(address) { + path.failed_at = Some(now); + } + } + + /// A send to `address` worked. + fn sent(&mut self, address: &str) { + if let Some(path) = self.paths.get_mut(address) { + path.failed_at = None; + } + } + + /// The PINGs due now, with their nonces (from `nonce`): to every path used within + /// `ACTIVE_FOR` that wasn't pinged within `PROBE_EVERY` and has no PING out. A PING + /// out for `PROBE_TIMEOUT` counts as unanswered. + fn due(&mut self, now: Instant, mut nonce: impl FnMut() -> u64) -> Vec<(String, u64)> { + self.forget_unused(now); + let mut due = Vec::new(); + for (address, path) in &mut self.paths { + if let Some((_, sent)) = path.probe + && now.saturating_duration_since(sent) >= PROBE_TIMEOUT + { + path.probe = None; + path.unanswered += 1; + self.counts.probes_unanswered += 1; + if path.unanswered == UNANSWERED_DOWN { + info!(address = %address, "A path stopped answering PINGs; using others"); + } + } + let active = path + .used + .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR); + let rested = path + .last_probe + .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY); + if active && rested && path.probe.is_none() { + let nonce = nonce(); + path.probe = Some((nonce, now)); + path.last_probe = Some(now); + self.counts.probes_sent += 1; + due.push((address.clone(), nonce)); + } + } + due + } + + /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out. + fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option { + let path = self.paths.get_mut(address)?; + let (expected, sent) = path.probe?; + if expected != nonce { + return None; + } + let sample = now.saturating_duration_since(sent); + path.probe = None; + if path.unanswered >= UNANSWERED_DOWN { + info!(address = %address, "A path answers PINGs again"); + } + path.unanswered = 0; + path.failed_at = None; + path.rtt = Some(match path.rtt { + Some(rtt) => (rtt * 3 + sample) / 4, + None => sample, + }); + self.counts.probes_answered += 1; + path.rtt + } + + /// Forgets the paths not used for twice `ACTIVE_FOR`, and choices among them. + fn forget_unused(&mut self, now: Instant) { + self.paths.retain(|_, path| { + path.used + .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR * 2) + }); + let paths = &self.paths; + self.chosen + .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address))); + } + + fn info(&self, address: &str, now: Instant) -> PathInfo { + let path = self.paths.get(address); + PathInfo { + address: address.to_owned(), + rtt_ms: path + .and_then(|path| path.rtt) + .map(|rtt| rtt.as_secs_f64() * 1000.0), + down: path.is_some_and(|path| path.is_down(now)), + } + } +} pub(crate) struct Paths { /// In order of preference. links: Vec, - /// Addresses whose last send failed, and when (`DOWN_FOR`). A std mutex: never held - /// across an await. - down: Mutex>, + /// PINGs and PONGs are counted here, under the address (`traffic`). + traffic: Option>, + /// A std mutex: never held across an await. + table: Mutex, } impl Paths { - /// Over `transports`, by scheme, in order of preference. - pub(crate) fn new(transports: Vec<(&'static str, Arc)>) -> Self { + /// Over `transports`, by scheme, in order of preference; PINGs and PONGs counted in + /// `traffic`. + pub(crate) fn new( + transports: Vec<(&'static str, Arc)>, + traffic: Option>, + ) -> Self { Self { links: transports .into_iter() .map(|(scheme, inner)| Link { scheme, inner }) .collect(), - down: Mutex::default(), + traffic, + table: Mutex::default(), } } @@ -123,26 +364,25 @@ impl Paths { .unwrap_or(usize::MAX) } - /// The usable ones of `addresses` (as paths keep them), best first: those whose - /// last send failed (within `DOWN_FOR`) last, and by the configured order. + /// The usable ones of `addresses` (as paths keep them), best first: paths that are + /// up before those down, the one in use first, then by round-trip time, then in + /// the configured order (module docs). Several addresses are paths to measure. pub(crate) fn rank(&self, addresses: &[String]) -> Vec { - let mut usable: Vec<(bool, usize, &String)> = addresses + let usable: Vec<(String, usize)> = addresses .iter() .filter(|address| self.usable(address)) - .map(|address| (false, self.preference(address), address)) + .map(|address| (address.clone(), self.preference(address))) .collect(); - { - let now = Instant::now(); - let mut down = self.down.lock_or_recover(); - down.retain(|_, at| now.duration_since(*at) < DOWN_FOR); - for (failed, _, address) in &mut usable { - *failed = down.contains_key(*address); - } - } - usable.sort_by_key(|(failed, preference, _)| (*failed, *preference)); - usable - .into_iter() - .map(|(_, _, address)| address.clone()) + self.table.lock_or_recover().rank(usable, Instant::now()) + } + + /// What paths knows about each of `addresses`. + pub(crate) fn info(&self, addresses: &[String]) -> Vec { + let now = Instant::now(); + let table = self.table.lock_or_recover(); + addresses + .iter() + .map(|address| table.info(address, now)) .collect() } @@ -277,8 +517,11 @@ impl Paths { }; match self.send(&encoded).await { Ok(()) => { - self.down.lock_or_recover().remove(address); + let mut table = self.table.lock_or_recover(); + table.sent(address); if !failures.is_empty() { + table.counts.failovers += 1; + drop(table); debug!( peer = %message.peer, address = %address, @@ -289,9 +532,7 @@ impl Paths { return Ok(address.clone()); } Err(error) => { - self.down - .lock_or_recover() - .insert(address.clone(), Instant::now()); + self.table.lock_or_recover().failed(address, Instant::now()); failures.push((address.clone(), error)); } } @@ -321,35 +562,136 @@ impl Paths { link.inner.connection(&target) } - /// Every transport's figures together. + /// Paths' own figures, and every transport's. pub(crate) fn stats(&self) -> TransportStats { let mut stats = TransportStats::default(); for link in &self.links { stats.merge(link.inner.stats()); } + let now = Instant::now(); + let table = self.table.lock_or_recover(); + let mut paths: Vec = table + .paths + .keys() + .map(|address| table.info(address, now)) + .collect(); + paths.sort_by(|a, b| a.address.cmp(&b.address)); + stats.paths = Some(PathsStats { + paths, + counts: table.counts, + }); stats } + /// Sends a PING or PONG with `nonce` to `address`, counted (`traffic`). + async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> { + let message = if pong { + ParsedMessage::Pong(nonce) + } else { + ParsedMessage::Ping(nonce) + }; + let encoded = self.encode(address, &message)?; + let sent = self.send(&encoded).await; + if let Some(traffic) = &self.traffic { + match &sent { + Ok(()) => traffic.sent(address, encoded.kind, encoded.payload_len), + Err(_) => traffic.refused(address, encoded.kind, encoded.payload_len), + } + } + sent + } + + /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. + async fn probe(self: Arc) { + let mut tick = tokio::time::interval(PROBE_TICK); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + let due = self + .table + .lock_or_recover() + .due(Instant::now(), random_nonce); + for (address, nonce) in due { + if let Err(error) = self.send_probe(&address, false, nonce).await { + debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING"); + self.table + .lock_or_recover() + .failed(&address, Instant::now()); + } + } + } + } + + /// A message from `address`: a PING is answered, a PONG noted; anything else goes to + /// `handler`. + async fn receive( + self: Arc, + address: String, + message: ParsedMessage, + payload_len: usize, + handler: &PathHandler, + ) -> Result<()> { + match message { + ParsedMessage::Ping(nonce) => { + if let Some(traffic) = &self.traffic { + traffic.received(&address, MessageKind::Ping, payload_len); + } + tokio::spawn(async move { + if let Err(error) = self.send_probe(&address, true, nonce).await { + debug!(to = %address, error = %format!("{error:#}"), "Couldn't answer a PING"); + } + }); + Ok(()) + } + ParsedMessage::Pong(nonce) => { + if let Some(traffic) = &self.traffic { + traffic.received(&address, MessageKind::Pong, payload_len); + } + self.table + .lock_or_recover() + .pong(&address, nonce, Instant::now()); + Ok(()) + } + message => handler(address, message, payload_len).await, + } + } + /// Every transport's consumer, each message handed to `handler` under the sender's - /// address; returns when the first one does. - pub(crate) async fn start_consumer(&self, handler: Arc) -> Result<()> { + /// address (PINGs and PONGs end here), and the probe task; returns when the first + /// consumer does. + pub(crate) async fn start_consumer(self: &Arc, handler: Arc) -> Result<()> { + let probing = tokio::spawn(Arc::clone(self).probe()); let mut consumers: Vec> + Send>>> = Vec::new(); for link in &self.links { let inner = Arc::clone(&link.inner); let scheme = link.scheme; let handler = Arc::clone(&handler); + let paths = Arc::clone(self); consumers.push(Box::pin(async move { inner .start_consumer(Box::new(move |sender, message, payload_len| { - handler(format!("{scheme}:{sender}"), message, payload_len) + let paths = Arc::clone(&paths); + let handler = Arc::clone(&handler); + Box::pin(async move { + paths + .receive( + format!("{scheme}:{sender}"), + message, + payload_len, + &*handler, + ) + .await + }) })) .await })); } - first(consumers) + let result = first(consumers) .await - .unwrap_or_else(|| Err(anyhow!("no transports"))) + .unwrap_or_else(|| Err(anyhow!("no transports"))); + probing.abort(); + result } /// When the first transport exits. @@ -369,6 +711,18 @@ impl Paths { } } +/// A PING's nonce: random, so a PONG can't be made up for a PING not seen. +fn random_nonce() -> u64 { + let mut bytes = [0u8; 8]; + if getrandom::fill(&mut bytes).is_err() { + // No randomness: a nonce still tells answers apart, if not unguessably. + return std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| since.as_nanos() as u64); + } + u64::from_le_bytes(bytes) +} + /// `address: error; …` for each failed send. fn describe(failures: &[(String, anyhow::Error)]) -> String { failures @@ -401,13 +755,16 @@ mod tests { use crate::transport::{HeadsSummary, Status}; /// "a" on two sim nets, as `sim` (preferred) and `other`; "b" on both too. - fn two_nets() -> (Paths, Paths) { + fn two_nets() -> (Arc, Arc) { let (first, second) = (SimNet::new(1), SimNet::new(2)); let on_both = |id: &str| { - Paths::new(vec![ - ("sim", first.transport(id) as Arc), - ("other", second.transport(id) as Arc), - ]) + Arc::new(Paths::new( + vec![ + ("sim", first.transport(id) as Arc), + ("other", second.transport(id) as Arc), + ], + None, + )) }; (on_both("a"), on_both("b")) } @@ -436,6 +793,109 @@ mod tests { assert!(a.resolve("nobody").await.is_err()); } + fn paths(addresses: &[&str]) -> Vec<(String, usize)> { + addresses + .iter() + .enumerate() + .map(|(preference, address)| (address.to_string(), preference)) + .collect() + } + + /// Answers every PING due `rtt` later, as from the path it went to (`rtts`). + fn answer(table: &mut Table, now: Instant, rtts: &[(&str, u64)]) -> Instant { + let mut nonces = 0; + let due = table.due(now, || { + nonces += 1; + nonces + }); + for (address, nonce) in due { + let rtt = rtts + .iter() + .find(|(to, _)| *to == address) + .map(|(_, ms)| *ms); + if let Some(ms) = rtt { + table.pong(&address, nonce, now + Duration::from_millis(ms)); + } + } + now + PROBE_EVERY + } + + #[test] + fn the_configured_order_until_measured_then_the_faster_path() { + let mut table = Table::default(); + let mut now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]); + // libp2p a bit slower: not enough to switch. + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:x", 110), ("mycelium:x", 100)]); + } + assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]); + assert_eq!(table.counts.switches, 0); + // Much slower: switched, and it stays switched while that holds. + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:x", 300), ("mycelium:x", 100)]); + } + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + assert_eq!(table.counts.switches, 1); + // Small differences under SWITCH_MIN never switch. + let lan = paths(&["libp2p:y", "mycelium:y"]); + table.rank(lan.clone(), now); + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:y", 4), ("mycelium:y", 1)]); + } + assert_eq!(table.rank(lan, now), ["libp2p:y", "mycelium:y"]); + } + + #[test] + fn unanswered_or_failing_paths_are_down() { + let mut table = Table::default(); + let mut now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + table.rank(set.clone(), now); + // libp2p stops answering: down after UNANSWERED_DOWN PINGs, then last. + for _ in 0..=UNANSWERED_DOWN { + now = answer(&mut table, now, &[("mycelium:x", 100)]); + table.rank(set.clone(), now); + } + assert!(table.info("libp2p:x", now).down); + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + // It answers again: up, and measured. + now = answer(&mut table, now, &[("libp2p:x", 50), ("mycelium:x", 100)]); + assert!(!table.info("libp2p:x", now).down); + + // A failed send: down for DOWN_FOR, unless a send there works. + table.failed("mycelium:x", now); + assert!(table.info("mycelium:x", now).down); + assert!(!table.info("mycelium:x", now + DOWN_FOR).down); + table.sent("mycelium:x"); + assert!(!table.info("mycelium:x", now).down); + } + + #[test] + fn only_paths_in_use_are_pinged() { + let mut table = Table::default(); + let now = Instant::now(); + let mut nonce = 0; + let mut next = || { + nonce += 1; + nonce + }; + // One address: nothing to choose, nothing to measure. + table.rank(paths(&["libp2p:z"]), now); + assert!(table.due(now, &mut next).is_empty()); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + assert_eq!(table.due(now, &mut next).len(), 2); + // Not again within PROBE_EVERY; a stale answer is ignored. + assert!(table.due(now + PROBE_TIMEOUT, &mut next).is_empty()); + assert_eq!(table.pong("libp2p:x", 99, now), None); + // Unused for ACTIVE_FOR: no more PINGs; for twice as long: forgotten. + assert!(table.due(now + ACTIVE_FOR, &mut next).is_empty()); + table.due(now + ACTIVE_FOR * 2, &mut next); + assert!(table.paths.is_empty() && table.chosen.is_empty()); + } + /// A transport whose sends all fail. struct Broken; @@ -482,11 +942,17 @@ mod tests { #[tokio::test] async fn a_failed_send_goes_to_the_next_address() { let net = SimNet::new(3); - let a = Paths::new(vec![ - ("broken", Arc::new(Broken) as Arc), - ("sim", net.transport("a") as Arc), - ]); - let b = Paths::new(vec![("sim", net.transport("b") as Arc)]); + let a = Paths::new( + vec![ + ("broken", Arc::new(Broken) as Arc), + ("sim", net.transport("a") as Arc), + ], + None, + ); + let b = Arc::new(Paths::new( + vec![("sim", net.transport("b") as Arc)], + None, + )); let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); let handler: Arc = Arc::new(move |from, _, _| { let sender = sender.clone(); -- cgit v1.3.1