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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
|
//! The peer worker: keepalives due, dormancy, HELLO retries and overdue sync replies,
//! when the earliest of them is due (or something may have made one due sooner,
//! `Due`); every `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready),
//! forget records, the member registry, stability, measurements and the state file;
//! forget records also at once when they changed.
use std::time::Duration;
use tokio::time::Instant;
use tracing::{debug, info, warn};
use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
use super::handshake::retry_hello_if_due;
use super::keepalive::{send_ping, send_status};
use crate::locks::MutexExt;
use crate::state::{AppState, Keepalive};
use crate::swarm::topology;
/// The least time between two passes, however often something is due.
const MIN_PASS_GAP: Duration = Duration::from_millis(20);
const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);
/// When the peer worker is next due. What makes something of its due sooner than it
/// knows (a HELLO sent, a sync message waiting for its answer, a keepalive received, a
/// new peer, new neighbours) says so (`by`); what only puts things off needn't.
#[derive(Default)]
pub(crate) struct Due {
/// Its next pass while it waits; None while it runs.
next: std::sync::Mutex<Option<Instant>>,
wake: tokio::sync::Notify,
}
impl Due {
/// Something is due at `at`: a pass then at the latest.
pub(crate) fn by(&self, at: Instant) {
if self.next.lock_or_recover().is_none_or(|next| at < next) {
self.wake.notify_one();
}
}
/// A pass now.
pub(crate) fn now(&self) {
self.wake.notify_one();
}
fn passing(&self) {
*self.next.lock_or_recover() = None;
}
/// Until `at`, or until something is due sooner.
async fn until(&self, at: Instant) {
*self.next.lock_or_recover() = Some(at);
tokio::select! {
() = self.wake.notified() => {}
() = tokio::time::sleep_until(at) => {}
}
}
}
pub(crate) async fn run_peer_worker(state: AppState) {
info!("Starting peer maintenance worker");
let mut last_forget_check: Option<Instant> = None;
let mut forgets_applied = AppliedForgets::new();
let mut was_ready = false;
loop {
state.peers_due.passing();
let mut next = Instant::now() + FORGET_CHECK_INTERVAL;
// While joining, keep handshakes fast; dormancy would slow probes to minutes.
let ready = state.is_ready().await;
if ready && !was_ready {
// Just became ready: register in the member list (and do the other periodic
// checks) now rather than at the next check. Until it is registered, other
// members don't count this node as a neighbour.
last_forget_check = None;
}
was_ready = ready;
let peers = state.peers_snapshot().await;
for (peer, peer_state_mutex) in &peers {
let now = Instant::now();
// Only neighbours are kept alive and judged; others we only answer.
let neighbour = topology::is_neighbour(&state, peer);
let (became_dormant, keepalive, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
let became_dormant = ready
&& neighbour
&& topology::change_peer(&state, &mut peer_state, |p| {
p.check_dormancy(&state.timing, now)
});
let keepalive = if neighbour {
peer_state.take_keepalive_due(&state.timing, now)
} else {
None
};
if !neighbour {
peer_state.stop_hello_retries();
}
let unanswered = peer_state.take_overdue_sync_reply(now);
(became_dormant, keepalive, unanswered)
};
if let Some(changes) = unanswered {
debug!(peer = %peer, changes, "No answer to our sync message; sending again");
crate::swarm::source::reply_timed_out(&state, peer);
state.request_sync(peer);
}
if became_dormant {
warn!(
peer = %peer,
silent_secs = state.timing.dormant_after.as_secs(),
probe_secs = state.timing.dormant_probe.as_secs(),
"Peer went dormant"
);
}
// The catch-up source gets STATUSes: we need its readiness, and whether it
// syncs with us.
let keepalive = match keepalive {
None if crate::swarm::source::probe_due(&state, peer, Instant::now()) => {
Some(Keepalive::Status)
}
keepalive => keepalive,
};
let sent = match keepalive {
Some(Keepalive::Ping) => send_ping(&state, peer).await,
Some(Keepalive::Status) => send_status(&state, peer).await,
None => Ok(()),
};
if let Err(error) = sent {
debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed");
}
if let Err(error) = retry_hello_if_due(&state, peer).await {
warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
}
let due = peer_state_mutex
.lock()
.await
.next_due(&state.timing, neighbour, ready);
let probe = crate::swarm::source::probe_at(&state, peer);
next = [Some(next), due, probe]
.into_iter()
.flatten()
.min()
.unwrap_or(next);
}
let records_changed = state
.forget_records_changed
.swap(false, std::sync::atomic::Ordering::AcqRel);
let periodic = last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL);
if records_changed || periodic {
apply_forget_records(&state, &mut forgets_applied).await;
}
if periodic {
forget_strangers(&state).await;
crate::swarm::members::ensure_registered(&state).await;
crate::swarm::stability::tick(&state).await;
crate::swarm::skew::check(&state).await;
crate::swarm::stats::measure_saved_size(&state);
crate::swarm::source::warn_if_stalled(&state);
crate::swarm::state_file::update(&state, false).await;
{
let known = state.peer_states.read().await;
// And their addresses, where the layers below the node count
// (`Traffic::node`).
let addresses: std::collections::HashSet<String> = known
.keys()
.flat_map(|peer| state.transport.identities_of(peer))
.collect();
state
.traffic
.retain(|peer| known.contains_key(peer) || addresses.contains(peer));
state
.throughput
.lock_or_recover()
.retain(|peer| known.contains_key(peer));
}
// Fallback for what doesn't notify (`topology::run_worker`).
state.topology_changed.notify_one();
last_forget_check = Some(Instant::now());
}
if let Some(checked) = last_forget_check {
next = next.min(checked + FORGET_CHECK_INTERVAL);
}
let passed = Instant::now();
state.peers_due.until(next).await;
tokio::time::sleep_until(passed + MIN_PASS_GAP).await;
}
}
|
