use std::ffi::{CStr, CString};
use std::io;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
use std::os::unix::ffi::OsStringExt;
use std::os::unix::process::CommandExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use tokio::process::{Child, Command};
use crate::Mechanism;
#[cfg(feature = "process-control")]
use crate::Signal;
#[cfg(feature = "limits")]
use crate::limits::ResourceLimits;
#[cfg(feature = "stats")]
use crate::stats::ProcessGroupStats;
use crate::sys::pgroup::ProcessGroup;
#[cfg(feature = "stats")]
use crate::sys::{ProcIdentity, ProcMetrics};
static NEXT_ID: AtomicU64 = AtomicU64::new(0);
fn cgroup_name_salt() -> u64 {
static SALT: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
*SALT.get_or_init(|| {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0)
})
}
pub(crate) struct Job {
backend: Backend,
skip_drop_kill: super::SkipDropKill,
}
enum Backend {
Cgroup(Cgroup),
ProcessGroup(ProcessGroup),
}
fn warn_containment_degraded_once() {
#[cfg(feature = "tracing")]
{
use std::sync::Once;
static WARNED: Once = Once::new();
WARNED.call_once(|| {
tracing::warn!(
target: "processkit",
"cgroup v2 unavailable — containment degraded to the POSIX \
process-group fallback; a child that calls setsid can escape \
teardown. Fires once per process (per-spawn detail is at debug)."
);
});
}
}
impl Job {
pub(crate) fn new(#[cfg(feature = "limits")] limits: &ResourceLimits) -> io::Result<Self> {
let backend = match Cgroup::create(
#[cfg(feature = "limits")]
limits,
) {
Ok(cg) => Backend::Cgroup(cg),
Err(_e) => {
#[cfg(feature = "limits")]
if limits.any() {
return Err(_e);
}
warn_containment_degraded_once();
Backend::ProcessGroup(ProcessGroup::new())
}
};
Ok(Job {
backend,
skip_drop_kill: super::SkipDropKill::new(),
})
}
pub(crate) fn spawn(
&self,
cmd: &mut Command,
opts: &crate::sys::SpawnOptions,
) -> io::Result<Child> {
let arm = |cmd: &mut Command| {
if opts.kill_on_parent_death {
let spawner_pid = std::process::id();
unsafe {
cmd.as_std_mut()
.pre_exec(move || arm_pdeathsig(spawner_pid));
}
}
};
match &self.backend {
Backend::Cgroup(cg) => {
let procs = CString::new(cg.path.join("cgroup.procs").into_os_string().into_vec())
.map_err(|_| {
io::Error::new(io::ErrorKind::InvalidInput, "cgroup path contains NUL")
})?;
unsafe {
cmd.as_std_mut()
.pre_exec(move || write_self_pid(procs.as_c_str()));
}
arm(cmd);
let child = cmd.spawn()?;
self.skip_drop_kill.clear();
Ok(child)
}
Backend::ProcessGroup(pg) => {
arm(cmd);
pg.spawn(cmd, opts)
}
}
}
#[cfg(feature = "process-control")]
pub(crate) fn adopt(&self, child: &Child) -> io::Result<()> {
let pid = child
.id()
.ok_or_else(|| io::Error::other("child has no pid (already exited?)"))?
as i32;
match &self.backend {
Backend::Cgroup(cg) => {
match std::fs::write(cg.path.join("cgroup.procs"), pid.to_string().as_bytes()) {
Ok(()) => {
self.skip_drop_kill.clear();
Ok(())
}
Err(e) if e.raw_os_error() == Some(libc::ESRCH) => Ok(()),
Err(e) => Err(e),
}
}
Backend::ProcessGroup(pg) => pg.adopt(child),
}
}
pub(crate) fn kill_all(&self) -> io::Result<()> {
match &self.backend {
Backend::Cgroup(cg) => cg.kill(),
Backend::ProcessGroup(pg) => pg.kill_all(),
}
}
#[cfg(feature = "process-control")]
pub(crate) fn signal(&self, sig: Signal) -> io::Result<()> {
match &self.backend {
Backend::Cgroup(cg) if sig.raw() == libc::SIGKILL => cg.kill(),
Backend::Cgroup(cg) => cg.signal(sig.raw()),
Backend::ProcessGroup(pg) => pg.signal(sig.raw()),
}
}
#[cfg(feature = "process-control")]
pub(crate) fn suspend(&self) -> io::Result<()> {
match &self.backend {
Backend::Cgroup(cg) => cg.freeze(true),
Backend::ProcessGroup(pg) => pg.suspend(),
}
}
#[cfg(feature = "process-control")]
pub(crate) fn resume(&self) -> io::Result<()> {
match &self.backend {
Backend::Cgroup(cg) => cg.freeze(false),
Backend::ProcessGroup(pg) => pg.resume(),
}
}
#[cfg(feature = "process-control")]
pub(crate) fn members(&self) -> io::Result<Vec<u32>> {
let pids = match &self.backend {
Backend::Cgroup(cg) => cg.members()?,
Backend::ProcessGroup(pg) => pg.members(),
};
Ok(pids.into_iter().map(|pid| pid as u32).collect())
}
pub(crate) async fn graceful_shutdown(
&self,
signal: i32,
timeout: Duration,
escalate: bool,
) -> io::Result<()> {
match &self.backend {
Backend::Cgroup(cg) => {
super::graceful::run(cg, &self.skip_drop_kill, signal, timeout, escalate).await
}
Backend::ProcessGroup(pg) => pg.graceful_shutdown(signal, timeout, escalate).await,
}
}
#[cfg(feature = "stats")]
pub(crate) fn stats(&self) -> io::Result<ProcessGroupStats> {
match &self.backend {
Backend::Cgroup(cg) => cg.stats(),
Backend::ProcessGroup(pg) => pg.stats(),
}
}
pub(crate) fn mechanism(&self) -> Mechanism {
match &self.backend {
Backend::Cgroup(_) => Mechanism::CgroupV2,
Backend::ProcessGroup(_) => Mechanism::ProcessGroup,
}
}
}
#[cfg(feature = "stats")]
fn read_proc_starttime(pid: u32) -> Option<u64> {
let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
let after = stat.rsplit_once(')')?.1;
after.split_whitespace().nth(19)?.parse::<u64>().ok()
}
#[cfg(feature = "stats")]
pub(crate) fn process_identity(pid: u32) -> Option<ProcIdentity> {
read_proc_starttime(pid).map(ProcIdentity::from_raw)
}
#[cfg(feature = "stats")]
pub(crate) fn process_metrics(pid: u32, expected: Option<ProcIdentity>) -> ProcMetrics {
let mut metrics = ProcMetrics::default();
let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok();
let fields: Option<Vec<&str>> = stat.as_deref().and_then(|s| {
let idx = s.rfind(')')?;
Some(s[idx + 1..].split_whitespace().collect())
});
if let Some(expected) = expected {
let current = fields
.as_ref()
.and_then(|f| f.get(19))
.and_then(|s| s.parse::<u64>().ok());
if current != Some(expected.raw()) {
return ProcMetrics::default();
}
}
if let Some(fields) = &fields {
if fields.len() > 12
&& let (Ok(utime), Ok(stime)) = (fields[11].parse::<u64>(), fields[12].parse::<u64>())
{
let hz = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
if hz > 0 {
let ticks = utime.saturating_add(stime);
let nanos = ticks as u128 * 1_000_000_000u128 / hz as u128;
metrics.cpu_time = Some(Duration::from_nanos(nanos.min(u64::MAX as u128) as u64));
}
}
}
if let Ok(status) = std::fs::read_to_string(format!("/proc/{pid}/status")) {
for line in status.lines() {
if let Some(rest) = line.strip_prefix("VmHWM:") {
if let Some(kb) = rest
.split_whitespace()
.next()
.and_then(|s| s.parse::<u64>().ok())
{
metrics.peak_memory_bytes = Some(kb.saturating_mul(1024));
}
break;
}
}
}
metrics
}
impl Drop for Job {
fn drop(&mut self) {
match &self.backend {
Backend::Cgroup(cg) => {
if !self.skip_drop_kill.is_set() {
let _ = cg.kill();
for _ in 0..50 {
if let Ok(true) = cg.is_empty() {
break;
}
std::thread::sleep(Duration::from_millis(2));
}
}
let _ = std::fs::remove_dir(&cg.path);
}
Backend::ProcessGroup(_) => {}
}
}
}
fn cgroup2_root() -> Option<PathBuf> {
for candidate in ["/sys/fs/cgroup", "/sys/fs/cgroup/unified"] {
let root = Path::new(candidate);
if root.join("cgroup.controllers").exists() {
return Some(root.to_path_buf());
}
}
None
}
struct Cgroup {
path: PathBuf,
}
impl Cgroup {
fn create(#[cfg(feature = "limits")] limits: &ResourceLimits) -> io::Result<Self> {
let root = cgroup2_root()
.ok_or_else(|| io::Error::new(io::ErrorKind::Unsupported, "cgroup v2 not mounted"))?;
let root = root.as_path();
let self_cgroup = std::fs::read_to_string("/proc/self/cgroup")?;
let rel = self_cgroup
.lines()
.find_map(|line| line.strip_prefix("0::"))
.unwrap_or("/")
.trim();
let parent = root.join(rel.trim_start_matches('/'));
let salt = cgroup_name_salt();
let mut created = None;
for _ in 0..32 {
let name = format!(
"processkit-{}-{:x}-{}",
std::process::id(),
salt,
NEXT_ID.fetch_add(1, Ordering::Relaxed)
);
let path = parent.join(name);
match std::fs::create_dir(&path) {
Ok(()) => {
created = Some(path);
break;
}
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
Err(e) => return Err(e),
}
}
let path = created.ok_or_else(|| {
io::Error::new(
io::ErrorKind::AlreadyExists,
"could not create a unique cgroup directory after retries",
)
})?;
let cg = Cgroup { path };
#[cfg(feature = "limits")]
if limits.any()
&& let Err(e) = cg.apply_limits(&parent, limits)
{
let _ = std::fs::remove_dir(&cg.path);
return Err(e);
}
Ok(cg)
}
#[cfg(feature = "limits")]
fn apply_limits(&self, parent: &Path, limits: &ResourceLimits) -> io::Result<()> {
let mut needed: Vec<&str> = Vec::new();
if limits.max_memory.is_some() {
needed.push("memory");
}
if limits.max_processes.is_some() {
needed.push("pids");
}
if limits.cpu_quota.is_some() {
needed.push("cpu");
}
let enabled =
std::fs::read_to_string(parent.join("cgroup.subtree_control")).unwrap_or_default();
let to_enable = controllers_to_enable(&needed, &enabled);
if !to_enable.is_empty() {
let spec = to_enable
.iter()
.map(|c| format!("+{c}"))
.collect::<Vec<_>>()
.join(" ");
let file = parent.join("cgroup.subtree_control");
std::fs::write(&file, &spec).map_err(|e| {
io::Error::new(
e.kind(),
format!(
"enabling cgroup controllers ({spec}) in {} failed: {e}. cgroup v2's \
'no internal processes' rule forbids enabling controllers in a cgroup \
that holds member processes (except the real hierarchy root), and this \
process is a member of that cgroup — so processkit's resource limits \
apply only when this process runs at the real cgroup-v2 root, not under \
a systemd session/scope/service nor an ordinary (private-cgroupns) \
container, both of which place it in a non-root cgroup. (A cgroup \
namespace root does not count — it only virtualizes the view.) processkit \
does not migrate your process into a sub-cgroup to satisfy the rule; \
arrange that externally (the create-leaf/migrate-self/enable dance) if \
you need limits there.",
file.display()
),
)
})?;
}
if let Some(bytes) = limits.max_memory {
std::fs::write(self.path.join("memory.max"), bytes.to_string())?;
}
if let Some(n) = limits.max_processes {
std::fs::write(self.path.join("pids.max"), n.to_string())?;
}
if let Some(cores) = limits.cpu_quota {
std::fs::write(self.path.join("cpu.max"), cpu_max_value(cores))?;
}
Ok(())
}
fn members(&self) -> io::Result<Vec<i32>> {
self.members_with(|path| std::fs::read_to_string(path))
}
fn members_with(&self, read: impl Fn(&Path) -> io::Result<String>) -> io::Result<Vec<i32>> {
match read(&self.path.join("cgroup.procs")) {
Ok(procs) => Ok(procs
.lines()
.filter_map(|l| l.trim().parse::<i32>().ok())
.filter(|&pid| pid > 0)
.collect()),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(Vec::new()),
Err(e) => Err(e),
}
}
fn is_empty(&self) -> io::Result<bool> {
Ok(self.members()?.is_empty())
}
#[cfg(feature = "stats")]
fn stats(&self) -> io::Result<ProcessGroupStats> {
self.stats_with(|path| std::fs::read_to_string(path))
}
#[cfg(feature = "stats")]
fn stats_with(
&self,
read: impl Fn(&Path) -> io::Result<String>,
) -> io::Result<ProcessGroupStats> {
let pids = self.members_with(&read)?;
let active = pids.len();
let mut cpu = Duration::ZERO;
let mut have_cpu = false;
let mut mem = 0u64;
let mut have_mem = false;
let mut last_err = None;
for pid in pids {
let sample = sample_member_identity_safe(
pid,
|p| process_identity(p as u32),
|p| Ok(self.members_with(&read)?.contains(&p)),
|p, id| process_metrics(p as u32, Some(id)),
);
match sample {
MemberSample::Folded(m) => {
if let Some(c) = m.cpu_time {
cpu = cpu.saturating_add(c);
have_cpu = true;
}
if let Some(p) = m.peak_memory_bytes {
mem = mem.saturating_add(p);
have_mem = true;
}
}
MemberSample::Skipped => {}
MemberSample::Failed(e) => last_err = Some(e),
}
}
if let Some(e) = last_err {
return Err(e);
}
Ok(ProcessGroupStats {
active_process_count: active,
total_cpu_time: have_cpu.then_some(cpu),
peak_memory_bytes: have_mem.then_some(mem),
})
}
fn signal(&self, sig: i32) -> io::Result<()> {
self.signal_with(sig, |path| std::fs::read_to_string(path))
}
fn signal_with(&self, sig: i32, read: impl Fn(&Path) -> io::Result<String>) -> io::Result<()> {
let still_member =
|pid: i32| -> io::Result<bool> { Ok(self.members_with(&read)?.contains(&pid)) };
let mut last_err = None;
for pid in self.members_with(&read)? {
match deliver_identity_safe(pid, sig, pidfd_open, still_member, pidfd_send_signal) {
Delivery::Delivered | Delivery::Skipped => {}
Delivery::Failed(err) => last_err = Some(err),
}
}
match last_err {
Some(err) => Err(err),
None => Ok(()),
}
}
#[cfg(feature = "process-control")]
fn freeze(&self, frozen: bool) -> io::Result<()> {
let val: &[u8] = if frozen { b"1" } else { b"0" };
match std::fs::write(self.path.join("cgroup.freeze"), val) {
Ok(()) => return Ok(()),
Err(e) if e.kind() != io::ErrorKind::NotFound => return Err(e),
Err(_) => {} }
let sig = if frozen { libc::SIGSTOP } else { libc::SIGCONT };
self.signal(sig)
}
fn kill(&self) -> io::Result<()> {
self.kill_with(|path| std::fs::read_to_string(path))
}
fn kill_with(&self, read: impl Fn(&Path) -> io::Result<String>) -> io::Result<()> {
if std::fs::write(self.path.join("cgroup.kill"), b"1").is_ok() {
return Ok(());
}
let _ = std::fs::write(self.path.join("cgroup.freeze"), b"1");
for _ in 0..50 {
if let Ok(members) = self.members_with(&read) {
if members.is_empty() {
break;
}
for pid in members {
unsafe {
libc::kill(pid, libc::SIGKILL);
}
}
}
std::thread::sleep(Duration::from_millis(2));
}
let _ = std::fs::write(self.path.join("cgroup.freeze"), b"0");
match self.members_with(&read) {
Ok(members) if members.is_empty() => Ok(()),
Ok(_) => Err(io::Error::other(
"cgroup did not drain after the bounded SIGKILL sweep (kernel < 5.14 fallback)",
)),
Err(e) => Err(e),
}
}
}
impl super::graceful::GracefulTarget for Cgroup {
fn signal_all(&self, signal: i32) {
let _ = self.signal(signal);
}
fn is_drained(&self) -> bool {
self.is_empty().unwrap_or(false)
}
fn hard_kill(&self) -> io::Result<()> {
self.kill()
}
}
enum Delivery {
Delivered,
Skipped,
Failed(io::Error),
}
fn deliver_identity_safe<H>(
pid: i32,
sig: i32,
open: impl Fn(i32) -> io::Result<H>,
still_member: impl Fn(i32) -> io::Result<bool>,
send: impl Fn(&H, i32) -> io::Result<()>,
) -> Delivery {
let handle = match open(pid) {
Ok(handle) => handle,
Err(e) if e.raw_os_error() == Some(libc::ESRCH) => return Delivery::Delivered,
Err(e) if e.raw_os_error() == Some(libc::ENOSYS) => {
return Delivery::Failed(pidfd_unsupported());
}
Err(e) => return Delivery::Failed(e),
};
match still_member(pid) {
Ok(true) => {}
Ok(false) => return Delivery::Skipped,
Err(e) => return Delivery::Failed(e),
}
match send(&handle, sig) {
Ok(()) => Delivery::Delivered,
Err(e) if e.raw_os_error() == Some(libc::ESRCH) => Delivery::Delivered,
Err(e) if e.raw_os_error() == Some(libc::ENOSYS) => Delivery::Failed(pidfd_unsupported()),
Err(e) => Delivery::Failed(e),
}
}
#[cfg(feature = "stats")]
enum MemberSample {
Folded(ProcMetrics),
Skipped,
Failed(io::Error),
}
#[cfg(feature = "stats")]
fn sample_member_identity_safe(
pid: i32,
capture_identity: impl Fn(i32) -> Option<ProcIdentity>,
still_member: impl Fn(i32) -> io::Result<bool>,
read_metrics: impl Fn(i32, ProcIdentity) -> ProcMetrics,
) -> MemberSample {
let Some(id) = capture_identity(pid) else {
return MemberSample::Skipped;
};
match still_member(pid) {
Ok(true) => {}
Ok(false) => return MemberSample::Skipped,
Err(e) => return MemberSample::Failed(e),
}
MemberSample::Folded(read_metrics(pid, id))
}
fn pidfd_unsupported() -> io::Error {
io::Error::new(
io::ErrorKind::Unsupported,
"identity-safe per-member signalling needs pidfd (pidfd_open/pidfd_send_signal, \
Linux >= 5.3); this kernel lacks it, so processkit refuses to fall back to a racy \
kill(pid, ...) that could hit a pid recycled by a process outside the cgroup — use \
SIGKILL teardown (atomic cgroup.kill) or run on a >= 5.3 kernel",
)
}
fn pidfd_open(pid: i32) -> io::Result<OwnedFd> {
let rc = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) };
if rc < 0 {
return Err(io::Error::last_os_error());
}
Ok(unsafe { OwnedFd::from_raw_fd(rc as RawFd) })
}
fn pidfd_send_signal(fd: &OwnedFd, sig: i32) -> io::Result<()> {
let rc = unsafe {
libc::syscall(
libc::SYS_pidfd_send_signal,
fd.as_raw_fd(),
sig,
std::ptr::null::<libc::siginfo_t>(),
0,
)
};
if rc < 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
#[cfg(feature = "limits")]
fn controllers_to_enable<'a>(needed: &[&'a str], subtree_control: &str) -> Vec<&'a str> {
let already: std::collections::HashSet<&str> = subtree_control.split_whitespace().collect();
needed
.iter()
.copied()
.filter(|c| !already.contains(c))
.collect()
}
#[cfg(feature = "limits")]
fn cpu_max_value(cores: f64) -> String {
const PERIOD: u64 = 100_000;
let quota = (cores * PERIOD as f64).round().max(1.0) as u64;
format!("{quota} {PERIOD}")
}
fn arm_pdeathsig(spawner_pid: u32) -> io::Result<()> {
unsafe {
if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL, 0, 0, 0) != 0 {
return Err(io::Error::last_os_error());
}
if libc::getppid() as u32 != spawner_pid {
libc::_exit(0);
}
}
Ok(())
}
fn write_self_pid(path: &CStr) -> io::Result<()> {
unsafe {
let fd = libc::open(path.as_ptr(), libc::O_WRONLY | libc::O_CLOEXEC);
if fd < 0 {
return Err(io::Error::last_os_error());
}
let mut buf = [0u8; 12];
let mut i = buf.len();
let mut v = libc::getpid() as u32;
loop {
i -= 1;
buf[i] = b'0' + (v % 10) as u8;
v /= 10;
if v == 0 {
break;
}
}
let bytes = &buf[i..];
let written = libc::write(fd, bytes.as_ptr().cast(), bytes.len());
let werr = io::Error::last_os_error();
libc::close(fd);
if written < 0 {
return Err(werr);
}
if (written as usize) != bytes.len() {
return Err(io::Error::from(io::ErrorKind::WriteZero));
}
Ok(())
}
}
#[cfg(test)]
mod cgroup_read_seam_tests {
use std::cell::Cell;
use std::io;
use std::path::{Path, PathBuf};
use super::{Cgroup, Delivery, deliver_identity_safe};
fn cgroup() -> Cgroup {
Cgroup {
path: PathBuf::from("/mock/processkit"),
}
}
#[test]
fn members_parses_readable_procs() {
let members = cgroup()
.members_with(|path| {
assert_eq!(path, Path::new("/mock/processkit/cgroup.procs"));
Ok("12\n0\ninvalid\n-3\n42\n".to_owned())
})
.expect("readable member list");
assert_eq!(members, [12, 42]);
}
#[test]
fn missing_procs_means_empty_cgroup() {
let members = cgroup()
.members_with(|_| Err(io::Error::from(io::ErrorKind::NotFound)))
.expect("a removed cgroup has no members");
assert!(members.is_empty());
}
#[test]
fn permission_denied_procs_is_unknown() {
let err = cgroup()
.members_with(|_| Err(io::Error::from(io::ErrorKind::PermissionDenied)))
.expect_err("an unreadable cgroup must not look empty");
assert_eq!(err.kind(), io::ErrorKind::PermissionDenied);
}
#[test]
fn io_error_procs_is_unknown() {
let err = cgroup()
.members_with(|_| Err(io::Error::from_raw_os_error(libc::EIO)))
.expect_err("an I/O failure must not look empty");
assert_eq!(err.raw_os_error(), Some(libc::EIO));
}
#[test]
fn signal_with_propagates_read_error_without_reaching_the_per_pid_loop() {
let err = cgroup()
.signal_with(libc::SIGTERM, |_| {
Err(io::Error::from(io::ErrorKind::PermissionDenied))
})
.expect_err("an unreadable member list must not look like a successful no-op signal");
assert_eq!(err.kind(), io::ErrorKind::PermissionDenied);
}
#[test]
fn signal_with_empty_member_list_is_a_no_op_success() {
cgroup()
.signal_with(libc::SIGTERM, |_| Ok(String::new()))
.expect("no members to signal is trivially successful");
}
#[test]
fn kill_with_persistent_read_error_reports_a_real_drain_failure() {
let err = cgroup()
.kill_with(|_| Err(io::Error::from_raw_os_error(libc::EIO)))
.expect_err("a cgroup.procs that never becomes readable must not report as drained");
assert_eq!(err.raw_os_error(), Some(libc::EIO));
}
#[test]
fn kill_with_empty_member_list_drains_immediately() {
cgroup()
.kill_with(|_| Ok(String::new()))
.expect("an already-empty cgroup is reported as drained by the fallback sweep");
}
#[cfg(feature = "stats")]
#[test]
fn stats_with_read_error_is_not_reported_as_zero_active_processes() {
let err = cgroup()
.stats_with(|_| Err(io::Error::from(io::ErrorKind::PermissionDenied)))
.expect_err("an unreadable member list must not look like an empty (0-process) group");
assert_eq!(err.kind(), io::ErrorKind::PermissionDenied);
}
#[cfg(feature = "stats")]
#[test]
fn stats_with_empty_member_list_reports_zero_active_processes() {
let stats = cgroup()
.stats_with(|_| Ok(String::new()))
.expect("an empty member list is a legitimate zero-active-process stats snapshot");
assert_eq!(stats.active_process_count, 0);
}
struct FakeHandle;
#[test]
fn reused_pid_outside_cgroup_is_never_signalled() {
let sent = Cell::new(false);
let outcome = deliver_identity_safe(
1234,
libc::SIGTERM,
|_| Ok(FakeHandle),
|_| Ok(false),
|_: &FakeHandle, _| {
sent.set(true);
Ok(())
},
);
assert!(matches!(outcome, Delivery::Skipped));
assert!(
!sent.get(),
"a pid recycled outside the cgroup must never be signalled"
);
}
#[test]
fn confirmed_member_is_signalled_with_the_requested_signal() {
let sent = Cell::new(None);
let outcome = deliver_identity_safe(
42,
libc::SIGTERM,
|_| Ok(FakeHandle),
|_| Ok(true),
|_: &FakeHandle, sig| {
sent.set(Some(sig));
Ok(())
},
);
assert!(matches!(outcome, Delivery::Delivered));
assert_eq!(
sent.get(),
Some(libc::SIGTERM),
"the requested signal reaches a confirmed member"
);
}
#[test]
fn member_gone_before_pin_is_a_benign_no_op() {
let sent = Cell::new(false);
let outcome = deliver_identity_safe(
7,
libc::SIGTERM,
|_| Err::<FakeHandle, _>(io::Error::from_raw_os_error(libc::ESRCH)),
|_| -> io::Result<bool> {
panic!("membership must not be checked once the pin fails ESRCH")
},
|_: &FakeHandle, _| {
sent.set(true);
Ok(())
},
);
assert!(matches!(outcome, Delivery::Delivered));
assert!(!sent.get());
}
#[test]
fn no_pidfd_support_fails_safe_instead_of_raw_kill() {
let sent = Cell::new(false);
let outcome = deliver_identity_safe(
7,
libc::SIGTERM,
|_| Err::<FakeHandle, _>(io::Error::from_raw_os_error(libc::ENOSYS)),
|_| Ok(true),
|_: &FakeHandle, _| {
sent.set(true);
Ok(())
},
);
match outcome {
Delivery::Failed(e) => assert_eq!(e.kind(), io::ErrorKind::Unsupported),
_ => panic!("a kernel without pidfd must fail safe, not signal"),
}
assert!(!sent.get(), "fail-safe must not send any signal");
}
#[test]
fn unreadable_membership_after_pin_fails_safe_without_sending() {
let sent = Cell::new(false);
let outcome = deliver_identity_safe(
7,
libc::SIGTERM,
|_| Ok(FakeHandle),
|_| Err(io::Error::from(io::ErrorKind::PermissionDenied)),
|_: &FakeHandle, _| {
sent.set(true);
Ok(())
},
);
match outcome {
Delivery::Failed(e) => assert_eq!(e.kind(), io::ErrorKind::PermissionDenied),
_ => panic!("an unreadable membership must fail safe"),
}
assert!(!sent.get());
}
#[test]
fn pinned_target_exiting_before_send_is_a_benign_esrch() {
let outcome = deliver_identity_safe(
7,
libc::SIGTERM,
|_| Ok(FakeHandle),
|_| Ok(true),
|_: &FakeHandle, _| Err(io::Error::from_raw_os_error(libc::ESRCH)),
);
assert!(matches!(outcome, Delivery::Delivered));
}
#[test]
fn eperm_on_send_is_a_real_failure_that_surfaces() {
let outcome = deliver_identity_safe(
7,
libc::SIGTERM,
|_| Ok(FakeHandle),
|_| Ok(true),
|_: &FakeHandle, _| Err(io::Error::from_raw_os_error(libc::EPERM)),
);
match outcome {
Delivery::Failed(e) => assert_eq!(e.raw_os_error(), Some(libc::EPERM)),
_ => panic!("EPERM is a real delivery failure and must surface"),
}
}
}
#[cfg(test)]
mod fail_safe_tests {
use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
use std::time::{Duration, Instant};
use super::{Backend, Cgroup, Job};
use crate::sys::SkipDropKill;
use crate::sys::graceful::GracefulTarget;
fn unreadable_procs_cgroup() -> Option<(Cgroup, PathBuf)> {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let dir = std::env::temp_dir().join(format!(
"processkit-failsafe-test-{}-{nanos}",
std::process::id()
));
std::fs::create_dir_all(&dir).expect("create temp cgroup dir");
let procs = dir.join("cgroup.procs");
std::fs::write(&procs, b"").expect("create cgroup.procs");
std::fs::set_permissions(&procs, std::fs::Permissions::from_mode(0o000))
.expect("revoke read permission on cgroup.procs");
let cg = Cgroup { path: dir.clone() };
if cg.is_empty().is_ok() {
let _ = std::fs::remove_dir_all(&dir);
eprintln!(
"skipping: this environment can read past chmod 000 (likely running as root) \
— the fail-safe path under test is not reachable here"
);
return None;
}
Some((cg, dir))
}
#[test]
fn is_drained_treats_unreadable_procs_as_not_drained() {
let Some((cg, dir)) = unreadable_procs_cgroup() else {
return;
};
assert!(
!cg.is_drained(),
"an unreadable member list is unknown, not drained — GracefulTarget::is_drained \
must not treat it as an empty cgroup (doing so would cancel the escalation)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn drop_keeps_waiting_out_the_bounded_drain_when_procs_is_unreadable() {
let Some((cg, dir)) = unreadable_procs_cgroup() else {
return;
};
let job = Job {
backend: Backend::Cgroup(cg),
skip_drop_kill: SkipDropKill::new(),
};
let start = Instant::now();
drop(job);
let elapsed = start.elapsed();
assert!(
elapsed >= Duration::from_millis(90),
"Job::drop exited its drain wait early ({elapsed:?}) — an unreadable member \
list must not be treated as an empty (drained) cgroup"
);
let _ = std::fs::remove_dir_all(&dir);
}
}
#[cfg(all(test, feature = "limits"))]
mod tests {
use super::{controllers_to_enable, cpu_max_value};
#[test]
fn cpu_max_formats_quota_and_period() {
assert_eq!(cpu_max_value(0.5), "50000 100000");
assert_eq!(cpu_max_value(2.0), "200000 100000");
assert_eq!(cpu_max_value(0.000_001), "1 100000");
}
#[test]
fn controllers_to_enable_skips_already_enabled_ones() {
assert!(controllers_to_enable(&["memory", "pids"], "cpu memory pids").is_empty());
assert_eq!(
controllers_to_enable(&["memory", "pids", "cpu"], "memory"),
["pids", "cpu"]
);
assert_eq!(controllers_to_enable(&["memory"], ""), ["memory"]);
assert!(controllers_to_enable(&["pids"], "pids io hugetlb").is_empty());
}
}
#[cfg(test)]
mod rearm_race_tests {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
struct RacingRearm<'a> {
latch: &'a crate::sys::SkipDropKill,
polls: AtomicUsize,
}
impl crate::sys::graceful::GracefulTarget for RacingRearm<'_> {
fn signal_all(&self, _signal: i32) {}
fn is_drained(&self) -> bool {
if self.polls.fetch_add(1, Ordering::Relaxed) == 1 {
self.latch.clear();
}
false
}
fn hard_kill(&self) -> std::io::Result<()> {
Ok(())
}
}
#[tokio::test(start_paused = true)]
async fn shutdown_request_does_not_override_a_concurrent_rearm() {
let skip = crate::sys::SkipDropKill::new();
skip.clear(); let target = RacingRearm {
latch: &skip,
polls: AtomicUsize::new(0),
};
crate::sys::graceful::run(
&target,
&skip,
libc::SIGTERM,
Duration::from_millis(100),
false,
)
.await
.expect("graceful run");
assert!(
!skip.is_set(),
"a child that joined the cgroup mid-shutdown must keep its Drop-kill \
backstop — the stale request must not re-spare it (Job::drop then \
cgroup.kill's the tree)"
);
}
}
#[cfg(test)]
mod pidfd_integration_tests {
use super::{Delivery, deliver_identity_safe, pidfd_open, pidfd_send_signal};
fn pidfd_available() -> bool {
pidfd_open(std::process::id() as i32).is_ok()
}
fn spawn_sleeper() -> std::process::Child {
std::process::Command::new("sleep")
.arg("30")
.spawn()
.expect("spawn `sleep 30`")
}
#[test]
fn pidfd_pins_identity_and_reports_exit_via_esrch() {
if !pidfd_available() {
eprintln!("skipping: pidfd_open unavailable on this kernel/sandbox");
return;
}
let mut child = spawn_sleeper();
let pid = child.id() as i32;
let fd = pidfd_open(pid).expect("pin the live child");
pidfd_send_signal(&fd, 0).expect("null-signal a live pinned child");
child.kill().expect("kill child");
child.wait().expect("reap child");
let err =
pidfd_send_signal(&fd, 0).expect_err("a reaped, pinned task must not be signallable");
assert_eq!(
err.raw_os_error(),
Some(libc::ESRCH),
"a pinned task that exited must report ESRCH, never signal a recycled pid"
);
}
#[test]
fn a_live_non_member_is_skipped_by_the_real_primitive() {
if !pidfd_available() {
eprintln!("skipping: pidfd_open unavailable on this kernel/sandbox");
return;
}
let mut child = spawn_sleeper();
let pid = child.id() as i32;
let outcome = deliver_identity_safe(
pid,
libc::SIGKILL,
pidfd_open,
|_| Ok(false),
pidfd_send_signal,
);
assert!(matches!(outcome, Delivery::Skipped));
assert!(
child.try_wait().expect("try_wait").is_none(),
"a non-member must receive no signal — the live child is untouched"
);
let _ = child.kill();
let _ = child.wait();
}
#[test]
fn a_confirmed_live_member_is_delivered_to() {
use std::os::unix::process::ExitStatusExt;
if !pidfd_available() {
eprintln!("skipping: pidfd_open unavailable on this kernel/sandbox");
return;
}
let mut child = spawn_sleeper();
let pid = child.id() as i32;
let outcome = deliver_identity_safe(
pid,
libc::SIGTERM,
pidfd_open,
|_| Ok(true),
pidfd_send_signal,
);
assert!(matches!(outcome, Delivery::Delivered));
let status = child.wait().expect("reap the signalled child");
assert_eq!(
status.signal(),
Some(libc::SIGTERM),
"the child exited on the SIGTERM we delivered through the pidfd"
);
}
}
#[cfg(all(test, feature = "stats"))]
mod member_sample_tests {
use std::cell::Cell;
use std::io;
use std::time::Duration;
use super::{
MemberSample, ProcIdentity, process_identity, process_metrics, read_proc_starttime,
sample_member_identity_safe,
};
use crate::sys::ProcMetrics;
fn some_metrics() -> ProcMetrics {
ProcMetrics {
cpu_time: Some(Duration::from_millis(10)),
peak_memory_bytes: Some(2048),
}
}
#[test]
fn reused_pid_outside_cgroup_is_never_folded() {
let read = Cell::new(false);
let outcome = sample_member_identity_safe(
1234,
|_| Some(ProcIdentity::from_raw(42)),
|_| Ok(false),
|_, _| {
read.set(true);
some_metrics()
},
);
assert!(matches!(outcome, MemberSample::Skipped));
assert!(
!read.get(),
"a pid recycled outside the cgroup must never have its counters folded"
);
}
#[test]
fn confirmed_member_is_folded_with_its_counters() {
let outcome = sample_member_identity_safe(
42,
|_| Some(ProcIdentity::from_raw(7)),
|_| Ok(true),
|_, _| some_metrics(),
);
match outcome {
MemberSample::Folded(m) => {
assert_eq!(m.cpu_time, Some(Duration::from_millis(10)));
assert_eq!(m.peak_memory_bytes, Some(2048));
}
_ => panic!("a confirmed member must be folded"),
}
}
#[test]
fn member_gone_before_pin_is_a_benign_skip() {
let read = Cell::new(false);
let outcome = sample_member_identity_safe(
7,
|_| None,
|_| -> io::Result<bool> { panic!("membership must not be checked once the pin fails") },
|_, _| {
read.set(true);
some_metrics()
},
);
assert!(matches!(outcome, MemberSample::Skipped));
assert!(!read.get(), "a gone member's counters must not be read");
}
#[test]
fn unreadable_membership_fails_safe_without_reading_counters() {
let read = Cell::new(false);
let outcome = sample_member_identity_safe(
7,
|_| Some(ProcIdentity::from_raw(1)),
|_| Err(io::Error::from(io::ErrorKind::PermissionDenied)),
|_, _| {
read.set(true);
some_metrics()
},
);
match outcome {
MemberSample::Failed(e) => assert_eq!(e.kind(), io::ErrorKind::PermissionDenied),
_ => panic!("an unreadable membership must fail safe"),
}
assert!(!read.get(), "fail-safe must not read any counters");
}
#[test]
fn recycle_after_reconfirm_folds_nothing() {
let outcome = sample_member_identity_safe(
7,
|_| Some(ProcIdentity::from_raw(1)),
|_| Ok(true),
|_, _| ProcMetrics::default(),
);
match outcome {
MemberSample::Folded(m) => {
assert!(
m.cpu_time.is_none() && m.peak_memory_bytes.is_none(),
"a recycle caught by the identity-gated read contributes nothing"
);
}
_ => panic!("a confirmed member is folded (with an all-None reading here)"),
}
}
#[test]
fn process_identity_matches_a_same_process_metrics_read() {
let me = std::process::id();
assert!(
read_proc_starttime(me).is_some(),
"our own /proc/<pid>/stat starttime must be readable"
);
let id = process_identity(me).expect("our own live process has a start identity");
let gated = process_metrics(me, Some(id));
assert!(
gated.cpu_time.is_some(),
"an identity-matched read of our own process reports CPU time"
);
}
#[test]
fn a_mismatched_identity_yields_defaults_not_the_live_process_counters() {
let me = std::process::id();
let real = process_identity(me).expect("our own live process has a start identity");
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"
);
assert!(
process_metrics(me, None).cpu_time.is_some(),
"an unchecked read (identity None) still reports metrics"
);
}
}