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..b9408e5 --- /dev/null +++ b/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); + +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 { + 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 = ::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(()) +} + +#[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()), + } +} 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;