use std::io;
use std::time::Duration;
use tokio::process::{Child, Command};
#[cfg(any(feature = "stats", feature = "process-control"))]
use windows_sys::Win32::Foundation::FILETIME;
use windows_sys::Win32::Foundation::{CloseHandle, HANDLE, HWND, INVALID_HANDLE_VALUE, LPARAM};
#[cfg(feature = "process-control")]
use windows_sys::Win32::Foundation::{ERROR_INVALID_PARAMETER, ERROR_MORE_DATA};
use windows_sys::Win32::System::Console::{CTRL_BREAK_EVENT, GenerateConsoleCtrlEvent};
use windows_sys::Win32::System::Diagnostics::ToolHelp::{
CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next,
};
#[cfg(feature = "process-control")]
use windows_sys::Win32::System::Diagnostics::ToolHelp::{
PROCESSENTRY32W, Process32FirstW, Process32NextW, TH32CS_SNAPPROCESS,
};
use windows_sys::Win32::System::JobObjects::IsProcessInJob;
use windows_sys::Win32::System::JobObjects::QueryInformationJobObject;
use windows_sys::Win32::System::JobObjects::{
AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JOBOBJECTINFOCLASS, JobObjectExtendedLimitInformation,
SetInformationJobObject, TerminateJobObject,
};
#[cfg(feature = "limits")]
use windows_sys::Win32::System::JobObjects::{
JOB_OBJECT_CPU_RATE_CONTROL_ENABLE, JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP,
JOB_OBJECT_LIMIT_ACTIVE_PROCESS, JOB_OBJECT_LIMIT_JOB_MEMORY,
JOBOBJECT_CPU_RATE_CONTROL_INFORMATION, JobObjectCpuRateControlInformation,
};
use windows_sys::Win32::System::JobObjects::{
JOBOBJECT_BASIC_ACCOUNTING_INFORMATION, JobObjectBasicAccountingInformation,
};
#[cfg(feature = "stats")]
use windows_sys::Win32::System::JobObjects::{
JOBOBJECT_BASIC_AND_IO_ACCOUNTING_INFORMATION, JobObjectBasicAndIoAccountingInformation,
};
#[cfg(feature = "process-control")]
use windows_sys::Win32::System::JobObjects::{
JOBOBJECT_BASIC_PROCESS_ID_LIST, JobObjectBasicProcessIdList,
};
use windows_sys::Win32::System::LibraryLoader::{GetModuleHandleW, GetProcAddress};
#[cfg(feature = "stats")]
use windows_sys::Win32::System::ProcessStatus::{K32GetProcessMemoryInfo, PROCESS_MEMORY_COUNTERS};
#[cfg(any(feature = "stats", feature = "process-control"))]
use windows_sys::Win32::System::Threading::GetProcessTimes;
use windows_sys::Win32::System::Threading::{
CREATE_NEW_PROCESS_GROUP, CREATE_SUSPENDED, OpenThread, ResumeThread, SetProcessAffinityMask,
THREAD_SUSPEND_RESUME,
};
#[cfg(feature = "process-control")]
use windows_sys::Win32::System::Threading::{
GetExitCodeProcess, GetProcessIdOfThread, SuspendThread, THREAD_QUERY_LIMITED_INFORMATION,
};
use windows_sys::Win32::System::Threading::{OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION};
#[cfg(feature = "process-control")]
use windows_sys::Win32::System::Threading::{PROCESS_SET_QUOTA, PROCESS_TERMINATE};
use windows_sys::Win32::UI::WindowsAndMessaging::{
EnumWindows, GetWindowThreadProcessId, PostMessageW, WM_CLOSE,
};
use windows_sys::core::BOOL;
use crate::Mechanism;
#[cfg(feature = "process-control")]
use crate::Signal;
#[cfg(feature = "limits")]
use crate::limits::{CappedAxes, LimitEvidence, LimitKind, LimitVerdict, ResourceLimits};
#[cfg(feature = "process-control")]
use crate::member::MemberInfo;
#[cfg(feature = "stats")]
use crate::stats::ProcessGroupStats;
#[cfg(feature = "stats")]
use crate::sys::{ProcIdentity, ProcMetrics};
pub(crate) struct Job {
handle: HANDLE,
suspend_lock: std::sync::Mutex<()>,
skip_drop_kill: super::SkipDropKill,
ctrl_break_leaders: std::sync::Mutex<Vec<u32>>,
}
unsafe impl Send for Job {}
unsafe impl Sync for Job {}
const EXTENDED_LIMIT_AXIS: &str = "extended-limit";
#[cfg(feature = "limits")]
const CPU_RATE_AXIS: &str = "cpu-rate";
fn set_information_job_object<T>(
handle: HANDLE,
class: JOBOBJECTINFOCLASS,
info: &T,
axis: &'static str,
) -> io::Result<()> {
#[cfg(test)]
if let Some(injected) = crate::sys::fault_injection::check(
crate::sys::fault_injection::Site::JobObjectSetInformation,
axis,
) {
return Err(injected);
}
#[cfg(not(test))]
let _ = axis;
let ok = unsafe {
SetInformationJobObject(
handle,
class,
std::ptr::from_ref(info).cast(),
std::mem::size_of::<T>() as u32,
)
};
if ok == 0 {
Err(io::Error::last_os_error())
} else {
Ok(())
}
}
impl Job {
pub(crate) fn new(#[cfg(feature = "limits")] limits: &ResourceLimits) -> io::Result<Self> {
let handle = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) };
if handle.is_null() {
return Err(io::Error::last_os_error());
}
let job = Job {
handle,
suspend_lock: std::sync::Mutex::new(()),
skip_drop_kill: super::SkipDropKill::new(),
ctrl_break_leaders: std::sync::Mutex::new(Vec::new()),
};
let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
#[cfg(feature = "limits")]
{
if let Some(bytes) = limits.max_memory {
info.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_JOB_MEMORY;
info.JobMemoryLimit = usize::try_from(bytes).unwrap_or(usize::MAX);
}
if let Some(n) = limits.max_processes {
info.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_ACTIVE_PROCESS;
info.BasicLimitInformation.ActiveProcessLimit = n;
}
}
set_information_job_object(
job.handle,
JobObjectExtendedLimitInformation,
&info,
EXTENDED_LIMIT_AXIS,
)?;
#[cfg(feature = "limits")]
if let Some(cores) = limits.cpu_quota {
let cpus = std::thread::available_parallelism().map_or(1.0, |n| n.get() as f64);
let rate = cpu_hard_cap_rate(cores, cpus);
let mut cpu: JOBOBJECT_CPU_RATE_CONTROL_INFORMATION = unsafe { std::mem::zeroed() };
cpu.ControlFlags =
JOB_OBJECT_CPU_RATE_CONTROL_ENABLE | JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP;
cpu.Anonymous.CpuRate = rate;
set_information_job_object(
job.handle,
JobObjectCpuRateControlInformation,
&cpu,
CPU_RATE_AXIS,
)?;
}
Ok(job)
}
#[cfg(feature = "limits")]
pub(crate) fn update_limits(&self, limits: &ResourceLimits) -> io::Result<()> {
let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
if let Some(bytes) = limits.max_memory {
info.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_JOB_MEMORY;
info.JobMemoryLimit = usize::try_from(bytes).unwrap_or(usize::MAX);
}
if let Some(n) = limits.max_processes {
info.BasicLimitInformation.LimitFlags |= JOB_OBJECT_LIMIT_ACTIVE_PROCESS;
info.BasicLimitInformation.ActiveProcessLimit = n;
}
if let Err(err) = set_information_job_object(
self.handle,
JobObjectExtendedLimitInformation,
&info,
EXTENDED_LIMIT_AXIS,
) {
return Err(io::Error::new(
err.kind(),
format!("{EXTENDED_LIMIT_AXIS} reissue: {err}"),
));
}
let mut cpu: JOBOBJECT_CPU_RATE_CONTROL_INFORMATION = unsafe { std::mem::zeroed() };
if let Some(cores) = limits.cpu_quota {
let cpus = std::thread::available_parallelism().map_or(1.0, |n| n.get() as f64);
cpu.ControlFlags =
JOB_OBJECT_CPU_RATE_CONTROL_ENABLE | JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP;
cpu.Anonymous.CpuRate = cpu_hard_cap_rate(cores, cpus);
}
if let Err(err) = set_information_job_object(
self.handle,
JobObjectCpuRateControlInformation,
&cpu,
CPU_RATE_AXIS,
) {
let benign_clear = limits.cpu_quota.is_none()
&& err.raw_os_error()
== Some(windows_sys::Win32::Foundation::ERROR_INVALID_PARAMETER as i32);
if !benign_clear {
return Err(io::Error::new(
err.kind(),
format!("{CPU_RATE_AXIS} reissue: {err}"),
));
}
}
Ok(())
}
pub(crate) fn spawn(
&self,
cmd: &mut Command,
opts: &crate::sys::SpawnOptions,
) -> io::Result<Child> {
use std::os::windows::process::CommandExt;
let mut flags = CREATE_SUSPENDED | opts.creation_flags;
if opts.windows_new_process_group {
flags |= CREATE_NEW_PROCESS_GROUP;
}
cmd.as_std_mut().creation_flags(flags);
let child = {
let _spawn_guard = crate::sys::process_spawn_lock();
cmd.spawn()?
};
let guard = UncontainedChildGuard::arm(child);
let pid = guard.child().id().ok_or_else(|| {
io::Error::other("child exited before it could be assigned to the job")
})?;
let handle = guard.child().raw_handle().ok_or_else(|| {
io::Error::other("child exited before it could be assigned to the job")
})?;
let _suspend_guard = self
.suspend_lock
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let ok = unsafe { AssignProcessToJobObject(self.handle, handle as HANDLE) };
if ok == 0 {
return Err(io::Error::last_os_error());
}
if let Some(mask) = opts.cpu_affinity
&& unsafe { SetProcessAffinityMask(handle as HANDLE, mask) } == 0
{
return Err(io::Error::last_os_error());
}
resume_process_threads(pid, handle as HANDLE)?;
self.skip_drop_kill.clear();
if opts.windows_new_process_group {
self.record_ctrl_break_leader(pid);
}
Ok(guard.disarm())
}
#[cfg(feature = "pty")]
pub(crate) fn contain_pty_child(
&self,
process: HANDLE,
primary_thread: HANDLE,
pid: u32,
opts: &crate::sys::SpawnOptions,
) -> io::Result<()> {
let _suspend_guard = self
.suspend_lock
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if unsafe { AssignProcessToJobObject(self.handle, process) } == 0 {
return Err(io::Error::last_os_error());
}
if let Some(mask) = opts.cpu_affinity
&& unsafe { SetProcessAffinityMask(process, mask) } == 0
{
return Err(io::Error::last_os_error());
}
resume_thread_handle(primary_thread)?;
self.skip_drop_kill.clear();
if opts.windows_new_process_group {
self.record_ctrl_break_leader(pid);
}
Ok(())
}
fn record_ctrl_break_leader(&self, pid: u32) {
let mut leaders = self
.ctrl_break_leaders
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
leaders.retain(|&leader| process_is_in_job(leader, self.handle));
if !leaders.contains(&pid) {
leaders.push(pid);
}
}
#[cfg(feature = "pty")]
pub(crate) fn spawn_pty(
&self,
cmd: &mut Command,
opts: &crate::sys::SpawnOptions,
env: Option<Vec<(std::ffi::OsString, std::ffi::OsString)>>,
) -> io::Result<crate::sys::pty::PtySpawn> {
crate::sys::pty::spawn_pty(cmd, opts, env, self)
}
#[cfg(feature = "process-control")]
pub(crate) fn adopt(&self, child: &Child) -> io::Result<()> {
let handle = child
.raw_handle()
.ok_or_else(|| io::Error::other("child has no handle (already exited?)"))?;
let ok = unsafe { AssignProcessToJobObject(self.handle, handle as HANDLE) };
if ok == 0 {
let err = io::Error::last_os_error();
if process_has_exited(handle as HANDLE) {
return Ok(());
}
return Err(err);
}
self.skip_drop_kill.clear();
Ok(())
}
#[cfg(feature = "process-control")]
pub(crate) fn adopt_external(&self, pid: u32) -> io::Result<()> {
let handle = unsafe {
OpenProcess(
PROCESS_SET_QUOTA | PROCESS_TERMINATE | PROCESS_QUERY_LIMITED_INFORMATION,
0,
pid,
)
};
if handle.is_null() {
let err = io::Error::last_os_error();
if err.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) {
return Err(io::Error::new(
io::ErrorKind::NotFound,
format!("no process with pid {pid} to adopt"),
));
}
return Err(err);
}
let outcome = self.assign_opened_process(handle);
unsafe { CloseHandle(handle) };
outcome
}
#[cfg(feature = "process-control")]
fn assign_opened_process(&self, handle: HANDLE) -> io::Result<()> {
if unsafe { AssignProcessToJobObject(self.handle, handle) } == 0 {
let err = io::Error::last_os_error();
if process_has_exited(handle) {
return Ok(());
}
return Err(err);
}
self.skip_drop_kill.clear();
Ok(())
}
pub(crate) fn kill_all(&self) -> io::Result<()> {
let ok = unsafe { TerminateJobObject(self.handle, 1) };
if ok == 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
#[cfg(feature = "process-control")]
pub(crate) fn signal(&self, sig: Signal) -> io::Result<()> {
match sig {
Signal::Kill => self.kill_all(),
Signal::Int | Signal::Term => {
let leaders = self
.ctrl_break_leaders
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone();
let signalled = ctrl_break_live_leaders(&leaders, self.handle);
let closed = close_member_windows(self.handle);
if signalled == 0 && closed == 0 {
return Err(io::Error::new(
io::ErrorKind::Unsupported,
format!(
"signal({sig:?}): no console-CTRL or windowed member to soft-close"
),
));
}
Ok(())
}
other => Err(io::Error::new(
io::ErrorKind::Unsupported,
format!("signal({other:?})"),
)),
}
}
#[cfg(feature = "process-control")]
pub(crate) fn soft_stop_scope(&self) -> crate::SoftStopScope {
let leaders = self
.ctrl_break_leaders
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone();
let has_live_leader = has_live_ctrl_break_leader(&leaders, self.handle);
if has_live_leader || job_has_windowed_member(self.handle) {
crate::SoftStopScope::OptInMembers
} else {
crate::SoftStopScope::Unsupported
}
}
#[cfg(feature = "process-control")]
pub(crate) fn suspend(&self) -> io::Result<()> {
self.for_each_member_thread(true)
}
#[cfg(feature = "process-control")]
pub(crate) fn resume(&self) -> io::Result<()> {
self.for_each_member_thread(false)
}
#[cfg(feature = "process-control")]
pub(crate) fn members(&self) -> io::Result<Vec<u32>> {
job_member_pids(self.handle)
}
#[cfg(feature = "process-control")]
pub(crate) fn members_info(&self) -> io::Result<Vec<MemberInfo>> {
let pids = job_member_pids(self.handle)?;
let meta = snapshot_process_metadata()?;
let mut out = Vec::with_capacity(pids.len());
for pid in pids {
let Some((ppid, exe)) = meta.get(&pid) else {
continue;
};
let start_time = process_start_time(pid);
out.push(MemberInfo::new(
pid,
Some(*ppid),
Some(exe.clone()),
start_time,
));
}
Ok(out)
}
#[cfg(feature = "process-control")]
fn for_each_member_thread(&self, suspend: bool) -> io::Result<()> {
let _guard = self
.suspend_lock
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let members: std::collections::HashSet<u32> =
job_member_pids(self.handle)?.into_iter().collect();
if members.is_empty() {
return Ok(());
}
let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) };
if snapshot == INVALID_HANDLE_VALUE {
return Err(io::Error::last_os_error());
}
let mut entry: THREADENTRY32 = unsafe { std::mem::zeroed() };
entry.dwSize = std::mem::size_of::<THREADENTRY32>() as u32;
let mut last_err = None;
let mut ok = unsafe { Thread32First(snapshot, &mut entry) };
while ok != 0 {
if members.contains(&entry.th32OwnerProcessID)
&& let Err(err) = suspend_or_resume_thread(
entry.th32ThreadID,
entry.th32OwnerProcessID,
self.handle,
suspend,
)
{
last_err = Some(err);
}
ok = unsafe { Thread32Next(snapshot, &mut entry) };
}
unsafe { CloseHandle(snapshot) };
match last_err {
Some(err) => Err(err),
None => Ok(()),
}
}
pub(crate) async fn graceful_shutdown(
&self,
signal: i32,
timeout: Duration,
escalate: bool,
) -> io::Result<super::graceful::GracefulOutcome> {
let leaders = self
.ctrl_break_leaders
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone();
let has_live_leader = has_live_ctrl_break_leader(&leaders, self.handle);
if has_live_leader || job_has_windowed_member(self.handle) {
let target = SoftShutdownTarget { job: self, leaders };
return super::graceful::run(&target, &self.skip_drop_kill, signal, timeout, escalate)
.await;
}
let started = tokio::time::Instant::now();
let members_before = job_active_count(self.handle);
let epoch = self.skip_drop_kill.begin_shutdown();
let already_empty = members_before == Some(0);
let (escalated, result) = if escalate {
(!already_empty, self.kill_all())
} else {
self.skip_drop_kill.request(epoch);
(false, Ok(()))
};
let members_after = job_active_count(self.handle);
result.map(|()| super::graceful::GracefulOutcome {
soft: super::graceful::SoftDelivery::Unsupported,
members_before,
members_after,
drained: already_empty,
escalated,
elapsed: started.elapsed(),
})
}
#[cfg(feature = "limits")]
pub(crate) fn limit_evidence(&self, capped: CappedAxes) -> LimitEvidence {
let verdict = |kind: LimitKind| {
if capped.has(kind) {
LimitVerdict::Unknown
} else {
LimitVerdict::NotTripped
}
};
LimitEvidence::new(
verdict(LimitKind::Memory),
verdict(LimitKind::Processes),
verdict(LimitKind::Cpu),
)
}
#[cfg(feature = "stats")]
pub(crate) fn stats(&self) -> io::Result<ProcessGroupStats> {
let mut acct: JOBOBJECT_BASIC_AND_IO_ACCOUNTING_INFORMATION = unsafe { std::mem::zeroed() };
let ok = unsafe {
QueryInformationJobObject(
self.handle,
JobObjectBasicAndIoAccountingInformation,
std::ptr::from_mut(&mut acct).cast(),
std::mem::size_of::<JOBOBJECT_BASIC_AND_IO_ACCOUNTING_INFORMATION>() as u32,
std::ptr::null_mut(),
)
};
if ok == 0 {
return Err(io::Error::last_os_error());
}
let io_counters = acct.IoInfo;
let acct = acct.BasicInfo;
let mut ext: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
let ok = unsafe {
QueryInformationJobObject(
self.handle,
JobObjectExtendedLimitInformation,
std::ptr::from_mut(&mut ext).cast(),
std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
std::ptr::null_mut(),
)
};
if ok == 0 {
return Err(io::Error::last_os_error());
}
let cpu_100ns = (acct.TotalUserTime as u64).saturating_add(acct.TotalKernelTime as u64);
Ok(ProcessGroupStats {
active_process_count: acct.ActiveProcesses as usize,
total_cpu_time: Some(Duration::from_nanos(cpu_100ns.saturating_mul(100))),
peak_memory_bytes: Some(ext.PeakJobMemoryUsed as u64),
io_read_bytes: Some(io_counters.ReadTransferCount),
io_write_bytes: Some(io_counters.WriteTransferCount),
peak_process_count: None,
})
}
pub(crate) fn mechanism(&self) -> Mechanism {
Mechanism::JobObject
}
}
pub(crate) fn detect_mechanism() -> Mechanism {
Mechanism::JobObject
}
struct SoftShutdownTarget<'a> {
job: &'a Job,
leaders: Vec<u32>,
}
impl super::graceful::GracefulTarget for SoftShutdownTarget<'_> {
fn signal_all(&self, _signal: i32) -> super::graceful::SoftDelivery {
let ctrl = ctrl_break_live_leaders(&self.leaders, self.job.handle);
let windows = close_member_windows(self.job.handle);
if ctrl + windows > 0 {
super::graceful::SoftDelivery::Sent
} else {
super::graceful::SoftDelivery::Failed
}
}
fn is_drained(&self) -> bool {
job_is_drained(self.job.handle)
}
fn alive_count(&self) -> Option<usize> {
job_active_count(self.job.handle)
}
fn hard_kill(&self) -> io::Result<()> {
self.job.kill_all()
}
}
fn ctrl_break_live_leaders(leaders: &[u32], job: HANDLE) -> usize {
let mut signalled = 0;
for &pid in leaders {
if !ctrl_break_leader_is_live(pid, job) {
continue;
}
unsafe {
GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pid);
}
signalled += 1;
}
signalled
}
fn ctrl_break_leader_is_live(pid: u32, job: HANDLE) -> bool {
pid != 0 && process_is_in_job(pid, job)
}
fn has_live_ctrl_break_leader(leaders: &[u32], job: HANDLE) -> bool {
leaders
.iter()
.any(|&pid| ctrl_break_leader_is_live(pid, job))
}
struct MemberWindowScan {
job: HANDLE,
close: bool,
matched: usize,
}
unsafe extern "system" fn scan_member_window(hwnd: HWND, lparam: LPARAM) -> BOOL {
let ctx = unsafe { &mut *(lparam as *mut MemberWindowScan) };
let mut pid: u32 = 0;
unsafe { GetWindowThreadProcessId(hwnd, &mut pid) };
if pid != 0 && process_is_in_job(pid, ctx.job) {
ctx.matched += 1;
if ctx.close {
unsafe {
PostMessageW(hwnd, WM_CLOSE, 0, 0);
}
}
}
1
}
fn scan_member_windows(job: HANDLE, close: bool) -> usize {
let mut ctx = MemberWindowScan {
job,
close,
matched: 0,
};
unsafe {
EnumWindows(
Some(scan_member_window),
std::ptr::from_mut(&mut ctx) as LPARAM,
);
}
ctx.matched
}
fn close_member_windows(job: HANDLE) -> usize {
scan_member_windows(job, true)
}
fn job_has_windowed_member(job: HANDLE) -> bool {
scan_member_windows(job, false) > 0
}
fn job_active_count(handle: HANDLE) -> Option<usize> {
let mut acct: JOBOBJECT_BASIC_ACCOUNTING_INFORMATION = unsafe { std::mem::zeroed() };
let ok = unsafe {
QueryInformationJobObject(
handle,
JobObjectBasicAccountingInformation,
std::ptr::from_mut(&mut acct).cast(),
std::mem::size_of::<JOBOBJECT_BASIC_ACCOUNTING_INFORMATION>() as u32,
std::ptr::null_mut(),
)
};
(ok != 0).then_some(acct.ActiveProcesses as usize)
}
fn job_is_drained(handle: HANDLE) -> bool {
job_active_count(handle) == Some(0)
}
#[cfg(feature = "process-control")]
fn process_has_exited(handle: HANDLE) -> bool {
const STILL_ACTIVE: u32 = 259;
let mut code: u32 = 0;
let ok = unsafe { GetExitCodeProcess(handle, &mut code) };
ok != 0 && code != STILL_ACTIVE
}
struct UncontainedChildGuard {
child: Option<Child>,
}
impl UncontainedChildGuard {
fn arm(child: Child) -> Self {
Self { child: Some(child) }
}
fn child(&self) -> &Child {
self.child
.as_ref()
.expect("the guarded child is present until disarm")
}
fn disarm(mut self) -> Child {
self.child
.take()
.expect("the guarded child is taken exactly once")
}
}
impl Drop for UncontainedChildGuard {
fn drop(&mut self) {
if let Some(mut child) = self.child.take() {
let _ = child.start_kill();
}
}
}
fn resume_process_threads(pid: u32, process: HANDLE) -> io::Result<()> {
if resume_process_threads_direct(process) {
return Ok(());
}
resume_process_threads_via_snapshot(pid)
}
type NtGetNextThread = unsafe extern "system" fn(
process: HANDLE,
thread: HANDLE,
desired_access: u32,
handle_attributes: u32,
flags: u32,
new_thread: *mut HANDLE,
) -> i32;
const NTDLL_UTF16: &[u16] = &[
b'n' as u16,
b't' as u16,
b'd' as u16,
b'l' as u16,
b'l' as u16,
b'.' as u16,
b'd' as u16,
b'l' as u16,
b'l' as u16,
0,
];
fn nt_get_next_thread() -> Option<NtGetNextThread> {
static ENTRY: std::sync::OnceLock<Option<NtGetNextThread>> = std::sync::OnceLock::new();
*ENTRY.get_or_init(|| {
let ntdll = unsafe { GetModuleHandleW(NTDLL_UTF16.as_ptr()) };
if ntdll.is_null() {
return None;
}
let symbol =
unsafe { GetProcAddress(ntdll, c"NtGetNextThread".to_bytes_with_nul().as_ptr()) };
symbol.map(|symbol| {
unsafe {
std::mem::transmute::<unsafe extern "system" fn() -> isize, NtGetNextThread>(symbol)
}
})
})
}
fn resume_process_threads_direct(process: HANDLE) -> bool {
let Some(next_thread) = nt_get_next_thread() else {
return false;
};
let mut resumed = 0u32;
let mut cursor: HANDLE = std::ptr::null_mut();
loop {
let mut thread: HANDLE = std::ptr::null_mut();
let status =
unsafe { next_thread(process, cursor, THREAD_SUSPEND_RESUME, 0, 0, &mut thread) };
if !cursor.is_null() {
unsafe { CloseHandle(cursor) };
}
if status < 0 || thread.is_null() {
break;
}
if resume_thread_handle(thread).is_ok() {
resumed += 1;
}
cursor = thread;
}
resumed > 0
}
fn resume_process_threads_via_snapshot(pid: u32) -> io::Result<()> {
let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) };
if snapshot == INVALID_HANDLE_VALUE {
return Err(io::Error::last_os_error());
}
let mut entry: THREADENTRY32 = unsafe { std::mem::zeroed() };
entry.dwSize = std::mem::size_of::<THREADENTRY32>() as u32;
let mut resumed = 0u32;
let mut last_err = None;
let mut ok = unsafe { Thread32First(snapshot, &mut entry) };
while ok != 0 {
if entry.th32OwnerProcessID == pid {
match resume_thread(entry.th32ThreadID) {
Ok(()) => resumed += 1,
Err(err) => last_err = Some(err),
}
}
ok = unsafe { Thread32Next(snapshot, &mut entry) };
}
unsafe { CloseHandle(snapshot) };
if resumed == 0 {
return Err(last_err
.unwrap_or_else(|| io::Error::other("no thread found to resume the contained child")));
}
Ok(())
}
fn resume_thread(tid: u32) -> io::Result<()> {
let thread = unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, tid) };
if thread.is_null() {
return Err(io::Error::last_os_error());
}
let result = resume_thread_handle(thread);
unsafe { CloseHandle(thread) };
result
}
fn resume_thread_handle(thread: HANDLE) -> io::Result<()> {
loop {
let previous = unsafe { ResumeThread(thread) };
if previous == u32::MAX {
return Err(io::Error::last_os_error());
}
if previous <= 1 {
return Ok(()); }
}
}
#[cfg(feature = "process-control")]
fn suspend_or_resume_thread(
tid: u32,
expected_pid: u32,
job: HANDLE,
suspend: bool,
) -> io::Result<()> {
let thread = unsafe {
OpenThread(
THREAD_SUSPEND_RESUME | THREAD_QUERY_LIMITED_INFORMATION,
0,
tid,
)
};
if thread.is_null() {
let err = io::Error::last_os_error();
if err.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) {
return Ok(());
}
return Err(err);
}
let owner = unsafe { GetProcessIdOfThread(thread) };
if owner != expected_pid {
unsafe { CloseHandle(thread) };
return Ok(());
}
if !process_is_in_job(owner, job) {
unsafe { CloseHandle(thread) };
return Ok(());
}
let prev = unsafe {
if suspend {
SuspendThread(thread)
} else {
ResumeThread(thread)
}
};
let err = (prev == u32::MAX).then(io::Error::last_os_error);
unsafe { CloseHandle(thread) };
match err {
Some(err) => Err(err),
None => Ok(()),
}
}
fn process_is_in_job(pid: u32, job: HANDLE) -> bool {
let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
if handle.is_null() {
return false;
}
let mut in_job: i32 = 0;
let ok = unsafe { IsProcessInJob(handle, job, &mut in_job) };
unsafe { CloseHandle(handle) };
ok != 0 && in_job != 0
}
#[cfg(feature = "process-control")]
fn job_member_pids(handle: HANDLE) -> io::Result<Vec<u32>> {
let mut cap: usize = 64;
loop {
let bytes = std::mem::size_of::<JOBOBJECT_BASIC_PROCESS_ID_LIST>()
+ cap.saturating_sub(1) * std::mem::size_of::<usize>();
let mut buf = vec![0u64; bytes.div_ceil(std::mem::size_of::<u64>())];
let ok = unsafe {
QueryInformationJobObject(
handle,
JobObjectBasicProcessIdList,
buf.as_mut_ptr().cast(),
bytes as u32,
std::ptr::null_mut(),
)
};
let list = buf.as_ptr().cast::<JOBOBJECT_BASIC_PROCESS_ID_LIST>();
if ok == 0 {
let err = io::Error::last_os_error();
if err.raw_os_error() == Some(ERROR_MORE_DATA as i32) {
let assigned = unsafe { (*list).NumberOfAssignedProcesses } as usize;
cap = assigned.max(cap).saturating_mul(2);
continue;
}
return Err(err);
}
let n = unsafe { (*list).NumberOfProcessIdsInList } as usize;
let ids =
unsafe { std::slice::from_raw_parts(std::ptr::addr_of!((*list).ProcessIdList[0]), n) };
return Ok(ids.iter().map(|&pid| pid as u32).collect());
}
}
#[cfg(feature = "process-control")]
fn snapshot_process_metadata() -> io::Result<std::collections::HashMap<u32, (u32, String)>> {
let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) };
if snapshot == INVALID_HANDLE_VALUE {
return Err(io::Error::last_os_error());
}
let mut entry: PROCESSENTRY32W = unsafe { std::mem::zeroed() };
entry.dwSize = std::mem::size_of::<PROCESSENTRY32W>() as u32;
let mut map = std::collections::HashMap::new();
let mut ok = unsafe { Process32FirstW(snapshot, &mut entry) };
while ok != 0 {
let len = entry
.szExeFile
.iter()
.position(|&c| c == 0)
.unwrap_or(entry.szExeFile.len());
let exe = String::from_utf16_lossy(&entry.szExeFile[..len]);
map.insert(entry.th32ProcessID, (entry.th32ParentProcessID, exe));
ok = unsafe { Process32NextW(snapshot, &mut entry) };
}
unsafe { CloseHandle(snapshot) };
Ok(map)
}
#[cfg(feature = "process-control")]
fn process_start_time(pid: u32) -> Option<u64> {
let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
if handle.is_null() {
return None;
}
let units = creation_time_units(handle);
unsafe { CloseHandle(handle) };
units
}
#[cfg(feature = "process-control")]
fn creation_time_units(handle: HANDLE) -> Option<u64> {
let mut creation = FILETIME {
dwLowDateTime: 0,
dwHighDateTime: 0,
};
let mut exit = creation;
let mut kernel = creation;
let mut user = creation;
let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };
(ok != 0).then(|| filetime_units(creation))
}
#[cfg(feature = "process-control")]
pub(crate) fn process_info(pid: u32) -> io::Result<Option<MemberInfo>> {
let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
if handle.is_null() {
let err = io::Error::last_os_error();
if err.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) {
return Ok(None);
}
return Err(err);
}
let start_time = creation_time_units(handle);
unsafe { CloseHandle(handle) };
let (ppid, exe) = match snapshot_process_metadata()?.get(&pid) {
Some((ppid, exe)) => (Some(*ppid), Some(exe.clone())),
None => (None, None),
};
Ok(Some(MemberInfo::new(pid, ppid, exe, start_time)))
}
#[cfg(any(feature = "stats", feature = "process-control"))]
fn filetime_units(ft: FILETIME) -> u64 {
((ft.dwHighDateTime as u64) << 32) | ft.dwLowDateTime as u64
}
#[cfg(feature = "stats")]
fn filetime_nanos(ft: FILETIME) -> u64 {
filetime_units(ft).saturating_mul(100)
}
#[cfg(feature = "limits")]
fn cpu_hard_cap_rate(cores: f64, cpus: f64) -> u32 {
let rate = ((cores / cpus) * 10_000.0).round();
rate.clamp(1.0, 10_000.0) as u32
}
#[cfg(feature = "stats")]
pub(crate) fn process_identity(pid: u32) -> Option<ProcIdentity> {
let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
if handle.is_null() {
return None;
}
let mut creation = FILETIME {
dwLowDateTime: 0,
dwHighDateTime: 0,
};
let mut exit = creation;
let mut kernel = creation;
let mut user = creation;
let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };
unsafe { CloseHandle(handle) };
(ok != 0).then(|| ProcIdentity::from_raw(filetime_units(creation)))
}
#[cfg(feature = "stats")]
pub(crate) fn process_metrics(pid: u32, expected: Option<ProcIdentity>) -> ProcMetrics {
let mut metrics = ProcMetrics::default();
let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
if handle.is_null() {
return metrics;
}
let mut creation = FILETIME {
dwLowDateTime: 0,
dwHighDateTime: 0,
};
let mut exit = creation;
let mut kernel = creation;
let mut user = creation;
let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };
if let Some(expected) = expected {
let confirmed = ok != 0 && filetime_units(creation) == expected.raw();
if !confirmed {
unsafe { CloseHandle(handle) };
return ProcMetrics::default();
}
}
if ok != 0 {
metrics.cpu_time = Some(Duration::from_nanos(
filetime_nanos(kernel).saturating_add(filetime_nanos(user)),
));
}
let mut counters: PROCESS_MEMORY_COUNTERS = unsafe { std::mem::zeroed() };
counters.cb = std::mem::size_of::<PROCESS_MEMORY_COUNTERS>() as u32;
let ok = unsafe { K32GetProcessMemoryInfo(handle, &mut counters, counters.cb) };
if ok != 0 {
metrics.peak_memory_bytes = Some(counters.PeakWorkingSetSize as u64);
}
unsafe { CloseHandle(handle) };
metrics
}
impl Drop for Job {
fn drop(&mut self) {
if self.skip_drop_kill.is_set() {
let info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
let _ = set_information_job_object(
self.handle,
JobObjectExtendedLimitInformation,
&info,
EXTENDED_LIMIT_AXIS,
);
#[cfg(feature = "limits")]
{
let cpu: JOBOBJECT_CPU_RATE_CONTROL_INFORMATION = unsafe { std::mem::zeroed() };
let _ = set_information_job_object(
self.handle,
JobObjectCpuRateControlInformation,
&cpu,
CPU_RATE_AXIS,
);
}
}
unsafe { CloseHandle(self.handle) };
}
}
#[cfg(all(test, feature = "process-control"))]
mod thread_tests {
use super::{process_is_in_job, suspend_or_resume_thread};
use windows_sys::Win32::Foundation::CloseHandle;
use windows_sys::Win32::System::JobObjects::CreateJobObjectW;
use windows_sys::Win32::System::Threading::GetCurrentProcessId;
#[test]
fn suspend_or_resume_a_stale_tid_is_ok() {
let job = std::ptr::null_mut();
assert!(suspend_or_resume_thread(1, 0, job, true).is_ok());
assert!(suspend_or_resume_thread(1, 0, job, false).is_ok());
}
#[test]
fn non_member_and_unopenable_pids_are_rejected() {
let job = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) };
assert!(!job.is_null(), "failed to create a test job object");
let me = unsafe { GetCurrentProcessId() };
assert!(
!process_is_in_job(me, job),
"a process not assigned to this job must not read as a member"
);
assert!(
!process_is_in_job(1, job),
"an unopenable pid must be treated as a non-member"
);
unsafe { CloseHandle(job) };
}
}
#[cfg(all(test, feature = "limits"))]
mod job_limit_error_paths {
use windows_sys::Win32::Foundation::{ERROR_ACCESS_DENIED, ERROR_INVALID_PARAMETER};
use super::{CPU_RATE_AXIS, EXTENDED_LIMIT_AXIS};
use crate::sys::fault_injection::{Faults, Site};
use crate::{ErrorKind, ErrorReason, LimitKind, LimitReason, ProcessGroup, ResourceLimits};
const SITE: Site = Site::JobObjectSetInformation;
#[test]
fn a_cpu_rate_rejection_after_the_extended_limit_axis_applied_is_reported() {
let mut group = ProcessGroup::new().expect("an uncapped Job Object group");
let faults = Faults::new()
.fail_from_nth(SITE, None, 2, ERROR_ACCESS_DENIED as i32)
.arm();
let err = group
.update_limits(ResourceLimits {
cpu_quota: Some(0.5),
..ResourceLimits::default()
})
.expect_err("a rejected CPU-rate write must not report the caps as applied");
assert_eq!(
faults.matched(SITE),
2,
"both axes must be attempted — the first for real, the second faulted"
);
assert_eq!(faults.fired(SITE), 1, "exactly the second axis was failed");
assert_eq!(err.kind(), ErrorKind::ResourceLimit);
match err.reason() {
ErrorReason::ResourceLimit {
kind,
reason,
detail,
} => {
assert_eq!(*kind, LimitKind::Cpu, "the only requested axis");
assert_eq!(
*reason,
LimitReason::Unenforceable,
"a Job Object exists and refused the request — not `Unsupported`"
);
assert!(
detail.contains(CPU_RATE_AXIS),
"the detail must name which axis was rejected: {detail}"
);
assert!(
!detail.contains(EXTENDED_LIMIT_AXIS),
"the axis that succeeded must not be blamed: {detail}"
);
}
other => panic!("expected a ResourceLimit failure, got {other:?}"),
}
}
#[test]
fn a_cpu_rate_clear_rejected_for_another_reason_still_fails() {
let mut group = ProcessGroup::new().expect("an uncapped Job Object group");
let faults = Faults::new()
.fail_every(SITE, Some(CPU_RATE_AXIS), ERROR_ACCESS_DENIED as i32)
.arm();
let err = group
.update_limits(ResourceLimits::default())
.expect_err("only ERROR_INVALID_PARAMETER is the benign 'nothing to disable' case");
assert_eq!(faults.fired(SITE), 1);
assert_eq!(
err.kind(),
ErrorKind::PermissionDenied,
"no cap was requested, so this is the OS refusal itself — not a \
ResourceLimit verdict"
);
match err.reason() {
ErrorReason::Io(source) => {
assert!(
source.to_string().contains(CPU_RATE_AXIS),
"the refusal must say which axis it was: {source}"
);
}
other => panic!("expected a plain Io failure, got {other:?}"),
}
}
#[test]
fn a_cpu_rate_clear_rejected_as_never_enabled_is_a_success() {
let mut group = ProcessGroup::new().expect("an uncapped Job Object group");
let faults = Faults::new()
.fail_every(SITE, Some(CPU_RATE_AXIS), ERROR_INVALID_PARAMETER as i32)
.arm();
group
.update_limits(ResourceLimits::default())
.expect("clearing a cap that was never enabled leaves the desired state in force");
assert_eq!(
faults.fired(SITE),
1,
"the clear really was rejected — the success is the crate's classification of it"
);
}
}
#[cfg(test)]
mod guard_tests {
fn pid_alive(pid: u32) -> bool {
use windows_sys::Win32::Foundation::CloseHandle;
use windows_sys::Win32::System::Threading::{
GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION,
};
const STILL_ACTIVE: u32 = 259;
let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
if handle.is_null() {
return false; }
let mut code: u32 = 0;
let ok = unsafe { GetExitCodeProcess(handle, &mut code) };
unsafe { CloseHandle(handle) };
ok != 0 && code == STILL_ACTIVE
}
fn spawn_suspended() -> tokio::process::Child {
use windows_sys::Win32::System::Threading::CREATE_SUSPENDED;
tokio::process::Command::new("cmd")
.args(["/C", "ping -n 30 127.0.0.1 > NUL"])
.creation_flags(CREATE_SUSPENDED)
.spawn()
.expect("spawn the suspended child")
}
#[tokio::test]
#[ignore = "spawns a real subprocess"]
async fn uncontained_guard_reaps_the_child_on_an_armed_drop() {
let child = spawn_suspended();
let pid = child.id().expect("the child has a pid");
assert!(
pid_alive(pid),
"the suspended child is alive right after spawn"
);
drop(super::UncontainedChildGuard::arm(child)); let mut dead = false;
for _ in 0..200 {
if !pid_alive(pid) {
dead = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(dead, "an armed guard drop must terminate the child");
}
#[tokio::test]
#[ignore = "spawns a real subprocess"]
async fn uncontained_guard_disarm_hands_back_a_live_child() {
let child = spawn_suspended();
let pid = child.id().expect("the child has a pid");
let mut kept = super::UncontainedChildGuard::arm(child).disarm();
assert!(pid_alive(pid), "disarm must leave the child running");
let _ = kept.start_kill();
let _ = kept.wait().await;
}
}
#[cfg(test)]
mod thread_resume_tests {
use std::time::Duration;
use windows_sys::Win32::Foundation::HANDLE;
use windows_sys::Win32::System::Threading::CREATE_SUSPENDED;
const RESUMED_EXIT_CODE: i32 = 7;
fn spawn_suspended_fixture() -> tokio::process::Child {
tokio::process::Command::new("cmd")
.args(["/C", "exit 7"])
.creation_flags(CREATE_SUSPENDED)
.spawn()
.expect("spawn the suspended fixture child")
}
async fn expect_resumed(child: &mut tokio::process::Child) {
let status = tokio::time::timeout(Duration::from_secs(30), child.wait())
.await
.expect("a resumed child must exit; a hang means it is still suspended")
.expect("wait for the resumed child");
assert_eq!(status.code(), Some(RESUMED_EXIT_CODE));
}
#[test]
fn resume_thread_handle_rejects_an_invalid_handle() {
let error = super::resume_thread_handle(std::ptr::null_mut())
.expect_err("ResumeThread failure must never look like a launched child");
assert_eq!(
error.raw_os_error(),
Some(windows_sys::Win32::Foundation::ERROR_INVALID_HANDLE as i32)
);
}
#[test]
fn the_direct_thread_walk_reports_failure_for_an_unusable_process_handle() {
assert!(
!super::resume_process_threads_direct(std::ptr::null_mut()),
"a refused enumeration must not be reported as a resumed child"
);
}
#[tokio::test]
#[ignore = "spawns a real subprocess"]
async fn the_direct_thread_walk_releases_a_suspended_child() {
let mut child = spawn_suspended_fixture();
let handle = child.raw_handle().expect("the child has a handle") as HANDLE;
assert!(
super::resume_process_threads_direct(handle),
"ntdll!NtGetNextThread must release the primary thread of a freshly \
suspended child; if this host lacks the export the spawn path still \
works through the snapshot fallback, but the fast path is gone"
);
expect_resumed(&mut child).await;
}
#[tokio::test]
#[ignore = "spawns a real subprocess"]
async fn the_snapshot_fallback_still_releases_a_suspended_child() {
let mut child = spawn_suspended_fixture();
let pid = child.id().expect("the child has a pid");
super::resume_process_threads_via_snapshot(pid).expect("the snapshot walk resumes");
expect_resumed(&mut child).await;
}
#[tokio::test]
#[ignore = "spawns a real subprocess"]
async fn the_combined_resume_releases_a_suspended_child() {
let mut child = spawn_suspended_fixture();
let pid = child.id().expect("the child has a pid");
let handle = child.raw_handle().expect("the child has a handle") as HANDLE;
super::resume_process_threads(pid, handle).expect("the contained child is resumed");
expect_resumed(&mut child).await;
}
}
#[cfg(all(test, feature = "limits"))]
mod tests {
use super::cpu_hard_cap_rate;
#[test]
fn cpu_rate_maps_per_core_fraction_to_total_system_percent() {
assert_eq!(cpu_hard_cap_rate(0.5, 8.0), 625);
assert_eq!(cpu_hard_cap_rate(1.0, 1.0), 10_000);
assert_eq!(cpu_hard_cap_rate(4.0, 4.0), 10_000);
assert_eq!(cpu_hard_cap_rate(8.0, 4.0), 10_000);
assert_eq!(cpu_hard_cap_rate(0.0001, 64.0), 1);
}
}
#[cfg(test)]
mod rearm_race_tests {
use std::time::Duration;
fn new_job() -> super::Job {
#[cfg(feature = "limits")]
{
super::Job::new(&crate::limits::ResourceLimits::default()).expect("create a test job")
}
#[cfg(not(feature = "limits"))]
{
super::Job::new().expect("create a test job")
}
}
#[tokio::test]
async fn non_escalating_shutdown_spares_then_a_rearm_re_arms() {
let job = new_job();
job.graceful_shutdown(0, Duration::ZERO, false)
.await
.expect("graceful shutdown");
assert!(
job.skip_drop_kill.is_set(),
"escalate=false spares survivors: Drop clears KILL_ON_JOB_CLOSE"
);
job.skip_drop_kill.clear();
assert!(
!job.skip_drop_kill.is_set(),
"a member that joined after the spare re-arms Drop's kill-on-close"
);
}
#[tokio::test(start_paused = true)]
async fn atomic_shutdown_elapsed_uses_tokio_clock() {
let job = new_job();
let before_advance = job
.graceful_shutdown(0, Duration::ZERO, true)
.await
.expect("atomic shutdown before clock advance");
tokio::time::advance(Duration::from_secs(60)).await;
let after_advance = job
.graceful_shutdown(0, Duration::ZERO, true)
.await
.expect("atomic shutdown after clock advance");
assert_eq!(before_advance.elapsed, Duration::ZERO);
assert_eq!(after_advance.elapsed, Duration::ZERO);
}
#[tokio::test]
async fn a_concurrent_rearm_wins_over_a_stale_non_escalating_request() {
let job = new_job();
job.skip_drop_kill.clear(); let epoch = job.skip_drop_kill.begin_shutdown();
job.skip_drop_kill.clear();
job.skip_drop_kill.request(epoch);
assert!(
!job.skip_drop_kill.is_set(),
"a child assigned to the job mid-shutdown must keep its kill-on-close \
backstop — the stale request must not re-spare it"
);
}
#[cfg(feature = "stats")]
#[test]
fn job_stats_report_the_io_counters_and_no_peak_process_count() {
let job = new_job();
let stats = job.stats().expect("an empty job reports its accounting");
assert_eq!(stats.active_process_count, 0);
assert_eq!(stats.io_read_bytes, Some(0));
assert_eq!(stats.io_write_bytes, Some(0));
assert_eq!(
stats.peak_process_count, None,
"a Job Object keeps no peak-concurrency counter; None is the honest answer"
);
assert_eq!(stats.total_cpu_time, Some(Duration::ZERO));
assert!(stats.peak_memory_bytes.is_some());
}
}
#[cfg(test)]
mod ctrl_break_tests {
use std::time::Duration;
use windows_sys::Win32::System::Threading::GetCurrentProcessId;
fn new_job() -> super::Job {
#[cfg(feature = "limits")]
{
super::Job::new(&crate::limits::ResourceLimits::default()).expect("create a test job")
}
#[cfg(not(feature = "limits"))]
{
super::Job::new().expect("create a test job")
}
}
#[test]
fn an_empty_job_reports_drained() {
let job = new_job();
assert!(
super::job_is_drained(job.handle),
"an empty job has zero active processes, so it is drained"
);
}
#[test]
fn an_empty_job_has_no_windowed_member_and_closes_nothing() {
let job = new_job();
assert!(
!super::job_has_windowed_member(job.handle),
"an empty job owns no top-level window"
);
assert_eq!(
super::close_member_windows(job.handle),
0,
"no member window means no WM_CLOSE is posted — never onto an unrelated \
app's window or the test runner's own"
);
}
#[cfg(feature = "process-control")]
#[test]
fn signal_int_or_term_without_a_soft_close_target_is_unsupported() {
let job = new_job();
for sig in [crate::Signal::Int, crate::Signal::Term] {
let err = job
.signal(sig)
.expect_err("no console/windowed target to reach");
assert_eq!(
err.kind(),
std::io::ErrorKind::Unsupported,
"{sig:?} on a group with no console/windowed member is Unsupported"
);
}
}
#[cfg(feature = "process-control")]
#[test]
fn signal_other_curated_variants_stay_unsupported() {
let job = new_job();
for sig in [
crate::Signal::Hup,
crate::Signal::Quit,
crate::Signal::Usr1,
crate::Signal::Usr2,
crate::Signal::Other(9),
] {
let err = job.signal(sig).expect_err("not a soft-close signal");
assert_eq!(
err.kind(),
std::io::ErrorKind::Unsupported,
"{sig:?} remains Unsupported on Windows"
);
}
}
#[cfg(feature = "process-control")]
#[test]
fn soft_stop_scope_on_an_empty_job_is_unsupported() {
let job = new_job();
assert_eq!(
job.soft_stop_scope(),
crate::SoftStopScope::Unsupported,
"an empty job can soft-stop nothing"
);
}
#[cfg(feature = "process-control")]
#[test]
fn soft_stop_scope_ignores_a_non_member_leader() {
let job = new_job();
let me = unsafe { GetCurrentProcessId() };
job.ctrl_break_leaders
.lock()
.expect("lock leaders")
.push(me);
assert_eq!(
job.soft_stop_scope(),
crate::SoftStopScope::Unsupported,
"a recorded but non-member leader is not a reachable soft-stop target"
);
assert_eq!(
job.ctrl_break_leaders.lock().expect("lock leaders").len(),
1,
"soft_stop_scope must not mutate the recorded-leader list"
);
}
#[cfg(feature = "process-control")]
fn opt_in_sleeper() -> tokio::process::Command {
let mut cmd = tokio::process::Command::new("ping");
cmd.args(["-n", "30", "127.0.0.1"]);
cmd
}
#[cfg(feature = "process-control")]
async fn wait_until_member(pid: u32, job: &super::Job) {
for _ in 0..500 {
if super::process_is_in_job(pid, job.handle) {
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
panic!("pid {pid} never became a live job member");
}
#[cfg(feature = "process-control")]
#[tokio::test]
#[ignore = "spawns a real opt-in console child and soft-signals it"]
async fn signal_term_and_int_reach_a_live_opt_in_leader() {
let job = new_job();
let opts = crate::sys::SpawnOptions {
windows_new_process_group: true,
..Default::default()
};
let mut child = job
.spawn(&mut opt_in_sleeper(), &opts)
.expect("spawn opt-in console child");
let pid = child.id().expect("child has a pid");
wait_until_member(pid, &job).await;
job.signal(crate::Signal::Term)
.expect("Term reaches a live opt-in console leader");
job.signal(crate::Signal::Int)
.expect("Int reaches a live opt-in console leader");
let _ = job.kill_all();
let _ = child.wait().await;
}
#[cfg(feature = "process-control")]
#[tokio::test]
#[ignore = "spawns a real opt-in console child and probes soft-stop availability"]
async fn soft_stop_scope_reports_opt_in_members_for_a_live_leader() {
let job = new_job();
let opts = crate::sys::SpawnOptions {
windows_new_process_group: true,
..Default::default()
};
let mut child = job
.spawn(&mut opt_in_sleeper(), &opts)
.expect("spawn opt-in console child");
let pid = child.id().expect("child has a pid");
wait_until_member(pid, &job).await;
assert_eq!(
job.soft_stop_scope(),
crate::SoftStopScope::OptInMembers,
"a live opt-in console leader makes a soft stop available"
);
job.signal(crate::Signal::Term)
.expect("Term reaches the live opt-in leader the probe reported");
let _ = job.kill_all();
let _ = child.wait().await;
}
#[tokio::test]
async fn a_recorded_non_member_leader_is_never_signalled() {
let job = new_job();
let me = unsafe { GetCurrentProcessId() };
job.ctrl_break_leaders
.lock()
.expect("lock leaders")
.push(me);
let start = std::time::Instant::now();
let outcome = job
.graceful_shutdown(crate::sys::SIGTERM_RAW, Duration::from_secs(30), true)
.await
.expect("ctrl-break graceful shutdown of an empty job");
assert_eq!(
outcome.soft,
crate::sys::graceful::SoftDelivery::Unsupported,
"a stale leader does not create a soft-shutdown tier"
);
assert!(
start.elapsed() < Duration::from_secs(5),
"a stale leader must not introduce a grace wait (took {:?})",
start.elapsed()
);
assert!(
job.ctrl_break_leaders.lock().expect("lock leaders").len() == 1,
"the recorded leader is retained across a shutdown"
);
}
#[tokio::test]
async fn a_stale_ctrl_leader_uses_atomic_non_escalating_shutdown() {
let job = new_job();
let me = unsafe { GetCurrentProcessId() };
job.ctrl_break_leaders
.lock()
.expect("lock leaders")
.push(me);
job.graceful_shutdown(crate::sys::SIGTERM_RAW, Duration::ZERO, false)
.await
.expect("non-escalating ctrl-break shutdown");
assert!(
job.skip_drop_kill.is_set(),
"a non-escalating atomic shutdown spares survivors: Drop clears KILL_ON_JOB_CLOSE"
);
job.skip_drop_kill.clear();
assert!(
!job.skip_drop_kill.is_set(),
"a reused job re-arms the backstop"
);
}
#[tokio::test]
#[ignore = "spawns real subprocesses"]
async fn stale_leaders_are_pruned_across_spawn_and_exit_cycles() {
let job = new_job();
let opts = crate::sys::SpawnOptions {
windows_new_process_group: true,
..Default::default()
};
fn short_lived_cmd() -> tokio::process::Command {
let mut cmd = tokio::process::Command::new("cmd");
cmd.args(["/C", "exit 0"]);
cmd
}
async fn wait_until_pruneable(pid: u32, job: &super::Job) {
for _ in 0..500 {
if !super::process_is_in_job(pid, job.handle) {
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
panic!("pid {pid} still reads as a job member after 5s — can't test pruning");
}
let mut child1 = job
.spawn(&mut short_lived_cmd(), &opts)
.expect("spawn first opt-in child");
let pid1 = child1.id().expect("first child has a pid");
child1.wait().await.expect("first child exits");
drop(child1);
assert_eq!(
job.ctrl_break_leaders.lock().expect("lock leaders").clone(),
vec![pid1],
"the first leader is recorded right after its spawn"
);
wait_until_pruneable(pid1, &job).await;
let mut child2 = job
.spawn(&mut short_lived_cmd(), &opts)
.expect("spawn second opt-in child");
let pid2 = child2.id().expect("second child has a pid");
assert_eq!(
job.ctrl_break_leaders.lock().expect("lock leaders").clone(),
vec![pid2],
"spawn prunes the stale first leader and records only the live second one"
);
child2.wait().await.expect("second child exits");
drop(child2);
wait_until_pruneable(pid2, &job).await;
let mut child3 = job
.spawn(&mut short_lived_cmd(), &opts)
.expect("spawn third opt-in child");
let pid3 = child3.id().expect("third child has a pid");
assert_eq!(
job.ctrl_break_leaders.lock().expect("lock leaders").clone(),
vec![pid3],
"the list stays bounded to the single live leader across a second \
spawn+exit cycle"
);
let _ = child3.kill().await;
}
}
#[cfg(all(test, feature = "stats"))]
mod metrics_identity_tests {
use super::{ProcIdentity, process_identity, process_metrics};
use windows_sys::Win32::System::Threading::GetCurrentProcessId;
#[test]
fn identity_is_captured_and_matches_a_same_process_read() {
let me = unsafe { GetCurrentProcessId() };
let id = process_identity(me).expect("our own live process has a creation time");
let gated = process_metrics(me, Some(id));
assert!(
gated.cpu_time.is_some(),
"an identity-matched read of our own process reports CPU time"
);
assert!(
gated.peak_memory_bytes.is_some(),
"an identity-matched read of our own process reports peak memory"
);
}
#[test]
fn a_mismatched_identity_yields_defaults_not_the_live_process_counters() {
let me = unsafe { GetCurrentProcessId() };
let real = process_identity(me).expect("our own live process has a creation time");
let bogus = ProcIdentity::from_raw(real.raw().wrapping_add(1));
let gated = process_metrics(me, Some(bogus));
assert!(
gated.cpu_time.is_none() && gated.peak_memory_bytes.is_none(),
"a mismatched identity must yield defaults, never the live process's \
CPU/memory — the recycled-pid fail-safe"
);
let ungated = process_metrics(me, None);
assert!(
ungated.cpu_time.is_some(),
"an unchecked read (identity None) still reports metrics"
);
}
}