diff --git a/README.md b/README.md index c530efd..0842c66 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,22 @@ 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`. +To terminate the job without the child, for example from another thread while the child is being +waited on, get a `JobObjectHandle` from the `JobObjectChild`. It can be cloned and sent across +threads, and keeps the job open for as long as it's held, even after the child is dropped: + +```rust +let mut child = CommandWrap::with_new("watch", |command| { command.arg("ls"); }) + .wrap(JobObject) + .spawn()?; +let job = (child.as_ref() as &dyn std::any::Any) + .downcast_ref::() + .expect("JobObject is the outermost wrapper") + .job_handle(); +std::thread::spawn(move || job.terminate(1)); +let status = child.wait().await?; +``` + ### Process group - Platforms: POSIX (Linux, Mac, BSDs...) diff --git a/src/std.rs b/src/std.rs index 8f9451a..b1f1b59 100644 --- a/src/std.rs +++ b/src/std.rs @@ -8,6 +8,9 @@ //! use process_wrap::std::*; //! ``` +#[cfg(all(windows, feature = "job-object"))] +#[doc(inline)] +pub use crate::windows::JobObjectHandle; #[doc(inline)] pub use core::{ChildWrapper, CommandWrap, CommandWrapper}; #[cfg(all(windows, feature = "creation-flags"))] diff --git a/src/std/job_object.rs b/src/std/job_object.rs index ad10743..7bad75b 100644 --- a/src/std/job_object.rs +++ b/src/std/job_object.rs @@ -18,7 +18,7 @@ use windows::Win32::{ use crate::{ ChildExitStatus, windows::{ - JobPort, job_creation_flags, make_job_object, resume_threads, terminate_job, wait_on_job, + JobObjectHandle, JobPort, job_creation_flags, make_job_object, resume_threads, wait_on_job, }, }; @@ -113,7 +113,7 @@ impl CommandWrapper for JobObject { if policy.resume_after_assignment { if let Err(error) = resume_threads(handle) { - let _ = terminate_job(job_port.job, 1); + let _ = job_port.job.terminate(1); terminate_child(&mut *inner); return Err(error); } @@ -140,6 +140,15 @@ impl JobObjectChild { job_port, } } + + /// Get a handle to the job object this child was assigned to. + /// + /// Unlike the child, the handle can be cloned and used from other threads, for example to + /// terminate the job while another thread waits on the child, and it keeps the job open after + /// the child is dropped. See [`JobObjectHandle`]. + pub fn job_handle(&self) -> JobObjectHandle { + self.job_port.job.clone() + } } impl ChildWrapper for JobObjectChild { @@ -164,7 +173,7 @@ impl ChildWrapper for JobObjectChild { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn start_kill(&mut self) -> Result<()> { - terminate_job(self.job_port.job, 1) + self.job_port.job.terminate(1) } #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] diff --git a/src/tokio.rs b/src/tokio.rs index 4eb8f75..2617989 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -8,6 +8,9 @@ //! use process_wrap::tokio::*; //! ``` +#[cfg(all(windows, feature = "job-object"))] +#[doc(inline)] +pub use crate::windows::JobObjectHandle; #[doc(inline)] pub use core::{ChildWrapper, CommandWrap, CommandWrapper}; #[cfg(all(windows, feature = "creation-flags"))] diff --git a/src/tokio/job_object.rs b/src/tokio/job_object.rs index 4433ef9..e3d2d96 100644 --- a/src/tokio/job_object.rs +++ b/src/tokio/job_object.rs @@ -18,7 +18,7 @@ use windows::Win32::{ use crate::{ ChildExitStatus, windows::{ - JobPort, job_creation_flags, make_job_object, resume_threads, terminate_job, wait_on_job, + JobObjectHandle, JobPort, job_creation_flags, make_job_object, resume_threads, wait_on_job, }, }; @@ -119,7 +119,7 @@ impl CommandWrapper for JobObject { if policy.resume_after_assignment { if let Err(error) = resume_threads(handle) { - let _ = terminate_job(job_port.job, 1); + let _ = job_port.job.terminate(1); terminate_child(&mut *inner); return Err(error); } @@ -146,6 +146,15 @@ impl JobObjectChild { job_port, } } + + /// Get a handle to the job object this child was assigned to. + /// + /// Unlike the child, the handle can be cloned and used from other threads, for example to + /// terminate the job while another thread waits on the child, and it keeps the job open after + /// the child is dropped. See [`JobObjectHandle`]. + pub fn job_handle(&self) -> JobObjectHandle { + self.job_port.job.clone() + } } impl ChildWrapper for JobObjectChild { @@ -170,7 +179,7 @@ impl ChildWrapper for JobObjectChild { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn start_kill(&mut self) -> Result<()> { - terminate_job(self.job_port.job, 1) + self.job_port.job.terminate(1) } #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] diff --git a/src/windows.rs b/src/windows.rs index 890bdb5..792909e 100644 --- a/src/windows.rs +++ b/src/windows.rs @@ -3,6 +3,10 @@ use std::{ io::{Error, Result}, ops::ControlFlow, + os::windows::io::{ + AsHandle, AsRawHandle, BorrowedHandle, FromRawHandle, OwnedHandle as OwnedJobHandle, + }, + sync::Arc, time::Duration, }; @@ -85,11 +89,37 @@ mod creation_flag_tests { } } -#[derive(Clone, Copy, Debug)] -pub struct JobHandle(pub HANDLE); +/// A handle to the job object of a `JobObjectChild`, shared with it. +/// +/// Get one with `JobObjectChild::job_handle()`. Unlike the child, it can be cloned and sent to other +/// threads, so the job can be terminated without the child: for example while another thread waits +/// on it. +/// +/// The job is closed once the child and every handle to it have been dropped. With `KillOnDrop` +/// (Tokio-only), that's when the processes still in it are killed, so holding a handle keeps them +/// running after the child is dropped. A child unwrapped with `into_inner()` leaves the job open. +/// +/// [`AsHandle`] borrows the job's handle, to use it with other job object APIs. +#[derive(Clone, Debug)] +pub struct JobObjectHandle(Arc); + +impl JobObjectHandle { + /// Terminate every process in the job, without waiting for them to exit. + /// + /// This is what `start_kill()` does on the child, but it doesn't need the child. Each process + /// exits with `exit_code`. See [TerminateJobObject](https://learn.microsoft.com/en-us/windows/win32/api/jobapi2/nf-jobapi2-terminatejobobject). + #[cfg_attr(feature = "tracing", instrument(level = "debug"))] + pub fn terminate(&self, exit_code: u32) -> Result<()> { + unsafe { TerminateJobObject(HANDLE(self.0.as_raw_handle()), exit_code) } + .map_err(Error::other) + } +} -unsafe impl Send for JobHandle {} -unsafe impl Sync for JobHandle {} +impl AsHandle for JobObjectHandle { + fn as_handle(&self) -> BorrowedHandle<'_> { + self.0.as_handle() + } +} #[derive(Clone, Copy, Debug)] pub struct PortHandle(pub HANDLE); @@ -99,16 +129,16 @@ unsafe impl Sync for PortHandle {} /// A JobObject and its associated completion port. /// -/// This struct closes the handles when dropped. +/// This struct closes the completion port when dropped; the job is closed when the last +/// `JobObjectHandle` to it is. #[derive(Debug)] pub(crate) struct JobPort { - pub job: JobHandle, + pub job: JobObjectHandle, pub completion_port: PortHandle, } impl Drop for JobPort { fn drop(&mut self) { - unsafe { CloseHandle(self.job.0) }.ok(); unsafe { CloseHandle(self.completion_port.0) }.ok(); } } @@ -173,7 +203,9 @@ pub(crate) fn make_job_object(process_handle: HANDLE, kill_on_drop: bool) -> Res debug!(?job, ?process_handle, "done AssignProcessToJobObject"); Ok(JobPort { - job: JobHandle(job.into_raw()), + job: JobObjectHandle(Arc::new(unsafe { + OwnedJobHandle::from_raw_handle(job.into_raw().0) + })), completion_port: PortHandle(completion_port.into_raw()), }) } @@ -234,12 +266,6 @@ pub(crate) fn resume_threads(child_process: HANDLE) -> Result<()> { unsafe { inner(child_id, tool_handle.0) } } -/// Terminate a job object without waiting for the processes to exit. -#[cfg_attr(feature = "tracing", instrument(level = "debug"))] -pub(crate) fn terminate_job(job: JobHandle, exit_code: u32) -> Result<()> { - unsafe { TerminateJobObject(job.0, exit_code) }.map_err(Error::other) -} - /// Wait for a job to complete. #[cfg_attr(feature = "tracing", instrument(level = "debug"))] pub(crate) fn wait_on_job( diff --git a/tests/std_windows/job_handle.rs b/tests/std_windows/job_handle.rs new file mode 100644 index 0000000..2fc4c26 --- /dev/null +++ b/tests/std_windows/job_handle.rs @@ -0,0 +1,152 @@ +use std::{ + any::Any, + io::{BufRead, BufReader, Error, ErrorKind}, + os::windows::io::{AsHandle, AsRawHandle}, + process::Command as StdCommand, + sync::mpsc, + time::Instant, +}; + +use windows::Win32::{ + Foundation::{CloseHandle, HANDLE, WAIT_OBJECT_0, WAIT_TIMEOUT}, + System::Threading::{ + OpenProcess, PROCESS_SYNCHRONIZE, PROCESS_TERMINATE, TerminateProcess, WaitForSingleObject, + }, +}; + +use super::prelude::*; + +const EXIT_TIMEOUT: Duration = Duration::from_secs(5); +const DESCENDANT_MARKER: &str = "PW-JOB-HANDLE-DESCENDANT:"; + +/// A process opened by ID, so its exit can be observed; terminated if a test fails before it exits. +struct ProcessGuard(HANDLE); + +impl ProcessGuard { + fn open(pid: u32) -> Result { + let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE | PROCESS_TERMINATE, false, pid) } + .map_err(Error::other)?; + Ok(Self(handle)) + } + + fn has_exited(&self) -> Result { + match unsafe { WaitForSingleObject(self.0, 0) } { + WAIT_OBJECT_0 => Ok(true), + WAIT_TIMEOUT => Ok(false), + _ => Err(Error::last_os_error()), + } + } + + fn wait_for_exit(&self) -> Result<()> { + let deadline = Instant::now() + EXIT_TIMEOUT; + while !self.has_exited()? { + if Instant::now() >= deadline { + return Err(Error::new(ErrorKind::TimedOut, "process did not exit")); + } + sleep(Duration::from_millis(10)); + } + Ok(()) + } +} + +impl Drop for ProcessGuard { + fn drop(&mut self) { + unsafe { TerminateProcess(self.0, 1) }.ok(); + unsafe { CloseHandle(self.0) }.ok(); + } +} + +/// A process which starts another, and reports its PID. +fn tree() -> CommandWrap { + CommandWrap::with_new(std::env::current_exe().unwrap(), |command| { + command + .args(["job_handle_parent", "--ignored", "--nocapture"]) + .stdout(Stdio::piped()); + }) +} + +fn descendant_pid(child: &mut dyn ChildWrapper) -> Result { + let stdout = child.stdout().take().expect("the helper's stdout is piped"); + for line in BufReader::new(stdout).lines() { + if let Some(pid) = line?.strip_prefix(DESCENDANT_MARKER) { + return pid.trim().parse().map_err(Error::other); + } + } + Err(Error::other( + "helper exited before reporting its descendant", + )) +} + +fn job_handle(child: &dyn ChildWrapper) -> JobObjectHandle { + (child as &dyn Any) + .downcast_ref::() + .expect("JobObject is the outermost wrapper") + .job_handle() +} + +#[test] +#[ignore = "subprocess helper"] +fn job_handle_leaf() { + sleep(Duration::from_secs(300)); +} + +#[test] +#[ignore = "subprocess helper"] +fn job_handle_parent() { + // The leaf's output would go to a pipe the test closes once it has read the PID. + let mut descendant = StdCommand::new(std::env::current_exe().unwrap()) + .args(["job_handle_leaf", "--ignored"]) + .stdout(Stdio::null()) + .spawn() + .unwrap(); + println!("{DESCENDANT_MARKER}{}", descendant.id()); + std::io::stdout().flush().unwrap(); + descendant.wait().unwrap(); +} + +#[test] +fn terminates_the_job_while_another_thread_waits_on_the_child() -> Result<()> { + let mut child = tree().wrap(JobObject).spawn()?; + let leader = ProcessGuard::open(child.id())?; + let descendant = ProcessGuard::open(descendant_pid(child.as_mut())?)?; + + let job = job_handle(child.as_ref()); + assert_eq!( + job.as_handle().as_raw_handle(), + job_handle(child.as_ref()).as_handle().as_raw_handle(), + "every job handle shares the one job handle" + ); + + let (sender, waited) = mpsc::channel(); + std::thread::spawn(move || { + let _ = sender.send(child.wait()); + }); + job.terminate(7)?; + let status = waited + .recv_timeout(EXIT_TIMEOUT) + .map_err(|_| Error::new(ErrorKind::TimedOut, "child did not exit"))??; + + assert_eq!(status.code(), Some(7)); + leader.wait_for_exit()?; + descendant.wait_for_exit()?; + Ok(()) +} + +#[test] +fn terminates_the_job_after_the_child_is_dropped() -> Result<()> { + let mut child = tree().wrap(JobObject).spawn()?; + let leader = ProcessGuard::open(child.id())?; + let descendant = ProcessGuard::open(descendant_pid(child.as_mut())?)?; + let job = job_handle(child.as_ref()); + + drop(child); + assert!( + !leader.has_exited()?, + "dropping a std child leaves it running" + ); + + job.terminate(7)?; + leader.wait_for_exit()?; + descendant.wait_for_exit()?; + Ok(()) +} diff --git a/tests/std_windows/mod.rs b/tests/std_windows/mod.rs index 857b0dd..ff22d06 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(feature = "job-object")] +mod job_handle; mod kill_and_try_wait; #[cfg(feature = "job-object")] mod process_handle; diff --git a/tests/tokio_windows/job_handle.rs b/tests/tokio_windows/job_handle.rs new file mode 100644 index 0000000..4bb11c4 --- /dev/null +++ b/tests/tokio_windows/job_handle.rs @@ -0,0 +1,157 @@ +use std::{ + any::Any, + io::{Error, ErrorKind, Write}, + os::windows::io::{AsHandle, AsRawHandle}, + process::Command as StdCommand, + time::Instant, +}; + +use tokio::io::{AsyncBufReadExt, BufReader}; +use windows::Win32::{ + Foundation::{CloseHandle, HANDLE, WAIT_OBJECT_0, WAIT_TIMEOUT}, + System::Threading::{ + OpenProcess, PROCESS_SYNCHRONIZE, PROCESS_TERMINATE, TerminateProcess, WaitForSingleObject, + }, +}; + +use super::prelude::*; + +const STARTUP_TIMEOUT: Duration = Duration::from_secs(10); +const EXIT_TIMEOUT: Duration = Duration::from_secs(5); +const DESCENDANT_MARKER: &str = "PW-JOB-HANDLE-DESCENDANT:"; + +/// A process opened by ID, so its exit can be observed; terminated if a test fails before it exits. +struct ProcessGuard(HANDLE); + +impl ProcessGuard { + fn open(pid: u32) -> Result { + let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE | PROCESS_TERMINATE, false, pid) } + .map_err(Error::other)?; + Ok(Self(handle)) + } + + fn has_exited(&self) -> Result { + match unsafe { WaitForSingleObject(self.0, 0) } { + WAIT_OBJECT_0 => Ok(true), + WAIT_TIMEOUT => Ok(false), + _ => Err(Error::last_os_error()), + } + } + + async fn wait_for_exit(&self) -> Result<()> { + let deadline = Instant::now() + EXIT_TIMEOUT; + while !self.has_exited()? { + if Instant::now() >= deadline { + return Err(Error::new(ErrorKind::TimedOut, "process did not exit")); + } + sleep(Duration::from_millis(10)).await; + } + Ok(()) + } +} + +impl Drop for ProcessGuard { + fn drop(&mut self) { + unsafe { TerminateProcess(self.0, 1) }.ok(); + unsafe { CloseHandle(self.0) }.ok(); + } +} + +/// A process which starts another, and reports its PID. +fn tree() -> CommandWrap { + CommandWrap::with_new(std::env::current_exe().unwrap(), |command| { + command + .args(["job_handle_parent", "--ignored", "--nocapture"]) + .stdout(Stdio::piped()); + }) +} + +async fn descendant_pid(child: &mut dyn ChildWrapper) -> Result { + let stdout = child.stdout().take().expect("the helper's stdout is piped"); + let mut lines = BufReader::new(stdout).lines(); + tokio::time::timeout(STARTUP_TIMEOUT, async { + while let Some(line) = lines.next_line().await? { + if let Some(pid) = line.strip_prefix(DESCENDANT_MARKER) { + return pid.trim().parse().map_err(Error::other); + } + } + Err(Error::other( + "helper exited before reporting its descendant", + )) + }) + .await + .map_err(|_| Error::new(ErrorKind::TimedOut, "helper did not report its descendant"))? +} + +fn job_handle(child: &dyn ChildWrapper) -> JobObjectHandle { + (child as &dyn Any) + .downcast_ref::() + .expect("JobObject is the outermost wrapper") + .job_handle() +} + +#[test] +#[ignore = "subprocess helper"] +fn job_handle_leaf() { + std::thread::sleep(Duration::from_secs(300)); +} + +#[test] +#[ignore = "subprocess helper"] +fn job_handle_parent() { + // The leaf's output would go to a pipe the test closes once it has read the PID. + let mut descendant = StdCommand::new(std::env::current_exe().unwrap()) + .args(["job_handle_leaf", "--ignored"]) + .stdout(Stdio::null()) + .spawn() + .unwrap(); + println!("{DESCENDANT_MARKER}{}", descendant.id()); + std::io::stdout().flush().unwrap(); + descendant.wait().unwrap(); +} + +#[tokio::test] +async fn terminates_the_job_from_another_thread_while_the_child_is_awaited() -> Result<()> { + let mut child = tree().wrap(JobObject).spawn()?; + let leader = ProcessGuard::open(child.id().expect("the helper is running"))?; + let descendant = ProcessGuard::open(descendant_pid(child.as_mut()).await?)?; + + let job = job_handle(child.as_ref()); + assert_eq!( + job.as_handle().as_raw_handle(), + job_handle(child.as_ref()).as_handle().as_raw_handle(), + "every job handle shares the one job handle" + ); + + let terminator = std::thread::spawn(move || job.terminate(7)); + let status = tokio::time::timeout(EXIT_TIMEOUT, child.wait()) + .await + .map_err(|_| Error::new(ErrorKind::TimedOut, "child did not exit"))??; + terminator + .join() + .expect("the terminating thread did not panic")?; + + assert_eq!(status.code(), Some(7)); + leader.wait_for_exit().await?; + descendant.wait_for_exit().await?; + Ok(()) +} + +#[cfg(feature = "kill-on-drop")] +#[tokio::test] +async fn keeps_a_kill_on_drop_job_open_after_the_child_is_dropped() -> Result<()> { + let mut child = tree().wrap(JobObject).wrap(KillOnDrop).spawn()?; + let descendant = ProcessGuard::open(descendant_pid(child.as_mut()).await?)?; + let job = job_handle(child.as_ref()); + + drop(child); + sleep(DIE_TIME).await; + assert!( + !descendant.has_exited()?, + "descendant outlived the child while the job handle was held" + ); + + drop(job); + descendant.wait_for_exit().await?; + Ok(()) +} diff --git a/tests/tokio_windows/mod.rs b/tests/tokio_windows/mod.rs index adb69bd..96e8f9f 100644 --- a/tests/tokio_windows/mod.rs +++ b/tests/tokio_windows/mod.rs @@ -15,6 +15,8 @@ mod creation_flags_job_object; mod id_same_as_inner; mod inner_read_stdout; mod into_inner_write_stdin; +#[cfg(feature = "job-object")] +mod job_handle; #[cfg(all(feature = "job-object", feature = "kill-on-drop"))] mod job_object_kill_on_drop; mod kill_and_try_wait;