strop_core/
process.rs

1//! Worker-only child ownership. Cancellation signals a private Unix process
2//! group promptly; only its owner may revoke the capability and reap the PID.
3mod capture;
4use crate::worker::{CancelToken, Failure, FailureKind};
5pub use capture::{
6    capture, capture_with, stream_with, CaptureError, CapturePolicy, CommandOutput, StdinPolicy,
7    StreamError, StreamOutput, StreamPolicy,
8};
9use parking_lot::Mutex;
10use std::io;
11use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, ExitStatus};
12use std::sync::Arc;
13
14#[derive(Default)]
15struct Group {
16    pid: Option<u32>,
17    cancelled: bool,
18}
19impl Group {
20    fn signal(&self) -> Result<(), Failure> {
21        #[cfg(unix)]
22        if let Some(pid) = self.pid {
23            // SAFETY: this positive PID belongs to our unreaped child, launched
24            // with process_group(0). The mutex serializes signalling and revocation;
25            // negation targets only that private group. std cannot signal groups.
26            if unsafe { libc::kill(-(pid as libc::pid_t), libc::SIGKILL) } == -1 {
27                let error = io::Error::last_os_error();
28                if error.raw_os_error() != Some(libc::ESRCH) {
29                    return Err(Failure::new(
30                        FailureKind::Io,
31                        format!("kill process group: {error}"),
32                    ));
33                }
34            }
35        }
36        Ok(())
37    }
38    fn cancel(&mut self) -> Result<(), Failure> {
39        self.cancelled = true;
40        self.signal()
41    }
42}
43
44/// Owns both the child and the cancellation capability. Never put this in editor
45/// state: Drop may wait, so it belongs exclusively to a worker stack.
46pub struct OwnedProcess {
47    child: Child,
48    group: Arc<Mutex<Group>>,
49    token: CancelToken,
50    reaped: bool,
51}
52impl OwnedProcess {
53    pub fn spawn(command: &mut Command, token: &CancelToken) -> Result<Self, Failure> {
54        #[cfg(not(unix))]
55        return Err(Failure::new(
56            FailureKind::Unavailable,
57            "process-group supervision requires Unix",
58        ));
59        #[cfg(unix)]
60        {
61            use std::os::unix::process::CommandExt;
62            let group = Arc::new(Mutex::new(Group::default()));
63            let callback = group.clone();
64            token.register_cancel_resource(move || callback.lock().cancel())?;
65            if token.is_cancelled() {
66                token.clear_cancel_resource();
67                return Err(Failure::new(
68                    FailureKind::Unavailable,
69                    "process cancelled before spawn",
70                ));
71            }
72            let child = match command.process_group(0).spawn() {
73                Ok(child) => child,
74                Err(error) => {
75                    token.clear_cancel_resource();
76                    return Err(Failure::new(FailureKind::Spawn, error.to_string()));
77                }
78            };
79            let process = Self {
80                child,
81                group,
82                token: token.clone(),
83                reaped: false,
84            };
85            {
86                let mut group = process.group.lock();
87                group.pid = Some(process.child.id());
88                // A callback may already have run before publication.
89                if group.cancelled || token.is_cancelled() {
90                    group.cancel()?;
91                }
92            }
93            Ok(process)
94        }
95    }
96    pub fn take_stdin(&mut self) -> Option<ChildStdin> {
97        self.child.stdin.take()
98    }
99    pub fn take_stdout(&mut self) -> Option<ChildStdout> {
100        self.child.stdout.take()
101    }
102    pub fn take_stderr(&mut self) -> Option<ChildStderr> {
103        self.child.stderr.take()
104    }
105    pub fn terminate(&mut self) -> Result<(), Failure> {
106        self.group.lock().signal()
107    }
108
109    /// Observe without reaping: descendants may still hold pipes, and cancellation
110    /// must retain its PID reservation until those descendants have been killed.
111    pub fn has_exited(&self) -> Result<bool, Failure> {
112        #[cfg(not(unix))]
113        return Err(Failure::new(
114            FailureKind::Unavailable,
115            "process supervision requires Unix",
116        ));
117        #[cfg(unix)]
118        {
119            // SAFETY: zero initializes siginfo_t; successful waitid writes it.
120            // WNOWAIT retains ownership of the child PID, WNOHANG never blocks.
121            let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() };
122            loop {
123                let result = unsafe {
124                    libc::waitid(
125                        libc::P_PID,
126                        self.child.id() as libc::id_t,
127                        &mut info,
128                        libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
129                    )
130                };
131                if result == 0 {
132                    // SAFETY: waitid successfully initialized the platform layout.
133                    return Ok(unsafe { info.si_pid() } != 0);
134                }
135                let error = io::Error::last_os_error();
136                if error.kind() != io::ErrorKind::Interrupted {
137                    return Err(Failure::new(FailureKind::Wait, error.to_string()));
138                }
139            }
140        }
141    }
142    /// Wait with cancellation intact, then revoke before reaping. A callback
143    /// detached from CancelToken can never signal a reused PID.
144    pub fn wait(&mut self) -> Result<ExitStatus, Failure> {
145        // Retain the signalling capability throughout the wait. Even a caller
146        // waiting before termination must remain cancellable.
147        while !self.reaped && !self.has_exited()? {
148            std::thread::park_timeout(std::time::Duration::from_millis(20));
149        }
150        self.group.lock().pid = None;
151        loop {
152            match self.child.wait() {
153                Ok(status) => {
154                    self.reaped = true;
155                    self.token.clear_cancel_resource();
156                    return Ok(status);
157                }
158                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
159                Err(error) => return Err(Failure::new(FailureKind::Wait, error.to_string())),
160            }
161        }
162    }
163}
164impl Drop for OwnedProcess {
165    fn drop(&mut self) {
166        if !self.reaped {
167            let _ = self.terminate();
168            // Direct-child fallback if process-group signalling failed.
169            let _ = self.child.kill();
170            let _ = self.wait();
171        }
172        self.token.clear_cancel_resource();
173    }
174}
strop_core/process.rs

strop_core/
process.rs
