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
|
//! Every member's clock against the swarm's, for the map: from the clocks each member
//! reports of its peers (`Link::clock`), chained from this node along the narrowest
//! bounds, then taken against their median. A member is off when it differs from the
//! median beyond its chained bound and the limit (`--clock-skew-warn`); with fewer than
//! three members compared, none is (two can't tell whose clock is off). An estimate for
//! the eye: each node judges its own clock by its peers' (`swarm::skew`).
use std::cmp::Reverse;
use std::collections::{BTreeMap, BinaryHeap, HashMap};
use serde::Serialize;
use crate::transport::Link;
/// A member's clock against the swarm's.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(super) struct MemberClock {
/// How far it is ahead of the median (behind if negative).
pub(super) offset_ms: i64,
/// Give or take: the bounds chained to it from this node.
pub(super) within_ms: u64,
pub(super) off: bool,
}
/// From `local` and `reports` (each member's links), with `limit_ms`; by member.
pub(super) fn swarm<'a>(
local: &str,
reports: impl IntoIterator<Item = (&'a str, &'a [Link])>,
limit_ms: u64,
) -> BTreeMap<String, MemberClock> {
// Each way between two members: how far the far end's clock is ahead, give or take.
let mut edges: HashMap<&str, Vec<(&str, i64, u64)>> = HashMap::new();
for (reporter, links) in reports {
for link in links {
if let Some(clock) = link.clock {
edges.entry(reporter).or_default().push((
link.peer.as_str(),
clock.offset_ms,
clock.within_ms,
));
edges.entry(link.peer.as_str()).or_default().push((
reporter,
-clock.offset_ms,
clock.within_ms,
));
}
}
}
// Ahead of this node's, along the narrowest chained bounds (Dijkstra).
let mut reached: HashMap<&str, (i64, u64)> = HashMap::new();
let mut queue = BinaryHeap::from([Reverse((0_u64, local, 0_i64))]);
while let Some(Reverse((within, member, offset))) = queue.pop() {
if reached.contains_key(member) {
continue;
}
reached.insert(member, (offset, within));
for &(next, step, step_within) in edges.get(member).into_iter().flatten() {
if !reached.contains_key(next) {
queue.push(Reverse((
within.saturating_add(step_within),
next,
offset.saturating_add(step),
)));
}
}
}
let mut offsets: Vec<i64> = reached.values().map(|(offset, _)| *offset).collect();
offsets.sort_unstable();
let median = offsets.get(offsets.len() / 2).copied().unwrap_or(0);
let judged = reached.len() >= 3 && limit_ms > 0;
reached
.into_iter()
.map(|(member, (offset, within))| {
let offset_ms = offset - median;
let off = judged && offset_ms.unsigned_abs() > within.saturating_add(limit_ms);
(
member.to_owned(),
MemberClock {
offset_ms,
within_ms: within,
off,
},
)
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::transport::paths::ClockOffset;
fn link(peer: &str, offset_ms: i64) -> Link {
Link {
peer: peer.into(),
transport: None,
relayed: false,
down: false,
rtt_us: None,
dormant: false,
others: Vec::new(),
clock: Some(ClockOffset {
offset_ms,
within_ms: 20,
}),
}
}
/// Chained through the reports: the one clock 10 s ahead is off, wherever it is.
#[test]
fn the_odd_clock_out() {
// a – b – c – d, d's clock 10 s ahead; a only knows b.
let a = [link("b", 0)];
let b = [link("a", 0), link("c", 5)];
let c = [link("d", 10_000)];
let clocks = swarm("a", [("a", &a[..]), ("b", &b[..]), ("c", &c[..])], 2_000);
assert_eq!(clocks.len(), 4);
assert!(clocks["d"].off);
assert!(clocks["d"].offset_ms > 9_900);
assert!(!clocks["a"].off && !clocks["b"].off && !clocks["c"].off);
assert_eq!(clocks["d"].within_ms, 60);
// Two alone: none.
let two = swarm("a", [("a", &[link("b", 10_000)][..])], 2_000);
assert!(two.values().all(|clock| !clock.off));
}
}
|
