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
use super::{ConnectionRuntime, Instant, Ordering, PoisonError, recover_lock};
impl ConnectionRuntime {
/// Bumps the delivery-quiescence generation under its mutex, then wakes the
/// flush barrier — the same lock-then-notify discipline as
/// [`Self::signal_connection_removed`], so a park published before the notify
/// can never be missed by a waiter holding the mutex across its re-check.
pub(super) fn signal_settle_changed(&self) {
{
let mut generation = recover_lock(&self.settle_generation);
*generation = generation.wrapping_add(1);
}
self.settle_changed.notify_all();
}
/// Reads the current delivery-quiescence generation under its mutex.
fn settle_generation_snapshot(&self) -> u64 {
*recover_lock(&self.settle_generation)
}
/// True when every tracked connection is parked with no pending READY edge —
/// i.e. every accepted publish has been pumped to its subscriber's outbound
/// and no fan-out wake is still in flight. An empty registry is trivially
/// quiescent.
fn all_connections_delivery_quiesced(&self) -> bool {
let Ok(records) = self.records.lock() else {
return false;
};
records.connections.values().all(|record| {
record.parked.load(Ordering::Acquire) && !record.ready_pending.load(Ordering::Acquire)
})
}
/// FIX A-ii: wakes every tracked connection once so it drains its socket and
/// pumps its subscriptions. This is what makes the flush barrier robust to the
/// readiness gap: a publisher whose fire-and-forget publish bytes have arrived
/// but whose readiness wake has not yet rescheduled it still looks "parked",
/// so without this it could be sampled as quiescent before it admits and fans
/// out those publishes. Firing sets each connection's `ready_pending` edge, so
/// the quiescence check below cannot pass until every woken connection has run
/// its slice (reading and admitting any buffered publish, whose admission then
/// fires its subscribers in turn) and re-parked.
fn wake_all_connections_for_flush(&self) {
let pids: Vec<u64> = self
.records
.lock()
.map(|records| records.connections.keys().copied().collect())
.unwrap_or_default();
for pid in pids {
if let Some(waker) = self.ready_waker(pid) {
waker.fire();
}
}
}
/// FIX A-ii: parks until every tracked connection has fanned out its accepted
/// publishes (delivery quiescence) or `deadline` elapses, returning `true` on
/// quiescence and `false` when the single admitted deadline won. The TOLD
/// shape mirrors [`Self::wait_for_active_connections_drained`]: arm the
/// barrier, then snapshot-before-observe so a park delivered between the
/// observation and the wait bumps a generation the wait detects. It samples
/// nothing on a timer — it wakes only on a delivered park (generation bump) or
/// the one deadline.
pub(in crate::server::connection) fn wait_for_delivery_quiesced(
&self,
deadline: Instant,
) -> bool {
self.settle_armed.store(true, Ordering::Release);
// Force every connection to run once so a publisher whose buffered publish
// bytes have not yet triggered a readiness wake still drains and admits
// them (and fires its subscribers) before the quiescence check can pass.
self.wake_all_connections_for_flush();
let quiesced = loop {
let snapshot = self.settle_generation_snapshot();
if self.all_connections_delivery_quiesced() {
break true;
}
let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
break false;
};
let (generation, outcome) = self
.settle_changed
.wait_timeout_while(
recover_lock(&self.settle_generation),
remaining,
|current| {
#[cfg(test)]
{
let observer = recover_lock(&self.settle_wait_observer).take();
if let Some(observer) = observer {
let _sent = observer.send(());
}
}
*current == snapshot
},
)
.unwrap_or_else(PoisonError::into_inner);
drop(generation);
// A predicate that became true without its completion signal must
// not turn a missed wake into success when the deadline wakes us.
if outcome.timed_out() {
break false;
}
};
self.settle_armed.store(false, Ordering::Release);
quiesced
}
}