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
use super::{
Arc, Atom, ConnectionRuntime, Ordering, RawFd, ReadinessRegistration, ReadinessToken,
ServerError, lock,
};
impl ConnectionRuntime {
/// The connection's single R6 `READY` wake atom.
pub(in crate::server::connection) const fn ready_atom(&self) -> Atom {
self.ready_atom
}
/// Builds a [`ReadyWaker`] targeting `pid` on the connection scheduler, if the
/// scheduler is still live. `None` when the scheduler is gone (teardown) or in
/// scheduler-free unit tests — a notifier with no waker simply never wakes,
/// which under the busy loop is redundant anyway (the every-slice pump still
/// services the source). This is the seam every wake source installs its
/// notifier through (R3/R1(vi)).
pub(in crate::server::connection) fn ready_waker(
&self,
pid: u64,
) -> Option<super::wake::ReadyWaker> {
let scheduler = self.scheduler.upgrade()?;
let ready_pending = self
.records
.lock()
.ok()?
.connections
.get(&pid)
.map(|record| Arc::clone(&record.ready_pending))?;
Some(super::wake::ReadyWaker::new(
&scheduler,
pid,
self.ready_atom,
ready_pending,
))
}
/// Acknowledges READY edges whose mailbox atoms were drained before this slice.
pub(in crate::server::connection) fn acknowledge_ready(&self, pid: u64) {
if let Ok(records) = self.records.lock()
&& let Some(record) = records.connections.get(&pid)
{
record.ready_pending.store(false, Ordering::Release);
}
}
/// Reports a READY edge queued while the current process snapshot is executing.
pub(in crate::server::connection) fn ready_pending(&self, pid: u64) -> bool {
self.records
.lock()
.ok()
.and_then(|records| {
records
.connections
.get(&pid)
.map(|record| record.ready_pending.load(Ordering::Acquire))
})
.unwrap_or(false)
}
/// FIX A-ii: marks `pid` as executing a slice — not parked, so not yet
/// delivery-quiescent. Reuses the registry lock the slice already takes for
/// `is_registered`; it never touches the barrier condvar.
pub(in crate::server::connection) fn mark_running(&self, pid: u64) {
if let Ok(records) = self.records.lock()
&& let Some(record) = records.connections.get(&pid)
{
record.parked.store(false, Ordering::Release);
}
}
/// FIX A-ii: marks `pid` as parked with every accepted publish already fanned
/// out to its socket, and — only while the shutdown flush barrier is armed —
/// bumps the settle generation and wakes it. The bump/notify is skipped
/// entirely in normal operation, so a park off the shutdown path is just one
/// flag store.
pub(in crate::server::connection) fn mark_parked(&self, pid: u64) {
if let Ok(records) = self.records.lock()
&& let Some(record) = records.connections.get(&pid)
{
record.parked.store(true, Ordering::Release);
}
if self.settle_armed.load(Ordering::Acquire) {
self.signal_settle_changed();
}
}
/// Removes a token minted in-slice when publishing it to the host record fails.
pub(in crate::server::connection) fn deregister_unpublished_readiness(
&self,
token: ReadinessToken,
) {
if let Some(scheduler) = self.scheduler.upgrade() {
scheduler.readiness_deregister(token);
}
}
/// Records the one readiness token minted for this connection. A live
/// connection never re-registers: later parked slices rearm this identity.
pub(in crate::server::connection) fn set_readiness_token_once(
&self,
pid: u64,
token: ReadinessToken,
fd: RawFd,
) -> Result<(), ServerError> {
let mut records = lock(&self.records, "connection registry")?;
let record =
records
.connections
.get_mut(&pid)
.ok_or_else(|| ServerError::ListenerAccept {
message: format!(
"connection {pid} has no host record for readiness registration"
),
})?;
if record.readiness.is_some() {
return Err(ServerError::ListenerAccept {
message: format!("connection {pid} attempted to replace its readiness token"),
});
}
record.readiness = Some(ReadinessRegistration { token, fd });
drop(records);
Ok(())
}
#[cfg(test)]
pub(super) fn readiness_registration_count(&self) -> usize {
self.records.lock().map_or(0, |records| {
records
.connections
.values()
.filter(|record| record.readiness.is_some())
.count()
})
}
#[cfg(test)]
pub(super) fn readiness_fd(&self, pid: u64) -> Option<RawFd> {
self.records
.lock()
.ok()?
.connections
.get(&pid)
.and_then(|record| record.readiness.map(|registration| registration.fd))
}
#[cfg(test)]
pub(super) fn readiness_token(&self, pid: u64) -> Option<ReadinessToken> {
self.records
.lock()
.ok()?
.connections
.get(&pid)
.and_then(|record| record.readiness.map(|registration| registration.token))
}
}