feat(windows): add shareable job handles to JobObjectChild

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::<JobObjectChild>()
+  .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...)

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"))]

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)))]

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"))]

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)))]

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<OwnedJobHandle>);
+
+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(

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<Self> {
+		let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE | PROCESS_TERMINATE, false, pid) }
+			.map_err(Error::other)?;
+		Ok(Self(handle))
+	}
+
+	fn has_exited(&self) -> Result<bool> {
+		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<u32> {
+	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::<JobObjectChild>()
+		.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(())
+}

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;