From 0e13c368f6214dabe99c987a98a990d4e4e0b3c8 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 03:37:29 +0100 Subject: looking at paths doesn't change them --- src/transport/identity.rs | 37 +++++++++------- src/transport/paths.rs | 111 ++++++++++++++++++++++++++++++++++------------ 2 files changed, 104 insertions(+), 44 deletions(-) diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 886f601..7ce8832 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -453,24 +453,25 @@ impl Shared { } /// Where to send to `peer`: an address it proved itself at, else one it is said to - /// be at; the most preferred transport's first. - fn destination(&self, peer: &str) -> Option { - self.proven_at(peer) + /// be at; the most preferred transport's first. `sending` as `Paths::rank` has it: + /// for a message about to go, or only a look. + fn destination(&self, peer: &str, sending: bool) -> Option { + self.proven_at(peer, sending) .into_iter() .next() - .or_else(|| self.hinted_at(peer).into_iter().next()) + .or_else(|| self.hinted_at(peer, sending).into_iter().next()) } /// The addresses `peer` proved itself at, best first (`Paths::rank`). - fn proven_at(&self, peer: &str) -> Vec { + fn proven_at(&self, peer: &str, sending: bool) -> Vec { let proven = self.lock().proven.get(peer).cloned().unwrap_or_default(); - self.paths.rank(&proven) + self.paths.rank(&proven, sending) } /// Where `peer` is said to be, best first. - fn hinted_at(&self, peer: &str) -> Vec { + fn hinted_at(&self, peer: &str, sending: bool) -> Vec { let hints = self.lock().hints.get(peer).cloned().unwrap_or_default(); - self.paths.rank(&hints) + self.paths.rank(&hints, sending) } } @@ -551,7 +552,7 @@ impl Transport for IdentityTransport { } fn can_reach(&self, peer: &str) -> bool { - self.shared.destination(peer).is_some() + self.shared.destination(peer, false).is_some() } fn transports(&self) -> Vec { @@ -566,10 +567,13 @@ impl Transport for IdentityTransport { /// 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 { - let proven = self.shared.proven_at(peer); + let proven = self.shared.proven_at(peer, false); let (address, is_proven) = match proven.first() { Some(address) => (address.clone(), true), - None => (self.shared.hinted_at(peer).into_iter().next()?, false), + None => ( + self.shared.hinted_at(peer, false).into_iter().next()?, + false, + ), }; let connection = self.shared.paths.connection(&address); let punching = match connection { @@ -592,7 +596,7 @@ impl Transport for IdentityTransport { /// The best path's, where messages to `peer` go. fn rtt(&self, peer: &str) -> Option { - let address = self.shared.proven_at(peer).into_iter().next()?; + let address = self.shared.proven_at(peer, false).into_iter().next()?; self.shared.paths.rtt(&address) } @@ -620,7 +624,7 @@ impl Transport for IdentityTransport { /// first that does), asked WHO. async fn resolve_seed(&self, seed: &str) -> Result { if identity::is_node_id(seed) { - if self.shared.destination(seed).is_some() { + if self.shared.destination(seed, false).is_some() { return Ok(seed.to_owned()); } bail!("a node ID alone doesn't say where that node is; give its address"); @@ -656,9 +660,10 @@ impl Transport for IdentityTransport { } /// For the address `destination` picks (a send there is refused until `peer` - /// proved itself there, `send_encoded`). + /// proved itself there, `send_encoded`). Only a look at the choice: the send makes + /// it (encoding again for another path if it changes). fn encode(&self, peer: &str, message: &ParsedMessage) -> Result { - let Some(address) = self.shared.destination(peer) else { + let Some(address) = self.shared.destination(peer, false) else { bail!("no address known for {peer} on this node's transports"); }; let mut encoded = self.shared.paths.encode(&address, message)?; @@ -669,7 +674,7 @@ impl Transport for IdentityTransport { /// To the addresses `peer` proved itself at, best first, the next when a send /// fails (`Paths::send_first`); refused if it proved itself nowhere yet. async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { - let proven = self.shared.proven_at(&message.peer); + let proven = self.shared.proven_at(&message.peer, true); if !proven.is_empty() { self.shared.ask_other_paths(&message.peer); } diff --git a/src/transport/paths.rs b/src/transport/paths.rs index b9dfc96..2e7ef76 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -147,19 +147,28 @@ struct Table { 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 { + /// `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 { 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); @@ -190,6 +199,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)) @@ -204,7 +214,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 @@ -456,14 +466,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 { + /// 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 { 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. @@ -1006,27 +1020,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] @@ -1034,14 +1060,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); @@ -1054,6 +1083,32 @@ mod tests { 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(); @@ -1064,9 +1119,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()); @@ -1099,7 +1154,7 @@ mod tests { 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). @@ -1117,7 +1172,7 @@ mod tests { 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); + 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)); @@ -1143,7 +1198,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); @@ -1312,13 +1367,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; -- cgit v1.3.1