1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
//! 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).
//!
//! 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;
use tokio::time::Instant;
#[derive(Debug, Default)]
pub(crate) struct RttTable {
hellos: HashMap<String, Outstanding>,
/// From the paths layer, as last measured.
from_paths: HashMap<String, Sample>,
/// From HELLOs, smoothed.
from_hellos: HashMap<String, Sample>,
}
#[derive(Debug, Clone, Copy)]
struct Sample {
rtt: Duration,
at: Instant,
}
#[derive(Debug, Clone, Copy)]
struct Outstanding {
sent: Instant,
count: u32,
}
impl RttTable {
/// 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.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) {
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 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 fresh = |sample: &&Sample| now.saturating_duration_since(sample.at) < fresh_for;
self.from_paths
.get(peer)
.filter(fresh)
.or_else(|| self.from_hellos.get(peer).filter(fresh))
.map(|sample| sample.rtt)
}
/// 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)> {
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))
})
}
}
#[cfg(test)]
mod tests {
use super::*;
const FRESH: Duration = Duration::from_secs(300);
#[test]
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.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.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 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))
);
// An unsolicited HELLO_REPLY changes nothing.
table.hello_reply_received("b", start);
assert_eq!(table.effective("b", start, FRESH), None);
}
#[test]
fn the_path_wins_nothing_waits_and_old_values_expire() {
let mut table = RttTable::default();
let start = Instant::now();
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", start, FRESH),
Some(Duration::from_millis(20))
);
// 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);
}
}
|
