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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
//! SRV-042 R1: a connection accepted while the incarnation start runs is held
//! here, off the accept thread, and run by the start thread once the start has
//! published, so no accept worker (TCP or WebSocket) and no listener stop ever
//! waits on the start. A held connection is admitted or refused by the start's
//! outcome through the ordinary spawn, in arrival order; the supervisor's
//! shutdown drops what is held, closing each socket, and runs nothing after.
//!
//! The start thread admits one held connection at a time under `admitting`,
//! and a stop takes that lock after marking the phase stopped: an admission
//! running when the stop begins finishes inside the stop, so the stop's
//! connection snapshot sees the connection it spawned, and none runs after
//! (Flapjack's post 108, 25 October: one spawned after the snapshot outlived
//! the stop and kept its store owners).
//!
//! Nothing is bounded here beyond the start's own duration: every held
//! connection still takes its admission reservation when it runs, so one over
//! `max_connections` is refused then, exactly as it would have been at accept.
use std::collections::VecDeque;
use std::sync::Mutex;
use super::ConnectionSupervisor;
/// One accepted connection waiting for the incarnation start.
pub(super) type HeldAdmission = Box<dyn FnOnce(&ConnectionSupervisor) + Send>;
enum Phase<A> {
/// No start is running: an admission runs on the caller's thread.
Started,
/// The start runs: admissions are held for it.
Starting(Vec<A>),
/// The start published: the start thread admits these in arrival order,
/// and one arriving now joins the end.
Admitting(VecDeque<A>),
/// The supervisor shut down: an admission is dropped, its socket closed.
Stopped,
}
/// The admissions held while the incarnation start runs.
pub(super) struct HeldAdmissions<A = HeldAdmission> {
phase: Mutex<Phase<A>>,
/// Held by the start thread while one admission runs; a stop takes it to
/// wait for that admission to end.
admitting: Mutex<()>,
/// Notified after each hold and at a stop, so a test waits for its
/// clients to be held before it stops the listener (SRV-046 R1).
#[cfg(test)]
arrived: std::sync::Condvar,
}
impl<A> Default for HeldAdmissions<A> {
fn default() -> Self {
Self {
phase: Mutex::new(Phase::Started),
admitting: Mutex::new(()),
#[cfg(test)]
arrived: std::sync::Condvar::new(),
}
}
}
impl<A> std::fmt::Debug for HeldAdmissions<A> {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let phase = match self.phase.lock().as_deref() {
Ok(Phase::Started) => "started".to_owned(),
Ok(Phase::Starting(held)) => format!("starting, {} held", held.len()),
Ok(Phase::Admitting(held)) => format!("admitting, {} held", held.len()),
Ok(Phase::Stopped) => "stopped".to_owned(),
Err(_) => "poisoned".to_owned(),
};
let mut debug = formatter.debug_struct("HeldAdmissions");
debug.field("phase", &phase);
debug.field("admitting", &self.admitting);
#[cfg(test)]
debug.field("arrived", &self.arrived);
debug.finish()
}
}
impl<A> HeldAdmissions<A> {
/// Holds every admission from now until [`Self::admitting`]; called before
/// the start thread begins, so none slips past it.
pub(super) fn starting(&self) {
if let Ok(mut phase) = self.phase.lock() {
if matches!(*phase, Phase::Started) {
*phase = Phase::Starting(Vec::new());
}
} else {
tracing::error!(
"the held admission lock was poisoned; admissions wait on the start on their own threads"
);
}
}
/// Holds `admit` while the start runs or admits, drops it once the
/// supervisor has stopped, and answers it back to run now otherwise.
pub(super) fn hold(&self, admit: A) -> Option<A> {
let Ok(mut phase) = self.phase.lock() else {
tracing::error!("the held admission lock was poisoned; the admission runs at once");
return Some(admit);
};
match &mut *phase {
Phase::Started => Some(admit),
Phase::Starting(held) => {
held.push(admit);
#[cfg(test)]
self.arrived.notify_all();
None
}
Phase::Admitting(held) => {
held.push_back(admit);
None
}
// Dropped when this returns, after the lock: its socket closes.
Phase::Stopped => None,
}
}
/// The start has published: what was held is admitted by
/// [`Self::admit_held`]. False once stopped, when nothing is admitted.
pub(super) fn admitting(&self) -> bool {
let Ok(mut phase) = self.phase.lock() else {
tracing::error!("the held admission lock was poisoned; nothing held is admitted");
return false;
};
match std::mem::replace(&mut *phase, Phase::Stopped) {
Phase::Starting(held) => {
*phase = Phase::Admitting(held.into());
true
}
Phase::Admitting(held) => {
*phase = Phase::Admitting(held);
true
}
Phase::Started => {
*phase = Phase::Started;
false
}
Phase::Stopped => false,
}
}
/// Runs each held admission through `admit`, one at a time and in arrival
/// order, until none is left, then lets later admissions run at once. A
/// stop between two ends it; what `admit` answers is dropped after the
/// admission's lock is released, so a supervisor owner it held may end
/// here without waiting on itself.
pub(super) fn admit_held<R>(&self, mut admit: impl FnMut(A) -> R) {
loop {
let Ok(running) = self.admitting.lock() else {
tracing::error!("the admission lock was poisoned; nothing more held is admitted");
return;
};
let Ok(mut phase) = self.phase.lock() else {
tracing::error!(
"the held admission lock was poisoned; nothing more held is admitted"
);
return;
};
let next = match &mut *phase {
Phase::Admitting(held) => held.pop_front(),
_ => None,
};
let Some(next) = next else {
if matches!(&*phase, Phase::Admitting(_)) {
*phase = Phase::Started;
}
return;
};
drop(phase);
let answered = admit(next);
drop(running);
drop(answered);
}
}
/// Whether `count` admissions are held while the start runs, waiting for
/// each hold's notify until `within` passes (SRV-046 R1).
#[cfg(test)]
pub(super) fn wait_held(&self, count: usize, within: std::time::Duration) -> bool {
let held = |phase: &Phase<A>| matches!(phase, Phase::Starting(held) if held.len() >= count);
self.phase.lock().is_ok_and(|phase| {
self.arrived
.wait_timeout_while(phase, within, |phase| !held(phase))
.is_ok_and(|(phase, _)| held(&phase))
})
}
/// Whether a stop has begun, waiting for its notify until `within` passes.
#[cfg(test)]
pub(super) fn wait_stopped(&self, within: std::time::Duration) -> bool {
self.phase.lock().is_ok_and(|phase| {
self.arrived
.wait_timeout_while(phase, within, |phase| !matches!(phase, Phase::Stopped))
.is_ok_and(|(phase, _)| matches!(*phase, Phase::Stopped))
})
}
/// The supervisor is shutting down: drops what is held, closing each
/// socket, and every admission after it, then waits for an admission the
/// start thread is running to end.
pub(super) fn stop(&self) {
let Ok(mut phase) = self.phase.lock() else {
tracing::error!(
"the held admission lock was poisoned; held sockets close with the supervisor"
);
return;
};
let held = std::mem::replace(&mut *phase, Phase::Stopped);
#[cfg(test)]
self.arrived.notify_all();
drop(phase);
drop(held);
// An admission's panic poisons the lock; the wait for it is the same.
match self.admitting.lock() {
Ok(running) => drop(running),
Err(poisoned) => drop(poisoned),
}
}
}
#[cfg(test)]
#[path = "held_admission_gate_tests.rs"]
mod gate_tests;