diff --git a/README.md b/README.md index da76607..78d0075 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,19 @@ CommandWrap::with_new("watch", |command| { command.arg("ls"); }) requested flag, temporarily adds `CREATE_SUSPENDED` while assigning the process, and resumes it after assignment unless the caller explicitly requested `CREATE_SUSPENDED`. +For the std frontend, add [`KillOnDrop`] to opt in to Windows' kill-on-job-close behavior. Without +the marker, the existing behavior is preserved and dropping the child wrapper does not terminate +the job. The marker and `JobObject` may be registered in either order: + +```rust +use process_wrap::std::*; + +let mut command = CommandWrap::with_new("watch", |command| { command.arg("ls"); }); +command.wrap(JobObject).wrap(KillOnDrop); +let child = command.spawn()?; +drop(child); // terminates every process still in the job +``` + ### Process group - Platforms: POSIX (Linux, Mac, BSDs...) @@ -189,11 +202,13 @@ after assignment unless the caller explicitly requested `CREATE_SUSPENDED`. ### Kill on drop -- Platforms: all (Tokio-only) +- Platforms: all; std opt-in is Windows-only and requires `JobObject` - Like command-group. - Feature: `kill-on-drop` (default) -This is a shim to allow wrappers to handle the kill-on-drop flag, as it can't be read from Command. +For Tokio, this shim sets the command's kill-on-drop flag so wrappers can read it. For std on +Windows, it opts `JobObject` into `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`; it has no effect without +`JobObject`. The std job-object default remains unchanged when the marker is absent. ```rust let child = CommandWrap::with_new("watch", |command| { command.arg("ls"); }) @@ -261,7 +276,7 @@ Both can exist at the same time, but generally you should use one or the other. - `creation-flags`: **default**, enables the [creation flags](#creation-flags) wrapper. - `job-object`: **default**, enables the [job object](#job-object) wrapper. -- `kill-on-drop`: **default**, enables the [kill on drop](#kill-on-drop) wrapper. +- `kill-on-drop`: **default**, enables the [kill on drop](#kill-on-drop) wrapper (including the std Windows `JobObject` opt-in). - `process-group`: **default**, enables the [process group](#process-group) wrapper. - `process-session`: **default**, enables the [process session](#process-session) wrapper. - `reset-sigmask`: enables the [reset signal mask](#reset-signal-mask) wrapper. diff --git a/src/lib.rs b/src/lib.rs index 428c605..7b6f8b4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -77,15 +77,15 @@ //! # KillOnDrop and CreationFlags //! //! The options set on an underlying `Command` are not queryable from library or user code. In most -//! cases this is not an issue; however on Windows, the `JobObject` wrapper needs to know the value -//! of `.kill_on_drop()` and any `.creation_flags()` set. The `KillOnDrop` and `CreationFlags` are -//! "shims" that _should_ be used instead of the aforementioned methods on `Command`. They will -//! internally set the values on the `Command` and also store them in the wrapper, so that wrappers -//! are able to access them. +//! cases this is not an issue; however on Windows, wrappers may need to know the value of +//! `.kill_on_drop()` and any `.creation_flags()` set. `KillOnDrop` and `CreationFlags` are shims +//! that store these options where wrappers can access them. Tokio's `KillOnDrop` also sets the +//! underlying command's kill-on-drop option. The std `KillOnDrop` is an opt-in for `JobObject` +//! kill-on-close behavior; without it, std `JobObject` keeps its existing behavior. //! //! In practice: //! -//! ## Instead of `.kill_on_drop(true)` (Tokio-only): +//! ## Tokio: instead of `.kill_on_drop(true)`: //! //! ```rust //! use process_wrap::tokio::*; @@ -93,6 +93,14 @@ //! command.wrap(KillOnDrop); //! ``` //! +//! ## std: opt in to Windows Job Object termination when the child wrapper is dropped: +//! +//! ```rust,ignore +//! use process_wrap::std::*; +//! let mut command = CommandWrap::with_new("watch", |command| { command.arg("ls"); }); +//! command.wrap(JobObject).wrap(KillOnDrop); +//! ``` +//! //! ## Instead of `.creation_flags(CREATE_NO_WINDOW)` (Windows-only): //! //! ```rust,ignore @@ -462,7 +470,7 @@ //! //! - `creation-flags`: **default**, enables the creation flags wrapper (Windows-only). //! - `job-object`: **default**, enables the job object wrapper (Windows-only). -//! - `kill-on-drop`: **default**, enables the kill on drop wrapper (Tokio-only). +//! - `kill-on-drop`: **default**, enables Tokio kill-on-drop and std Windows JobObject kill-on-close. //! - `process-group`: **default**, enables the process group wrapper (Unix-only). //! - `process-session`: **default**, enables the process session wrapper (Unix-only). //! - `reset-sigmask`: enables the sigmask reset wrapper (Unix-only). diff --git a/src/std.rs b/src/std.rs index 8f9451a..5107b30 100644 --- a/src/std.rs +++ b/src/std.rs @@ -16,6 +16,9 @@ pub use creation_flags::CreationFlags; #[cfg(all(windows, feature = "job-object"))] #[doc(inline)] pub use job_object::{JobObject, JobObjectChild}; +#[cfg(all(windows, feature = "job-object", feature = "kill-on-drop"))] +#[doc(inline)] +pub use kill_on_drop::KillOnDrop; #[cfg(all(unix, feature = "process-group"))] #[doc(inline)] pub use process_group::{ProcessGroup, ProcessGroupChild}; @@ -31,6 +34,8 @@ mod core; mod creation_flags; #[cfg(all(windows, feature = "job-object"))] mod job_object; +#[cfg(all(windows, feature = "job-object", feature = "kill-on-drop"))] +mod kill_on_drop; #[cfg(all(unix, feature = "process-group"))] mod process_group; #[cfg(all(unix, feature = "process-session"))] diff --git a/src/std/job_object.rs b/src/std/job_object.rs index ad10743..fbbc1dc 100644 --- a/src/std/job_object.rs +++ b/src/std/job_object.rs @@ -24,6 +24,8 @@ use crate::{ #[cfg(feature = "creation-flags")] use super::CreationFlags; +#[cfg(feature = "kill-on-drop")] +use super::KillOnDrop; use super::{ChildWrapper, CommandWrap, CommandWrapper}; /// Wrapper which creates a job object context for a `Command`. @@ -82,6 +84,11 @@ impl CommandWrapper for JobObject { mut inner: Box, core: &CommandWrap, ) -> Result> { + #[cfg(feature = "kill-on-drop")] + let kill_on_drop = core.has_wrap::(); + #[cfg(not(feature = "kill-on-drop"))] + let kill_on_drop = false; + let policy = job_creation_flags(user_creation_flags(core)); #[cfg(feature = "tracing")] @@ -103,7 +110,7 @@ impl CommandWrapper for JobObject { } }; - let job_port = match make_job_object(handle, false) { + let job_port = match make_job_object(handle, kill_on_drop) { Ok(job_port) => job_port, Err(error) => { terminate_child(&mut *inner); diff --git a/src/std/kill_on_drop.rs b/src/std/kill_on_drop.rs new file mode 100644 index 0000000..49a0e45 --- /dev/null +++ b/src/std/kill_on_drop.rs @@ -0,0 +1,12 @@ +use super::CommandWrapper; + +/// Opt in to terminating every process in a Windows [`super::JobObject`] when its child wrapper is +/// dropped. +/// +/// This marker has no effect without [`super::JobObject`]. Without this marker, the std +/// [`super::JobObject`] keeps its existing behavior and closing its child wrapper does not +/// terminate the job. +#[derive(Clone, Copy, Debug)] +pub struct KillOnDrop; + +impl CommandWrapper for KillOnDrop {} diff --git a/tests/std_windows/job_object_kill_on_drop.rs b/tests/std_windows/job_object_kill_on_drop.rs new file mode 100644 index 0000000..a83c2b7 --- /dev/null +++ b/tests/std_windows/job_object_kill_on_drop.rs @@ -0,0 +1,387 @@ +use std::{ + io::{BufRead, BufReader, Error, ErrorKind, Result, Write}, + process::{Child, Command}, + sync::{Arc, Mutex, mpsc}, + thread::{self, sleep}, + time::{Duration, Instant}, +}; + +use windows::{ + Win32::{ + Foundation::{CloseHandle, HANDLE, WAIT_FAILED, WAIT_OBJECT_0, WAIT_TIMEOUT}, + System::Threading::{ + CreateEventW, OpenProcess, PROCESS_SYNCHRONIZE, PROCESS_TERMINATE, WaitForSingleObject, + }, + }, + core::PCWSTR, +}; + +use super::prelude::*; + +const STARTUP_TIMEOUT: Duration = Duration::from_secs(10); +const EXIT_TIMEOUT: Duration = Duration::from_secs(5); +const DESCENDANT_MARKER: &str = "PW-DESCENDANT:"; + +#[derive(Clone, Copy)] +enum RegistrationOrder { + Default, + KillOnDropFirst, + JobObjectFirst, +} + +struct ProcessGuard(Option); + +impl ProcessGuard { + fn open(pid: u32) -> Result { + let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE | PROCESS_TERMINATE, false, pid) } + .map_err(Error::other)?; + Ok(Self(Some(handle))) + } + + fn handle(&self) -> HANDLE { + self.0 + .expect("ProcessGuard is disarmed only after the process exits") + } + + fn is_running(&self) -> Result { + match unsafe { WaitForSingleObject(self.handle(), 0) } { + WAIT_TIMEOUT => Ok(true), + WAIT_OBJECT_0 => Ok(false), + WAIT_FAILED => Err(Error::last_os_error()), + result => Err(Error::other(format!("unexpected wait result: {result:?}"))), + } + } + + fn wait_for_exit(&self) -> Result<()> { + let deadline = Instant::now() + EXIT_TIMEOUT; + loop { + match unsafe { WaitForSingleObject(self.handle(), 0) } { + WAIT_OBJECT_0 => return Ok(()), + WAIT_FAILED => return Err(Error::last_os_error()), + WAIT_TIMEOUT if Instant::now() < deadline => sleep(Duration::from_millis(10)), + WAIT_TIMEOUT => { + return Err(Error::new( + ErrorKind::TimedOut, + "process remained alive after its job was dropped", + )); + } + result => return Err(Error::other(format!("unexpected wait result: {result:?}"))), + } + } + } + + fn terminate(&self) -> Result<()> { + unsafe { windows::Win32::System::Threading::TerminateProcess(self.handle(), 1) } + .map_err(Error::other)?; + self.wait_for_exit() + } + + fn disarm(mut self) -> Result<()> { + if let Some(handle) = self.0.take() { + unsafe { CloseHandle(handle) }.map_err(Error::other)?; + } + Ok(()) + } +} + +impl Drop for ProcessGuard { + fn drop(&mut self) { + if let Some(handle) = self.0.take() { + unsafe { windows::Win32::System::Threading::TerminateProcess(handle, 1) }.ok(); + unsafe { CloseHandle(handle) }.ok(); + } + } +} + +fn descendant_pid(child: &mut dyn ChildWrapper) -> Result { + let stdout = child.stdout().take().ok_or_else(|| { + Error::new( + ErrorKind::BrokenPipe, + "descendant helper stdout was not piped", + ) + })?; + let (sender, receiver) = mpsc::channel(); + let reader = thread::spawn(move || { + let mut reader = BufReader::new(stdout); + let mut transcript = String::new(); + loop { + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) => { + let _ = sender.send(Err(Error::other(format!( + "descendant helper exited before reporting its process ID; output: {transcript}" + )))); + return; + } + Ok(_) => { + if let Some(pid) = line.strip_prefix(DESCENDANT_MARKER) { + let _ = sender.send(pid.trim().parse::().map_err(Error::other)); + return; + } + transcript.push_str(&line); + } + Err(error) => { + let _ = sender.send(Err(error)); + return; + } + } + } + }); + + let result = match receiver.recv_timeout(STARTUP_TIMEOUT) { + Ok(result) => result, + Err(mpsc::RecvTimeoutError::Timeout) => Err(Error::new( + ErrorKind::TimedOut, + "descendant helper did not report its process ID", + )), + Err(mpsc::RecvTimeoutError::Disconnected) => Err(Error::other( + "descendant helper output reader exited before reporting a process ID", + )), + }; + + if result.is_err() { + let _ = child.start_kill(); + let _ = child.wait(); + } + reader + .join() + .map_err(|_| Error::other("descendant output reader panicked"))?; + result +} + +fn command(order: RegistrationOrder) -> CommandWrap { + let mut command = CommandWrap::with_new( + std::env::current_exe().expect("test executable path"), + |command| { + command + .args(["descendant_parent", "--ignored", "--nocapture"]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + }, + ); + match order { + RegistrationOrder::Default => { + command.wrap(JobObject); + } + RegistrationOrder::KillOnDropFirst => { + command.wrap(KillOnDrop).wrap(JobObject); + } + RegistrationOrder::JobObjectFirst => { + command.wrap(JobObject).wrap(KillOnDrop); + } + } + command +} + +#[test] +#[ignore = "subprocess helper"] +fn descendant_leaf() { + thread::sleep(Duration::from_secs(300)); +} + +#[test] +#[ignore = "subprocess helper"] +fn descendant_parent() { + let mut descendant = Command::new(std::env::current_exe().expect("test executable path")) + .args(["descendant_leaf", "--ignored", "--nocapture"]) + .spawn() + .expect("spawn descendant helper"); + println!("{DESCENDANT_MARKER}{}", descendant.id()); + std::io::stdout() + .flush() + .expect("flush descendant process ID"); + descendant.wait().expect("wait for descendant helper"); +} + +#[test] +fn kill_on_drop_is_opt_in_and_registration_order_independent() -> Result<()> { + for (order, should_kill) in [ + (RegistrationOrder::Default, false), + (RegistrationOrder::KillOnDropFirst, true), + (RegistrationOrder::JobObjectFirst, true), + ] { + let mut command = command(order); + let mut child = command.spawn()?; + let descendant_id = descendant_pid(child.as_mut())?; + let parent = ProcessGuard::open(child.id())?; + let descendant = ProcessGuard::open(descendant_id)?; + + drop(child); + if should_kill { + parent.wait_for_exit()?; + descendant.wait_for_exit()?; + } else { + assert!(parent.is_running()?); + assert!(descendant.is_running()?); + descendant.terminate()?; + parent.wait_for_exit()?; + } + + parent.disarm()?; + descendant.disarm()?; + } + Ok(()) +} + +#[derive(Debug)] +struct EventHandle(usize); + +impl EventHandle { + fn new() -> Result { + let handle = + unsafe { CreateEventW(None, false, false, PCWSTR::null()) }.map_err(Error::other)?; + Ok(Self(handle.0 as usize)) + } +} + +impl Drop for EventHandle { + fn drop(&mut self) { + unsafe { CloseHandle(HANDLE(self.0 as _)) }.ok(); + } +} + +#[derive(Debug)] +struct ObservedProcessHandle(usize); + +impl ObservedProcessHandle { + fn open(pid: u32) -> Result { + let handle = + unsafe { OpenProcess(PROCESS_SYNCHRONIZE, false, pid) }.map_err(Error::other)?; + Ok(Self(handle.0 as usize)) + } + + fn wait_for_exit(&self) -> Result<()> { + let deadline = Instant::now() + EXIT_TIMEOUT; + loop { + match unsafe { WaitForSingleObject(HANDLE(self.0 as _), 0) } { + WAIT_OBJECT_0 => return Ok(()), + WAIT_FAILED => return Err(Error::last_os_error()), + WAIT_TIMEOUT if Instant::now() < deadline => sleep(Duration::from_millis(10)), + WAIT_TIMEOUT => { + return Err(Error::new( + ErrorKind::TimedOut, + "process remained alive after job assignment failed", + )); + } + result => return Err(Error::other(format!("unexpected wait result: {result:?}"))), + } + } + } +} + +impl Drop for ObservedProcessHandle { + fn drop(&mut self) { + unsafe { CloseHandle(HANDLE(self.0 as _)) }.ok(); + } +} + +#[derive(Clone, Debug)] +struct EventAsProcessHandle { + event: Arc, + observed_process: Arc>>, +} + +#[derive(Debug)] +struct MisreportedProcessHandleChild { + child: Child, + event: Arc, +} + +impl ChildWrapper for MisreportedProcessHandleChild { + fn inner(&self) -> &dyn ChildWrapper { + self + } + + fn inner_mut(&mut self) -> &mut dyn ChildWrapper { + self + } + + fn into_inner(self: Box) -> Box { + self + } + + fn process_handle(&self) -> Option> { + // SAFETY: the test-owned event remains open through the spawn attempt; it is a valid handle, + // but intentionally not a process handle, so AssignProcessToJobObject rejects it. + Some(unsafe { + std::os::windows::io::BorrowedHandle::borrow_raw(self.event.0 as *mut std::ffi::c_void) + }) + } + + fn id(&self) -> u32 { + self.child.id() + } + + fn start_kill(&mut self) -> Result<()> { + self.child.kill() + } + + fn wait(&mut self) -> Result { + self.child.wait() + } +} + +impl CommandWrapper for EventAsProcessHandle { + fn wrap_child( + &mut self, + child: Box, + _core: &CommandWrap, + ) -> Result> { + let child = unsafe { child.try_into_inner_child() } + .map_err(|_| Error::other("test expected a native child"))?; + *self + .observed_process + .lock() + .expect("process observation lock") = Some(ObservedProcessHandle::open(child.id())?); + Ok(Box::new(MisreportedProcessHandleChild { + child, + event: Arc::clone(&self.event), + })) + } +} + +#[test] +fn failed_job_assignment_kills_the_suspended_process_before_its_marker_runs() -> Result<()> { + let directory = tempfile::tempdir()?; + let marker = directory.path().join("executed.txt"); + let mut control = Command::new("cmd.exe") + .current_dir(directory.path()) + .args(["/D", "/S", "/C", "echo executed > executed.txt"]) + .spawn()?; + assert!(control.wait()?.success()); + assert!(marker.exists(), "control command did not write its marker"); + std::fs::remove_file(&marker)?; + + let event = Arc::new(EventHandle::new()?); + let observed_process = Arc::new(Mutex::new(None)); + let mut command = CommandWrap::with_new("cmd.exe", |command| { + command.current_dir(directory.path()).args([ + "/D", + "/S", + "/C", + "echo executed > executed.txt", + ]); + }); + command + .wrap(EventAsProcessHandle { + event: Arc::clone(&event), + observed_process: Arc::clone(&observed_process), + }) + .wrap(JobObject); + + let error = command + .spawn() + .expect_err("an event handle cannot be assigned to a job"); + assert_eq!(error.kind(), ErrorKind::Other); + let observed_process = observed_process + .lock() + .expect("process observation lock") + .take() + .expect("test wrapper observed the spawned process"); + observed_process.wait_for_exit()?; + assert!( + !marker.exists(), + "the suspended command ran before assignment failed" + ); + Ok(()) +} diff --git a/tests/std_windows/mod.rs b/tests/std_windows/mod.rs index 857b0dd..85925ea 100644 --- a/tests/std_windows/mod.rs +++ b/tests/std_windows/mod.rs @@ -16,6 +16,8 @@ mod creation_flags_job_object; mod id_same_as_inner; mod inner_read_stdout; mod into_inner_write_stdin; +#[cfg(all(feature = "job-object", feature = "kill-on-drop"))] +mod job_object_kill_on_drop; mod kill_and_try_wait; #[cfg(feature = "job-object")] mod process_handle;