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
use std::{
io,
ops::{Deref, DerefMut},
process::{Child, ExitStatus},
};
#[cfg(feature = "local")]
use crate::local::ProcessHandle;
/// A newly spawned candidate is disposable until its selection commits.
/// Dropping a selected candidate only reaps it; it never terminates the daemon.
pub struct Candidate {
process: Process,
selected: bool,
}
enum Process {
Child {
child: Reaped,
#[cfg(feature = "local")]
exit: Option<ProcessHandle>,
},
#[cfg(feature = "launch")]
Daemon(crate::launch::DaemonChild),
}
impl Candidate {
/// Take ownership immediately after spawning a staged candidate.
///
/// With `local` this also opens the candidate's exit handle, while
/// nothing has waited for the child, so the handle follows this process
/// even if its PID is reused after a later reap.
pub fn new(process: Child) -> Self {
Self {
process: Process::Child {
#[cfg(feature = "local")]
exit: ProcessHandle::for_child(&process).ok(),
child: Reaped::new(process),
},
selected: false,
}
}
/// OS process identity used by the live verifier.
pub fn id(&self) -> crate::ProcessId {
match &self.process {
Process::Child { child, .. } => crate::ProcessId::of(child),
#[cfg(feature = "launch")]
Process::Daemon(daemon) => daemon.id(),
}
}
/// Observe whether the candidate exited before readiness.
pub fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
match &mut self.process {
Process::Child { child, .. } => child.try_wait(),
#[cfg(feature = "launch")]
Process::Daemon(daemon) => daemon.try_wait(),
}
}
/// The candidate's exit as an event, for a verifier that waits on it
/// between probes instead of sleeping. The handle is opened by
/// [`Self::new`]; this fails if that could not be done, and
/// [`Self::wait_until`] then asks the child directly.
#[cfg(feature = "local")]
pub fn exit_handle(&mut self) -> io::Result<&mut ProcessHandle> {
match &mut self.process {
Process::Child { exit, .. } => exit
.as_mut()
.ok_or_else(|| io::Error::other("the candidate's exit handle could not be opened")),
#[cfg(feature = "launch")]
Process::Daemon(daemon) => Ok(daemon.exit_handle()),
}
}
/// Block until the candidate exits or `deadline` passes: its exit status,
/// or `None` while it still runs.
#[cfg(feature = "local")]
pub fn wait_until(&mut self, deadline: std::time::Instant) -> io::Result<Option<ExitStatus>> {
match &mut self.process {
Process::Child { child, exit } => {
if !crate::local::wait_child(|| child.try_wait(), exit.as_mut(), deadline)? {
return Ok(None);
}
child.wait().map(Some)
}
#[cfg(feature = "launch")]
Process::Daemon(daemon) => daemon.wait_until(deadline),
}
}
/// Mark the selection boundary before doing repairable follow-up work.
pub fn selected(&mut self) {
self.selected = true;
}
/// Kill the candidate and wait for it to exit.
fn stop(&mut self) -> io::Result<ExitStatus> {
match &mut self.process {
Process::Child { child, .. } => {
child.kill()?;
child.wait()
}
#[cfg(feature = "launch")]
Process::Daemon(daemon) => {
daemon.kill()?;
daemon.wait()
}
}
}
}
/// A daemon started by [`DaemonCommand`](crate::launch::DaemonCommand) as a
/// candidate: dropped before [`Candidate::selected`], it is stopped. Take its
/// readiness channel first; the candidate keeps it but does not give it out.
#[cfg(feature = "launch")]
impl From<crate::launch::DaemonChild> for Candidate {
fn from(daemon: crate::launch::DaemonChild) -> Self {
Self {
process: Process::Daemon(daemon),
selected: false,
}
}
}
impl Drop for Candidate {
/// Stop an unselected candidate. Dropping its process then reaps it.
fn drop(&mut self) {
if !self.selected {
let _ = self.stop();
}
}
}
/// A child that is reaped, never stopped, when dropped: at once if it has
/// exited, otherwise by a thread that waits for it. On Unix a child nothing
/// waits for stays a zombie after it exits. Reaping a running child is best
/// effort: if no thread can be created, nothing waits for it.
#[derive(Debug)]
pub(crate) struct Reaped(Option<Child>);
impl Reaped {
pub(crate) fn new(child: Child) -> Self {
Self(Some(child))
}
}
impl Deref for Reaped {
type Target = Child;
fn deref(&self) -> &Child {
self.0
.as_ref()
.expect("the child is taken only when dropped")
}
}
impl DerefMut for Reaped {
fn deref_mut(&mut self) -> &mut Child {
self.0
.as_mut()
.expect("the child is taken only when dropped")
}
}
impl Drop for Reaped {
fn drop(&mut self) {
let Some(mut child) = self.0.take() else {
return;
};
if child.try_wait().is_ok_and(|status| status.is_some()) {
return;
}
// A failed spawn drops the closure and the child with it; the
// `try_wait` above was the last chance to reap it.
let _ = std::thread::Builder::new()
.name("daemon-reaper".into())
.spawn(move || {
let _ = child.wait();
});
}
}
#[cfg(test)]
mod tests {
#[cfg(feature = "local")]
#[test]
fn a_candidate_watches_its_process_from_the_moment_it_is_taken() {
// Opened later, after a `try_wait` had reaped the child, the handle
// could follow whatever process reused the PID.
let child = std::process::Command::new(std::env::current_exe().unwrap())
.arg("--list")
.stdout(std::process::Stdio::null())
.spawn()
.unwrap();
let pid = crate::ProcessId::of(&child);
let mut candidate = super::Candidate::new(child);
assert_eq!(candidate.exit_handle().unwrap().pid(), pid);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
assert!(candidate.wait_until(deadline).unwrap().unwrap().success());
}
}