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
//! 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.
//!
//! 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::sync::Mutex;
use super::ConnectionSupervisor;
/// One accepted connection waiting for the incarnation start.
pub(super) type HeldAdmission = Box<dyn FnOnce(&ConnectionSupervisor) + Send>;
#[derive(Default)]
enum Phase {
/// No start is running: an admission runs on the caller's thread.
#[default]
Started,
/// The start runs: admissions are held for it.
Starting(Vec<HeldAdmission>),
/// The supervisor shut down: an admission is dropped, its socket closed.
Stopped,
}
/// The admissions held while the incarnation start runs.
#[derive(Default)]
pub(super) struct HeldAdmissions {
phase: Mutex<Phase>,
/// Notified after each hold, 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 std::fmt::Debug for HeldAdmissions {
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::Stopped) => "stopped".to_owned(),
Err(_) => "poisoned".to_owned(),
};
let mut debug = formatter.debug_struct("HeldAdmissions");
debug.field("phase", &phase);
#[cfg(test)]
debug.field("arrived", &self.arrived);
debug.finish()
}
}
impl HeldAdmissions {
/// Holds every admission from now until [`Self::started`]; 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, drops it once the supervisor has
/// stopped, and answers it back to run now otherwise.
pub(super) fn hold(&self, admit: HeldAdmission) -> Option<HeldAdmission> {
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
}
// Dropped when this returns, after the lock: its socket closes.
Phase::Stopped => None,
}
}
/// The start has published: answers what was held, in arrival order, and
/// lets later admissions run at once. Answers nothing once stopped.
pub(super) fn started(&self) -> Vec<HeldAdmission> {
let Ok(mut phase) = self.phase.lock() else {
tracing::error!("the held admission lock was poisoned; nothing held is admitted");
return Vec::new();
};
match std::mem::take(&mut *phase) {
Phase::Starting(held) => held,
Phase::Stopped => {
*phase = Phase::Stopped;
Vec::new()
}
Phase::Started => Vec::new(),
}
}
/// 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| 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))
})
}
/// The supervisor is shutting down: drops what is held, closing each
/// socket, and every admission after it.
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);
drop(phase);
drop(held);
}
}