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
#[cfg(test)]
use super::Ordering;
use super::{ConnectionRuntime, Duration, Instant, PoisonError, ServerError, lock, recover_lock};
impl ConnectionRuntime {
/// Bumps the drain-completion generation and wakes the drain/settle waiter.
/// Called from [`Self::remove`] on every route that actually drops a record.
pub(super) fn signal_connection_removed(&self) {
// Bump the generation under the lock, release it, THEN notify. A waiter
// holds this lock continuously from its generation re-check through the
// atomic release inside `wait_timeout`, so it can never miss a bump
// published before the notify — the Condvar lost-wakeup contract holds
// without notifying under the guard.
{
let mut generation = recover_lock(&self.drain_generation);
*generation = generation.wrapping_add(1);
}
self.drain_removed.notify_all();
}
/// Parks the calling thread until every tracked connection has been removed
/// or `deadline` elapses, returning `true` when the drain completed and
/// `false` when the single admitted deadline won. This is the TOLD
/// replacement (W4 leg 3, §4.3) for the retired reap/count/sleep drain loop:
/// it never samples completion on a timer. Completion is observed only on a
/// delivered connection-removal wake (composed from the one `remove()` funnel,
/// so a Died/Detached/crash exit and an orderly close decrement through the
/// same path — oracle 15) or the one deadline. Force-close settle reuses this
/// same waiter with its own deadline rather than a second poll loop (oracle
/// 14).
pub(in crate::server::connection) fn wait_for_active_connections_drained(
&self,
deadline: Instant,
) -> bool {
loop {
// ARM before OBSERVE: snapshot the removal generation first, so an exit
// delivered after the completion observation below and before the park
// bumps a generation the park detects — it is never lost (oracle 18).
let snapshot = self.drain_generation_snapshot();
// OBSERVE completion first: a last exit that reaches zero wins a tie
// with a simultaneously elapsed deadline (oracle 19).
if self.active_count() == 0 {
return true;
}
let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
#[cfg(test)]
self.drain_deadline_hits.fetch_add(1, Ordering::SeqCst);
return false;
};
#[cfg(test)]
self.run_drain_park_barrier();
self.park_until_removed_or(snapshot, remaining);
}
}
/// Reads the current removal generation under its mutex.
fn drain_generation_snapshot(&self) -> u64 {
*recover_lock(&self.drain_generation)
}
/// The removal generation, read as a departure count. A poisoned lock is
/// an error here, not recovered: a caller waiting on departures must not
/// read a count a panicking holder may have left half-written.
pub(in crate::server::connection) fn departures(&self) -> Result<u64, ServerError> {
Ok(*lock(&self.drain_generation, "connection departures")?)
}
/// Parks on the removal `Condvar` until the generation has advanced by
/// `count` since `since`, or until `deadline`. `wait_timeout_while` checks
/// the predicate under the lock before it parks, so departures already
/// counted return at once, and spurious wakes park again inside the call.
pub(in crate::server::connection) fn await_departures(
&self,
since: u64,
count: u64,
deadline: Instant,
) -> Result<bool, ServerError> {
let remaining = deadline.saturating_duration_since(Instant::now());
let (generation, _timed_out) = self
.drain_removed
.wait_timeout_while(
lock(&self.drain_generation, "connection departures")?,
remaining,
|current| current.wrapping_sub(since) < count,
)
.map_err(|error| ServerError::ListenerAccept {
message: format!("connection departures unavailable: {error}"),
})?;
let departed = generation.wrapping_sub(since) >= count;
drop(generation);
Ok(departed)
}
/// Parks on the removal `Condvar` for at most `timeout`, but only while no
/// removal has bumped the generation since `snapshot` — the arm-before-observe
/// barrier. `wait_timeout_while` evaluates the predicate under the lock BEFORE
/// waiting, so a generation already advanced (an exit landed between the
/// observation and here) returns immediately without sleeping; spurious wakes
/// re-wait inside the call and never return early.
fn park_until_removed_or(&self, snapshot: u64, timeout: Duration) {
let outcome = self
.drain_removed
.wait_timeout_while(recover_lock(&self.drain_generation), timeout, |current| {
*current == snapshot
})
.unwrap_or_else(PoisonError::into_inner);
// A non-timed-out return means the predicate went false — a real removal
// bumped the generation and woke this park (as opposed to the deadline).
#[cfg(test)]
if !outcome.1.timed_out() {
self.drain_exit_wakes.fetch_add(1, Ordering::SeqCst);
}
// Release the guard promptly; the caller re-loops unlocked.
drop(outcome);
}
}