diff options
Diffstat (limited to 'src/swarm/rtt.rs')
| -rw-r--r-- | src/swarm/rtt.rs | 108 |
1 files changed, 76 insertions, 32 deletions
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs index 01d10ee..466fc98 100644 --- a/src/swarm/rtt.rs +++ b/src/swarm/rtt.rs @@ -1,9 +1,11 @@ -//! Round-trip times to peers, measured from traffic we send anyway: STATUS → STATUS_REPLY and -//! HELLO → HELLO_REPLY. No probes of our own. +//! Round-trip times to peers, measured from traffic we send anyway: the keepalive PING +//! → PONG (`swarm::peer::keepalive`), and HELLO → HELLO_REPLY. //! -//! 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. +//! 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 the oldest request still unanswered (the round trip takes at least that long), so @@ -18,11 +20,20 @@ use tokio::time::Instant; #[derive(Debug, Default)] pub(crate) struct RttTable { - statuses: HashMap<String, Outstanding>, + 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, + /// When the oldest PING still unanswered went out (an earlier one, replaced). + waiting_since: Instant, +} + #[derive(Debug, Clone, Copy)] struct Sample { rtt: Duration, @@ -36,13 +47,26 @@ struct Outstanding { } impl RttTable { - pub(crate) fn status_sent(&mut self, peer: &str, now: Instant) { - Self::sent(&mut self.statuses, peer, now); + pub(crate) fn ping_sent(&mut self, peer: &str, nonce: u64, now: Instant) { + let waiting_since = self + .pings + .get(peer) + .map_or(now, |earlier| earlier.waiting_since); + self.pings.insert( + peer.to_owned(), + Ping { + nonce, + sent: now, + waiting_since, + }, + ); } - pub(crate) fn status_reply_received(&mut self, peer: &str, now: Instant) { - if let Some(sample) = Self::answered(&mut self.statuses, peer, now) { - self.record(peer, sample, 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); } } @@ -68,12 +92,15 @@ impl RttTable { .smoothed .get(peer) .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?; - let waiting = [self.statuses.get(peer), self.hellos.get(peer)] - .into_iter() - .flatten() - .map(|outstanding| now.saturating_duration_since(outstanding.sent)) - .max() - .unwrap_or_default(); + let waiting = [ + self.pings.get(peer).map(|ping| ping.waiting_since), + self.hellos.get(peer).map(|outstanding| outstanding.sent), + ] + .into_iter() + .flatten() + .map(|sent| now.saturating_duration_since(sent)) + .max() + .unwrap_or_default(); Some(sample.rtt.max(waiting)) } @@ -128,31 +155,47 @@ mod tests { let mut table = RttTable::default(); let start = Instant::now(); let at = |ms| start + Duration::from_millis(ms); - table.status_sent("a", start); - table.status_reply_received("a", at(100)); + table.ping_sent("a", 1, start); + table.pong_received("a", 1, at(100)); assert_eq!( table.effective("a", at(100), FRESH), Some(Duration::from_millis(100)) ); - table.status_sent("a", at(100)); - table.status_reply_received("a", at(300)); + table.ping_sent("a", 2, at(100)); + table.pong_received("a", 2, at(300)); assert_eq!( table.effective("a", at(300), FRESH), Some(Duration::from_millis(125)) ); - // Two STATUSes out: the STATUS_REPLY can't be attributed. - table.status_sent("a", at(300)); - table.status_sent("a", at(400)); - table.status_reply_received("a", at(450)); + // 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)); assert_eq!( - table.effective("a", at(450), FRESH), + 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 STATUS_REPLY changes nothing. - table.status_reply_received("b", start); + // An unsolicited PONG changes nothing. + table.pong_received("b", 9, start); assert_eq!(table.effective("b", start, FRESH), None); } @@ -160,12 +203,13 @@ mod tests { fn unanswered_requests_raise_it_and_old_samples_expire() { let mut table = RttTable::default(); let start = Instant::now(); - table.status_sent("a", start); - table.status_reply_received("a", start + Duration::from_millis(50)); + table.ping_sent("a", 1, start); + table.pong_received("a", 1, start + Duration::from_millis(50)); - // A STATUS unanswered for 40 s: at least 40 s. + // PINGs unanswered for 40 s: at least 40 s, from the oldest. let sent = start + Duration::from_secs(60); - table.status_sent("a", sent); + table.ping_sent("a", 2, sent); + table.ping_sent("a", 3, sent + Duration::from_secs(30)); let later = sent + Duration::from_secs(40); assert_eq!( table.effective("a", later, FRESH), |