diff options
| -rw-r--r-- | docs/openapi.yaml | 52 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 76 | ||||
| -rw-r--r-- | src/simtest/identity.rs | 7 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 5 | ||||
| -rw-r--r-- | src/transport/identity.rs | 25 | ||||
| -rw-r--r-- | src/transport/mod.rs | 20 | ||||
| -rw-r--r-- | src/transport/paths.rs | 568 | ||||
| -rw-r--r-- | src/transport/stats.rs | 43 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 3 | ||||
| -rw-r--r-- | src/transport/wire.rs | 29 |
10 files changed, 751 insertions, 77 deletions
diff --git a/docs/openapi.yaml b/docs/openapi.yaml index be21300..58faa9c 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1141,6 +1141,21 @@ components: type: [string, 'null'] enum: [direct, relayed, null] description: The libp2p connection to it; null if none, or on a transport without connections (Mycelium). + paths: + type: array + description: Every address it proved itself at, best first, as the paths layer measures them. + items: + type: object + required: [address, down] + properties: + address: + type: string + rttMs: + type: [number, 'null'] + description: Smoothed round-trip time of its PINGs; null before an answer (and on a peer's only path, which isn't measured). + down: + type: boolean + description: Its PINGs went unanswered three times in a row, or a send there failed within a minute. validatedSwarm: type: [string, 'null'] status: @@ -1190,7 +1205,7 @@ components: ByKind: type: object - required: [hello, helloReply, sync, status, statusReply, blob, identity] + required: [hello, helloReply, sync, status, statusReply, blob, identity, paths] properties: hello: $ref: '#/components/schemas/Counter' @@ -1208,6 +1223,9 @@ components: identity: $ref: '#/components/schemas/Counter' description: WHO and ME, counted under the address they went to or came from (so only in the totals). + paths: + $ref: '#/components/schemas/Counter' + description: PING and PONG, counted under the address (so only in the totals). Traffic: type: object @@ -1414,6 +1432,38 @@ components: type: object description: How the transports find and reach peers (`measurements.md`, Transports). properties: + paths: + description: The paths layer's; null without one. + oneOf: + - type: 'null' + - type: object + required: [paths, probesSent, probesAnswered, probesUnanswered, sendFailures, failovers, switches] + properties: + paths: + type: array + description: Every path measured (of peers with several addresses, used lately). + items: + type: object + required: [address, down] + properties: + address: + type: string + rttMs: + type: [number, 'null'] + down: + type: boolean + probesSent: + type: integer + probesAnswered: + type: integer + probesUnanswered: + type: integer + sendFailures: + type: integer + failovers: + type: integer + switches: + type: integer identity: description: The identity layer's; null without one. oneOf: diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 11645a3..9a4f162 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -103,6 +103,11 @@ fn escape(value: &str) -> String { .replace('\n', "\\n") } +/// The transport of an address. +fn scheme(address: &str) -> &str { + address.split_once(':').map_or("", |(scheme, _)| scheme) +} + fn flag(value: bool) -> f64 { if value { 1.0 } else { 0.0 } } @@ -177,6 +182,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou (direction, "status_reply", by.status_reply), (direction, "blob", by.blob), (direction, "identity", by.identity), + (direction, "paths", by.paths), ] }; let mut series = Vec::new(); @@ -238,6 +244,40 @@ fn transports(out: &mut Out, stats: &TransportStats) { out.single(name, "counter", help, value as f64); } } + if let Some(paths) = &stats.paths { + let counts = &paths.counts; + out.family( + "paths_probes_total", + "counter", + "PINGs to paths of peers with several, by result.", + ); + for (result, count) in [ + ("sent", counts.probes_sent), + ("answered", counts.probes_answered), + ("unanswered", counts.probes_unanswered), + ] { + out.sample("paths_probes_total", &[("result", result)], count as f64); + } + for (name, help, value) in [ + ( + "paths_send_failures_total", + "Sends that failed on a path, each tried at the next.", + counts.send_failures, + ), + ( + "paths_failovers_total", + "Sends that went over another path after one failed.", + counts.failovers, + ), + ( + "paths_switches_total", + "Changes of the path in use for a faster one.", + counts.switches, + ), + ] { + out.single(name, "counter", help, value as f64); + } + } if let Some(libp2p) = &stats.libp2p { out.family( "libp2p_reachability", @@ -675,10 +715,7 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); for peer in &peers { if let Some(destination) = &peer.destination { - let transport = destination - .address - .split_once(':') - .map_or("", |(scheme, _)| scheme); + let transport = scheme(&destination.address); out.sample( "peer_destination_info", &[ @@ -695,6 +732,37 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { } } + out.family( + "peer_path_rtt_seconds", + "gauge", + "Smoothed round-trip time of the PINGs on each path to a peer with several.", + ); + for peer in &peers { + for path in peer.destination.iter().flat_map(|d| &d.paths) { + if let Some(ms) = path.rtt_ms { + out.sample( + "peer_path_rtt_seconds", + &[("peer", &peer.key), ("transport", scheme(&path.address))], + ms / 1000.0, + ); + } + } + } + out.family( + "peer_path_down", + "gauge", + "1 while a path to the peer is down (PINGs unanswered, or a send failed).", + ); + for peer in &peers { + for path in peer.destination.iter().flat_map(|d| &d.paths) { + out.sample( + "peer_path_down", + &[("peer", &peer.key), ("transport", scheme(&path.address))], + flag(path.down), + ); + } + } + out.family("peer_neighbour", "gauge", "1 if the peer is a neighbour."); for peer in &peers { out.sample( diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs index b789bda..e93b1da 100644 --- a/src/simtest/identity.rs +++ b/src/simtest/identity.rs @@ -14,6 +14,7 @@ use crate::swarm::bootstrap::Start; use crate::swarm::members; use crate::transport::identity::{WHO_LABEL, signed}; use crate::transport::sim::SimNet; +use crate::transport::stats::PathInfo; use crate::transport::{ Destination, HeadsSummary, IdentityProof, ParsedMessage, Status, Transport, }; @@ -50,6 +51,12 @@ async fn nodes_are_their_node_ids() { address: "sim:b".into(), proven: true, connection: None, + // One path: nothing to choose, so nothing measured. + paths: vec![PathInfo { + address: "sim:b".into(), + rtt_ms: None, + down: false, + }], }) ); let stats = a.transport.stats(); diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 4c7e7a8..92a1a5e 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -160,7 +160,10 @@ pub(crate) async fn handle_message( .status_reply_received(&peer, Instant::now()); on_status(state, &peer, &remote, reactivated).await; } - transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) => {} + transport::ParsedMessage::Who(_) + | transport::ParsedMessage::Me(_) + | transport::ParsedMessage::Ping(_) + | transport::ParsedMessage::Pong(_) => {} } Ok(()) diff --git a/src/transport/identity.rs b/src/transport/identity.rs index dd3950b..5857fb0 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -95,7 +95,7 @@ pub(crate) struct IdentityTransport { } struct Shared { - paths: Paths, + paths: Arc<Paths>, key: Arc<NodeKey>, node_id: String, /// WHO and ME are counted here, under the address (`traffic`). @@ -214,11 +214,12 @@ impl IdentityTransport { traffic: Arc<Traffic>, ) -> Self { let node_id = config.key.id(); - let paths = Paths::new( + let paths = Arc::new(Paths::new( std::iter::once((config.scheme, transport)) .chain(config.more) .collect(), - ); + Some(Arc::clone(&traffic)), + )); Self { shared: Arc::new(Shared { paths, @@ -443,20 +444,20 @@ impl Transport for IdentityTransport { self.shared.paths.transports() } - /// The address `encode` picks, and the transport's connection there. + /// The address `encode` picks, the transport's connection there, and every path + /// to `peer` as paths measures it, best first. fn destination(&self, peer: &str) -> Option<Destination> { - let address = self.shared.destination(peer)?; - let proven = self - .shared - .lock() - .proven - .get(peer) - .is_some_and(|addresses| addresses.contains(&address)); + let proven = self.shared.proven_at(peer); + let (address, is_proven) = match proven.first() { + Some(address) => (address.clone(), true), + None => (self.shared.hinted_at(peer).into_iter().next()?, false), + }; let connection = self.shared.paths.connection(&address); Some(Destination { address, - proven, + proven: is_proven, connection, + paths: self.shared.paths.info(&proven), }) } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index f20110a..147e560 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -147,6 +147,10 @@ pub(crate) enum ParsedMessage { Who(IdentityProof), /// Answers a WHO. Me(IdentityProof), + /// Asks for a PONG with this nonce, over the same path (`transport::paths`). + Ping(u64), + /// Answers a PING. + Pong(u64), } /// A node's proof that it holds the key of its node ID (`identity`), sending from its @@ -270,6 +274,8 @@ pub(crate) enum MessageKind { Blob, Who, Me, + Ping, + Pong, } impl ParsedMessage { @@ -283,11 +289,13 @@ impl ParsedMessage { ParsedMessage::Blob(_) => MessageKind::Blob, ParsedMessage::Who(_) => MessageKind::Who, ParsedMessage::Me(_) => MessageKind::Me, + ParsedMessage::Ping(_) => MessageKind::Ping, + ParsedMessage::Pong(_) => MessageKind::Pong, } } /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them - /// but WHO and ME (which the node never sees). + /// but WHO, ME, PING and PONG (which the node never sees). pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> { match self { ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)), @@ -297,7 +305,10 @@ impl ParsedMessage { Some((status.epoch, status.epoch_seq)) } ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)), - ParsedMessage::Who(_) | ParsedMessage::Me(_) => None, + ParsedMessage::Who(_) + | ParsedMessage::Me(_) + | ParsedMessage::Ping(_) + | ParsedMessage::Pong(_) => None, } } } @@ -334,7 +345,7 @@ pub(crate) struct ListedPeer { } /// Where messages to a peer go now (`Transport::destination`). -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct Destination { /// As the identity layer keeps it: `<scheme>:<transport identity>`. @@ -344,6 +355,9 @@ pub(crate) struct Destination { pub(crate) proven: bool, /// The transport's connection to it, if the transport has connections. pub(crate) connection: Option<Connection>, + /// Every address it proved itself at, best first, as the paths layer measures them + /// (`transport::paths`). + pub(crate) paths: Vec<stats::PathInfo>, } /// How a transport with connections (libp2p) is connected to a peer. 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<Box<dyn Future<Output = Result<()>> + 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<Instant>, + /// Smoothed round-trip time of its PINGs. + rtt: Option<Duration>, + /// The PING out, by nonce, and when it went. + probe: Option<(u64, Instant)>, + last_probe: Option<Instant>, + /// PINGs in a row unanswered. + unanswered: u32, + /// When a send there last failed. + failed_at: Option<Instant>, +} + +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<String, PathState>, + /// By the set's addresses (sorted, joined): the one in use. + chosen: HashMap<String, String>, + 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<String> { + 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<String> = 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<Duration> { + 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<Link>, - /// Addresses whose last send failed, and when (`DOWN_FOR`). A std mutex: never held - /// across an await. - down: Mutex<HashMap<String, Instant>>, + /// PINGs and PONGs are counted here, under the address (`traffic`). + traffic: Option<Arc<Traffic>>, + /// A std mutex: never held across an await. + table: Mutex<Table>, } impl Paths { - /// Over `transports`, by scheme, in order of preference. - pub(crate) fn new(transports: Vec<(&'static str, Arc<dyn Transport>)>) -> Self { + /// Over `transports`, by scheme, in order of preference; PINGs and PONGs counted in + /// `traffic`. + pub(crate) fn new( + transports: Vec<(&'static str, Arc<dyn Transport>)>, + traffic: Option<Arc<Traffic>>, + ) -> 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<String> { - 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<PathInfo> { + 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<PathInfo> = 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<Self>) { + 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<Self>, + 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<PathHandler>) -> 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<Self>, handler: Arc<PathHandler>) -> Result<()> { + let probing = tokio::spawn(Arc::clone(self).probe()); let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + 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<Paths>, Arc<Paths>) { let (first, second) = (SimNet::new(1), SimNet::new(2)); let on_both = |id: &str| { - Paths::new(vec![ - ("sim", first.transport(id) as Arc<dyn Transport>), - ("other", second.transport(id) as Arc<dyn Transport>), - ]) + Arc::new(Paths::new( + vec![ + ("sim", first.transport(id) as Arc<dyn Transport>), + ("other", second.transport(id) as Arc<dyn Transport>), + ], + 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<dyn Transport>), - ("sim", net.transport("a") as Arc<dyn Transport>), - ]); - let b = Paths::new(vec![("sim", net.transport("b") as Arc<dyn Transport>)]); + let a = Paths::new( + vec![ + ("broken", Arc::new(Broken) as Arc<dyn Transport>), + ("sim", net.transport("a") as Arc<dyn Transport>), + ], + None, + ); + let b = Arc::new(Paths::new( + vec![("sim", net.transport("b") as Arc<dyn Transport>)], + None, + )); let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); let handler: Arc<PathHandler> = Arc::new(move |from, _, _| { let sender = sender.clone(); diff --git a/src/transport/stats.rs b/src/transport/stats.rs index 35238f5..6b1a42f 100644 --- a/src/transport/stats.rs +++ b/src/transport/stats.rs @@ -11,6 +11,8 @@ use serde::Serialize; pub(crate) struct TransportStats { /// The identity layer's (`transport::identity`); absent without one. pub(crate) identity: Option<IdentityStats>, + /// The paths layer's (`transport::paths`); absent without one. + pub(crate) paths: Option<PathsStats>, /// The libp2p transport's; absent if it isn't one of this node's. pub(crate) libp2p: Option<Libp2pStats>, } @@ -21,12 +23,53 @@ impl TransportStats { if other.identity.is_some() { self.identity = other.identity; } + if other.paths.is_some() { + self.paths = other.paths; + } if other.libp2p.is_some() { self.libp2p = other.libp2p; } } } +/// How the paths to peers with several addresses do (`transport::paths`). +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PathsStats { + /// Every path measured: those of peers this node chose among lately. + pub(crate) paths: Vec<PathInfo>, + #[serde(flatten)] + pub(crate) counts: PathCounts, +} + +/// One path, as `transport::paths` measures it. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PathInfo { + /// `<scheme>:<transport identity>`. + pub(crate) address: String, + /// Smoothed round-trip time of its PINGs; absent before an answer. + pub(crate) rtt_ms: Option<f64>, + /// Its PINGs went unanswered three times in a row, or a send there failed within a + /// minute: used only when no other path works. + pub(crate) down: bool, +} + +/// Counts since start. +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PathCounts { + pub(crate) probes_sent: u64, + pub(crate) probes_answered: u64, + pub(crate) probes_unanswered: u64, + /// Sends that failed on a path (each tried at the next). + pub(crate) send_failures: u64, + /// Sends that went over another path after one failed. + pub(crate) failovers: u64, + /// Changes of the path in use for one faster. + pub(crate) switches: u64, +} + /// Who is where (`transport::identity`): the books now, and counts since start. #[derive(Debug, Clone, Default, Serialize)] #[serde(rename_all = "camelCase")] diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 2134521..9b5f9c9 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -43,6 +43,8 @@ pub(crate) struct ByKind { /// WHO and ME (`transport::identity`), counted by the identity layer under the /// address they went to or came from. pub(crate) identity: Counter, + /// PING and PONG (`transport::paths`), counted by the paths layer under the address. + pub(crate) paths: Counter, } impl ByKind { @@ -55,6 +57,7 @@ impl ByKind { MessageKind::StatusReply => &mut self.status_reply, MessageKind::Blob => &mut self.blob, MessageKind::Who | MessageKind::Me => &mut self.identity, + MessageKind::Ping | MessageKind::Pong => &mut self.paths, } } } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 62aa3ca..fcbee85 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 5; +pub(crate) const VERSION: u8 = 6; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -27,6 +27,8 @@ fn tag(kind: MessageKind) -> u8 { MessageKind::Blob => 5, MessageKind::Who => 6, MessageKind::Me => 7, + MessageKind::Ping => 8, + MessageKind::Pong => 9, } } @@ -62,6 +64,9 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> { ParsedMessage::Who(proof) | ParsedMessage::Me(proof) => { rmp_serde::encode::write(&mut payload, proof)? } + ParsedMessage::Ping(nonce) | ParsedMessage::Pong(nonce) => { + rmp_serde::encode::write(&mut payload, nonce)? + } } Ok(payload) } @@ -90,6 +95,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> { ParsedMessage::Who(rmp_serde::from_slice(body)?) } else if is(MessageKind::Me) { ParsedMessage::Me(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Ping) { + ParsedMessage::Ping(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Pong) { + ParsedMessage::Pong(rmp_serde::from_slice(body)?) } else { return Err(WireError::UnknownKind(kind)); }) @@ -470,7 +479,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("050097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("060097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -490,7 +499,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "050199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "060199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -505,7 +514,7 @@ mod tests { }); assert_eq!( hex(&encode(&status).unwrap()), - format!("0503959290c400{epoch_hex}01c3c2") + format!("0603959290c400{epoch_hex}01c3c2") ); let who = ParsedMessage::Who(IdentityProof { @@ -514,7 +523,7 @@ mod tests { }); assert_eq!( hex(&encode(&who).unwrap()), - format!("050692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + format!("060692c420{}c440{}", "08".repeat(32), "09".repeat(64)) ); let ParsedMessage::Me(proof) = decode( &encode(&ParsedMessage::Me(IdentityProof { @@ -527,5 +536,15 @@ mod tests { panic!("not a ME"); }; assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); + + assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "060807"); + assert_eq!( + hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()), + "0609cfffffffffffffffff" + ); + let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else { + panic!("not a PONG"); + }; + assert_eq!(nonce, 42); } } |