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        {
77            // JobObject Replaces Native Flags; Its Wrapper Preserves Explicit Policy.
78            let mut flags = process_wrap::std::CreationFlags(Default::default());
79            flags.0.0 = 0x08000000;
80            command.wrap(flags).wrap(process_wrap::std::JobObject);
81        }
82        let child = Arc::new(Mutex::new(command.spawn()?));
83        let retained = child.clone();
84        let finished = Arc::new(AtomicBool::new(false));
85        let completed = finished.clone();
86        let cancellation = cancellation.clone();
87        let monitor = std::thread::spawn(move || {
88            while !completed.load(Ordering::Acquire) {
89                if cancellation.is_cancelled() {
90                    return retained.lock().expect("process owner is healthy").start_kill();
91                }
92                std::thread::park_timeout(Duration::from_millis(20));
93            }
94            Ok(())
95        });
96        Ok(Self { child, finished, monitor: Some(monitor) })
97    }
98
99    pub(super) fn stdout(&self) -> std::process::ChildStdout {
100        self.child.lock().expect("process owner is healthy").stdout().take().expect("Cargo stdout was piped")
101    }
102
103    pub(super) fn kill(&self) -> Result<(), Error> {
104        let mut child = self.child.lock().expect("process owner is healthy");
105        if child.try_wait()?.is_none() {
106            child.start_kill()?;
107        }
108        Ok(())
109    }
110
111    pub(super) fn wait(&mut self) -> Result<ExitStatus, Error> {
112        loop {
113            let status = self.child.lock().expect("process owner is healthy").try_wait()?;
114            if let Some(status) = status {
115                self.finished.store(true, Ordering::Release);
116                if let Some(monitor) = self.monitor.take() {
117                    monitor.thread().unpark();
118                    monitor.join().map_err(|_| std::io::Error::other("process cancellation monitor panicked"))??;
119                }
120                return Ok(status);
121            }
122            std::thread::sleep(Duration::from_millis(20));
123        }
124    }
125
126    pub(super) fn output(mut self) -> Result<Output, Error> {
127        let mut stdout = self.stdout();
128        let mut stderr = self.child.lock().expect("process owner is healthy").stderr().take().expect("Cargo stderr was piped");
129        let stdout = std::thread::spawn(move || {
130            let mut bytes = Vec::new();
131            stdout.read_to_end(&mut bytes).map(|_| bytes)
132        });
133        let stderr = std::thread::spawn(move || {
134            let mut bytes = Vec::new();
135            stderr.read_to_end(&mut bytes).map(|_| bytes)
136        });
137        let status = self.wait();
138        let cleanup = if status.is_err() { self.kill().and_then(|_| self.wait().map(|_| ())) } else { Ok(()) };
139        let stdout = stdout.join().map_err(|_| std::io::Error::other("Cargo stdout reader panicked"));
140        let stderr = stderr.join().map_err(|_| std::io::Error::other("Cargo stderr reader panicked"));
141        cleanup?;
142        Ok(Output { status: status?, stdout: stdout??, stderr: stderr?? })
143    }
144}
145
146impl Drop for Process {
147    fn drop(&mut self) {
148        if let Err(error) = self.kill().and_then(|_| self.wait().map(|_| ())) {
149            tracing::error!(%error, "retained Cargo process cleanup failed");
150        }
151    }
152}
153
154/* === Public Functions === */
155
156/* === Private Functions === */
157
158/* === Tests === */
harmos_packaging/process.rs

harmos_packaging/
process.rs
