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
//! Worker-only child ownership. Cancellation signals a private Unix process
//! group promptly; only its owner may revoke the capability and reap the PID.
mod capture;
use crate::worker::{CancelToken, Failure, FailureKind};
pub use capture::{capture, capture_with, CaptureError, CapturePolicy, CommandOutput, StdinPolicy};
use parking_lot::Mutex;
use std::io;
use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, ExitStatus};
use std::sync::Arc;
#[derive(Default)]
struct Group {
pid: Option<u32>,
cancelled: bool,
}
impl Group {
fn signal(&self) -> Result<(), Failure> {
#[cfg(unix)]
if let Some(pid) = self.pid {
// SAFETY: this positive PID belongs to our unreaped child, launched
// with process_group(0). The mutex serializes signalling and revocation;
// negation targets only that private group. std cannot signal groups.
if unsafe { libc::kill(-(pid as libc::pid_t), libc::SIGKILL) } == -1 {
let error = io::Error::last_os_error();
if error.raw_os_error() != Some(libc::ESRCH) {
return Err(Failure::new(
FailureKind::Io,
format!("kill process group: {error}"),
));
}
}
}
Ok(())
}
fn cancel(&mut self) -> Result<(), Failure> {
self.cancelled = true;
self.signal()
}
}
/// Owns both the child and the cancellation capability. Never put this in editor
/// state: Drop may wait, so it belongs exclusively to a worker stack.
pub struct OwnedProcess {
child: Child,
group: Arc<Mutex<Group>>,
token: CancelToken,
reaped: bool,
}
impl OwnedProcess {
pub fn spawn(command: &mut Command, token: &CancelToken) -> Result<Self, Failure> {
#[cfg(not(unix))]
return Err(Failure::new(
FailureKind::Unavailable,
"process-group supervision requires Unix",
));
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
let group = Arc::new(Mutex::new(Group::default()));
let callback = group.clone();
token.register_cancel_resource(move || callback.lock().cancel())?;
if token.is_cancelled() {
token.clear_cancel_resource();
return Err(Failure::new(
FailureKind::Unavailable,
"process cancelled before spawn",
));
}
let child = match command.process_group(0).spawn() {
Ok(child) => child,
Err(error) => {
token.clear_cancel_resource();
return Err(Failure::new(FailureKind::Spawn, error.to_string()));
}
};
let process = Self {
child,
group,
token: token.clone(),
reaped: false,
};
{
let mut group = process.group.lock();
group.pid = Some(process.child.id());
// A callback may already have run before publication.
if group.cancelled || token.is_cancelled() {
group.cancel()?;
}
}
Ok(process)
}
}
pub fn take_stdin(&mut self) -> Option<ChildStdin> {
self.child.stdin.take()
}
pub fn take_stdout(&mut self) -> Option<ChildStdout> {
self.child.stdout.take()
}
pub fn take_stderr(&mut self) -> Option<ChildStderr> {
self.child.stderr.take()
}
pub fn terminate(&mut self) -> Result<(), Failure> {
self.group.lock().signal()
}
/// Observe without reaping: descendants may still hold pipes, and cancellation
/// must retain its PID reservation until those descendants have been killed.
pub fn has_exited(&self) -> Result<bool, Failure> {
#[cfg(not(unix))]
return Err(Failure::new(
FailureKind::Unavailable,
"process supervision requires Unix",
));
#[cfg(unix)]
{
// SAFETY: zero initializes siginfo_t; successful waitid writes it.
// WNOWAIT retains ownership of the child PID, WNOHANG never blocks.
let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() };
loop {
let result = unsafe {
libc::waitid(
libc::P_PID,
self.child.id() as libc::id_t,
&mut info,
libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
)
};
if result == 0 {
// SAFETY: waitid successfully initialized the platform layout.
return Ok(unsafe { info.si_pid() } != 0);
}
let error = io::Error::last_os_error();
if error.kind() != io::ErrorKind::Interrupted {
return Err(Failure::new(FailureKind::Wait, error.to_string()));
}
}
}
}
/// Wait with cancellation intact, then revoke before reaping. A callback
/// detached from CancelToken can never signal a reused PID.
pub fn wait(&mut self) -> Result<ExitStatus, Failure> {
// Retain the signalling capability throughout the wait. Even a caller
// waiting before termination must remain cancellable.
while !self.reaped && !self.has_exited()? {
std::thread::park_timeout(std::time::Duration::from_millis(20));
}
self.group.lock().pid = None;
loop {
match self.child.wait() {
Ok(status) => {
self.reaped = true;
self.token.clear_cancel_resource();
return Ok(status);
}
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
Err(error) => return Err(Failure::new(FailureKind::Wait, error.to_string())),
}
}
}
}
impl Drop for OwnedProcess {
fn drop(&mut self) {
if !self.reaped {
let _ = self.terminate();
// Direct-child fallback if process-group signalling failed.
let _ = self.child.kill();
let _ = self.wait();
}
self.token.clear_cancel_resource();
}
}