From bc73cde98416ef4377cc10fbf67e2ff122c6691d Mon Sep 17 00:00:00 2001 From: Peter Krenesky Date: Wed, 14 Oct 2026 12:23:59 -0700 Subject: [PATCH 1/3] Expose Windows Job Object CPU accounting --- Cargo.toml | 6 ++ src/std/core.rs | 42 ++++++++++ src/std/job_object.rs | 8 ++ src/windows.rs | 46 ++++++++++- tests/std_windows/job_object_accounting.rs | 79 +++++++++++++++++++ tests/std_windows/mod.rs | 2 + tests/support/job_object_accounting_helper.rs | 44 +++++++++++ 7 files changed, 223 insertions(+), 4 deletions(-) create mode 100644 tests/std_windows/job_object_accounting.rs create mode 100644 tests/support/job_object_accounting_helper.rs diff --git a/Cargo.toml b/Cargo.toml index 2b45b84..986e2af 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,12 @@ remoteprocess = "0.5.0" tempfile = { version = "3.20.0", default-features = false } tokio = { version = "1.38.2", features = ["io-util", "macros", "process", "rt", "rt-multi-thread", "time"] } +[[bin]] +name = "job-object-accounting-helper" +path = "tests/support/job_object_accounting_helper.rs" +test = false +bench = false + [features] default = ["creation-flags", "job-object", "kill-on-drop", "process-group", "process-session", "tracing"] diff --git a/src/std/core.rs b/src/std/core.rs index 90c000b..31181b1 100644 --- a/src/std/core.rs +++ b/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 { + 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 { + 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 diff --git a/src/std/job_object.rs b/src/std/job_object.rs index ad10743..011c219 100644 --- a/src/std/job_object.rs +++ b/src/std/job_object.rs @@ -161,6 +161,14 @@ impl ChildWrapper for JobObjectChild { fn process_handle(&self) -> Option> { child_process_handle(self.inner.as_ref()) } + #[cfg(windows)] + fn job_cpu_time(&self) -> Result { + crate::windows::job_cpu_time(self.job_port.job) + } + #[cfg(windows)] + fn job_resident_bytes(&self) -> Result { + crate::windows::job_resident_bytes(self.job_port.job) + } #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn start_kill(&mut self) -> Result<()> { diff --git a/src/windows.rs b/src/windows.rs index 890bdb5..f9930da 100644 --- a/src/windows.rs +++ b/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 { + 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 { + 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 { + 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 diff --git a/tests/std_windows/job_object_accounting.rs b/tests/std_windows/job_object_accounting.rs new file mode 100644 index 0000000..a39f49f --- /dev/null +++ b/tests/std_windows/job_object_accounting.rs @@ -0,0 +1,79 @@ +use std::{ + io::{BufRead, BufReader, ErrorKind}, + process::{Command, Stdio}, + time::Duration, +}; + +use super::prelude::*; + +struct ChildGuard(Box); + +impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.start_kill(); + let _ = self.0.wait(); + } +} + +fn wrapped_helper(mode: &str) -> Result { + let mut command = CommandWrap::with_new( + env!("CARGO_BIN_EXE_job-object-accounting-helper"), + |command| { + command.arg(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 line = String::new(); + BufReader::new(stdout).read_line(&mut line)?; + if line.trim() == "child-exited" || line.trim() == "ready" { + Ok(()) + } else { + Err(std::io::Error::other( + "helper returned an unexpected readiness line", + )) + } +} + +#[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 = ::job_cpu_time(&child).unwrap_err(); + let memory_error = + ::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(()) +} diff --git a/tests/std_windows/mod.rs b/tests/std_windows/mod.rs index 857b0dd..e08f068 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_object_accounting; mod kill_and_try_wait; #[cfg(feature = "job-object")] mod process_handle; diff --git a/tests/support/job_object_accounting_helper.rs b/tests/support/job_object_accounting_helper.rs new file mode 100644 index 0000000..9bed97a --- /dev/null +++ b/tests/support/job_object_accounting_helper.rs @@ -0,0 +1,44 @@ +use std::{ + black_box, + io::{self, Write}, + process::{Command, Stdio}, + thread, + time::{Duration, Instant}, +}; + +fn cpu_root() -> io::Result<()> { + let mut worker = Command::new(std::env::current_exe()?) + .arg("cpu-worker") + .stdout(Stdio::null()) + .spawn()?; + worker.wait()?; + writeln!(io::stdout().lock(), "child-exited")?; + loop { + thread::sleep(Duration::from_secs(60)); + } +} + +fn 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); + } +} + +fn run() -> io::Result<()> { + match std::env::args().nth(1).as_deref() { + Some("cpu-root") => cpu_root()?, + Some("cpu-worker") => cpu_worker(), + _ => return Err(io::Error::other("expected helper mode").into()), + } + Ok(()) +} + +fn main() { + if let Err(error) = run() { + eprintln!("{error}"); + std::process::exit(1); + } +} From f88b9bb84323592aa5bd5498163af324d94885d9 Mon Sep 17 00:00:00 2001 From: Peter Krenesky Date: Wed, 14 Oct 2026 12:51:53 -0700 Subject: [PATCH 2/3] Gate accounting helper behind test-only feature --- Cargo.toml | 6 -- tests/std_windows/job_object_accounting.rs | 76 +++++++++++++++---- tests/support/job_object_accounting_helper.rs | 44 ----------- 3 files changed, 61 insertions(+), 65 deletions(-) delete mode 100644 tests/support/job_object_accounting_helper.rs diff --git a/Cargo.toml b/Cargo.toml index 986e2af..2b45b84 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,12 +40,6 @@ remoteprocess = "0.5.0" tempfile = { version = "3.20.0", default-features = false } tokio = { version = "1.38.2", features = ["io-util", "macros", "process", "rt", "rt-multi-thread", "time"] } -[[bin]] -name = "job-object-accounting-helper" -path = "tests/support/job_object_accounting_helper.rs" -test = false -bench = false - [features] default = ["creation-flags", "job-object", "kill-on-drop", "process-group", "process-session", "tracing"] diff --git a/tests/std_windows/job_object_accounting.rs b/tests/std_windows/job_object_accounting.rs index a39f49f..c9377c6 100644 --- a/tests/std_windows/job_object_accounting.rs +++ b/tests/std_windows/job_object_accounting.rs @@ -1,7 +1,9 @@ use std::{ - io::{BufRead, BufReader, ErrorKind}, + hint::black_box, + io::{BufRead, BufReader, ErrorKind, Write}, process::{Command, Stdio}, - time::Duration, + thread, + time::{Duration, Instant}, }; use super::prelude::*; @@ -15,13 +17,19 @@ impl Drop for ChildGuard { } } +fn helper_test_filter() -> String { + format!("{}::helper_entrypoint", module_path!()) +} + fn wrapped_helper(mode: &str) -> Result { - let mut command = CommandWrap::with_new( - env!("CARGO_BIN_EXE_job-object-accounting-helper"), - |command| { - command.arg(mode).stdout(Stdio::piped()); - }, - ); + let executable = std::env::current_exe()?; + let filter = helper_test_filter(); + let mut command = CommandWrap::with_new(executable, |command| { + command + .args(["--exact", &filter]) + .env("PROCESS_WRAP_ACCOUNTING_HELPER", mode) + .stdout(Stdio::piped()); + }); command.wrap(JobObject); Ok(ChildGuard(command.spawn()?)) } @@ -31,14 +39,18 @@ fn read_ready(child: &mut dyn ChildWrapper) -> Result<()> { .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(); - BufReader::new(stdout).read_line(&mut line)?; - if line.trim() == "child-exited" || line.trim() == "ready" { - Ok(()) - } else { - Err(std::io::Error::other( - "helper returned an unexpected readiness line", - )) + 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(()); + } } } @@ -77,3 +89,37 @@ fn child_without_job_object_reports_accounting_unavailable() -> Result<()> { 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]) + .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()), + } +} diff --git a/tests/support/job_object_accounting_helper.rs b/tests/support/job_object_accounting_helper.rs deleted file mode 100644 index 9bed97a..0000000 --- a/tests/support/job_object_accounting_helper.rs +++ /dev/null @@ -1,44 +0,0 @@ -use std::{ - black_box, - io::{self, Write}, - process::{Command, Stdio}, - thread, - time::{Duration, Instant}, -}; - -fn cpu_root() -> io::Result<()> { - let mut worker = Command::new(std::env::current_exe()?) - .arg("cpu-worker") - .stdout(Stdio::null()) - .spawn()?; - worker.wait()?; - writeln!(io::stdout().lock(), "child-exited")?; - loop { - thread::sleep(Duration::from_secs(60)); - } -} - -fn 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); - } -} - -fn run() -> io::Result<()> { - match std::env::args().nth(1).as_deref() { - Some("cpu-root") => cpu_root()?, - Some("cpu-worker") => cpu_worker(), - _ => return Err(io::Error::other("expected helper mode").into()), - } - Ok(()) -} - -fn main() { - if let Err(error) = run() { - eprintln!("{error}"); - std::process::exit(1); - } -} From 85083bdeee83fe0305e0c3acadbc0330788dae69 Mon Sep 17 00:00:00 2001 From: Peter Krenesky Date: Wed, 14 Oct 2026 12:56:15 -0700 Subject: [PATCH 3/3] Disable capture for accounting helper subprocesses --- tests/std_windows/job_object_accounting.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/std_windows/job_object_accounting.rs b/tests/std_windows/job_object_accounting.rs index c9377c6..b9408e5 100644 --- a/tests/std_windows/job_object_accounting.rs +++ b/tests/std_windows/job_object_accounting.rs @@ -26,7 +26,7 @@ fn wrapped_helper(mode: &str) -> Result { let filter = helper_test_filter(); let mut command = CommandWrap::with_new(executable, |command| { command - .args(["--exact", &filter]) + .args(["--exact", &filter, "--nocapture"]) .env("PROCESS_WRAP_ACCOUNTING_HELPER", mode) .stdout(Stdio::piped()); }); @@ -101,7 +101,7 @@ fn helper_entrypoint() -> Result<()> { let executable = std::env::current_exe()?; let filter = helper_test_filter(); let mut worker = Command::new(executable) - .args(["--exact", &filter]) + .args(["--exact", &filter, "--nocapture"]) .env("PROCESS_WRAP_ACCOUNTING_HELPER", "cpu-worker") .stdout(Stdio::null()) .spawn()?;