diff options
Diffstat (limited to 'src/swarm/rtt.rs')
| -rw-r--r-- | src/swarm/rtt.rs | 209 |
1 files changed, 82 insertions, 127 deletions
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs index 04d7c8d..cf91ac2 100644 --- a/src/swarm/rtt.rs +++ b/src/swarm/rtt.rs @@ -1,19 +1,16 @@ -//! Round-trip times to peers, measured from traffic we send anyway: the keepalive PING -//! → PONG (`swarm::peer::keepalive`), and HELLO → HELLO_REPLY. +//! Round-trip times to peers. Mostly the paths layer's (`transport::paths`): the +//! smoothed round-trip time of the path a peer's messages go over, measured by PINGs +//! (keepalives, and the paths layer's own), taken whenever a PONG from the peer arrives +//! (`measured`). Before a peer's path is measured (while joining: no keepalives yet), +//! HELLO → HELLO_REPLY gives one, smoothed (exponential moving average, weight 1/4); +//! a HELLO_REPLY only counts when exactly one HELLO was outstanding: with two, we can't +//! tell which one the answer belongs to (an answer to the older one would look too +//! fast). //! -//! A PONG answers the PING with its nonce, so it is attributed exactly; a PONG to an -//! older PING (one a newer one replaced) is no sample. A HELLO_REPLY only counts when -//! exactly one HELLO was outstanding: with two, we can't tell which one the answer -//! belongs to (an answer to the older one would look too fast). Samples are smoothed -//! (exponential moving average, weight 1/4) to ride out jitter. -//! -//! What we use is the *effective* round-trip time: the smoothed value, raised to the age -//! of a HELLO still unanswered (the round trip takes at least that long), so a peer that -//! stops answering its handshake drifts away by itself. An unanswered PING isn't: a lost -//! PING or PONG is no sample, not a slow one (the next keepalive measures again, and a -//! peer that is gone goes dormant). Without a recent sample (nodes move; an old value -//! says little) it is unknown. Silence alone doesn't count: we don't ask non-neighbours -//! anything, so their quiet is ours. +//! Without a recent value (nodes move; an old value says little) it is unknown. A lost +//! PING or HELLO is no sample: a peer that stops answering goes dormant, and is given up +//! as a catch-up source, by itself. Silence alone doesn't count: we don't ask +//! non-neighbours anything, so their quiet is ours. use std::collections::HashMap; use std::time::Duration; @@ -22,16 +19,11 @@ use tokio::time::Instant; #[derive(Debug, Default)] pub(crate) struct RttTable { - pings: HashMap<String, Ping>, hellos: HashMap<String, Outstanding>, - smoothed: HashMap<String, Sample>, -} - -/// The PING out to a peer. -#[derive(Debug, Clone, Copy)] -struct Ping { - nonce: u64, - sent: Instant, + /// From the paths layer, as last measured. + from_paths: HashMap<String, Sample>, + /// From HELLOs, smoothed. + from_hellos: HashMap<String, Sample>, } #[derive(Debug, Clone, Copy)] @@ -47,88 +39,73 @@ struct Outstanding { } impl RttTable { - /// A PING to `peer`: it replaces one still unanswered. - pub(crate) fn ping_sent(&mut self, peer: &str, nonce: u64, now: Instant) { - self.pings - .insert(peer.to_owned(), Ping { nonce, sent: now }); - } - - pub(crate) fn pong_received(&mut self, peer: &str, nonce: u64, now: Instant) { - if self.pings.get(peer).is_some_and(|ping| ping.nonce == nonce) - && let Some(ping) = self.pings.remove(peer) - { - self.record(peer, now.saturating_duration_since(ping.sent), now); - } + /// The paths layer's round-trip time to `peer` (`Transport::rtt`), as a PONG from + /// it arrived. + pub(crate) fn measured(&mut self, peer: &str, rtt: Duration, now: Instant) { + self.from_paths + .insert(peer.to_owned(), Sample { rtt, at: now }); } pub(crate) fn hello_sent(&mut self, peer: &str, now: Instant) { - Self::sent(&mut self.hellos, peer, now); + self.hellos + .entry(peer.to_owned()) + .and_modify(|outstanding| outstanding.count += 1) + .or_insert(Outstanding { + sent: now, + count: 1, + }); } pub(crate) fn hello_reply_received(&mut self, peer: &str, now: Instant) { - if let Some(sample) = Self::answered(&mut self.hellos, peer, now) { - self.record(peer, sample, now); + let Some(outstanding) = self.hellos.remove(peer) else { + return; + }; + if outstanding.count != 1 { + return; } + let sample = now.saturating_duration_since(outstanding.sent); + let rtt = match self.from_hellos.get(peer) { + Some(old) => (old.rtt * 3 + sample) / 4, + None => sample, + }; + self.from_hellos + .insert(peer.to_owned(), Sample { rtt, at: now }); } - /// The effective round-trip time: the smoothed value if sampled within `fresh_for`, - /// raised to the age of the oldest unanswered request; None without a recent sample. + /// The round-trip time: the paths layer's if measured within `fresh_for`, else + /// HELLOs' if sampled within it; None without either. pub(crate) fn effective( &self, peer: &str, now: Instant, fresh_for: Duration, ) -> Option<Duration> { - let sample = self - .smoothed - .get(peer) - .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?; - let waiting = self - .hellos + let fresh = |sample: &&Sample| now.saturating_duration_since(sample.at) < fresh_for; + self.from_paths .get(peer) - .map(|outstanding| now.saturating_duration_since(outstanding.sent)) - .unwrap_or_default(); - Some(sample.rtt.max(waiting)) + .filter(fresh) + .or_else(|| self.from_hellos.get(peer).filter(fresh)) + .map(|sample| sample.rtt) } - /// The effective round-trip time of every peer that has one. + /// The round-trip time of every peer that has one. pub(crate) fn all_effective( &self, now: Instant, fresh_for: Duration, ) -> impl Iterator<Item = (&str, Duration)> { - self.smoothed.keys().filter_map(move |peer| { + let mut peers: Vec<&String> = self + .from_paths + .keys() + .chain(self.from_hellos.keys()) + .collect(); + peers.sort_unstable(); + peers.dedup(); + peers.into_iter().filter_map(move |peer| { self.effective(peer, now, fresh_for) .map(|rtt| (peer.as_str(), rtt)) }) } - - fn sent(map: &mut HashMap<String, Outstanding>, peer: &str, now: Instant) { - map.entry(peer.to_owned()) - .and_modify(|outstanding| outstanding.count += 1) - .or_insert(Outstanding { - sent: now, - count: 1, - }); - } - - fn answered( - map: &mut HashMap<String, Outstanding>, - peer: &str, - now: Instant, - ) -> Option<Duration> { - let outstanding = map.remove(peer)?; - (outstanding.count == 1).then(|| now.saturating_duration_since(outstanding.sent)) - } - - fn record(&mut self, peer: &str, sample: Duration, now: Instant) { - let rtt = match self.smoothed.get(peer) { - Some(old) => (old.rtt * 3 + sample) / 4, - None => sample, - }; - self.smoothed - .insert(peer.to_owned(), Sample { rtt, at: now }); - } } #[cfg(test)] @@ -138,79 +115,57 @@ mod tests { const FRESH: Duration = Duration::from_secs(300); #[test] - fn samples_only_unambiguous_answers_and_smooths() { + fn hellos_sample_only_unambiguous_answers_and_smooth() { let mut table = RttTable::default(); let start = Instant::now(); let at = |ms| start + Duration::from_millis(ms); - table.ping_sent("a", 1, start); - table.pong_received("a", 1, at(100)); + table.hello_sent("a", start); + table.hello_reply_received("a", at(100)); assert_eq!( table.effective("a", at(100), FRESH), Some(Duration::from_millis(100)) ); - - table.ping_sent("a", 2, at(100)); - table.pong_received("a", 2, at(300)); + table.hello_sent("a", at(100)); + table.hello_reply_received("a", at(300)); assert_eq!( table.effective("a", at(300), FRESH), Some(Duration::from_millis(125)) ); - - // A newer PING replaced it: a PONG to the older one is no sample. - table.ping_sent("a", 3, at(300)); - table.ping_sent("a", 4, at(400)); - table.pong_received("a", 3, at(410)); + // Two HELLOs out: the HELLO_REPLY can't be attributed. + table.hello_sent("a", at(300)); + table.hello_sent("a", at(400)); + table.hello_reply_received("a", at(410)); assert_eq!( table.effective("a", at(410), FRESH), Some(Duration::from_millis(125)) ); - table.pong_received("a", 4, at(500)); - assert_eq!( - table.effective("a", at(500), FRESH), - Some(Duration::from_micros(118_750)) // (125 * 3 + 100) / 4 - ); - - // Two HELLOs out: the HELLO_REPLY can't be attributed. - table.hello_sent("c", start); - table.hello_reply_received("c", at(100)); - table.hello_sent("c", at(100)); - table.hello_sent("c", at(200)); - table.hello_reply_received("c", at(210)); - assert_eq!( - table.effective("c", at(210), FRESH), - Some(Duration::from_millis(100)) - ); - - // An unsolicited PONG changes nothing. - table.pong_received("b", 9, start); + // An unsolicited HELLO_REPLY changes nothing. + table.hello_reply_received("b", start); assert_eq!(table.effective("b", start, FRESH), None); } #[test] - fn unanswered_hellos_raise_it_and_old_samples_expire() { + fn the_path_wins_nothing_waits_and_old_values_expire() { let mut table = RttTable::default(); let start = Instant::now(); - table.ping_sent("a", 1, start); - table.pong_received("a", 1, start + Duration::from_millis(50)); - - // PINGs unanswered (lost): no sample, nothing raised. - let sent = start + Duration::from_secs(60); - table.ping_sent("a", 2, sent); - table.ping_sent("a", 3, sent + Duration::from_secs(30)); - let later = sent + Duration::from_secs(40); + table.hello_sent("a", start); + table.hello_reply_received("a", start + Duration::from_millis(80)); + table.measured("a", Duration::from_millis(20), start); assert_eq!( - table.effective("a", later, FRESH), - Some(Duration::from_millis(50)) + table.effective("a", start, FRESH), + Some(Duration::from_millis(20)) ); - - // A HELLO unanswered for 40 s: at least 40 s. - table.hello_sent("a", sent); + // A HELLO unanswered for long raises nothing. + table.hello_sent("a", start); assert_eq!( - table.effective("a", later, FRESH), - Some(Duration::from_secs(40)) + table.effective("a", start + Duration::from_secs(40), FRESH), + Some(Duration::from_millis(20)) ); - - // No sample within the freshness window: unknown. + assert_eq!( + table.all_effective(start, FRESH).collect::<Vec<_>>(), + [("a", Duration::from_millis(20))] + ); + // No value within the freshness window: unknown. let stale = start + FRESH + Duration::from_secs(1); assert_eq!(table.effective("a", stale, FRESH), None); } |