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
//! Running a child process under a deadline
//!
//! [`run_bounded`] is [`std::process::Command::output`] with a deadline, for
//! the `.js` Flockfile bridge: a config module that leaves a server or timer
//! running keeps node's event loop alive after `require` returned.
//!
//! It takes no stdin and offers no cancellation. Both streams drain on their
//! own threads, since a pipe holds 64 KiB before a write blocks and a child
//! blocked writing never exits.
use std::io::Read;
use std::process::{Command, Output, Stdio};
use std::sync::mpsc::{self, Receiver};
use std::time::{Duration, Instant};
/// How often [`run_bounded`] asks whether the child has exited.
///
/// The budgets this runs under are seconds long, so 10ms costs a few hundred
/// wakeups and adds at most one interval to an honest run.
const POLL_INTERVAL: Duration = Duration::from_millis(10);
/// What a bounded run came back with.
#[derive(Debug)]
pub(crate) enum Bounded {
/// The child exited on its own inside the budget, and both its streams
/// arrived.
Exited(Output),
/// The child was still running at the deadline, and was killed.
Killed,
/// The child exited on its own, but something it left behind still holds
/// a captured pipe, so its output never arrived inside the budget.
/// Nothing was killed: whatever holds the pipe is not shep's child.
OutputHeldOpen,
}
/// One of a child's output streams, read to the end on its own thread.
struct Draining(Receiver<std::io::Result<Vec<u8>>>);
impl Draining {
/// Starts draining `source`, returning the handle its bytes arrive on.
fn of<R: Read + Send + 'static>(mut source: R) -> Self {
let (sender, receiver) = mpsc::channel();
std::thread::spawn(move || {
let mut buffer = Vec::new();
let read = source.read_to_end(&mut buffer).map(|_count| buffer);
// The receiver is already gone whenever the run timed out, which
// is an ordinary end for this thread.
let _ = sender.send(read);
});
Self(receiver)
}
/// The stream's bytes, or `None` if they did not arrive inside `budget`.
///
/// A reader thread that died without sending also answers `None`, folding
/// a panic into the timeout verdict.
fn collect_within(&self, budget: Duration) -> Option<std::io::Result<Vec<u8>>> {
self.0.recv_timeout(budget).ok()
}
}
/// Runs `command` to completion, killing it once it outlives `budget`.
///
/// Takes both output pipes for itself, so the caller must not set stdout or
/// stderr. stdin is left as it was.
///
/// The budget covers the reads as well as the wait: a grandchild can hold the
/// inherited pipe open after the child exits, and calling that
/// [`Bounded::Killed`] would claim a kill that never happened.
///
/// # Errors
/// - The child could not be spawned, killed, or waited for.
/// - A stream could not be read.
pub(crate) fn run_bounded(command: &mut Command, budget: Duration) -> std::io::Result<Bounded> {
let deadline = Instant::now() + budget;
let mut child = command
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let stdout = Draining::of(
child
.stdout
.take()
.expect("stdout is piped on the line that spawned this child"),
);
let stderr = Draining::of(
child
.stderr
.take()
.expect("stderr is piped on the line that spawned this child"),
);
// `try_wait` first, deadline second: a child that exited in the last
// interval has already done its work, and killing it over the microseconds
// since would throw away output shep asked for.
let status = loop {
if let Some(status) = child.try_wait()? {
break status;
}
let left = deadline.saturating_duration_since(Instant::now());
if left.is_zero() {
child.kill()?;
// Reaped here rather than left to the `Child` drop, which does
// not wait: a killed child nobody waits for is a zombie.
child.wait()?;
return Ok(Bounded::Killed);
}
std::thread::sleep(POLL_INTERVAL.min(left));
};
let Some(stdout) = stdout.collect_within(deadline.saturating_duration_since(Instant::now()))
else {
return Ok(Bounded::OutputHeldOpen);
};
let Some(stderr) = stderr.collect_within(deadline.saturating_duration_since(Instant::now()))
else {
return Ok(Bounded::OutputHeldOpen);
};
Ok(Bounded::Exited(Output {
status,
stdout: stdout?,
stderr: stderr?,
}))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_child_that_outlives_its_budget_is_killed_rather_than_waited_for() {
let started = Instant::now();
let outcome = run_bounded(Command::new("sleep").arg("30"), Duration::from_millis(100))
.expect("the child spawns, and killing it is the only other syscall");
assert!(
matches!(outcome, Bounded::Killed),
"a 30s sleep cannot finish inside 100ms: {outcome:?}"
);
assert!(
started.elapsed() < Duration::from_secs(10),
"the sleep was waited out rather than killed, in {:?}",
started.elapsed()
);
}
#[test]
fn a_child_that_finishes_comes_back_with_both_streams() {
let outcome = run_bounded(
Command::new("sh")
.arg("-c")
.arg("printf wool; printf bleat >&2"),
Duration::from_secs(30),
)
.expect("sh is on every host this crate compiles for");
let Bounded::Exited(output) = outcome else {
panic!("a printf finishes well inside 30s: {outcome:?}");
};
assert!(output.status.success());
assert_eq!(String::from_utf8_lossy(&output.stdout), "wool");
assert_eq!(String::from_utf8_lossy(&output.stderr), "bleat");
}
/// `sleep 5 &` in a shell that exits straight after leaves a process
/// holding the inherited pipes, so the wait ends at once and only the
/// reads run out of budget. 2s is long enough that a slow `sh` start is
/// not a kill, short enough to stay inside the backgrounded `sleep 5`.
#[test]
fn a_child_whose_output_outlives_it_is_not_reported_as_killed() {
let started = Instant::now();
let outcome = run_bounded(
Command::new("sh").arg("-c").arg("sleep 5 & exit 0"),
Duration::from_secs(2),
)
.expect("sh is on every host this crate compiles for");
assert!(
matches!(outcome, Bounded::OutputHeldOpen),
"the shell exits at once and the backgrounded sleep holds both \
pipes: {outcome:?}"
);
assert!(
started.elapsed() < Duration::from_secs(5),
"the budget did not bound the reads, in {:?}",
started.elapsed()
);
}
/// 200_000 bytes is past the 64 KiB a pipe holds, so a child writing this
/// much blocks against a reader that has not started.
#[test]
fn a_child_louder_than_one_pipeful_is_not_deadlocked_against_its_own_budget() {
let outcome = run_bounded(
Command::new("sh")
.arg("-c")
.arg("head -c 200000 /dev/zero; head -c 200000 /dev/zero >&2"),
Duration::from_secs(30),
)
.expect("sh is on every host this crate compiles for");
let Bounded::Exited(output) = outcome else {
panic!("the writes finish in milliseconds against a draining reader: {outcome:?}");
};
assert_eq!(output.stdout.len(), 200_000);
assert_eq!(output.stderr.len(), 200_000);
}
}