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
//! Stopping a [`ProcessRunner`]: an explicit kill, or the
//! [`RunOptions::timeout`](crate::RunOptions::timeout) deadline (issue #227).
use std::future::Future;
use std::time::Duration;
use tokio::time::Instant;
use crate::{signal, utils, ProcessRunner, Result};
impl ProcessRunner {
/// Stop the process using the configured kill signal
/// ([`RunOptions::kill_signal`](crate::RunOptions::kill_signal), default `SIGTERM`).
///
/// Mirrors the JavaScript `kill()` with no argument.
pub fn kill(&mut self) -> Result<()> {
let signal = self.options.kill_signal.clone();
self.kill_with(&signal)
}
/// Stop the process using an explicit signal, overriding
/// [`RunOptions::kill_signal`](crate::RunOptions::kill_signal) for this call.
///
/// Mirrors the JavaScript `kill(signal)`. The signal is delivered to the
/// child and its process group, so grandchildren behind a `sh -c` wrapper
/// are stopped too - except for a child sharing the caller's terminal,
/// which stays in the caller's group so CTRL+C keeps reaching it. The child
/// then has [`RunOptions::kill_grace_ms`](crate::RunOptions::kill_grace_ms) to run its own handler before
/// `SIGKILL` follows, so a process that ignores the signal still
/// terminates.
///
/// ```no_run
/// use command_stream::{ProcessRunner, RunOptions};
///
/// # #[tokio::main]
/// # async fn main() -> command_stream::Result<()> {
/// let mut runner = ProcessRunner::new("sleep 30", RunOptions::default());
/// runner.start().await?;
/// runner.kill_with("SIGINT")?; // the CTRL+C signal
/// # Ok(())
/// # }
/// ```
pub fn kill_with(&mut self, signal: &str) -> Result<()> {
self.cancelled = true;
utils::trace_lazy("ProcessRunner", || format!("kill | signal={signal}"));
let Some(child) = self.child.as_mut() else {
return Ok(());
};
// Windows stops the tree before the parent exits, while its descendants
// can still be found. Direct-child termination is the fallback.
// The `#[cfg(unix)]` block below is stripped on Windows, which leaves
// this one as the function's tail expression - hence no `return`.
#[cfg(not(unix))]
{
if let Some(pid) = child.id() {
signal::send_signal_to_process(pid, signal, signal::Delivery::ProcessAndGroup);
}
child.start_kill()?;
utils::trace_lazy("ProcessRunner", || {
"Direct-child termination requested".to_string()
});
Ok(())
}
// Without a pid the process never spawned (or was already reaped);
// fall back to the forceful stop so `kill()` still terminates it.
#[cfg(unix)]
{
let Some(pid) = child.id() else {
child.start_kill()?;
return Ok(());
};
// `SIGKILL` cannot be handled, so there is nothing to wait for.
//
// A zero grace period means the child is given no opportunity to
// handle the signal either, so the requested signal is not
// delivered at all. Anything done between it and `SIGKILL` - even a
// single syscall - is a window the child can be scheduled in, which
// made "no grace" a race the child occasionally won rather than a
// guarantee. The reported exit code still comes from the signal
// that was requested.
let grace = self.options.kill_grace_ms;
let delivery = if self.own_process_group {
signal::Delivery::ProcessAndGroup
} else {
signal::Delivery::ProcessOnly
};
if grace == 0 || signal == "SIGKILL" {
signal::send_signal_to_process(pid, "SIGKILL", delivery);
let _ = child.start_kill();
return Ok(());
}
signal::send_signal_to_process(pid, signal, delivery);
// Otherwise escalate in the background so the child keeps its grace
// period without blocking the caller, which may not be inside an
// await point.
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(grace)).await;
// Best effort: if the child already exited on the first signal
// this delivery simply fails, and the pid has not been reused
// because the `Child` handle above has not reaped it yet. That
// unreaped leader is also what keeps the group id alive, so the
// group delivery still reaches a grandchild that outlived it.
signal::send_signal_to_process(pid, "SIGKILL", delivery);
});
Ok(())
}
}
/// Whether the [`RunOptions::timeout`](crate::RunOptions::timeout) elapsed
/// and stopped the command.
///
/// Mirrors the JavaScript `result.timedOut`. An explicit kill, or a
/// command finishing on its own, leaves it `false`.
pub fn timed_out(&self) -> bool {
self.timed_out
}
/// How a timed-out child is stopped: the configured signal, delivered the
/// way [`kill_with`](Self::kill_with) would deliver it.
pub(crate) fn stop_plan(&self) -> StopPlan {
#[cfg(unix)]
let delivery = if self.own_process_group {
signal::Delivery::ProcessAndGroup
} else {
signal::Delivery::ProcessOnly
};
#[cfg(not(unix))]
let delivery = signal::Delivery::ProcessAndGroup;
StopPlan {
pid: self.pid,
signal: self.options.kill_signal.clone(),
grace_ms: self.options.kill_grace_ms,
delivery,
}
}
}
/// How to stop a child whose deadline has passed.
pub(crate) struct StopPlan {
pid: Option<u32>,
signal: String,
grace_ms: u64,
delivery: signal::Delivery,
}
impl StopPlan {
fn send(&self, signal: &str) {
if let Some(pid) = self.pid {
signal::send_signal_to_process(pid, signal, self.delivery);
}
}
}
/// Await `work` - draining and reaping the child - but stop the child if
/// `deadline` passes first.
///
/// Returns the work's output, or `None` when the child and its pipes did not
/// let go even after `SIGKILL` and the exit-pump grace period, and whether the
/// deadline fired. The signals are sent while `work` still holds the unreaped
/// child, so its pid and process group cannot have been reused.
pub(crate) async fn bounded<T>(
work: impl Future<Output = T>,
deadline: Option<Instant>,
plan: StopPlan,
) -> (Option<T>, bool) {
tokio::pin!(work);
let Some(deadline) = deadline else {
return (Some(work.await), false);
};
tokio::select! {
output = &mut work => return (Some(output), false),
() = tokio::time::sleep_until(deadline) => {}
}
utils::trace_lazy("ProcessRunner", || {
format!(
"Timeout elapsed | pid={:?} signal={}",
plan.pid, plan.signal
)
});
// As in `kill_with`, no grace means `SIGKILL` alone.
if cfg!(unix) && plan.grace_ms > 0 && plan.signal != "SIGKILL" {
plan.send(&plan.signal);
let grace = Duration::from_millis(plan.grace_ms);
if let Ok(output) = tokio::time::timeout(grace, &mut work).await {
return (Some(output), true);
}
}
plan.send("SIGKILL");
let pump = Duration::from_millis(crate::stream::DEFAULT_EXIT_PUMP_GRACE_MS);
(tokio::time::timeout(pump, &mut work).await.ok(), true)
}