diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs index 6fddb93bf7..b24af61b45 100644 --- a/nativelink-config/src/cas_server.rs +++ b/nativelink-config/src/cas_server.rs @@ -916,7 +916,7 @@ pub struct EndpointConfig { pub enum MemoryEnforcement { /// Measure only. None, - /// Kill the action's process group once two consecutive samples exceed + /// Kill the action's processes once two consecutive samples exceed /// the reservation plus headroom, or one sample exceeds twice it. #[default] Soft, diff --git a/nativelink-proto/com/github/trace_machina/nativelink/remote_execution/worker_api.proto b/nativelink-proto/com/github/trace_machina/nativelink/remote_execution/worker_api.proto index 0fd3d61660..b98361657c 100644 --- a/nativelink-proto/com/github/trace_machina/nativelink/remote_execution/worker_api.proto +++ b/nativelink-proto/com/github/trace_machina/nativelink/remote_execution/worker_api.proto @@ -178,8 +178,8 @@ message ActionResourceUsage { string worker_id = 4; /// Total CPU time consumed by the action process tree, in milliseconds. - /// User and system time combined, summed across every process in the - /// action's process group. + /// User and system time combined, of every process the action started: + /// the kernel's total of everything the action's stub reaped. uint64 cpu_time_ms = 5; /// Wall time from spawn to exit, in milliseconds; cores used is diff --git a/nativelink-proto/genproto/com.github.trace_machina.nativelink.remote_execution.pb.rs b/nativelink-proto/genproto/com.github.trace_machina.nativelink.remote_execution.pb.rs index 62926da037..745a638dbf 100644 --- a/nativelink-proto/genproto/com.github.trace_machina.nativelink.remote_execution.pb.rs +++ b/nativelink-proto/genproto/com.github.trace_machina.nativelink.remote_execution.pb.rs @@ -146,8 +146,8 @@ pub struct ActionResourceUsage { #[prost(string, tag = "4")] pub worker_id: ::prost::alloc::string::String, /// / Total CPU time consumed by the action process tree, in milliseconds. - /// / User and system time combined, summed across every process in the - /// / action's process group. + /// / User and system time combined, of every process the action started: + /// / the kernel's total of everything the action's stub reaped. #[prost(uint64, tag = "5")] pub cpu_time_ms: u64, /// / Wall time from spawn to exit, in milliseconds; cores used is diff --git a/nativelink-worker/BUILD.bazel b/nativelink-worker/BUILD.bazel index b32036527e..b58eb0db40 100644 --- a/nativelink-worker/BUILD.bazel +++ b/nativelink-worker/BUILD.bazel @@ -19,13 +19,17 @@ rust_library( "src/persistent_worker/mod.rs", "src/persistent_worker/pool.rs", "src/persistent_worker/protocol.rs", + "src/process_tree.rs", "src/qos.rs", "src/reaper.rs", "src/running_actions_manager.rs", "src/worker_api_client_wrapper.rs", "src/worker_utils.rs", ] + select({ - "@platforms//os:linux": ["src/namespace_utils.rs"], + "@platforms//os:linux": [ + "src/action_stub.rs", + "src/namespace_utils.rs", + ], "//conditions:default": [], }), proc_macro_deps = [ diff --git a/nativelink-worker/src/action_stub.rs b/nativelink-worker/src/action_stub.rs new file mode 100644 index 0000000000..7bbdb22a5e --- /dev/null +++ b/nativelink-worker/src/action_stub.rs @@ -0,0 +1,369 @@ +// Copyright 2026 The NativeLink Authors. All rights reserved. +// +// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// See LICENSE file for details +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! The process every action runs under: the worker's child, which forks the +//! action and stays as its parent until everything the action started is +//! gone, as Bazel's `process-wrapper` does for a local action +//! (`src/main/tools/process-tools-linux.cc`, `TerminateAndWaitForAll`). +//! +//! The stub is a child subreaper (`PR_SET_CHILD_SUBREAPER`), so a process the +//! action orphans is reparented to it, not to the worker, whatever session +//! or group the process moved to: the action's processes are exactly the +//! stub's descendants (`crate::process_tree`). When the action's own +//! process exits, the stub kills what is left, reaps it, and exits with the +//! action's status, so nothing outlives its action to hold a port, a file +//! or memory that the next action would be charged for. +//! +//! Before it exits it writes a [`StubReport`] to a pipe the worker holds: +//! the CPU time of everything it reaped, which the kernel totals exactly +//! (`getrusage(RUSAGE_CHILDREN)`), including processes too short-lived for +//! any sample to see. +//! +//! Everything here runs between `fork` and `exec` in a copy of a +//! multi-threaded process, so it is async-signal-safe: system calls only, +//! no allocation, no locks. + +use std::io::Error; +use std::os::fd::{AsRawFd, OwnedFd}; + +/// What the stub reports when the action is over. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct StubReport { + /// User CPU time of every process the stub reaped, and of what those + /// reaped, in microseconds. + pub user_cpu_us: u64, + /// System CPU time of the same, in microseconds. + pub system_cpu_us: u64, + /// How many processes were still running, or waiting to be reaped, when + /// the action's own process exited. + pub stragglers: u32, +} + +/// The report's size on the pipe: two little-endian `u64`s and a `u32`. +pub const REPORT_LEN: usize = 20; + +impl StubReport { + pub fn to_bytes(self) -> [u8; REPORT_LEN] { + let mut bytes = [0u8; REPORT_LEN]; + bytes[0..8].copy_from_slice(&self.user_cpu_us.to_le_bytes()); + bytes[8..16].copy_from_slice(&self.system_cpu_us.to_le_bytes()); + bytes[16..20].copy_from_slice(&self.stragglers.to_le_bytes()); + bytes + } + + pub fn from_bytes(bytes: &[u8; REPORT_LEN]) -> Self { + let u64_at = |at: usize| { + let mut word = [0u8; 8]; + word.copy_from_slice(&bytes[at..at + 8]); + u64::from_le_bytes(word) + }; + let mut stragglers = [0u8; 4]; + stragglers.copy_from_slice(&bytes[16..20]); + Self { + user_cpu_us: u64_at(0), + system_cpu_us: u64_at(8), + stragglers: u32::from_le_bytes(stragglers), + } + } + + /// The total CPU time, in milliseconds. + pub const fn cpu_time_ms(&self) -> u64 { + self.user_cpu_us.saturating_add(self.system_cpu_us) / 1_000 + } +} + +/// The worker's end of the report pipe, and the stub's. +#[derive(Debug)] +pub struct ReportPipe { + /// Non-blocking, so a stub that died without writing reads as no + /// report instead of a wait. + pub read: OwnedFd, + /// Close-on-exec, so the stub (which never execs) keeps it and the + /// action loses it at its `exec`. The worker drops its copy once the + /// action is spawned. + pub write: OwnedFd, +} + +impl ReportPipe { + pub fn new() -> Result { + let mut fds: [libc::c_int; 2] = [0; 2]; + // SAFETY: fds is a two-element array on the stack, as pipe2 needs. + if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) } != 0 { + return Err(Error::last_os_error()); + } + // SAFETY: pipe2 returned two new descriptors that nothing else owns. + let (read, write) = unsafe { + use std::os::fd::FromRawFd; + (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) + }; + // SAFETY: fcntl on a descriptor we own, with integer arguments. + let flags = unsafe { libc::fcntl(read.as_raw_fd(), libc::F_GETFL) }; + // SAFETY: as above. + if flags < 0 + || unsafe { libc::fcntl(read.as_raw_fd(), libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 + { + return Err(Error::last_os_error()); + } + Ok(Self { read, write }) + } + + /// The stub's report, once the stub has exited: `None` when it wrote + /// none (it was killed, or it could not be started). + pub fn read_report(read: &OwnedFd) -> Option { + let mut bytes = [0u8; REPORT_LEN]; + // SAFETY: bytes is on the stack and REPORT_LEN long. + let n = unsafe { libc::read(read.as_raw_fd(), bytes.as_mut_ptr().cast(), REPORT_LEN) }; + (usize::try_from(n).ok() == Some(REPORT_LEN)).then(|| StubReport::from_bytes(&bytes)) + } +} + +fn exit(status: i32) -> ! { + // SAFETY: It is always safe to _exit. + unsafe { libc::_exit(status) }; +} + +fn errno() -> libc::c_int { + // SAFETY: reading errno is always safe. + unsafe { *libc::__errno_location() } +} + +/// Closes every descriptor except `keep`. +/// +/// # Safety +/// +/// Only for a process about to run nothing but the stub's own loop. +unsafe fn close_all_fds_except(keep: libc::c_int) { + // SAFETY: closing descriptors has no memory safety considerations; the + // stub uses none but `keep`. + unsafe { + if keep > 0 { + libc::syscall(libc::SYS_close_range, 0, keep - 1, 0); + } + if libc::syscall(libc::SYS_close_range, keep + 1, libc::c_uint::MAX, 0) == 0 { + return; + } + // Before Linux 5.9: every descriptor up to the limit, one by one. + let mut rlim = core::mem::MaybeUninit::::uninit(); + let max_fd = if libc::getrlimit(libc::RLIMIT_NOFILE, rlim.as_mut_ptr()) == 0 { + let cur = rlim.assume_init().rlim_cur; + if cur == libc::RLIM_INFINITY { + 0x0001_0000 + } else { + libc::c_int::try_from(cur).unwrap_or(0x0001_0000) + } + } else { + 4096 + }; + for fd in 0..max_fd { + if fd != keep { + libc::close(fd); + } + } + } +} + +/// Sends SIGKILL to every child of this process, as listed by +/// `/proc/thread-self/children` (the stub is single-threaded, so its one +/// thread's children are all of its children). Returns `false` when the +/// list cannot be read (a kernel without `CONFIG_PROC_CHILDREN`). +fn kill_children() -> bool { + // SAFETY: open with a static path and integer flags. + let fd = unsafe { + libc::open( + c"/proc/thread-self/children".as_ptr(), + libc::O_RDONLY | libc::O_CLOEXEC, + ) + }; + if fd < 0 { + return false; + } + let mut buffer = [0u8; 512]; + let mut pid: libc::pid_t = 0; + let mut in_number = false; + loop { + // SAFETY: buffer is on the stack and its length is passed. + let n = unsafe { libc::read(fd, buffer.as_mut_ptr().cast(), buffer.len()) }; + if n < 0 && errno() == libc::EINTR { + continue; + } + let Ok(n) = usize::try_from(n) else { break }; + if n == 0 { + break; + } + for &byte in &buffer[..n] { + if byte.is_ascii_digit() { + pid = pid + .saturating_mul(10) + .saturating_add(libc::pid_t::from(byte - b'0')); + in_number = true; + } else if in_number { + // SAFETY: an unreaped child's pid cannot be reused, so this + // signals only our own child. + unsafe { libc::kill(pid, libc::SIGKILL) }; + pid = 0; + in_number = false; + } + } + } + if in_number { + // SAFETY: as above. + unsafe { libc::kill(pid, libc::SIGKILL) }; + } + // SAFETY: fd was opened above. + unsafe { libc::close(fd) }; + true +} + +/// Kills and reaps everything left under the stub once the action's own +/// process is gone: each pass kills the stub's children, whose own children +/// are then reparented to the stub for the next pass, until `waitpid` +/// reports no child left. Returns how many processes it reaped. +fn terminate_and_wait_for_all() -> u32 { + let mut reaped = 0u32; + loop { + if !kill_children() { + // Without the list there is nothing to kill by; what is left is + // the worker's to end by the session. + return reaped; + } + let mut status = 0; + // SAFETY: status is on the stack. + let res = unsafe { libc::waitpid(-1, &raw mut status, 0) }; + if res > 0 { + reaped = reaped.saturating_add(1); + } else if errno() == libc::ECHILD { + return reaped; + } + // EINTR: list and wait again. + } +} + +/// Microseconds in a `timeval`. +fn micros(time: libc::timeval) -> u64 { + u64::try_from(time.tv_sec) + .unwrap_or(0) + .saturating_mul(1_000_000) + .saturating_add(u64::try_from(time.tv_usec).unwrap_or(0)) +} + +/// Writes the report and exits as the action did: with its exit code, or by +/// the signal that killed it. +fn report_and_exit(report_fd: libc::c_int, status: libc::c_int, stragglers: u32) -> ! { + let mut usage = core::mem::MaybeUninit::::zeroed(); + // SAFETY: usage is on the stack; getrusage only writes it, and a failed + // call leaves the zeroes, which report no CPU. + unsafe { libc::getrusage(libc::RUSAGE_CHILDREN, usage.as_mut_ptr()) }; + // SAFETY: an all-zero rusage is valid, and getrusage writes a valid one. + let usage = unsafe { usage.assume_init() }; + let bytes = StubReport { + user_cpu_us: micros(usage.ru_utime), + system_cpu_us: micros(usage.ru_stime), + stragglers, + } + .to_bytes(); + // SAFETY: bytes is on the stack. A short or failed write leaves the + // worker with no report, which it treats as such. + unsafe { libc::write(report_fd, bytes.as_ptr().cast(), bytes.len()) }; + + if libc::WIFSIGNALED(status) { + let mut sigset = core::mem::MaybeUninit::::uninit(); + // SAFETY: sigset is on the stack; raise is always safe, and _exit + // below is the fallback. + unsafe { + libc::sigemptyset(sigset.as_mut_ptr()); + libc::sigaddset(sigset.as_mut_ptr(), libc::WTERMSIG(status)); + libc::sigprocmask(libc::SIG_UNBLOCK, sigset.as_ptr(), core::ptr::null_mut()); + libc::signal(libc::WTERMSIG(status), libc::SIG_DFL); + libc::raise(libc::WTERMSIG(status)); + } + exit(128 + libc::WTERMSIG(status)); + } + exit(libc::WEXITSTATUS(status)); +} + +/// Forks the action under a stub. For `pre_exec`, after any namespace has +/// been entered: in the child, which goes on to exec the action, it +/// returns `Ok`; the parent becomes the stub and never returns. +/// +/// The stub passes a SIGTERM on to the action, so the action gets the grace +/// the worker's `kill_grace_ms` promises, reaps as children exit, and when +/// the action's process exits, ends the rest (see the module docs). The +/// action gets `PR_SET_PDEATHSIG`, so a stub that is killed takes the +/// action's own process with it. +/// +/// # Safety +/// +/// Only in `pre_exec`. `report_fd` is the write end of a [`ReportPipe`], +/// or -1 for no report. +pub unsafe fn fork_stub(report_fd: libc::c_int) -> Result<(), Error> { + // SAFETY: We are in pre_exec, an async-signal-safe context, and stay + // in it. + match unsafe { libc::fork() } { + 0 => { + // SAFETY: prctl with integer arguments is async-signal-safe. + if unsafe { libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) } != 0 { + exit(1); + } + Ok(()) + } + pid if pid > 0 => { + // SAFETY: prctl with integer arguments is async-signal-safe. If + // it fails, orphans go to the worker as they did before stubs. + unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) }; + // SAFETY: the stub uses no descriptor but the report's from here. + unsafe { close_all_fds_except(report_fd) }; + + let mut sigset = core::mem::MaybeUninit::::uninit(); + // SAFETY: sigset is on the stack and initialized here. + unsafe { + libc::sigemptyset(sigset.as_mut_ptr()); + libc::sigaddset(sigset.as_mut_ptr(), libc::SIGTERM); + libc::sigaddset(sigset.as_mut_ptr(), libc::SIGCHLD); + libc::sigprocmask(libc::SIG_BLOCK, sigset.as_ptr(), core::ptr::null_mut()); + } + + loop { + // Reap whatever has exited; the action's own process ends + // the action. + loop { + let mut status = 0; + // SAFETY: status is on the stack. + let res = unsafe { libc::waitpid(-1, &raw mut status, libc::WNOHANG) }; + if res == pid { + let stragglers = terminate_and_wait_for_all(); + report_and_exit(report_fd, status, stragglers); + } else if res == 0 { + break; + } else if res < 0 { + if errno() == libc::EINTR { + continue; + } + // ECHILD with the action never reaped cannot happen; + // treat it as an action that vanished. + exit(255); + } + } + + let mut siginfo = core::mem::MaybeUninit::::uninit(); + // SAFETY: sigset is initialized and siginfo is on the stack. + let sig = unsafe { libc::sigwaitinfo(sigset.as_ptr(), siginfo.as_mut_ptr()) }; + if sig == libc::SIGTERM { + // SAFETY: pid is our unreaped child. + unsafe { libc::kill(pid, libc::SIGTERM) }; + } + } + } + _ => Err(Error::last_os_error()), + } +} diff --git a/nativelink-worker/src/lib.rs b/nativelink-worker/src/lib.rs index 48997ca77a..42b693ce14 100644 --- a/nativelink-worker/src/lib.rs +++ b/nativelink-worker/src/lib.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[cfg(target_os = "linux")] +pub mod action_stub; pub mod buck2_file_capture; pub mod capacity; pub mod directory_cache; @@ -19,6 +21,8 @@ pub mod local_worker; #[cfg(target_os = "linux")] pub mod namespace_utils; pub mod persistent_worker; +#[cfg(target_family = "unix")] +pub mod process_tree; pub mod qos; pub mod reaper; pub mod running_actions_manager; diff --git a/nativelink-worker/src/namespace_utils.rs b/nativelink-worker/src/namespace_utils.rs index 206759a92e..0cf81b375e 100644 --- a/nativelink-worker/src/namespace_utils.rs +++ b/nativelink-worker/src/namespace_utils.rs @@ -16,60 +16,50 @@ use std::io::Error; use tracing::error; -/// A wrapper around a Child that may be the stub of a PID namespace. A -/// SIGTERM to the stub is passed on to the action; a SIGKILL to the stub -/// ends the whole namespace, which is the hard kill. +/// The action's stub (`crate::action_stub`), which may be outside a PID +/// namespace the action is the init of. A SIGTERM to the stub is passed on +/// to the action; the hard kill ends every descendant of the stub, and then +/// the stub. #[derive(Debug)] -pub struct MaybeNamespacedChild { - namespaced: bool, +pub struct ActionStubChild { child: tokio::process::Child, } -/// SIGKILL to every process in the group `pgid` leads. ESRCH means the -/// group is already gone, which is the outcome wanted; anything else, EPERM -/// above all, means a survivor and is reported. -#[cfg(target_os = "linux")] -fn kill_process_group(pgid: u32) { - let Ok(pgid) = i32::try_from(pgid) else { - return; - }; - // SAFETY: killpg only takes integers and has no memory safety - // considerations; a stale group id is reported as ESRCH, not acted on. - let rc = unsafe { libc::killpg(pgid, libc::SIGKILL) }; - if rc != 0 { - let err = Error::last_os_error(); - if err.raw_os_error() != Some(libc::ESRCH) { - error!( - pgid, - ?err, - "Could not kill the action's process group; a descendant may have survived" - ); - } +/// SIGKILL to every descendant of the stub, repeated until none is left +/// alive (`crate::process_tree::signal_descendants`). A descendant that +/// survives that is reported. +fn kill_descendants(stub: u32) { + crate::process_tree::signal_descendants(stub, libc::SIGKILL); + let survivors = crate::process_tree::tree_members(stub) + .iter() + .filter(|(pid, stat)| *pid != stub && !crate::process_tree::is_zombie_stat(stat)) + .count(); + if survivors > 0 { + error!( + stub, + survivors, "Could not kill the action's processes; a descendant may have survived" + ); } } -impl MaybeNamespacedChild { - pub const fn new(namespaced: bool, child: tokio::process::Child) -> Self { - Self { namespaced, child } +impl ActionStubChild { + pub const fn new(child: tokio::process::Child) -> Self { + Self { child } } pub fn id(&self) -> Option { self.child.id() } - /// SIGKILL the child and wait for it. Namespaced, the child is the - /// stub and the init of the action's PID namespace, so its death takes - /// the action and everything the action spawned with it; the stub also - /// set `PR_SET_PDEATHSIG` on the action for the same end. Not - /// namespaced, the child is its own group leader, so the group gets the - /// SIGKILL first: a timed-out or cancelled action used to lose only its - /// direct child while its descendants ran on (issue #225). + /// SIGKILL the action and wait for its stub. The stub's descendants go + /// first, so none is orphaned to the worker by the stub's death; a + /// timed-out or cancelled action used to lose only its direct child + /// while its descendants ran on (issue #225). Namespaced, the action is + /// the init of its PID namespace, whose death takes the namespace with + /// it as well. pub async fn kill(&mut self) -> Result<(), Error> { - #[cfg(target_os = "linux")] - if !self.namespaced - && let Some(pgid) = self.child.id() - { - kill_process_group(pgid); + if let Some(stub) = self.child.id() { + kill_descendants(stub); } self.child.kill().await } @@ -292,40 +282,6 @@ fn write_signal_safe(file_name: &core::ffi::CStr, data: &[u8]) -> Result<(), cor } } -/// An async-signal-safe method to close all open file descriptors for the -/// current process. This function is unsafe as any existing handles to -/// file descriptors will be invalidated. None may be used after calling -/// this function. -unsafe fn close_all_fds() { - // SAFETY: It is safe to call close on all file descriptors as this is - // the purpose of the function. - if unsafe { libc::syscall(libc::SYS_close_range, 0, libc::INT_MAX, 0) } == 0 { - return; - } - // Since we're <5.9 kernel, we need to get the max FD count. - let mut rlim = core::mem::MaybeUninit::::uninit(); - // SAFETY: We just allocated the memory for this and getrlimit is async-signal-safe. - let max_fd = if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, rlim.as_mut_ptr()) } == 0 { - // SAFETY: We just initialised this in getrlimit above that succeeded. - let cur = unsafe { rlim.assume_init().rlim_cur }; - if cur == libc::RLIM_INFINITY { - // Sane fallback for unlimited environments - 0x0001_0000 - } else { - core::ffi::c_int::try_from(cur).unwrap_or(0x0001_0000) - } - } else { - // Fallback for getrlimit failure. - 4096 - }; - for fd in 0..max_fd { - // SAFETY: It is safe to close a file descriptor that is not open and - // we also want to close all, so there's no issue with closing file - // descriptors that others may have handles to. - unsafe { libc::close(fd) }; - } -} - /// Write the value n to the given slice as a decimal string. fn u32_to_bytes(mut n: u32, buf: &mut [u8]) -> usize { if n == 0 { @@ -584,6 +540,7 @@ pub fn configure_namespace( tmp_directory: Option<&core::ffi::CStr>, root_action_directory: &core::ffi::CStr, action_directory: &core::ffi::CStr, + report_fd: core::ffi::c_int, ) -> std::io::Result<()> { // SAFETY: It is always safe to call geteuid on Posix. let uid = unsafe { libc::geteuid() }; @@ -641,89 +598,8 @@ pub fn configure_namespace( } } - // Fork to enter the PID namespace. - // SAFETY: We are already in a required async-signal-safe environment, we - // will continue to ensure that ongoing. - match unsafe { libc::fork() } { - 0 => { - // SAFETY: This function is async-signal-safe and references no memory or resources. - if unsafe { libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) } != 0 { - exit(1); - } - Ok(()) - } - pid if pid > 0 => { - // Ensure that any children spawned by the action are re-parented to - // this process if their parent dies. This is effectively a sub-reaper. - // SAFETY: prctl is async-signal-safe. - unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) }; - - // SAFETY: All operations below simply _exit and therefore there - // are no issues with dangling file descriptor handles. - unsafe { close_all_fds() }; - - let mut sigset = core::mem::MaybeUninit::::uninit(); - // SAFETY: sigset is on the stack and we are initializing it. - unsafe { - libc::sigemptyset(sigset.as_mut_ptr()); - libc::sigaddset(sigset.as_mut_ptr(), libc::SIGTERM); - libc::sigaddset(sigset.as_mut_ptr(), libc::SIGCHLD); - libc::sigprocmask(libc::SIG_BLOCK, sigset.as_ptr(), core::ptr::null_mut()); - } - - loop { - // Reap all exited children. - loop { - let mut status = 0; - // SAFETY: The status is on the stack and waitpid is otherwise - // safe to call. - let res = unsafe { libc::waitpid(-1, &raw mut status, libc::WNOHANG) }; - if res == pid { - if libc::WIFEXITED(status) { - exit(libc::WEXITSTATUS(status)); - } else if libc::WIFSIGNALED(status) { - // Try to exit with the same signal as the child. - // SAFETY: The sigset was previously allocated and used on the stack. - unsafe { - libc::sigprocmask( - libc::SIG_UNBLOCK, - sigset.as_ptr(), - core::ptr::null_mut(), - ) - }; - // SAFETY: It's always safe to raise and as a fallback we _exit below. - unsafe { libc::raise(libc::WTERMSIG(status)) }; - // We shouldn't get here, but it's a fallback in case. - exit(libc::WTERMSIG(status)); - } - } else if res <= 0 { - // SAFETY: We just called a libc function that failed. - if res == -1 && unsafe { *libc::__errno_location() } != libc::EINTR { - exit(255); - } - // Break the reaping loop to wait for signals. - break; - } - } - - let mut siginfo = core::mem::MaybeUninit::::uninit(); - // SAFETY: sigset is initialized and siginfo is on the stack. - let sig = unsafe { libc::sigwaitinfo(sigset.as_ptr(), siginfo.as_mut_ptr()) }; - - if sig == libc::SIGTERM { - // Pass the SIGTERM on, so the action gets the grace the - // worker's kill_grace_ms promises to write its own - // cleanup. The stub used to turn it into SIGKILL, which - // ended the action at once under namespaces while the - // same action got its grace without them. The hard kill - // is a SIGKILL to the stub itself: as this PID - // namespace's init, its death takes every process in - // the namespace with it. - // SAFETY: pid is valid and we are sending a signal. - unsafe { libc::kill(pid, libc::SIGTERM) }; - } - } - } - _ => Err(Error::last_os_error()), - } + // Fork to enter the PID namespace: the action is its init, and the + // stub outside it is the action's parent, as without namespaces. + // SAFETY: We are in pre_exec and report_fd is the report pipe's write end. + unsafe { crate::action_stub::fork_stub(report_fd) } } diff --git a/nativelink-worker/src/persistent_worker/live_worker.rs b/nativelink-worker/src/persistent_worker/live_worker.rs index b8501191cf..12dc818922 100644 --- a/nativelink-worker/src/persistent_worker/live_worker.rs +++ b/nativelink-worker/src/persistent_worker/live_worker.rs @@ -100,10 +100,16 @@ impl LiveWorker { // safely to a single request. .stderr(Stdio::null()) .kill_on_drop(true); - // Its own process group, so the resource sampler can attribute the - // process and its children to the request being served. + // The leader of its own session, so the resource sampler can + // attribute the process and its children to the request being + // served (`crate::process_tree`). Registered before the + // namespace hook, so the stub and what it starts are in the session. #[cfg(unix)] - cmd.process_group(0); + // SAFETY: become_session_leader only calls setsid, which is + // async-signal-safe. + unsafe { + cmd.pre_exec(crate::process_tree::become_session_leader); + } #[cfg(target_os = "linux")] if namespaced { use std::os::unix::ffi::OsStrExt; @@ -118,6 +124,9 @@ impl LiveWorker { None, &working_dir_c, &action_dir_c, + // No report: a pooled process outlives every + // request, and each is measured from its own pid. + -1, ) }); } @@ -166,7 +175,7 @@ impl LiveWorker { self.wire_format } - /// The worker process's id, which is also its process group. + /// The worker process's id, which is also its session's. pub fn pid(&self) -> Option { self.child.id() } diff --git a/nativelink-worker/src/process_tree.rs b/nativelink-worker/src/process_tree.rs new file mode 100644 index 0000000000..9a67034979 --- /dev/null +++ b/nativelink-worker/src/process_tree.rs @@ -0,0 +1,266 @@ +// Copyright 2026 The NativeLink Authors. All rights reserved. +// +// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// See LICENSE file for details +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! An action's processes, as the worker measures and ends them: every +//! descendant of the action's stub (`crate::action_stub`). +//! +//! Neither a process group nor a session is enough. A job-control shell puts +//! each job in a group of its own (`set -m`, which Bazel's `test-setup.sh` +//! does to every test), and a process that detaches its children calls +//! `setsid` for each (Node's `spawn(..., { detached: true })`, a daemonizing +//! server), so a test and everything under it can leave both. Ancestry is +//! the one membership a process cannot leave: the stub is a child +//! subreaper, so an orphan in the action is reparented to it rather than to +//! the worker, and every process of the action stays a descendant of the +//! stub for as long as the stub lives. +//! +//! The stub leads a new session too (`become_session_leader`), so the +//! action never shares the worker's, and if the stub itself is killed the +//! session is what the worker can still reach. + +use std::collections::{HashMap, HashSet}; +use std::io::Error; + +/// Makes the calling process the leader of a new session, and of a new +/// process group in it. For `pre_exec`: `setsid` is async-signal-safe. It +/// fails for a process that already leads a group, so the command must not +/// also ask for `process_group(0)`. +#[cfg(target_family = "unix")] +pub fn become_session_leader() -> Result<(), Error> { + // SAFETY: setsid takes no arguments and has no memory safety + // considerations. + if unsafe { libc::setsid() } == -1 { + return Err(Error::last_os_error()); + } + Ok(()) +} + +/// The fields of a `/proc//stat` line after `comm` (field 2), which can +/// contain spaces and parentheses, so the fields are counted from the final +/// `)`: state(0) ppid(1) pgrp(2) session(3) ... +fn after_comm(stat: &str) -> Option> { + Some(stat.rsplit_once(')')?.1.split_whitespace()) +} + +/// Parses the parent pid (`ppid`, field 4) from the contents of a +/// `/proc//stat` line. Returns `None` when the line is malformed. +pub fn parse_ppid_from_stat(stat: &str) -> Option { + after_comm(stat)?.nth(1)?.parse().ok() +} + +/// Parses the session id (`session`, field 6) from the contents of a +/// `/proc//stat` line. Returns `None` when the line is malformed. +pub fn parse_session_from_stat(stat: &str) -> Option { + after_comm(stat)?.nth(3)?.parse().ok() +} + +/// Whether a `/proc//stat` line is a zombie's: it has exited and holds +/// nothing but its entry until it is reaped. +pub fn is_zombie_stat(stat: &str) -> bool { + after_comm(stat).and_then(|mut fields| fields.next()) == Some("Z") +} + +/// Every process on the host this worker can see, as its pid and +/// `/proc//stat` line. A process that exits between the listing and +/// the read is skipped. +#[cfg(target_os = "linux")] +fn all_processes() -> Vec<(u32, String)> { + let Ok(entries) = std::fs::read_dir("/proc") else { + return Vec::new(); + }; + entries + .flatten() + .filter_map(|entry| entry.file_name().to_string_lossy().parse::().ok()) + .filter_map(|pid| { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + Some((pid, stat)) + }) + .collect() +} + +/// `root` and every descendant of it, as its pid and `/proc//stat` +/// line, `root` first when it is still there. One pass over `/proc`: a +/// process forked after its parent was read is seen by the next pass, which +/// is why the kills below repeat. +#[cfg(target_os = "linux")] +pub fn tree_members(root: u32) -> Vec<(u32, String)> { + let processes = all_processes(); + let mut children: HashMap> = HashMap::new(); + let mut root_index = None; + for (index, (pid, stat)) in processes.iter().enumerate() { + if *pid == root { + root_index = Some(index); + } else if let Some(ppid) = parse_ppid_from_stat(stat) { + children.entry(ppid).or_default().push(index); + } + } + let mut members = Vec::new(); + if let Some(index) = root_index { + members.push(processes[index].clone()); + } + let mut pending = vec![root]; + let mut seen = HashSet::from([root]); + while let Some(parent) = pending.pop() { + for &index in children.get(&parent).map_or(&[][..], Vec::as_slice) { + let (pid, stat) = &processes[index]; + if seen.insert(*pid) { + members.push((*pid, stat.clone())); + pending.push(*pid); + } + } + } + members +} + +/// Every process in session `sid`, as its pid and `/proc//stat` line. +/// What the worker can still reach of an action whose stub was killed: its +/// orphans then belong to the worker, and the session is all that marks +/// them as the action's. +#[cfg(target_os = "linux")] +pub fn session_members(sid: u32) -> Vec<(u32, String)> { + all_processes() + .into_iter() + .filter(|(_, stat)| parse_session_from_stat(stat) == Some(sid)) + .collect() +} + +/// How many times a SIGKILL lists the processes again: a member can fork +/// between the listing and its kill, and its child is only seen by the next +/// listing. Whatever is still alive after this many is reported by the +/// caller's own checks. +#[cfg(target_os = "linux")] +const KILL_PASSES: usize = 16; + +/// Sends `signal` to `pid` through a pidfd, after reading again that its +/// parent is `parent`: between the listing and the signal the process could +/// have been reaped and its pid given to an unrelated process, and the +/// pidfd pins the process the check read. Falls back to `kill` on a kernel +/// without pidfds (before 5.3), where a pid is not reused while its parent, +/// in the tree, has not reaped it. +#[cfg(target_os = "linux")] +fn signal_checked(pid: u32, parent: u32, signal: i32) -> bool { + let Ok(raw_pid) = libc::pid_t::try_from(pid) else { + return false; + }; + // SAFETY: pidfd_open takes integers and returns a new descriptor or -1. + let pidfd = unsafe { libc::syscall(libc::SYS_pidfd_open, raw_pid, 0) }; + if pidfd < 0 { + // SAFETY: kill only takes integers and has no memory safety + // considerations; a pid that exited is reported as ESRCH. + return unsafe { libc::kill(raw_pid, signal) } == 0; + } + let Ok(pidfd) = libc::c_int::try_from(pidfd) else { + return false; + }; + let still_ours = std::fs::read_to_string(format!("/proc/{pid}/stat")) + .ok() + .and_then(|stat| parse_ppid_from_stat(&stat)) + == Some(parent); + let signalled = still_ours + // SAFETY: the pidfd is ours and open; a null siginfo is the plain + // kill(2) form. + && unsafe { + libc::syscall( + libc::SYS_pidfd_send_signal, + pidfd, + signal, + core::ptr::null::(), + 0, + ) + } == 0; + // SAFETY: the pidfd was opened above and is closed once. + unsafe { libc::close(pidfd) }; + signalled +} + +/// Sends `signal` to every live descendant of `root`, not to `root` itself, +/// and returns whether any process got it. For SIGKILL the tree is listed +/// again until no live descendant is left (or [`KILL_PASSES`]), since a +/// descendant can fork while it is being killed. `root` is the action's +/// stub, which ends the action itself once its child is gone, or a pooled +/// worker process that outlives the request. +#[cfg(target_os = "linux")] +pub fn signal_descendants(root: u32, signal: i32) -> bool { + let passes = if signal == libc::SIGKILL { + KILL_PASSES + } else { + 1 + }; + let mut signalled = false; + for _ in 0..passes { + let mut live = false; + for (pid, stat) in tree_members(root) { + if pid == root || is_zombie_stat(&stat) { + continue; + } + let Some(parent) = parse_ppid_from_stat(&stat) else { + continue; + }; + live = true; + signalled |= signal_checked(pid, parent, signal); + } + if !live { + break; + } + } + signalled +} + +/// Sends `signal` to every live process in session `sid`, and returns +/// whether any process got it, listing the session again for SIGKILL as +/// [`signal_descendants`] does. The fallback for an action whose stub is +/// gone. +#[cfg(target_os = "linux")] +pub fn signal_session(sid: u32, signal: i32) -> bool { + let passes = if signal == libc::SIGKILL { + KILL_PASSES + } else { + 1 + }; + let mut signalled = false; + for _ in 0..passes { + let mut live = false; + for (pid, stat) in session_members(sid) { + if is_zombie_stat(&stat) { + continue; + } + let Ok(pid) = libc::pid_t::try_from(pid) else { + continue; + }; + live = true; + // SAFETY: kill only takes integers and has no memory safety + // considerations; a pid that exited since the listing is + // reported as ESRCH and not acted on. + if unsafe { libc::kill(pid, signal) } == 0 { + signalled = true; + } + } + if !live { + break; + } + } + signalled +} + +/// Without `/proc`, the session's own group: what a group kill reached +/// before, which is every process that did not move to a group of its own. +#[cfg(all(target_family = "unix", not(target_os = "linux")))] +pub fn signal_session(sid: u32, signal: i32) -> bool { + let Ok(pgid) = i32::try_from(sid) else { + return false; + }; + // SAFETY: killpg only takes integers and has no memory safety + // considerations; a stale group id is reported as ESRCH, not acted on. + unsafe { libc::killpg(pgid, signal) == 0 } +} diff --git a/nativelink-worker/src/running_actions_manager.rs b/nativelink-worker/src/running_actions_manager.rs index 9c76c6e17d..431c42d65f 100644 --- a/nativelink-worker/src/running_actions_manager.rs +++ b/nativelink-worker/src/running_actions_manager.rs @@ -155,8 +155,8 @@ fn tag_missing_input(err: Error) -> Error { /// How long a killed action's output pipes, or those of one that left /// stragglers behind, get to drain. They close when every process holding -/// them is gone, which the SIGKILL to the group sees to; a process that -/// left the group could still hold them, so that wait is bounded, and what +/// them is gone, which the SIGKILL to the session sees to; a process that +/// left the session could still hold them, so that wait is bounded, and what /// was read by then is kept. An action that exited cleanly with nothing /// left behind is read to the end. const OUTPUT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); @@ -192,19 +192,19 @@ pub fn classify_outcome( } #[cfg(target_os = "linux")] -type SpawnedChild = crate::namespace_utils::MaybeNamespacedChild; +type SpawnedChild = crate::namespace_utils::ActionStubChild; #[cfg(not(target_os = "linux"))] type SpawnedChild = process::Child; -/// SIGKILL through the child wrapper, which takes the whole process group -/// with it, logged if it fails. +/// SIGKILL through the child wrapper, which takes every process of the +/// action with it, logged if it fails. async fn hard_kill(child: &mut SpawnedChild, why: &str) { if let Err(err) = child.kill().await { error!(?err, why, "Could not kill process in RunningActionsManager"); } } -/// Starts stopping an action. With a grace period the process group gets +/// Starts stopping an action. With a grace period the action's processes get /// SIGTERM and the returned future fires when SIGKILL is due; without one /// SIGKILL goes out now and the future never fires. async fn stop_action( @@ -215,9 +215,9 @@ async fn stop_action( debug!(grace_ms = grace.as_millis(), why, "Stopping action"); #[cfg(target_os = "linux")] if !grace.is_zero() - && let Some(pgid) = child.id() + && let Some(stub) = child.id() { - signal_process_group(pgid, libc::SIGTERM); + crate::process_tree::signal_descendants(stub, libc::SIGTERM); return tokio::time::sleep(grace).boxed().fuse(); } hard_kill(child, why).await; @@ -288,16 +288,27 @@ struct Ceilings { #[cfg(target_os = "linux")] const DISK_SAMPLE_EVERY: u32 = 40; +/// The processes a sample reads: every descendant of `root`, and `root` +/// itself when it is the tool. An action's root is its stub, a fork of the +/// worker whose pages are the worker's, so it is left out; a pooled worker's +/// root is the tool. +#[cfg(target_os = "linux")] +#[derive(Debug, Clone, Copy)] +struct SampledTree { + root: u32, + include_root: bool, +} + #[cfg(target_os = "linux")] fn start_action_resource_usage_sampler( - pgid: u32, + tree: SampledTree, ceilings: Ceilings, disk_directory: Option<(PathBuf, SystemTime)>, ) -> ActionResourceUsageSampler { let (stop_tx, stop_rx) = watch::channel(false); let handle = background_spawn!( "action_resource_usage_sampler", - sample_action_resource_usage(pgid, stop_rx, ceilings, disk_directory) + sample_action_resource_usage(tree, stop_rx, ceilings, disk_directory) ); ActionResourceUsageSampler { stop_tx, handle } } @@ -386,22 +397,6 @@ fn start_disk_sample( })) } -/// A signal to every process in the action's group. The action is its own -/// group leader, so this reaches the children a shell forked. The SIGKILL -/// at the end of a kill goes through the child wrapper, which takes the -/// group with it (or tells the stub, namespaced); this is for the SIGTERM -/// that gives an action its grace, and for the stragglers left once the -/// action's own process has exited. Returns whether any process got it. -#[cfg(target_family = "unix")] -fn signal_process_group(pgid: u32, signal: i32) -> bool { - let Ok(pgid) = i32::try_from(pgid) else { - return false; - }; - // SAFETY: killpg only takes integers and has no memory safety - // considerations; a stale group id is reported as ESRCH, not acted on. - unsafe { libc::killpg(pgid, signal) == 0 } -} - #[cfg(target_os = "linux")] async fn finish_action_resource_usage_sampler( sampler: ActionResourceUsageSampler, @@ -412,7 +407,7 @@ async fn finish_action_resource_usage_sampler( #[cfg(target_os = "linux")] async fn sample_action_resource_usage( - pgid: u32, + tree: SampledTree, mut stop_rx: watch::Receiver, mut ceilings: Ceilings, disk_directory: Option<(PathBuf, SystemTime)>, @@ -425,17 +420,18 @@ async fn sample_action_resource_usage( // The walk in flight, if one is; polled, never awaited, from the loop. let mut disk_sample: Option> = None; // CPU time is cumulative per process and a process that exits stops - // appearing, so summing the live group at the end would lose everything + // appearing, so summing the live tree at the end would lose everything // short-lived. Keep the last figure seen for each pid and total them at - // the end instead. + // the end instead. An action's stub reports the exact total when it + // exits, which replaces this figure; a pooled worker has no stub. let mut cpu_ticks_by_pid = HashMap::new(); - // What the group had already spent when sampling began. A one-shot + // What the tree had already spent when sampling began. A one-shot // process starts at nothing; a pooled process carries every request it // served before, which is not this action's. let mut baseline_ticks: Option = None; let sample = |peak_memory_kb: &mut u64, cpu_ticks_by_pid: &mut HashMap| { - let observed = sample_process_group(pgid, cpu_ticks_by_pid); + let observed = sample_tree(tree, cpu_ticks_by_pid); if let Some(memory_kb) = observed { *peak_memory_kb = (*peak_memory_kb).max(memory_kb); } @@ -450,9 +446,9 @@ async fn sample_action_resource_usage( if let Some(memory_kb) = observed { last_memory_kb = memory_kb; } - if observed.is_none() && !Path::new(&format!("/proc/{pgid}")).exists() { - // The group leader has been reaped and no member process remains, - // so the action is finished. + if observed.is_none() && !Path::new(&format!("/proc/{}", tree.root)).exists() { + // The root has been reaped and nothing under it remains, so the + // action is finished. break; } // Two consecutive samples over the ceiling, so one spike between @@ -556,33 +552,17 @@ fn ticks_to_millis(ticks: u64) -> u64 { ticks.saturating_mul(1_000) / hz } -/// Sums the resident memory of every process in the action's process group. -/// -/// The action is spawned as its own process-group leader (see -/// `command_builder.process_group(0)` in `inner_execute`), so `pgid` equals -/// the spawned child's pid and every descendant inherits it. Sampling by -/// process group — rather than walking the parent/child tree from the spawned -/// pid — is required because an intermediate shell frequently exits and -/// reparents the real workload to the worker (PID 1); the tree walk then sees -/// only a childless zombie and reports zero. Process-group membership is -/// inherited and survives reparenting, and excludes the worker's own group. -/// Returns `None` when no member process can be read (the group is empty). +/// Sums the resident memory of every process in the sampled tree +/// (`crate::process_tree`): the stub's descendants, which an orphan cannot +/// leave because the stub is its subreaper, whatever session or group it +/// moved to. +/// Returns `None` when no process can be read (the tree is empty). #[cfg(target_os = "linux")] -fn sample_process_group(pgid: u32, cpu_ticks_by_pid: &mut HashMap) -> Option { - let Ok(entries) = std::fs::read_dir("/proc") else { - return None; - }; - +fn sample_tree(tree: SampledTree, cpu_ticks_by_pid: &mut HashMap) -> Option { let mut total_kb = 0; let mut found_any_process = false; - for entry in entries.flatten() { - let Ok(member_pid) = entry.file_name().to_string_lossy().parse::() else { - continue; - }; - let Ok(stat) = std::fs::read_to_string(format!("/proc/{member_pid}/stat")) else { - continue; - }; - if parse_pgid_from_stat(&stat) != Some(pgid) { + for (member_pid, stat) in crate::process_tree::tree_members(tree.root) { + if member_pid == tree.root && !tree.include_root { continue; } if let Some(ticks) = parse_cpu_ticks_from_stat(&stat) { @@ -601,7 +581,7 @@ fn sample_process_group(pgid: u32, cpu_ticks_by_pid: &mut HashMap) -> } /// The process's share of its resident pages: `Pss` from `smaps_rollup` -/// divides a page shared by N processes N ways, so summed over the group it +/// divides a page shared by N processes N ways, so summed over the session it /// counts a forked parent's pages, shared libraries and mapped inputs once. /// `VmRSS` counts them once per process, and a test runner that forks eight /// children would sample at several times what the cgroup charges. Falls @@ -625,29 +605,16 @@ pub fn parse_kb_field(text: &str, field: &str) -> Option { }) } -/// Parses the process group id (`pgrp`, field 5) from the contents of a -/// `/proc//stat` line. -/// -/// `comm` (field 2) can contain spaces and parentheses, so fields are parsed -/// relative to the final `)` to avoid miscounting. Returns `None` when the -/// line is malformed. -#[cfg(target_os = "linux")] -pub fn parse_pgid_from_stat(stat: &str) -> Option { - let after_comm = stat.rsplit_once(')')?.1; - // Fields after the final ')': state(0) ppid(1) pgrp(2) ... - after_comm.split_whitespace().nth(2)?.parse().ok() -} - /// Sums user and system CPU ticks (`utime` + `stime`, fields 14 and 15) from /// the contents of a `/proc//stat` line. /// -/// Parsed relative to the final `)` for the same reason as the pgid above: +/// Parsed relative to the final `)`, as `parse_session_from_stat` does: /// `comm` can contain spaces and parentheses. Returns `None` when the line is /// malformed. /// /// `cutime`/`cstime` are deliberately not included. They only cover children /// this process has already reaped, so adding them here would double count -/// every child that is still its own entry in the group. +/// every child that is still its own entry in the session. #[cfg(target_os = "linux")] pub fn parse_cpu_ticks_from_stat(stat: &str) -> Option { let after_comm = stat.rsplit_once(')')?.1; @@ -2536,9 +2503,12 @@ impl RunningActionImpl { // peak memory is the whole process, which holds // what earlier requests left in it. #[cfg(target_os = "linux")] - let sampler = lease.worker().pid().map(|pgid| { + let sampler = lease.worker().pid().map(|pid| { start_action_resource_usage_sampler( - pgid, + SampledTree { + root: pid, + include_root: true, + }, Ceilings { memory_limit_kb: None, disk_limit_kb: None, @@ -2698,13 +2668,37 @@ impl RunningActionImpl { command_builder.env(name, value); } + // The spawned child leads a new session on every unix, so the action + // never shares the worker's. On Linux it then forks the action and + // stays as the action's stub (`crate::action_stub`): the sampler + // and the kills reach every descendant of the stub, which holds + // everything the action started whatever session or group it moved + // to, and the session is what is left to reach if the stub dies. + #[cfg(target_family = "unix")] + // SAFETY: become_session_leader only calls setsid, which is + // async-signal-safe. + unsafe { + command_builder.pre_exec(crate::process_tree::become_session_leader); + } + // Sandboxing of the command if we are running on Linux, this resolves issues where // children can spawn children and also provides better reproducibility. #[cfg(target_os = "linux")] + let report_pipe = crate::action_stub::ReportPipe::new() + .err_tip(|| "Creating the action stub's report pipe")?; + #[cfg(target_os = "linux")] { + use std::os::fd::AsRawFd; + let report_fd = report_pipe.write.as_raw_fd(); let use_namespaces = self.running_actions_manager.use_namespaces; - if !matches!(use_namespaces, UseNamespaces::No) { + if matches!(use_namespaces, UseNamespaces::No) { + // SAFETY: fork_stub is async-signal-safe and meant for + // pre_exec; report_fd is the pipe's write end. + unsafe { + command_builder.pre_exec(move || crate::action_stub::fork_stub(report_fd)); + } + } else { let (mount, isolate_tmp) = match use_namespaces { UseNamespaces::YesAndMount { isolate_tmp } => (true, isolate_tmp), UseNamespaces::No | UseNamespaces::Yes => (false, false), @@ -2739,23 +2733,23 @@ impl RunningActionImpl { tmp_directory.as_deref(), &root_action_directory, &action_directory, + report_fd, ) }); } } } - // Run the action as its own process-group leader (pgid == child - // pid) on every unix. The resource-usage sampler attributes memory - // by process group, and the kill at exit ends the group, so this - // keeps the whole action together, including processes reparented - // to the worker when an intermediate shell exits, and never - // conflates it with the worker's own group. - #[cfg(target_family = "unix")] - command_builder.process_group(0); - let mut child_process = command_builder .spawn() .err_tip(|| format!("Could not execute command {args:?}"))?; + // The stub holds the write end now; the worker's copy would keep the + // pipe open past the stub. + #[cfg(target_os = "linux")] + let report_read = { + let crate::action_stub::ReportPipe { read, write } = report_pipe; + drop(write); + read + }; // Ours until the wait below collects it; the reaper leaves it alone. let _owned_child = crate::reaper::OwnedChild::new(child_process.id()); let stdout_reader = child_process @@ -2768,15 +2762,7 @@ impl RunningActionImpl { .err_tip(|| "Expected stderr to exist on command this should never happen")?; #[cfg(target_os = "linux")] - // Wrap the child process to send SIGTERM rather than SIGKILL if namespaced to - // prevent zombie processes. - let child_process = crate::namespace_utils::MaybeNamespacedChild::new( - !matches!( - self.running_actions_manager.use_namespaces, - UseNamespaces::No, - ), - child_process, - ); + let child_process = crate::namespace_utils::ActionStubChild::new(child_process); // The reservation the scheduler placed this action by, if the // worker is asked to hold it to that number. @@ -2823,7 +2809,7 @@ impl RunningActionImpl { // below stays pending instead of resolving closed. let mut over_limit_keepalive = Some(over_limit_tx); #[cfg(target_os = "linux")] - let mut maybe_resource_usage_sampler = child_process.id().map(|pgid| { + let mut maybe_resource_usage_sampler = child_process.id().map(|stub| { let memory_limit_kb = memory_reservation.map(|(_, limit_kb)| limit_kb); let disk_limit_kb = disk_enforcement.and_then(|(_, limit_kb)| limit_kb); let over_limit_tx = (memory_limit_kb.is_some() || disk_limit_kb.is_some()) @@ -2838,7 +2824,10 @@ impl RunningActionImpl { (directory, since) }); start_action_resource_usage_sampler( - pgid, + SampledTree { + root: stub, + include_root: false, + }, Ceilings { memory_limit_kb, disk_limit_kb, @@ -2851,10 +2840,10 @@ impl RunningActionImpl { let _ = &disk_enforcement; let mut over_limit_fut = over_limit_rx.fuse(); - // The group to end when the action's own process has exited; taken - // before the guard owns the child. + // The session the spawned child leads, to end when the action's own + // process has exited; taken before the guard owns the child. #[cfg(target_family = "unix")] - let action_pgid = child_process.id(); + let action_sid = child_process.id(); let mut child_process_guard = guard(child_process, |mut child_process| { let result: Result, std::io::Error> = child_process.try_wait(); @@ -3029,24 +3018,35 @@ impl RunningActionImpl { } else { timer.measure(); } - // The action is over; anything still running in its - // group is a straggler (a background job, a helper that - // daemonized) holding the pipes open, and without this - // the action would sit here until its timeout. The - // sandboxes Bazel runs locally end them the same way. - #[cfg(target_family = "unix")] - let had_stragglers = action_pgid - .is_some_and(|pgid| signal_process_group(pgid, libc::SIGKILL)); + // The action is over; anything it left running is a + // straggler (a background job, a helper that daemonized) + // holding the pipes open, and without an end to it the + // action would sit here until its timeout. On Linux the + // stub has already killed and reaped every one before it + // exited, as the sandboxes Bazel runs locally do, and + // says how many there were; the session is what is left + // to end when the stub itself was killed and its orphans + // came to the worker. Elsewhere the session is all there + // is. + #[cfg(target_os = "linux")] + let stub_report = crate::action_stub::ReportPipe::read_report(&report_read); + #[cfg(target_os = "linux")] + let had_stragglers = stub_report.is_some_and(|report| report.stragglers > 0) + | action_sid + .is_some_and(|sid| crate::process_tree::signal_session(sid, libc::SIGKILL)); + #[cfg(all(target_family = "unix", not(target_os = "linux")))] + let had_stragglers = action_sid + .is_some_and(|sid| crate::process_tree::signal_session(sid, libc::SIGKILL)); #[cfg(not(target_family = "unix"))] let had_stragglers = false; // An action that exited on its own with nothing left in - // its group closes its pipes itself, and its output is + // its session closes its pipes itself, and its output is // the result, so that read runs to the end: a reader // slowed by a busy spill must not lose bytes to a clock. // The wait is bounded only when there was something to // hold the pipes: a killed action, whose output is a // bonus, or stragglers, one of which may have left the - // group. What was read by then is kept. + // session. What was read by then is kept. let mut drain = core::pin::pin!(async { tokio::join!(all_stdout_fut, all_stderr_fut) }); let (maybe_all_stdout, maybe_all_stderr) = if killed_action || had_stragglers { tokio::select! { @@ -3072,11 +3072,20 @@ impl RunningActionImpl { ); #[cfg(target_os = "linux")] - let sampled_usage = match maybe_resource_usage_sampler.take() { - Some(sampler) => finish_action_resource_usage_sampler(sampler) - .await - .unwrap_or_default(), - None => SampledResourceUsage::default(), + let sampled_usage = { + let mut sampled_usage = match maybe_resource_usage_sampler.take() { + Some(sampler) => finish_action_resource_usage_sampler(sampler) + .await + .unwrap_or_default(), + None => SampledResourceUsage::default(), + }; + // The kernel's total of everything the stub reaped, + // exact where the samples miss what ran between + // two of them. + if let Some(report) = stub_report { + sampled_usage.cpu_time_ms = report.cpu_time_ms(); + } + sampled_usage }; #[cfg(not(target_os = "linux"))] let sampled_usage = SampledResourceUsage::default(); diff --git a/nativelink-worker/tests/namespace_utils_test.rs b/nativelink-worker/tests/namespace_utils_test.rs index 5648f7f846..3316c2f23a 100644 --- a/nativelink-worker/tests/namespace_utils_test.rs +++ b/nativelink-worker/tests/namespace_utils_test.rs @@ -62,6 +62,7 @@ fn namespaced_sh(script: &str, isolate_tmp: bool, root_path: &Path, action_path: tmp_dir_c.as_deref(), &root_dir_c, &action_dir_c, + -1, ) }); } @@ -288,7 +289,7 @@ async fn test_configure_namespace() -> Result<(), Error> { // child, and the original child process will continue to execute the command. unsafe { command.pre_exec(move || { - namespace_utils::configure_namespace(false, None, &root_dir, &action_dir) + namespace_utils::configure_namespace(false, None, &root_dir, &action_dir, -1) }); } @@ -347,7 +348,7 @@ async fn test_configure_namespace_mount_isolation() -> Result<(), Error> { unsafe { command.pre_exec(move || { - namespace_utils::configure_namespace(true, None, &root_dir_c, &action1_dir_c) + namespace_utils::configure_namespace(true, None, &root_dir_c, &action1_dir_c, -1) }); } @@ -365,7 +366,7 @@ async fn test_configure_namespace_mount_isolation() -> Result<(), Error> { let action1_dir_c = CString::new(action1_path.to_str().unwrap()).unwrap(); unsafe { command_access.pre_exec(move || { - namespace_utils::configure_namespace(true, None, &root_dir_c, &action1_dir_c) + namespace_utils::configure_namespace(true, None, &root_dir_c, &action1_dir_c, -1) }); } let output_access = command_access.output()?; @@ -409,12 +410,12 @@ async fn test_namespaced_action_gets_sigterm_not_sigkill() -> Result<(), Error> // SAFETY: configure_namespace is async-signal-safe and intended for pre_exec. unsafe { command.pre_exec(move || { - namespace_utils::configure_namespace(false, None, &root_dir, &action_dir) + namespace_utils::configure_namespace(false, None, &root_dir, &action_dir, -1) }); } let child = command.spawn()?; let pgid = i32::try_from(child.id().unwrap()).unwrap(); - let mut namespaced_child = namespace_utils::MaybeNamespacedChild::new(true, child); + let mut namespaced_child = namespace_utils::ActionStubChild::new(child); // Let the shell install its trap and fork the sleep. tokio::time::sleep(Duration::from_millis(300)).await; @@ -459,12 +460,12 @@ async fn test_maybe_namespaced_child_kill_reaps_orphans() -> Result<(), Error> { // SAFETY: configure_namespace is async-signal-safe and intended for pre_exec. unsafe { command.pre_exec(move || { - namespace_utils::configure_namespace(false, None, &root_dir, &action_dir) + namespace_utils::configure_namespace(false, None, &root_dir, &action_dir, -1) }); } let child = command.spawn()?; - let mut namespaced_child = namespace_utils::MaybeNamespacedChild::new(true, child); + let mut namespaced_child = namespace_utils::ActionStubChild::new(child); // Give the shell time to fork the background sleep processes. tokio::time::sleep(Duration::from_millis(200)).await; @@ -509,7 +510,7 @@ async fn test_maybe_namespaced_child_non_namespaced_kill() -> Result<(), Error> command.arg("10"); // Sleep for 10 seconds, should be killed before then. let child = command.spawn()?; - let mut non_namespaced_child = namespace_utils::MaybeNamespacedChild::new(false, child); + let mut non_namespaced_child = namespace_utils::ActionStubChild::new(child); // Give the process a moment to start. tokio::time::sleep(Duration::from_millis(50)).await; @@ -541,12 +542,12 @@ async fn test_maybe_namespaced_child_namespaced_natural_exit() -> Result<(), Err // SAFETY: configure_namespace is async-signal-safe and intended for pre_exec. unsafe { command.pre_exec(move || { - namespace_utils::configure_namespace(false, None, &root_dir, &action_dir) + namespace_utils::configure_namespace(false, None, &root_dir, &action_dir, -1) }); } let child = command.spawn()?; - let mut namespaced_child = namespace_utils::MaybeNamespacedChild::new(true, child); + let mut namespaced_child = namespace_utils::ActionStubChild::new(child); // Wait for the child to exit naturally. let status = namespaced_child.wait().await?; @@ -570,12 +571,11 @@ async fn test_maybe_namespaced_child_try_wait() -> Result<(), Error> { let action_dir = CString::new("/tmp/action").unwrap(); unsafe { command_running.pre_exec(move || { - namespace_utils::configure_namespace(false, None, &root_dir, &action_dir) + namespace_utils::configure_namespace(false, None, &root_dir, &action_dir, -1) }); } let child_running = command_running.spawn()?; - let mut namespaced_child_running = - namespace_utils::MaybeNamespacedChild::new(true, child_running); + let mut namespaced_child_running = namespace_utils::ActionStubChild::new(child_running); // Give it a moment to start tokio::time::sleep(Duration::from_millis(50)).await; @@ -595,12 +595,11 @@ async fn test_maybe_namespaced_child_try_wait() -> Result<(), Error> { command_exited.args(["-c", &format!("exit {expected_exit_code}")]); unsafe { command_exited.pre_exec(move || { - namespace_utils::configure_namespace(false, None, &root_dir, &action_dir) + namespace_utils::configure_namespace(false, None, &root_dir, &action_dir, -1) }); } let child_exited = command_exited.spawn()?; - let mut namespaced_child_exited = - namespace_utils::MaybeNamespacedChild::new(true, child_exited); + let mut namespaced_child_exited = namespace_utils::ActionStubChild::new(child_exited); // Wait for it to exit tokio::time::sleep(Duration::from_millis(50)).await; diff --git a/nativelink-worker/tests/running_actions_manager_test.rs b/nativelink-worker/tests/running_actions_manager_test.rs index 9a2de33764..0adcfa3f76 100644 --- a/nativelink-worker/tests/running_actions_manager_test.rs +++ b/nativelink-worker/tests/running_actions_manager_test.rs @@ -6853,20 +6853,398 @@ exit 1 Ok(()) } - /// Regression for skipping the `pre_exec` hook when namespaces are off. - /// With namespaces disabled (the default) the action spawn goes through - /// `posix_spawn` instead of `fork`, and `process_group(0)` must still make - /// the child its own process-group leader (pgid == pid). + /// The action runs under its stub, which leads a session and a group of + /// its own: with namespaces off (the default) the spawn still runs the + /// `pre_exec` hooks that call `setsid` and fork the stub, so the + /// action's parent is neither the worker nor anything of the worker's + /// session, and its sid == pgid == its parent's pid. #[cfg(target_os = "linux")] #[nativelink_test] - async fn no_namespace_action_is_process_group_leader() -> Result<(), Box> - { + async fn no_namespace_action_runs_under_its_stub() -> Result<(), Box> { + const WORKER_ID: &str = "foo_worker_id"; + + fn test_monotonic_clock() -> SystemTime { + static CLOCK: AtomicU64 = AtomicU64::new(0); + monotonic_clock(&CLOCK) + } + + let (_, _, cas_store, ac_store) = setup_stores().await?; + let root_action_directory = make_temp_path("root_action_directory"); + fs::create_dir_all(&root_action_directory).await?; + + let running_actions_manager = Arc::new(RunningActionsManagerImpl::new_with_callbacks( + RunningActionsManagerArgs { + root_action_directory, + cas_store: cas_store.clone(), + ac_store: Some(Store::new(ac_store.clone())), + execution_configuration: ExecutionConfiguration::default(), + historical_store: Store::new(cas_store.clone()), + upload_action_result_config: &UploadActionResultConfig { + upload_ac_results_strategy: UploadCacheResultsStrategy::Never, + ..Default::default() + }, + max_action_timeout: Duration::MAX, + max_upload_timeout: Duration::from_secs(DEFAULT_MAX_UPLOAD_TIMEOUT), + max_download_timeout: Duration::from_secs(DEFAULT_MAX_DOWNLOAD_TIMEOUT), + max_cleanup_wait: Duration::from_secs(DEFAULT_MAX_CLEANUP_WAIT), + max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), + timeout_handled_externally: false, + active_input_leases: false, + directory_cache: None, + // Pin namespaces off so this exercises the no-pre_exec/posix_spawn + // path regardless of what the host kernel supports. + use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::No, + }, + Callbacks { + now_fn: test_monotonic_clock, + sleep_fn: |_duration| Box::pin(future::pending()), + }, + )?); + + // Print the shell's own pid (field 1), group (field 5) and session (field 6). + let command = Command { + arguments: vec![ + "sh".to_string(), + "-c".to_string(), + "read -r _ _ _ ppid pgrp sid _ < /proc/$$/stat; printf '%s %s %s' \"$ppid\" \"$pgrp\" \"$sid\"" + .to_string(), + ], + output_paths: vec![], + working_directory: ".".to_string(), + environment_variables: vec![EnvironmentVariable { + name: "PATH".to_string(), + value: env::var("PATH").unwrap(), + }], + ..Default::default() + }; + let command_digest = serialize_and_upload_message( + &command, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let input_root_digest = serialize_and_upload_message( + &Directory::default(), + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let action = Action { + command_digest: Some(command_digest.into()), + input_root_digest: Some(input_root_digest.into()), + ..Default::default() + }; + let action_digest = serialize_and_upload_message( + &action, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + + let execute_request = ExecuteRequest { + action_digest: Some(action_digest.into()), + digest_function: ProtoDigestFunction::Sha256.into(), + ..Default::default() + }; + let operation_id = OperationId::default().to_string(); + let running_action_impl = running_actions_manager + .create_and_add_action( + WORKER_ID.to_string(), + StartExecute { + request_metadata: None, + execute_request: Some(execute_request), + operation_id, + ..Default::default() + }, + ) + .await?; + + let action_result = run_action(running_action_impl.clone()).await?; + assert_eq!(action_result.exit_code, 0, "{action_result:?}"); + let stdout = cas_store + .as_ref() + .get_part_unchunked(action_result.stdout_digest, 0, None) + .await?; + let stdout = from_utf8(&stdout)?.trim().to_string(); + let ids: Vec<&str> = stdout.split(' ').collect(); + assert_eq!(ids.len(), 3, "expected 'ppid pgrp sid' output: {stdout}"); + assert!( + ids[0] == ids[1] && ids[1] == ids[2], + "the action's stub should lead its group and session (ppid pgrp sid = {stdout})" + ); + assert_ne!( + ids[0], + std::process::id().to_string(), + "the action's parent is its stub, not the worker" + ); + Ok(()) + } + + /// A job-control shell (`set -m`, as Bazel's `test-setup.sh` runs every + /// test) moves each job to a process group of its own. The job is still + /// the action's: the CPU it spends is in the action's measurement. When + /// the sampler matched the action's group, a test's usage read as the + /// waiting shell's, about nothing. + #[cfg(target_os = "linux")] + #[nativelink_test] + async fn job_control_job_is_measured() -> Result<(), Box> { + const WORKER_ID: &str = "foo_worker_id"; + + fn test_monotonic_clock() -> SystemTime { + static CLOCK: AtomicU64 = AtomicU64::new(0); + monotonic_clock(&CLOCK) + } + + let (_, _, cas_store, ac_store) = setup_stores().await?; + let root_action_directory = make_temp_path("root_action_directory"); + fs::create_dir_all(&root_action_directory).await?; + + let running_actions_manager = Arc::new(RunningActionsManagerImpl::new_with_callbacks( + RunningActionsManagerArgs { + root_action_directory, + cas_store: cas_store.clone(), + ac_store: Some(Store::new(ac_store.clone())), + execution_configuration: ExecutionConfiguration::default(), + historical_store: Store::new(cas_store.clone()), + upload_action_result_config: &UploadActionResultConfig { + upload_ac_results_strategy: UploadCacheResultsStrategy::Never, + ..Default::default() + }, + max_action_timeout: Duration::MAX, + max_upload_timeout: Duration::from_secs(DEFAULT_MAX_UPLOAD_TIMEOUT), + max_download_timeout: Duration::from_secs(DEFAULT_MAX_DOWNLOAD_TIMEOUT), + max_cleanup_wait: Duration::from_secs(DEFAULT_MAX_CLEANUP_WAIT), + max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), + timeout_handled_externally: false, + active_input_leases: false, + directory_cache: None, + // Pin namespaces off so this exercises the no-pre_exec/posix_spawn + // path regardless of what the host kernel supports. + use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::No, + }, + Callbacks { + now_fn: test_monotonic_clock, + sleep_fn: |_duration| Box::pin(future::pending()), + }, + )?); + + // Two seconds of busy loop in a background job, which `set -m` puts in its own group. + let command = Command { + arguments: vec![ + "bash".to_string(), + "-c".to_string(), + "set -m; (end=$((SECONDS + 2)); while [ $SECONDS -lt $end ]; do :; done) & wait" + .to_string(), + ], + output_paths: vec![], + working_directory: ".".to_string(), + environment_variables: vec![EnvironmentVariable { + name: "PATH".to_string(), + value: env::var("PATH").unwrap(), + }], + ..Default::default() + }; + let command_digest = serialize_and_upload_message( + &command, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let input_root_digest = serialize_and_upload_message( + &Directory::default(), + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let action = Action { + command_digest: Some(command_digest.into()), + input_root_digest: Some(input_root_digest.into()), + ..Default::default() + }; + let action_digest = serialize_and_upload_message( + &action, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + + let execute_request = ExecuteRequest { + action_digest: Some(action_digest.into()), + digest_function: ProtoDigestFunction::Sha256.into(), + ..Default::default() + }; + let operation_id = OperationId::default().to_string(); + let running_action_impl = running_actions_manager + .create_and_add_action( + WORKER_ID.to_string(), + StartExecute { + request_metadata: None, + execute_request: Some(execute_request), + operation_id, + ..Default::default() + }, + ) + .await?; + + let action_result = run_action(running_action_impl.clone()).await?; + assert_eq!(action_result.exit_code, 0, "{action_result:?}"); + let usage = running_action_impl + .resource_usage() + .expect("usage is reported for every executed action"); + assert!( + usage.cpu_time_ms >= 500, + "the job's busy loop is the action's CPU: {usage:?}" + ); + Ok(()) + } + + /// A job the action left running in a process group of its own (a + /// job-control shell's background job) is ended with the action: the + /// straggler kill reaches the whole session, not only the action's + /// group, which the job had left. + #[cfg(target_os = "linux")] + #[nativelink_test] + async fn job_control_straggler_is_killed() -> Result<(), Box> { + const WORKER_ID: &str = "foo_worker_id"; + + fn test_monotonic_clock() -> SystemTime { + static CLOCK: AtomicU64 = AtomicU64::new(0); + monotonic_clock(&CLOCK) + } + + let (_, _, cas_store, ac_store) = setup_stores().await?; + let root_action_directory = make_temp_path("root_action_directory"); + fs::create_dir_all(&root_action_directory).await?; + + let running_actions_manager = Arc::new(RunningActionsManagerImpl::new_with_callbacks( + RunningActionsManagerArgs { + root_action_directory, + cas_store: cas_store.clone(), + ac_store: Some(Store::new(ac_store.clone())), + execution_configuration: ExecutionConfiguration::default(), + historical_store: Store::new(cas_store.clone()), + upload_action_result_config: &UploadActionResultConfig { + upload_ac_results_strategy: UploadCacheResultsStrategy::Never, + ..Default::default() + }, + max_action_timeout: Duration::MAX, + max_upload_timeout: Duration::from_secs(DEFAULT_MAX_UPLOAD_TIMEOUT), + max_download_timeout: Duration::from_secs(DEFAULT_MAX_DOWNLOAD_TIMEOUT), + max_cleanup_wait: Duration::from_secs(DEFAULT_MAX_CLEANUP_WAIT), + max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), + timeout_handled_externally: false, + active_input_leases: false, + directory_cache: None, + // Pin namespaces off so this exercises the no-pre_exec/posix_spawn + // path regardless of what the host kernel supports. + use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::No, + }, + Callbacks { + now_fn: test_monotonic_clock, + sleep_fn: |_duration| Box::pin(future::pending()), + }, + )?); + + // Leave a sleep behind in a group of its own, and name its pid. + let command = Command { + arguments: vec![ + "bash".to_string(), + "-c".to_string(), + "set -m; sleep 300 >/dev/null 2>&1 & echo $!".to_string(), + ], + output_paths: vec![], + working_directory: ".".to_string(), + environment_variables: vec![EnvironmentVariable { + name: "PATH".to_string(), + value: env::var("PATH").unwrap(), + }], + ..Default::default() + }; + let command_digest = serialize_and_upload_message( + &command, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let input_root_digest = serialize_and_upload_message( + &Directory::default(), + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let action = Action { + command_digest: Some(command_digest.into()), + input_root_digest: Some(input_root_digest.into()), + ..Default::default() + }; + let action_digest = serialize_and_upload_message( + &action, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + + let execute_request = ExecuteRequest { + action_digest: Some(action_digest.into()), + digest_function: ProtoDigestFunction::Sha256.into(), + ..Default::default() + }; + let operation_id = OperationId::default().to_string(); + let running_action_impl = running_actions_manager + .create_and_add_action( + WORKER_ID.to_string(), + StartExecute { + request_metadata: None, + execute_request: Some(execute_request), + operation_id, + ..Default::default() + }, + ) + .await?; + + let action_result = run_action(running_action_impl.clone()).await?; + assert_eq!(action_result.exit_code, 0, "{action_result:?}"); + let stdout = cas_store + .as_ref() + .get_part_unchunked(action_result.stdout_digest, 0, None) + .await?; + let stdout = from_utf8(&stdout)?.trim().to_string(); + let pid: u32 = stdout.parse()?; + // Killed, it is gone or a zombie waiting to be reaped by whoever + // adopted it. + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + let alive = std::fs::read_to_string(format!("/proc/{pid}/stat")) + .is_ok_and(|stat| !nativelink_worker::process_tree::is_zombie_stat(&stat)); + if !alive { + break; + } + assert!( + std::time::Instant::now() < deadline, + "the job the action left behind (pid {pid}) is still running" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + Ok(()) + } + + /// A job that calls `setsid` (a detached spawn) is still the action's: its + /// CPU is the action's CPU. + #[cfg(target_os = "linux")] + #[nativelink_test] + async fn detached_job_is_measured() -> Result<(), Box> { const WORKER_ID: &str = "foo_worker_id"; fn test_monotonic_clock() -> SystemTime { static CLOCK: AtomicU64 = AtomicU64::new(0); monotonic_clock(&CLOCK) } + // The job leaves the session with util-linux's `setsid`; an environment + // without it (the coverage build's nix sandbox) has nothing to measure. + if which::which("setsid").is_err() { + eprintln!("skipping: setsid is not on PATH"); + return Ok(()); + } let (_, _, cas_store, ac_store) = setup_stores().await?; let root_action_directory = make_temp_path("root_action_directory"); @@ -6901,13 +7279,13 @@ exit 1 }, )?); - // Print the shell's own pid (field 1) and process-group id (field 5) - // from its /proc stat line; process_group(0) makes them equal. + // Two seconds of busy loop in a job that leaves the action's session, as a + // `detached: true` spawn does. let command = Command { arguments: vec![ - "sh".to_string(), + "bash".to_string(), "-c".to_string(), - "read -r pid _ _ _ pgrp _ < /proc/$$/stat; printf '%s %s' \"$pid\" \"$pgrp\"" + "setsid bash -c 'end=$((SECONDS + 2)); while [ $SECONDS -lt $end ]; do :; done' & wait" .to_string(), ], output_paths: vec![], @@ -6961,21 +7339,263 @@ exit 1 .await?; let action_result = run_action(running_action_impl.clone()).await?; - assert_eq!( - action_result.exit_code, 0, - "action should run to completion via posix_spawn" + assert_eq!(action_result.exit_code, 0, "{action_result:?}"); + let usage = running_action_impl + .resource_usage() + .expect("usage is reported for every executed action"); + assert!( + usage.cpu_time_ms >= 500, + "the detached job's busy loop is the action's CPU: {usage:?}" + ); + Ok(()) + } + + /// A job that calls `setsid` is still sampled: its memory is the action's. + #[cfg(target_os = "linux")] + #[nativelink_test] + async fn detached_job_memory_is_measured() -> Result<(), Box> { + const WORKER_ID: &str = "foo_worker_id"; + + fn test_monotonic_clock() -> SystemTime { + static CLOCK: AtomicU64 = AtomicU64::new(0); + monotonic_clock(&CLOCK) + } + // The job leaves the session with util-linux's `setsid` and holds its + // memory from python3; an environment without either (the coverage + // build's nix sandbox) has nothing to measure. + if which::which("setsid").is_err() || which::which("python3").is_err() { + eprintln!("skipping: setsid or python3 is not on PATH"); + return Ok(()); + } + + let (_, _, cas_store, ac_store) = setup_stores().await?; + let root_action_directory = make_temp_path("root_action_directory"); + fs::create_dir_all(&root_action_directory).await?; + + let running_actions_manager = Arc::new(RunningActionsManagerImpl::new_with_callbacks( + RunningActionsManagerArgs { + root_action_directory, + cas_store: cas_store.clone(), + ac_store: Some(Store::new(ac_store.clone())), + execution_configuration: ExecutionConfiguration::default(), + historical_store: Store::new(cas_store.clone()), + upload_action_result_config: &UploadActionResultConfig { + upload_ac_results_strategy: UploadCacheResultsStrategy::Never, + ..Default::default() + }, + max_action_timeout: Duration::MAX, + max_upload_timeout: Duration::from_secs(DEFAULT_MAX_UPLOAD_TIMEOUT), + max_download_timeout: Duration::from_secs(DEFAULT_MAX_DOWNLOAD_TIMEOUT), + max_cleanup_wait: Duration::from_secs(DEFAULT_MAX_CLEANUP_WAIT), + max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), + timeout_handled_externally: false, + active_input_leases: false, + directory_cache: None, + // Pin namespaces off so this exercises the no-pre_exec/posix_spawn + // path regardless of what the host kernel supports. + use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::No, + }, + Callbacks { + now_fn: test_monotonic_clock, + sleep_fn: |_duration| Box::pin(future::pending()), + }, + )?); + + // 256 MiB held for two seconds by a job that leaves the action's session. + let command = Command { + arguments: vec![ + "bash".to_string(), + "-c".to_string(), + "setsid python3 -c 'import time; x = b\"x\" * (256 << 20); time.sleep(2)' & wait" + .to_string(), + ], + output_paths: vec![], + working_directory: ".".to_string(), + environment_variables: vec![EnvironmentVariable { + name: "PATH".to_string(), + value: env::var("PATH").unwrap(), + }], + ..Default::default() + }; + let command_digest = serialize_and_upload_message( + &command, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let input_root_digest = serialize_and_upload_message( + &Directory::default(), + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let action = Action { + command_digest: Some(command_digest.into()), + input_root_digest: Some(input_root_digest.into()), + ..Default::default() + }; + let action_digest = serialize_and_upload_message( + &action, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + + let execute_request = ExecuteRequest { + action_digest: Some(action_digest.into()), + digest_function: ProtoDigestFunction::Sha256.into(), + ..Default::default() + }; + let operation_id = OperationId::default().to_string(); + let running_action_impl = running_actions_manager + .create_and_add_action( + WORKER_ID.to_string(), + StartExecute { + request_metadata: None, + execute_request: Some(execute_request), + operation_id, + ..Default::default() + }, + ) + .await?; + + let action_result = run_action(running_action_impl.clone()).await?; + assert_eq!(action_result.exit_code, 0, "{action_result:?}"); + let usage = running_action_impl + .resource_usage() + .expect("usage is reported for every executed action"); + assert!( + usage.peak_memory_kb >= 200 * 1024, + "the detached job's 256 MiB is the action's memory: {usage:?}" ); + Ok(()) + } + + /// A process the action detached into a session of its own and left + /// running is killed with the action, not left to hold what it holds. + #[cfg(target_os = "linux")] + #[nativelink_test] + async fn detached_straggler_is_killed() -> Result<(), Box> { + const WORKER_ID: &str = "foo_worker_id"; + + fn test_monotonic_clock() -> SystemTime { + static CLOCK: AtomicU64 = AtomicU64::new(0); + monotonic_clock(&CLOCK) + } + + let (_, _, cas_store, ac_store) = setup_stores().await?; + let root_action_directory = make_temp_path("root_action_directory"); + fs::create_dir_all(&root_action_directory).await?; + + let running_actions_manager = Arc::new(RunningActionsManagerImpl::new_with_callbacks( + RunningActionsManagerArgs { + root_action_directory, + cas_store: cas_store.clone(), + ac_store: Some(Store::new(ac_store.clone())), + execution_configuration: ExecutionConfiguration::default(), + historical_store: Store::new(cas_store.clone()), + upload_action_result_config: &UploadActionResultConfig { + upload_ac_results_strategy: UploadCacheResultsStrategy::Never, + ..Default::default() + }, + max_action_timeout: Duration::MAX, + max_upload_timeout: Duration::from_secs(DEFAULT_MAX_UPLOAD_TIMEOUT), + max_download_timeout: Duration::from_secs(DEFAULT_MAX_DOWNLOAD_TIMEOUT), + max_cleanup_wait: Duration::from_secs(DEFAULT_MAX_CLEANUP_WAIT), + max_cleanup_backoff: Duration::from_millis(DEFAULT_MAX_CLEANUP_BACKOFF), + timeout_handled_externally: false, + active_input_leases: false, + directory_cache: None, + // Pin namespaces off so this exercises the no-pre_exec/posix_spawn + // path regardless of what the host kernel supports. + use_namespaces: nativelink_worker::running_actions_manager::UseNamespaces::No, + }, + Callbacks { + now_fn: test_monotonic_clock, + sleep_fn: |_duration| Box::pin(future::pending()), + }, + )?); + + // Leave a sleep behind in a session of its own, and name its pid. + let command = Command { + arguments: vec![ + "bash".to_string(), + "-c".to_string(), + "setsid sleep 300 >/dev/null 2>&1 & echo $!".to_string(), + ], + output_paths: vec![], + working_directory: ".".to_string(), + environment_variables: vec![EnvironmentVariable { + name: "PATH".to_string(), + value: env::var("PATH").unwrap(), + }], + ..Default::default() + }; + let command_digest = serialize_and_upload_message( + &command, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let input_root_digest = serialize_and_upload_message( + &Directory::default(), + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + let action = Action { + command_digest: Some(command_digest.into()), + input_root_digest: Some(input_root_digest.into()), + ..Default::default() + }; + let action_digest = serialize_and_upload_message( + &action, + cas_store.as_pin(), + &mut DigestHasherFunc::Sha256.hasher(), + ) + .await?; + + let execute_request = ExecuteRequest { + action_digest: Some(action_digest.into()), + digest_function: ProtoDigestFunction::Sha256.into(), + ..Default::default() + }; + let operation_id = OperationId::default().to_string(); + let running_action_impl = running_actions_manager + .create_and_add_action( + WORKER_ID.to_string(), + StartExecute { + request_metadata: None, + execute_request: Some(execute_request), + operation_id, + ..Default::default() + }, + ) + .await?; + let action_result = run_action(running_action_impl.clone()).await?; + assert_eq!(action_result.exit_code, 0, "{action_result:?}"); let stdout = cas_store .as_ref() .get_part_unchunked(action_result.stdout_digest, 0, None) .await?; - let stdout = from_utf8(&stdout)?; - let (pid, pgrp) = stdout.split_once(' ').expect("expected 'pid pgrp' output"); - assert_eq!( - pid, pgrp, - "spawned process should be its own process-group leader (pid={pid} pgrp={pgrp})" - ); + let stdout = from_utf8(&stdout)?.trim().to_string(); + let pid: u32 = stdout.parse()?; + // Killed, it is gone or a zombie waiting to be reaped by whoever + // adopted it. + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + let alive = std::fs::read_to_string(format!("/proc/{pid}/stat")) + .is_ok_and(|stat| !nativelink_worker::process_tree::is_zombie_stat(&stat)); + if !alive { + break; + } + assert!( + std::time::Instant::now() < deadline, + "the job the action left behind (pid {pid}) is still running" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } Ok(()) } @@ -8610,29 +9230,58 @@ done Ok(()) } - #[cfg(target_os = "linux")] + #[cfg(target_family = "unix")] #[test] - fn parse_pgid_from_stat_extracts_field_after_comm() { - use nativelink_worker::running_actions_manager::parse_pgid_from_stat; - // /proc//stat layout: pid (comm) state ppid pgrp ... + fn parse_session_from_stat_extracts_field_after_comm() { + use nativelink_worker::process_tree::parse_session_from_stat; + // /proc//stat layout: pid (comm) state ppid pgrp session ... assert_eq!( - parse_pgid_from_stat("315 (perl) S 1 60 60 0 -1 0"), - Some(60) + parse_session_from_stat("315 (perl) S 1 60 42 0 -1 0"), + Some(42) ); } - #[cfg(target_os = "linux")] + #[cfg(target_family = "unix")] + #[test] + fn parse_ppid_from_stat_reads_the_parent_after_comm() { + use nativelink_worker::process_tree::parse_ppid_from_stat; + assert_eq!( + parse_ppid_from_stat("315 (a) b (c)) S 77 60 42 0 -1 0"), + Some(77) + ); + assert_eq!(parse_ppid_from_stat("315 (perl"), None); + } + + #[cfg(target_family = "unix")] #[test] - fn parse_pgid_from_stat_handles_comm_with_spaces_and_parens() { - use nativelink_worker::running_actions_manager::parse_pgid_from_stat; + fn parse_session_from_stat_handles_comm_with_spaces_and_parens() { + use nativelink_worker::process_tree::parse_session_from_stat; // `comm` may contain spaces and parentheses; parsing must be relative - // to the final ')'. Here pgrp == 777. + // to the final ')'. Here session == 778. assert_eq!( - parse_pgid_from_stat("1234 (weird )( name) R 1 777 777 0 -1 0"), - Some(777) + parse_session_from_stat("1234 (weird )( name) R 1 777 778 0 -1 0"), + Some(778) ); } + #[cfg(target_family = "unix")] + #[test] + fn parse_session_from_stat_rejects_malformed_input() { + use nativelink_worker::process_tree::parse_session_from_stat; + assert_eq!(parse_session_from_stat("no parenthesis here"), None); + assert_eq!(parse_session_from_stat("123 (only) S 1 2"), None); // too few fields + assert_eq!(parse_session_from_stat(""), None); + } + + #[cfg(target_family = "unix")] + #[test] + fn is_zombie_stat_reads_the_state_after_comm() { + use nativelink_worker::process_tree::is_zombie_stat; + assert!(is_zombie_stat("99 (a) Z) Z 1 99 99 0")); + assert!(!is_zombie_stat("99 (Z) S 1 99 99 0")); + assert!(!is_zombie_stat("no parenthesis here")); + } + #[cfg(target_os = "linux")] #[test] fn parse_cpu_ticks_from_stat_sums_user_and_system_time() { @@ -8675,15 +9324,6 @@ done assert_eq!(parse_cpu_ticks_from_stat(""), None); } - #[cfg(target_os = "linux")] - #[test] - fn parse_pgid_from_stat_rejects_malformed_input() { - use nativelink_worker::running_actions_manager::parse_pgid_from_stat; - assert_eq!(parse_pgid_from_stat("no parenthesis here"), None); - assert_eq!(parse_pgid_from_stat("123 (only) S"), None); // too few fields - assert_eq!(parse_pgid_from_stat(""), None); - } - // Regression test for #2636: deeply nested directories with a single // semaphore permit previously deadlocked because dir_futures and // file_futures competed for the same permit inside try_join3. diff --git a/web/apps/docs/content/docs/reference/nativelink-config/index.mdx b/web/apps/docs/content/docs/reference/nativelink-config/index.mdx index 82d3d394d6..d416846aad 100644 --- a/web/apps/docs/content/docs/reference/nativelink-config/index.mdx +++ b/web/apps/docs/content/docs/reference/nativelink-config/index.mdx @@ -1593,7 +1593,7 @@ The possible error codes that might occur on an upstream request. | Value | Description | | --- | --- | | `"none"` | Measure only. | -| `"soft"` | Kill the action's process group once two consecutive samples exceed the reservation plus headroom, or one sample exceeds twice it. | +| `"soft"` | Kill the action's processes once two consecutive samples exceed the reservation plus headroom, or one sample exceeds twice it. | ## DiskEnforcement diff --git a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx index 1e18d69b7a..c7c4390ad4 100644 --- a/web/apps/docs/content/docs/reference/nativelink-config/main.mdx +++ b/web/apps/docs/content/docs/reference/nativelink-config/main.mdx @@ -1593,7 +1593,7 @@ The possible error codes that might occur on an upstream request. | Value | Description | | --- | --- | | `"none"` | Measure only. | -| `"soft"` | Kill the action's process group once two consecutive samples exceed the reservation plus headroom, or one sample exceeds twice it. | +| `"soft"` | Kill the action's processes once two consecutive samples exceed the reservation plus headroom, or one sample exceeds twice it. | ## DiskEnforcement