diff options
Diffstat (limited to 'src/transport/paths.rs')
| -rw-r--r-- | src/transport/paths.rs | 464 |
1 files changed, 397 insertions, 67 deletions
diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 3baa092..164574d 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -8,7 +8,7 @@ //! transport of its address, hands every transport's messages to one handler under the //! sender's address, and puts together what the transports say about themselves //! (addresses, seeds they found, figures). The transports are in order of preference, -//! as configured (`--transports`). +//! as configured (`--transport-order`). //! //! A message to a node with several addresses goes to the first of them that takes it //! (`send_first`), in the order `rank` gives. A send that fails is tried at once at the @@ -16,7 +16,8 @@ //! next carries on within the same send. //! //! Paths measures the addresses of such nodes itself (and every path the node's -//! keepalive PINGs go over, below): every `PROBE_EVERY`, each address it had a choice of +//! keepalive PINGs go over, below): every `PROBE_EVERY`, or every keepalive interval if +//! that is longer, each address it had a choice of //! within `ACTIVE_FOR` gets a PING with a random nonce, which the //! other side's paths answers at once with a PONG, over the same path. Answers give each //! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a @@ -30,6 +31,11 @@ //! alive. The node's keepalive is a PING too (`swarm::peer::keepalive`), which paths //! sends like any message, and notes as the probe of the path it went over. //! +//! Every message on every transport passes here, so this is where messages are logged +//! (debug): `Sent message` once a transport took one, `Received message` as one +//! arrives, each with its address and what it says. Failed sends are logged by those who +//! sent them, with what they do about it. +//! //! The debug firewall (`transport::firewall`) works here, below everything else, through //! a `Gate`: messages from an address it blocks are dropped as they arrive, and sends to //! one are dropped silently, as by a lossy network. @@ -42,6 +48,7 @@ use anyhow::{Context, Result, anyhow}; use tokio::time::Instant; use tracing::{debug, info}; +use crate::clock::{Clock, UnixMillis}; use crate::locks::MutexExt; use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats}; use crate::transport::traffic::Traffic; @@ -84,15 +91,16 @@ pub(crate) fn scheme_of(address: &str) -> &str { /// 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. +/// How often a path in use is pinged at most (every keepalive interval, if that is +/// longer: `Table::probe_every`), 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. +/// Paths this node had a choice of, or sent over, 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`. @@ -100,14 +108,48 @@ 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; +/// A path's clock sample with the narrowest bound is kept this long at most; then the +/// next one takes its place. (A clock set anew shows at once: `Table::clock`.) +const CLOCK_KEPT: Duration = Duration::from_secs(300); + +/// How far a peer's clock is ahead of ours (behind if negative), from a PING and its +/// PONG: `offset_ms`, give or take `within_ms`. The PONG says its sender's time when it +/// answered, which was after our PING went and before the PONG came: within half the +/// round trip of its middle, however differently long the two ways took. +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ClockOffset { + pub(crate) offset_ms: i64, + pub(crate) within_ms: u64, +} + +impl ClockOffset { + /// From a PONG stamped `at` by its sender, which came at `heard` by our clock, `rtt` + /// after its PING went. + pub(crate) fn of(at: UnixMillis, heard: UnixMillis, rtt: Duration) -> Self { + let half_us = rtt.as_micros() / 2; + let middle_us = i128::from(heard.as_u64()) * 1000 - half_us as i128; + let offset_us = i128::from(at.as_u64()) * 1000 - middle_us; + Self { + offset_ms: i64::try_from(offset_us / 1000).unwrap_or(i64::MAX), + // And a millisecond for the stamps' resolution. + within_ms: u64::try_from(half_us.div_ceil(1000)).unwrap_or(u64::MAX) + 1, + } + } +} /// 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>, + /// When a message last went over it: a peer's only path is measured too, if its + /// keepalives don't (a busy pair has none: `PeerState::take_keepalive_due`). + carried: Option<Instant>, /// Smoothed round-trip time of its PINGs. rtt: Option<Duration>, + /// Where the transport went when `rtt` was measured (`Paths::location`). + measured_at: Option<String>, /// The PING out, by nonce, and when it went. probe: Option<(u64, Instant)>, last_probe: Option<Instant>, @@ -115,6 +157,9 @@ struct PathState { unanswered: u32, /// When a send there last failed. failed_at: Option<Instant>, + /// The peer's clock as its PONGs on this path say (the narrowest bound lately), and + /// when that sample came. + clock: Option<(ClockOffset, Instant)>, } impl PathState { @@ -127,29 +172,58 @@ impl PathState { } /// Every path's state, and which one is in use for each set to choose from. -#[derive(Debug, Default)] +#[derive(Debug)] struct Table { paths: HashMap<String, PathState>, /// By the set's addresses (sorted, joined): the one in use. chosen: HashMap<String, String>, counts: PathCounts, + /// How often a path in use is pinged: `PROBE_EVERY`, or the node's keepalive + /// interval if that is longer, so that the probes don't add to the keepalives a + /// long interval is chosen to spare. A busy pair, whose keepalives don't go, still + /// gets its paths measured this often. + probe_every: Duration, +} + +impl Default for Table { + fn default() -> Self { + Self::new(PROBE_EVERY) + } } impl Table { + fn new(keepalive: Duration) -> Self { + Self { + paths: HashMap::new(), + chosen: HashMap::new(), + counts: PathCounts::default(), + probe_every: PROBE_EVERY.max(keepalive), + } + } + /// `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> { + /// `Paths::rank` says). With `sending`, for a message about to go: they count as + /// used (`ACTIVE_FOR`), and the one in use may change for a faster one. Without, a + /// look at the choice as it stands, changing nothing (`/peers`, figures, LINKS). + fn rank( + &mut self, + mut usable: Vec<(String, usize)>, + now: Instant, + sending: bool, + ) -> 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); + if sending { + 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); @@ -180,6 +254,7 @@ impl Table { .min_by_key(|(_, rtt)| *rtt); if let (Some((fastest, fastest_rtt)), Some(current_rtt)) = (fastest, rtt(self, ¤t)) + && sending && *fastest != current && current_rtt.saturating_sub(fastest_rtt) >= SWITCH_MIN.max(current_rtt.mul_f64(SWITCH_GAIN)) @@ -194,7 +269,7 @@ impl Table { self.counts.switches += 1; current = fastest.clone(); } - if self.chosen.len() < MAX_PATHS || self.chosen.contains_key(&key) { + if sending && (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 @@ -228,15 +303,22 @@ impl Table { } /// A send to `address` worked. - fn sent(&mut self, address: &str) { - if let Some(path) = self.paths.get_mut(address) { - path.failed_at = None; + fn sent(&mut self, address: &str, now: Instant) { + if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) { + self.forget_unused(now); + if self.paths.len() >= MAX_PATHS { + return; + } } + let path = self.paths.entry(address.to_owned()).or_default(); + path.failed_at = None; + path.carried = Some(now); } - /// 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. + /// The PINGs due now, with their nonces (from `nonce`): to every path chosen among + /// others or sent over within `ACTIVE_FOR` that wasn't pinged (by us, or by a + /// keepalive) 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(); @@ -251,12 +333,13 @@ impl Table { 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 lately = |at: Option<Instant>| { + at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR) + }; + let active = lately(path.used) || lately(path.carried); let rested = path .last_probe - .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY); + .is_none_or(|last| now.saturating_duration_since(last) >= self.probe_every); if active && rested && path.probe.is_none() { let nonce = nonce(); path.probe = Some((nonce, now)); @@ -269,8 +352,8 @@ impl Table { } /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that - /// path, which its PONG measures, even if it is a peer's only path (then it isn't - /// pinged otherwise: `due` pings only paths there was a choice of). + /// path, which its PONG measures, even if it is a peer's only path; `due` then + /// doesn't ping it again within `probe_every`. fn pinged(&mut self, address: &str, nonce: u64, now: Instant) { if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) { self.forget_unused(now); @@ -295,7 +378,23 @@ impl Table { } } - /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out. + /// The transport of `address` now goes to `location` (a libp2p peer over another + /// connection: through a relay before, directly now): round trips measured before + /// were of another way, and would weigh on the next ones for minutes (`pong` + /// smooths), so they are dropped. + fn located(&mut self, address: &str, location: &str) { + let Some(path) = self.paths.get_mut(address) else { + return; + }; + if path.measured_at.as_deref() != Some(location) { + if path.measured_at.is_some() { + path.rtt = None; + } + path.measured_at = Some(location.to_owned()); + } + } + + /// A PONG with `nonce` from `address`: its 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?; @@ -314,7 +413,26 @@ impl Table { None => sample, }); self.counts.probes_answered += 1; - path.rtt + Some(sample) + } + + /// A clock sample from a PONG over `address`: kept if its bound is as narrow as the + /// one kept, or that one is older than `CLOCK_KEPT`, or the two can't both be right + /// (their bounds don't overlap: a clock was set anew). The narrowest bound is the + /// round trip least delayed by queues and relays. + fn clock(&mut self, address: &str, offset: ClockOffset, now: Instant) { + let Some(path) = self.paths.get_mut(address) else { + return; + }; + let replaces = path.clock.is_none_or(|(kept, at)| { + offset.within_ms <= kept.within_ms + || now.saturating_duration_since(at) >= CLOCK_KEPT + || offset.offset_ms.abs_diff(kept.offset_ms) + > offset.within_ms.saturating_add(kept.within_ms) + }); + if replaces { + path.clock = Some((offset, now)); + } } /// Forgets the paths neither used nor pinged for twice `ACTIVE_FOR`, and choices @@ -324,7 +442,7 @@ impl Table { at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR * 2) }; self.paths - .retain(|_, path| recent(path.used) || recent(path.last_probe)); + .retain(|_, path| recent(path.used) || recent(path.carried) || recent(path.last_probe)); let paths = &self.paths; self.chosen .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address))); @@ -338,6 +456,7 @@ impl Table { .and_then(|path| path.rtt) .map(|rtt| rtt.as_secs_f64() * 1000.0), down: path.is_some_and(|path| path.is_down(now)), + connection: None, } } } @@ -349,17 +468,22 @@ pub(crate) struct Paths { traffic: Option<Arc<Traffic>>, /// The debug firewall, if any. gate: Option<Weak<dyn Gate>>, + /// The node's wall clock, which its PONGs say the time by. + clock: Arc<Clock>, /// A std mutex: never held across an await. table: Mutex<Table>, } impl Paths { /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs - /// counted in `traffic`, the firewall through `gate`. + /// counted in `traffic`, the firewall through `gate`, PONGs stamped by `clock`; + /// paths pinged every `keepalive` at least (`Table::probe_every`). pub(crate) fn new( transports: Vec<(&'static str, Arc<dyn Transport>)>, traffic: Option<Arc<Traffic>>, gate: Option<Weak<dyn Gate>>, + clock: Arc<Clock>, + keepalive: Duration, ) -> Self { Self { links: transports @@ -368,7 +492,8 @@ impl Paths { .collect(), traffic, gate, - table: Mutex::default(), + clock, + table: Mutex::new(Table::new(keepalive)), } } @@ -421,14 +546,18 @@ impl Paths { /// 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> { + /// the configured order (module docs). For a send (`sending`): several addresses + /// are paths to measure, and the one in use may change; else only a look + /// (`Table::rank`). + pub(crate) fn rank(&self, addresses: &[String], sending: bool) -> Vec<String> { let usable: Vec<(String, usize)> = addresses .iter() .filter(|address| self.usable(address)) .map(|address| (address.clone(), self.preference(address))) .collect(); - self.table.lock_or_recover().rank(usable, Instant::now()) + self.table + .lock_or_recover() + .rank(usable, Instant::now(), sending) } /// The smoothed round-trip time of the path to `address`, if measured. @@ -436,6 +565,12 @@ impl Paths { self.table.lock_or_recover().paths.get(address)?.rtt } + /// The peer's clock at `address`, as its PONGs there say (`ClockOffset`). + pub(crate) fn clock_offset(&self, address: &str) -> Option<ClockOffset> { + let table = self.table.lock_or_recover(); + table.paths.get(address)?.clock.map(|(offset, _)| offset) + } + /// What paths knows about each of `addresses`. pub(crate) fn info(&self, addresses: &[String]) -> Vec<PathInfo> { let now = Instant::now(); @@ -449,9 +584,11 @@ impl Paths { self.located(infos) } - /// `infos` with where each path goes (`located`); asked outside the table's lock. + /// `infos` with where each path goes (`located`) and the transport's connection + /// there; asked outside the table's lock. fn located(&self, mut infos: Vec<PathInfo>) -> Vec<PathInfo> { for info in &mut infos { + info.connection = self.connection(&info.address); info.address = self.location(&info.address); } infos @@ -507,17 +644,40 @@ impl Paths { .unwrap_or_default() } - /// Each transport's (`Transport::local_addresses`), with its scheme. - pub(crate) fn local_addresses(&self) -> Vec<String> { + /// The longest of the transports' (`Transport::sync_reply_timeout`): an answer may + /// come over any of them. + pub(crate) fn sync_reply_timeout(&self) -> Duration { self.links .iter() - .flat_map(|link| { + .map(|link| link.inner.sync_reply_timeout()) + .max() + .unwrap_or(crate::state::SYNC_REPLY_TIMEOUT) + } + + /// Each transport's (`Transport::local_addresses`), with its scheme: taken in turns, + /// each transport's first in the configured order, then each one's second, and so on, + /// so a list cut short (`members::MAX_ADDRESSES`) still has every transport's best. + pub(crate) fn local_addresses(&self) -> Vec<String> { + let mut each: Vec<std::vec::IntoIter<String>> = self + .links + .iter() + .map(|link| { link.inner .local_addresses() .into_iter() .map(|target| link.address(&target)) + .collect::<Vec<_>>() + .into_iter() }) - .collect() + .collect(); + let mut addresses = Vec::new(); + loop { + let before = addresses.len(); + addresses.extend(each.iter_mut().filter_map(Iterator::next)); + if addresses.len() == before { + return addresses; + } + } } /// The schemes of the transports, in order of preference. @@ -541,6 +701,20 @@ impl Paths { .collect() } + /// Each transport's (`Transport::on_seeds_found`). + pub(crate) fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) { + for link in &self.links { + link.inner.on_seeds_found(found.clone()); + } + } + + /// Each transport's (`Transport::on_reachable`). + pub(crate) fn on_reachable(&self, reachable: std::sync::Arc<tokio::sync::Notify>) { + for link in &self.links { + link.inner.on_reachable(reachable.clone()); + } + } + /// The address of the node a seed leads to, as paths keep it: an address of ours /// (`<scheme>:...`) whose transport reads it at once; with a scheme of ours, what /// that transport resolves the rest to (an address the transport can't read as an @@ -623,7 +797,7 @@ impl Paths { match self.send(&encoded).await { Ok(()) => { let mut table = self.table.lock_or_recover(); - table.sent(address); + table.sent(address, Instant::now()); if !failures.is_empty() { table.counts.failovers += 1; drop(table); @@ -665,7 +839,16 @@ impl Paths { let mut inner = message.clone(); inner.peer = target.clone(); inner.address = target; - link.inner.send_encoded(&inner).await + link.inner.send_encoded(&inner).await?; + match &message.message { + Some(parsed) => { + debug!(to = %message.address, bytes = message.payload_len, message = ?parsed, "Sent message"); + } + None => { + debug!(to = %message.address, bytes = message.payload_len, kind = ?message.kind, "Sent message"); + } + } + Ok(()) } /// How the transport of `address` is connected there (`Transport::connection`). @@ -674,6 +857,12 @@ impl Paths { link.inner.connection(&target) } + /// How hole punching to `address` went (`Transport::punching`). + pub(crate) fn punching(&self, address: &str) -> Option<crate::transport::Punching> { + let (link, target) = self.split(address)?; + link.inner.punching(&target) + } + /// Paths' own figures, and every transport's. pub(crate) fn stats(&self) -> TransportStats { let mut stats = TransportStats::default(); @@ -696,11 +885,14 @@ impl Paths { stats } - /// Sends a PING (or with `pong`, a PONG) with `nonce` to `address`, counted - /// (`traffic`). + /// Sends a PING (or with `pong`, a PONG) with `nonce` to `address`, counted and + /// logged (`traffic`). async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> { let message = if pong { - ParsedMessage::Pong(nonce) + ParsedMessage::Pong { + nonce, + at: self.clock.now_millis(), + } } else { ParsedMessage::Ping(nonce) }; @@ -711,6 +903,12 @@ impl Paths { Ok(()) => traffic.sent(address, encoded.kind, encoded.payload_len), Err(_) => traffic.refused(address, encoded.kind, encoded.payload_len), } + // Logged too, under the address: the PINGs and PONGs that come back go up to + // the node and are logged there, and an exchange is seen whole. + let way = if sent.is_ok() { "sent" } else { "refused" }; + traffic + .log + .note(way, address, &message, encoded.payload_len); } sent } @@ -762,6 +960,7 @@ impl Paths { tracing::trace!(from = %address, "Firewall dropped inbound message"); return Ok(()); } + debug!(from = %address, bytes = payload_len, message = ?message, "Received message"); match message { ParsedMessage::Ping(nonce) => { let paths = Arc::clone(&self); @@ -772,10 +971,15 @@ impl Paths { } }); } - ParsedMessage::Pong(nonce) => { - self.table - .lock_or_recover() - .pong(&address, nonce, Instant::now()); + ParsedMessage::Pong { nonce, at } => { + let heard = self.clock.now_millis(); + let location = self.location(&address); + let now = Instant::now(); + let mut table = self.table.lock_or_recover(); + table.located(&address, &location); + if let Some(rtt) = table.pong(&address, nonce, now) { + table.clock(&address, ClockOffset::of(at, heard, rtt), now); + } } _ => {} } @@ -891,6 +1095,8 @@ mod tests { ], None, None, + Arc::default(), + Duration::ZERO, )) }; (on_both("a"), on_both("b")) @@ -952,27 +1158,39 @@ mod tests { 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"]); + assert_eq!( + table.rank(set.clone(), now, true), + ["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.rank(set.clone(), now, true), + ["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.rank(set.clone(), now, true), + ["mycelium:x", "libp2p:x"] + ); + assert_eq!( + table.rank(set.clone(), now, true), + ["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); + table.rank(lan.clone(), now, true); 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"]); + assert_eq!(table.rank(lan, now, true), ["libp2p:y", "mycelium:y"]); } #[test] @@ -980,14 +1198,17 @@ mod tests { let mut table = Table::default(); let mut now = Instant::now(); let set = paths(&["libp2p:x", "mycelium:x"]); - table.rank(set.clone(), now); + table.rank(set.clone(), now, true); // 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); + table.rank(set.clone(), now, true); } assert!(table.info("libp2p:x", now).down); - assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + assert_eq!( + table.rank(set.clone(), now, true), + ["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); @@ -996,10 +1217,36 @@ mod tests { 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"); + table.sent("mycelium:x", now); assert!(!table.info("mycelium:x", now).down); } + /// A look at the choice (`/peers`, figures) changes nothing: no path counts as + /// used, and no faster one is switched to. + #[test] + fn a_look_changes_nothing() { + let mut table = Table::default(); + let mut now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + assert_eq!( + table.rank(set.clone(), now, false), + ["libp2p:x", "mycelium:x"] + ); + assert!(table.paths.is_empty() && table.chosen.is_empty()); + table.rank(set.clone(), now, true); + for _ in 0..4 { + now = answer(&mut table, now, &[("libp2p:x", 500), ("mycelium:x", 10)]); + } + // Much faster, but only looked at: still the one in use. + assert_eq!( + table.rank(set.clone(), now, false), + ["libp2p:x", "mycelium:x"] + ); + assert_eq!(table.counts.switches, 0); + assert_eq!(table.rank(set, now, true), ["mycelium:x", "libp2p:x"]); + assert_eq!(table.counts.switches, 1); + } + #[test] fn only_paths_in_use_are_pinged() { let mut table = Table::default(); @@ -1010,9 +1257,9 @@ mod tests { nonce }; // One address: nothing to choose, nothing to measure. - table.rank(paths(&["libp2p:z"]), now); + table.rank(paths(&["libp2p:z"]), now, true); assert!(table.due(now, &mut next).is_empty()); - table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now, true); 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()); @@ -1023,12 +1270,65 @@ mod tests { assert!(table.paths.is_empty() && table.chosen.is_empty()); } + /// A PONG's time against the middle of its round trip: the peer's clock, give or + /// take half the round trip; the narrowest bound kept, until it is old. + #[test] + fn clocks_compare_by_the_middle_of_the_round_trip() { + let rtt = Duration::from_millis(200); + let heard = UnixMillis(10_000); + let ahead = ClockOffset::of(UnixMillis(14_900), heard, rtt); + assert_eq!((ahead.offset_ms, ahead.within_ms), (5_000, 101)); + let behind = ClockOffset::of(UnixMillis(6_900), heard, rtt); + assert_eq!(behind.offset_ms, -3_000); + + let mut table = Table::default(); + let now = Instant::now(); + table.sent("libp2p:z", now); + table.clock("libp2p:z", ahead, now); + let wider = ClockOffset::of(UnixMillis(15_000), heard, rtt * 4); + table.clock("libp2p:z", wider, now); + assert_eq!( + table.paths["libp2p:z"].clock.map(|(kept, _)| kept), + Some(ahead) + ); + table.clock("libp2p:z", wider, now + CLOCK_KEPT); + assert_eq!( + table.paths["libp2p:z"].clock.map(|(kept, _)| kept), + Some(wider) + ); + // A clock set anew: a sample the kept one contradicts takes its place at once, + // its bound wider or not. + let set_anew = ClockOffset::of(UnixMillis(25_000), heard, rtt * 8); + table.clock("libp2p:z", set_anew, now + CLOCK_KEPT); + assert_eq!( + table.paths["libp2p:z"].clock.map(|(kept, _)| kept), + Some(set_anew) + ); + } + + /// A peer's only path is measured too once messages go over it, unless its + /// keepalives just did. + #[test] + fn a_path_sent_over_is_pinged() { + let mut table = Table::default(); + let now = Instant::now(); + table.sent("libp2p:z", now); + assert_eq!(table.due(now, || 1), [("libp2p:z".to_owned(), 1)]); + table.pong("libp2p:z", 1, now + Duration::from_millis(20)); + assert!(table.info("libp2p:z", now).rtt_ms.is_some()); + // A keepalive's PING within PROBE_EVERY: none of ours. + let later = now + PROBE_EVERY; + table.pinged("libp2p:z", 2, later); + table.sent("libp2p:z", later); + assert!(table.due(later, || 3).is_empty()); + } + /// A PING's round trip counts from when it goes, not from when it was found due. #[test] fn round_trips_count_from_the_send() { let mut table = Table::default(); let now = Instant::now(); - table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now, true); let due = table.due(now, || 7); assert_eq!(due.len(), 2); // Sent a second late (after another path's slow send). @@ -1040,6 +1340,27 @@ mod tests { assert!((rtt - 100.0).abs() < 1e-6, "{rtt}"); } + /// Round trips of another way to the peer (through a relay, then directly) don't + /// weigh on those of the new one. + #[test] + fn a_path_that_goes_elsewhere_is_measured_anew() { + let mut table = Table::default(); + let now = Instant::now(); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now, true); + let ping = |table: &mut Table, nonce: u64, at: Instant, ms: u64| { + table.pinged("libp2p:x", nonce, at); + table.pong("libp2p:x", nonce, at + Duration::from_millis(ms)); + }; + table.located("libp2p:x", "libp2p:/relayed"); + ping(&mut table, 1, now, 120); + table.located("libp2p:x", "libp2p:/relayed"); + ping(&mut table, 2, now + Duration::from_secs(30), 120); + table.located("libp2p:x", "libp2p:/lan"); + ping(&mut table, 3, now + Duration::from_secs(60), 2); + let rtt = table.info("libp2p:x", now).rtt_ms.unwrap(); + assert!((rtt - 2.0).abs() < 1e-6, "{rtt}"); + } + /// The node's keepalive PING on a path is its probe: paths doesn't ping it too, and /// the PONG measures it. #[test] @@ -1051,7 +1372,7 @@ mod tests { nonce += 1; nonce }; - table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now, true); table.pinged("libp2p:x", 7, now); let due = table.due(now, &mut next); assert_eq!(due.len(), 1); @@ -1105,11 +1426,15 @@ mod tests { vec![("sim", net.transport("b") as Arc<dyn Transport>)], None, Some(Arc::downgrade(&gate)), + Arc::default(), + Duration::ZERO, )); let a = Arc::new(Paths::new( vec![("sim", net.transport("a") as Arc<dyn Transport>)], None, None, + Arc::default(), + Duration::ZERO, )); let received = |paths: &Arc<Paths>| { let (sender, received) = tokio::sync::mpsc::unbounded_channel(); @@ -1167,7 +1492,7 @@ mod tests { Ok(seed.to_owned()) } fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { - crate::transport::mycelium::encode_message(peer, message) + (crate::transport::sim::WIRE.encode)(peer, message) } async fn send_encoded(&self, _message: &EncodedMessage) -> Result<()> { Err(anyhow!("broken")) @@ -1187,6 +1512,7 @@ mod tests { epoch_seq: 0, ready: false, syncs_with_you: false, + leaving: false, }) } @@ -1202,11 +1528,15 @@ mod tests { ], None, None, + Arc::default(), + Duration::ZERO, ); let b = Arc::new(Paths::new( vec![("sim", net.transport("b") as Arc<dyn Transport>)], None, None, + Arc::default(), + Duration::ZERO, )); let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); let handler: Arc<PathHandler> = Arc::new(move |from, _, _| { @@ -1219,13 +1549,13 @@ mod tests { tokio::spawn(async move { b.start_consumer(handler).await }); let addresses = vec!["sim:b".to_owned(), "broken:b".to_owned()]; - let ranked = a.rank(&addresses); + let ranked = a.rank(&addresses, true); assert_eq!(ranked, ["broken:b", "sim:b"]); // The configured order. let mut encoded = a.encode(&ranked[0], &some_status()).unwrap(); encoded.peer = "node-b".to_owned(); assert_eq!(a.send_first(&encoded, &ranked).await.unwrap(), "sim:b"); assert_eq!(received.recv().await.unwrap(), "sim:a"); - assert_eq!(a.rank(&addresses), ["sim:b", "broken:b"]); + assert_eq!(a.rank(&addresses, true), ["sim:b", "broken:b"]); // Without the message kept, nothing else can be tried: the transport's error. encoded.message = None; |