harmos_packaging/
process.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Retain Cargo And Its Process Tree Until Successful Or Cancelled Settlement.
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::io::Read;
20use std::process::{Command, ExitStatus, Output};
21use std::sync::atomic::{AtomicBool, Ordering};
22use std::sync::{Arc, Mutex};
23use std::thread::JoinHandle;
24use std::time::Duration;
25
26// Third Party Imports
27use process_wrap::std::{ChildWrapper, CommandWrap};
28
29// Local Imports
30use super::Error;
31
32/* === Constants === */
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38/* === Structs === */
39
40#[derive(Clone, Default)]
41pub struct BuildCancellation {
42    stopped: Arc<AtomicBool>,
43}
44
45pub(super) struct Process {
46    child: Arc<Mutex<Box<dyn ChildWrapper>>>,
47    finished: Arc<AtomicBool>,
48    monitor: Option<JoinHandle<std::io::Result<()>>>,
49}
50
51/* === Enums === */
52
53/* === Traits === */
54
55/* === Implementations === */
56
57impl BuildCancellation {
58    pub fn cancel(&self) {
59        self.stopped.store(true, Ordering::Release);
60    }
61    pub fn is_cancelled(&self) -> bool {
62        self.stopped.load(Ordering::Acquire)
63    }
64    pub(super) fn check(&self) -> Result<(), Error> {
65        if self.is_cancelled() { Err(Error::Cancelled) } else { Ok(()) }
66    }
67}
68
69impl Process {
70    pub(super) fn spawn(command: Command, cancellation: &BuildCancellation) -> Result<Self, Error> {
71        cancellation.check()?;
72        let mut command = CommandWrap::from(command);
73        #[cfg(unix)]
74        command.wrap(process_wrap::std::ProcessGroup::leader());
75        #[cfg(windows)]
76        command.wrap(process_wrap::std::JobObject);
77        let child = Arc::new(Mutex::new(command.spawn()?));
78        let retained = child.clone();
79        let finished = Arc::new(AtomicBool::new(false));
80        let completed = finished.clone();
81        let cancellation = cancellation.clone();
82        let monitor = std::thread::spawn(move || {
83            while !completed.load(Ordering::Acquire) {
84                if cancellation.is_cancelled() {
85                    return retained.lock().expect("process owner is healthy").start_kill();
86                }
87                std::thread::park_timeout(Duration::from_millis(20));
88            }
89            Ok(())
90        });
91        Ok(Self { child, finished, monitor: Some(monitor) })
92    }
93
94    pub(super) fn stdout(&self) -> std::process::ChildStdout {
95        self.child.lock().expect("process owner is healthy").stdout().take().expect("Cargo stdout was piped")
96    }
97
98    pub(super) fn kill(&self) -> Result<(), Error> {
99        let mut child = self.child.lock().expect("process owner is healthy");
100        if child.try_wait()?.is_none() {
101            child.start_kill()?;
102        }
103        Ok(())
104    }
105
106    pub(super) fn wait(&mut self) -> Result<ExitStatus, Error> {
107        loop {
108            let status = self.child.lock().expect("process owner is healthy").try_wait()?;
109            if let Some(status) = status {
110                self.finished.store(true, Ordering::Release);
111                if let Some(monitor) = self.monitor.take() {
112                    monitor.thread().unpark();
113                    monitor.join().map_err(|_| std::io::Error::other("process cancellation monitor panicked"))??;
114                }
115                return Ok(status);
116            }
117            std::thread::sleep(Duration::from_millis(20));
118        }
119    }
120
121    pub(super) fn output(mut self) -> Result<Output, Error> {
122        let mut stdout = self.stdout();
123        let mut stderr = self.child.lock().expect("process owner is healthy").stderr().take().expect("Cargo stderr was piped");
124        let stdout = std::thread::spawn(move || {
125            let mut bytes = Vec::new();
126            stdout.read_to_end(&mut bytes).map(|_| bytes)
127        });
128        let stderr = std::thread::spawn(move || {
129            let mut bytes = Vec::new();
130            stderr.read_to_end(&mut bytes).map(|_| bytes)
131        });
132        let status = self.wait();
133        let cleanup = if status.is_err() { self.kill().and_then(|_| self.wait().map(|_| ())) } else { Ok(()) };
134        let stdout = stdout.join().map_err(|_| std::io::Error::other("Cargo stdout reader panicked"));
135        let stderr = stderr.join().map_err(|_| std::io::Error::other("Cargo stderr reader panicked"));
136        cleanup?;
137        Ok(Output { status: status?, stdout: stdout??, stderr: stderr?? })
138    }
139}
140
141impl Drop for Process {
142    fn drop(&mut self) {
143        if let Err(error) = self.kill().and_then(|_| self.wait().map(|_| ())) {
144            tracing::error!(%error, "retained Cargo process cleanup failed");
145        }
146    }
147}
148
149/* === Public Functions === */
150
151/* === Private Functions === */
152
153/* === Tests === */
harmos_packaging/process.rs

harmos_packaging/
process.rs
