Expose Windows Job Object CPU accounting

src/std/core.rs

@@ -6,6 +6,8 @@ use std::{
 
 #[cfg(windows)]
 use std::os::windows::io::{AsHandle, BorrowedHandle};
+#[cfg(windows)]
+use std::time::Duration;
 
 #[cfg(unix)]
 use nix::{
@@ -92,6 +94,46 @@ pub trait ChildWrapper: Any + std::fmt::Debug + Send + Sync {
 		None
 	}
 
+	/// Return cumulative user plus kernel CPU time for this child's Windows Job Object.
+	///
+	/// This includes terminated members and is available only when wrapped with `JobObject`.
+	/// Transparent wrappers forward to their child; other children return `Unsupported`.
+	#[cfg(windows)]
+	fn job_cpu_time(&self) -> Result<Duration> {
+		let inner = self.inner();
+		let self_pointer = self as *const Self as *const ();
+		let inner_pointer = inner as *const dyn ChildWrapper as *const ();
+		if self_pointer != inner_pointer {
+			inner.job_cpu_time()
+		} else {
+			Err(std::io::Error::new(
+				std::io::ErrorKind::Unsupported,
+				"child has no Windows Job Object accounting boundary",
+			))
+		}
+	}
+
+	/// Return a coherent current working-set total for this child's Windows Job Object.
+	///
+	/// If a complete aggregate cannot be sampled, this returns an error instead of a partial sum.
+	/// Implementations must not approximate a coherent aggregate by adding separate per-process
+	/// snapshots. Transparent wrappers forward to their child; unsupported children return
+	/// `Unsupported`.
+	#[cfg(windows)]
+	fn job_resident_bytes(&self) -> Result<u64> {
+		let inner = self.inner();
+		let self_pointer = self as *const Self as *const ();
+		let inner_pointer = inner as *const dyn ChildWrapper as *const ();
+		if self_pointer != inner_pointer {
+			inner.job_resident_bytes()
+		} else {
+			Err(std::io::Error::new(
+				std::io::ErrorKind::Unsupported,
+				"child has no Windows Job Object accounting boundary",
+			))
+		}
+	}
+
 	/// Obtain a clone if possible.
 	///
 	/// Some implementations may make it possible to clone the implementing structure, even though

src/std/job_object.rs

@@ -161,6 +161,14 @@ impl ChildWrapper for JobObjectChild {
 	fn process_handle(&self) -> Option<BorrowedHandle<'_>> {
 		child_process_handle(self.inner.as_ref())
 	}
+	#[cfg(windows)]
+	fn job_cpu_time(&self) -> Result<Duration> {
+		crate::windows::job_cpu_time(self.job_port.job)
+	}
+	#[cfg(windows)]
+	fn job_resident_bytes(&self) -> Result<u64> {
+		crate::windows::job_resident_bytes(self.job_port.job)
+	}
 
 	#[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))]
 	fn start_kill(&mut self) -> Result<()> {

src/windows.rs

@@ -1,7 +1,7 @@
 //! Windows API support functions.
 
 use std::{
-	io::{Error, Result},
+	io::{Error, ErrorKind, Result},
 	ops::ControlFlow,
 	time::Duration,
 };
@@ -19,9 +19,10 @@ use windows::{
 			IO::{CreateIoCompletionPort, GetQueuedCompletionStatus, OVERLAPPED},
 			JobObjects::{
 				AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
-				JOBOBJECT_ASSOCIATE_COMPLETION_PORT, JOBOBJECT_EXTENDED_LIMIT_INFORMATION,
-				JobObjectAssociateCompletionPortInformation, JobObjectExtendedLimitInformation,
-				SetInformationJobObject, TerminateJobObject,
+				JOBOBJECT_ASSOCIATE_COMPLETION_PORT, JOBOBJECT_BASIC_ACCOUNTING_INFORMATION,
+				JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectAssociateCompletionPortInformation,
+				JobObjectBasicAccountingInformation, JobObjectExtendedLimitInformation,
+				QueryInformationJobObject, SetInformationJobObject, TerminateJobObject,
 			},
 			Threading::{
 				CREATE_SUSPENDED, GetProcessId, INFINITE, OpenThread, PROCESS_CREATION_FLAGS,
@@ -113,6 +114,43 @@ impl Drop for JobPort {
 	}
 }
 
+fn job_accounting(job: JobHandle) -> Result<JOBOBJECT_BASIC_ACCOUNTING_INFORMATION> {
+	let mut accounting = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION::default();
+	let size = u32::try_from(std::mem::size_of_val(&accounting)).map_err(Error::other)?;
+	// SAFETY: `accounting` is a writable, correctly sized instance of the information class.
+	unsafe {
+		QueryInformationJobObject(
+			Some(job.0),
+			JobObjectBasicAccountingInformation,
+			(&mut accounting as *mut JOBOBJECT_BASIC_ACCOUNTING_INFORMATION).cast(),
+			size,
+			None,
+		)
+	}
+	.map_err(Error::other)?;
+	Ok(accounting)
+}
+
+pub(crate) fn job_cpu_time(job: JobHandle) -> Result<Duration> {
+	let accounting = job_accounting(job)?;
+	let user = u64::try_from(accounting.TotalUserTime).map_err(Error::other)?;
+	let kernel = u64::try_from(accounting.TotalKernelTime).map_err(Error::other)?;
+	let ticks = user
+		.checked_add(kernel)
+		.ok_or_else(|| Error::new(ErrorKind::InvalidData, "job CPU counter overflowed"))?;
+	let nanos = ticks
+		.checked_mul(100)
+		.ok_or_else(|| Error::new(ErrorKind::InvalidData, "job CPU duration overflowed"))?;
+	Ok(Duration::from_nanos(nanos))
+}
+
+pub(crate) fn job_resident_bytes(_job: JobHandle) -> Result<u64> {
+	Err(Error::new(
+		ErrorKind::Unsupported,
+		"Windows Job Object has no coherent current resident-memory aggregate",
+	))
+}
+
 /// Create a JobObject and an associated completion port.
 ///
 /// If `kill_on_drop` is true, we opt into the `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE` flag, which

tests/std_windows/job_object_accounting.rs

@@ -0,0 +1,125 @@
+use std::{
+	hint::black_box,
+	io::{BufRead, BufReader, ErrorKind, Write},
+	process::{Command, Stdio},
+	thread,
+	time::{Duration, Instant},
+};
+
+use super::prelude::*;
+
+struct ChildGuard(Box<dyn ChildWrapper>);
+
+impl Drop for ChildGuard {
+	fn drop(&mut self) {
+		let _ = self.0.start_kill();
+		let _ = self.0.wait();
+	}
+}
+
+fn helper_test_filter() -> String {
+	format!("{}::helper_entrypoint", module_path!())
+}
+
+fn wrapped_helper(mode: &str) -> Result<ChildGuard> {
+	let executable = std::env::current_exe()?;
+	let filter = helper_test_filter();
+	let mut command = CommandWrap::with_new(executable, |command| {
+		command
+			.args(["--exact", &filter, "--nocapture"])
+			.env("PROCESS_WRAP_ACCOUNTING_HELPER", mode)
+			.stdout(Stdio::piped());
+	});
+	command.wrap(JobObject);
+	Ok(ChildGuard(command.spawn()?))
+}
+
+fn read_ready(child: &mut dyn ChildWrapper) -> Result<()> {
+	let stdout = child
+		.stdout()
+		.take()
+		.ok_or_else(|| std::io::Error::other("helper stdout was not piped"))?;
+	let mut reader = BufReader::new(stdout);
+	let mut line = String::new();
+	loop {
+		line.clear();
+		if reader.read_line(&mut line)? == 0 {
+			return Err(std::io::Error::other(
+				"helper exited before reporting readiness",
+			));
+		}
+		if line.trim() == "child-exited" || line.trim() == "ready" {
+			return Ok(());
+		}
+	}
+}
+
+#[test]
+fn job_cpu_time_includes_descendant_that_exited_between_samples() -> Result<()> {
+	let mut child = wrapped_helper("cpu-root")?;
+	read_ready(child.0.as_mut())?;
+	let cpu_time = child.0.job_cpu_time()?;
+	assert!(
+		cpu_time >= Duration::from_millis(250),
+		"terminated descendant CPU was missing from the Job Object total: {cpu_time:?}"
+	);
+	Ok(())
+}
+
+#[test]
+fn job_resident_bytes_fail_closed_without_a_coherent_aggregate() -> Result<()> {
+	let mut child = wrapped_helper("cpu-root")?;
+	read_ready(child.0.as_mut())?;
+	let error = child.0.job_resident_bytes().unwrap_err();
+	assert_eq!(error.kind(), ErrorKind::Unsupported);
+	Ok(())
+}
+
+#[test]
+fn child_without_job_object_reports_accounting_unavailable() -> Result<()> {
+	let mut child = Command::new("cmd.exe")
+		.args(["/D", "/S", "/C", "ping -n 6 127.0.0.1 >NUL"])
+		.spawn()?;
+	let cpu_error = <std::process::Child as ChildWrapper>::job_cpu_time(&child).unwrap_err();
+	let memory_error =
+		<std::process::Child as ChildWrapper>::job_resident_bytes(&child).unwrap_err();
+	child.kill()?;
+	let _ = child.wait()?;
+	assert_eq!(cpu_error.kind(), ErrorKind::Unsupported);
+	assert_eq!(memory_error.kind(), ErrorKind::Unsupported);
+	Ok(())
+}
+
+#[test]
+fn helper_entrypoint() -> Result<()> {
+	let Some(mode) = std::env::var_os("PROCESS_WRAP_ACCOUNTING_HELPER") else {
+		return Ok(());
+	};
+
+	match mode.to_str() {
+		Some("cpu-root") => {
+			let executable = std::env::current_exe()?;
+			let filter = helper_test_filter();
+			let mut worker = Command::new(executable)
+				.args(["--exact", &filter, "--nocapture"])
+				.env("PROCESS_WRAP_ACCOUNTING_HELPER", "cpu-worker")
+				.stdout(Stdio::null())
+				.spawn()?;
+			worker.wait()?;
+			writeln!(std::io::stdout().lock(), "child-exited")?;
+			loop {
+				thread::sleep(Duration::from_secs(60));
+			}
+		}
+		Some("cpu-worker") => {
+			let deadline = Instant::now() + Duration::from_millis(700);
+			let mut value = 0_u64;
+			while Instant::now() < deadline {
+				value = value.wrapping_mul(6364136223846793005).wrapping_add(1);
+				black_box(value);
+			}
+			Ok(())
+		}
+		_ => Err(std::io::Error::other("expected helper mode").into()),
+	}
+}

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_object_accounting;
 mod kill_and_try_wait;
 #[cfg(feature = "job-object")]
 mod process_handle;