summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/peers.rs4
-rw-r--r--src/api/stats/prometheus.rs2
-rw-r--r--src/swarm/peer/keepalive.rs15
-rw-r--r--src/swarm/rtt.rs209
-rw-r--r--src/transport/identity.rs6
-rw-r--r--src/transport/mod.rs7
-rw-r--r--src/transport/paths.rs5
-rw-r--r--src/transport/traffic.rs4
8 files changed, 116 insertions, 136 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs
index f2281c2..a653f7d 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -69,8 +69,8 @@ pub(crate) struct PeerStatus {
heads: Option<usize>,
/// We keep it alive and sync with it on our own; other peers we only answer.
neighbour: bool,
- /// Effective round-trip time in milliseconds (smoothed, raised by a HELLO still
- /// unanswered); absent without a recent sample.
+ /// Round-trip time in milliseconds of the path messages to it go over (`swarm::rtt`);
+ /// absent without a recent value.
rtt_ms: Option<f64>,
/// Messages and payload bytes exchanged with it, by kind, since we started (or
/// since it was last listed).
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 911228c..559a23e 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -808,7 +808,7 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
out.family(
"peer_rtt_seconds",
"gauge",
- "Effective round-trip time (smoothed, raised by an unanswered HELLO).",
+ "Round-trip time of the path messages to the peer go over (before that, of HELLOs).",
);
for peer in &peers {
if let Some(rtt) = peer.rtt_secs {
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index efb8a65..c8a93eb 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -2,7 +2,7 @@
//!
//! A quiet pair's keepalive (`PeerState::take_keepalive_due`) is a PING, answered with a
//! PONG by the peer's paths layer (`transport::paths`), over the same path: it tells both
-//! sides the other is alive, and gives the round trip (`swarm::rtt`). Every `Timing::status_refresh` it is a
+//! sides the other is alive, and measures the path (`swarm::rtt`). Every `Timing::status_refresh` it is a
//! STATUS instead, answered with a STATUS_REPLY: each side's state (heads, readiness,
//! epoch), which also go out as they change (relay announcements, the epoch after a
//! switch, readiness: `announce`). When a STATUS shows our documents differ while sync
@@ -33,18 +33,17 @@ pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
.await
}
-/// Sends `peer` a PING, noting it for its round trip.
+/// Sends `peer` a PING (the paths layer measures the path with it).
pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
let nonce = crate::transport::paths::random_nonce();
- state.lock_rtt().ping_sent(peer, nonce, Instant::now());
state
.transport
.send_message(peer, &ParsedMessage::Ping(nonce))
.await
}
-/// A PING or PONG from `peer`: it is alive, and a PONG may answer our PING (its round
-/// trip). A transport without the paths layer (`Transport::answers_pings`: in tests)
+/// A PING or PONG from `peer`: it is alive, and after a PONG the paths layer has a
+/// new round-trip time to it (`swarm::rtt`). A transport without the paths layer (`Transport::answers_pings`: in tests)
/// leaves answering a PING to us. Only peers we know already: a PING makes none.
pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMessage) {
let now = Instant::now();
@@ -55,7 +54,11 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes
debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
}
}
- ParsedMessage::Pong(nonce) => state.lock_rtt().pong_received(peer, *nonce, now),
+ ParsedMessage::Pong(_) => {
+ if let Some(rtt) = state.transport.rtt(peer) {
+ state.lock_rtt().measured(peer, rtt, now);
+ }
+ }
_ => {}
}
let Some(peer_state) = state.get_peer_state(peer).await else {
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);
}
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 153dacc..0c4cda4 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -496,6 +496,12 @@ impl Transport for IdentityTransport {
true
}
+ /// The best path's, where messages to `peer` go.
+ fn rtt(&self, peer: &str) -> Option<Duration> {
+ let address = self.shared.proven_at(peer).into_iter().next()?;
+ self.shared.paths.rtt(&address)
+ }
+
/// The identity layer's, and each transport's.
fn stats(&self) -> TransportStats {
let mut stats = self.shared.paths.stats();
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index f60a9f6..80b71cd 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -649,6 +649,13 @@ pub(crate) trait Transport: Send + Sync {
None
}
+ /// The smoothed round-trip time of the path messages to `peer` go over, if
+ /// measured (`transport::paths`). By default none.
+ fn rtt(&self, peer: &str) -> Option<std::time::Duration> {
+ let _ = peer;
+ None
+ }
+
/// Whether the transport answers PINGs with PONGs itself (the paths layer,
/// `transport::paths`); if not (a bare transport, in tests), the node does. By
/// default not.
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 643a238..a9bac49 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -420,6 +420,11 @@ impl Paths {
self.table.lock_or_recover().rank(usable, Instant::now())
}
+ /// The smoothed round-trip time of the path to `address`, if measured.
+ pub(crate) fn rtt(&self, address: &str) -> Option<Duration> {
+ self.table.lock_or_recover().paths.get(address)?.rtt
+ }
+
/// What paths knows about each of `addresses`.
pub(crate) fn info(&self, addresses: &[String]) -> Vec<PathInfo> {
let now = Instant::now();
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index b070023..64595cb 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -192,6 +192,10 @@ impl Transport for CountingTransport {
self.inner.answers_pings()
}
+ fn rtt(&self, peer: &str) -> Option<std::time::Duration> {
+ self.inner.rtt(peer)
+ }
+
async fn resolve_seed(&self, seed: &str) -> Result<String> {
self.inner.resolve_seed(seed).await
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 15:09:41 +0000