diff options
Diffstat (limited to 'src/swarm/rtt.rs')
| -rw-r--r-- | src/swarm/rtt.rs | 177 |
1 files changed, 85 insertions, 92 deletions
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs index ed6cc73..cf91ac2 100644 --- a/src/swarm/rtt.rs +++ b/src/swarm/rtt.rs @@ -1,15 +1,16 @@ -//! Round-trip times to peers, measured from traffic we send anyway: PING → PONG and -//! HELLO → HELLO_REPLY. No probes of our own. +//! 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 sample only counts when exactly one request 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 the oldest request still unanswered (the round trip takes at least that long), so -//! a peer that stops answering drifts away by itself. 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; @@ -18,9 +19,11 @@ use tokio::time::Instant; #[derive(Debug, Default)] pub(crate) struct RttTable { - pings: HashMap<String, Outstanding>, hellos: HashMap<String, Outstanding>, - smoothed: HashMap<String, Sample>, + /// From the paths layer, as last measured. + from_paths: HashMap<String, Sample>, + /// From HELLOs, smoothed. + from_hellos: HashMap<String, Sample>, } #[derive(Debug, Clone, Copy)] @@ -36,85 +39,73 @@ struct Outstanding { } impl RttTable { - pub(crate) fn ping_sent(&mut self, peer: &str, now: Instant) { - Self::sent(&mut self.pings, peer, now); - } - - pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) { - if let Some(sample) = Self::answered(&mut self.pings, peer, now) { - self.record(peer, sample, 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 + let fresh = |sample: &&Sample| now.saturating_duration_since(sample.at) < fresh_for; + self.from_paths .get(peer) - .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?; - let waiting = [self.pings.get(peer), self.hellos.get(peer)] - .into_iter() - .flatten() - .map(|outstanding| now.saturating_duration_since(outstanding.sent)) - .max() - .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)] @@ -124,55 +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", start); - table.pong_received("a", 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", at(100)); - table.pong_received("a", 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)) ); - - // Two PINGs out: the PONG can't be attributed. - table.ping_sent("a", at(300)); - table.ping_sent("a", at(400)); - table.pong_received("a", at(450)); + // 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(450), FRESH), + table.effective("a", at(410), FRESH), Some(Duration::from_millis(125)) ); - - // An unsolicited PONG changes nothing. - table.pong_received("b", start); + // An unsolicited HELLO_REPLY changes nothing. + table.hello_reply_received("b", start); assert_eq!(table.effective("b", start, FRESH), None); } #[test] - fn unanswered_requests_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", start); - table.pong_received("a", start + Duration::from_millis(50)); - - // A PING unanswered for 40 s: at least 40 s. - let sent = start + Duration::from_secs(60); - table.ping_sent("a", sent); - 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_secs(40)) + table.effective("a", start, FRESH), + Some(Duration::from_millis(20)) ); - - // No sample within the freshness window: unknown. + // A HELLO unanswered for long raises nothing. + table.hello_sent("a", start); + assert_eq!( + table.effective("a", start + Duration::from_secs(40), FRESH), + Some(Duration::from_millis(20)) + ); + 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); } |