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
//! 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>,
}
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(),
};
formatter
.debug_struct("HeldAdmissions")
.field("phase", &phase)
.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);
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(),
}
}
/// 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);
}
}