From b982f6d553e85749f8c169d95dc83147d0d9fc22 Mon Sep 17 00:00:00 2001 From: Johnny Date: Sat, 10 Oct 2026 14:04:58 +0000 Subject: [PATCH 1/3] fix(worker): measure and kill an action by its process tree The worker started each action as its own process-group leader, sampled the group's memory and CPU, and killed the group. Processes leave a group freely: a job-control shell moves every job to a group of its own (Bazel's test-setup.sh runs `set -m` before every test), and a detached spawn (Node's `spawn(..., { detached: true })`, a daemonizing server) calls `setsid`, which leaves the session too. So the sampler measured the shell waiting for the test (a two-second busy test read as 0 ms of CPU and 194 KiB peak; a web e2e test whose harness detaches its server read as 60 MB while using gigabytes), soft memory and disk enforcement could not see the real work, and a kill left it running. Run every action under a stub, as Bazel's process-wrapper does for a local action. The spawned child calls `setsid` and forks: the child execs the action, the parent stays as its stub. The stub is a child subreaper, so a process the action orphans is reparented to it, never to the worker, whatever session or group it moved to: the action's processes are exactly the stub's descendants. The sampler walks that tree in /proc each tick (not the stub, a fork of the worker whose pages are the worker's). The SIGTERM grace and the hard kill signal every descendant, through a pidfd after checking the parent again, so a reused pid is never signalled, and the hard kill repeats until none is left alive. When the action's own process exits the stub kills what is left, reaps it, and exits with the action's status, as process-wrapper's TerminateAndWaitForAll does, so nothing outlives its action to hold a port or memory the next action is charged for. Before exiting it writes the kernel's CPU total of everything it reaped (getrusage RUSAGE_CHILDREN) to a pipe the worker reads, which replaces the sampled CPU: exact, including processes too short-lived for any sample. With namespaces on, the same stub sits outside the PID namespace the action is the init of; it replaces the namespace's own fork. Pooled persistent workers are sampled from their own pid. If the stub itself is killed, its orphans go to the worker, and the session it leads is still signalled. Without /proc (macOS) there is no stub and the session's group is signalled, as before. Tests: the action runs under a stub that leads its group and session; a busy loop in a `set -m` job and in a `setsid` job is the action's CPU; 256 MiB held by a `setsid` job is the action's memory; a job left in a group of its own and one left in a session of its own are killed when the action ends. The three detached tests fail on the code before this change. --- nativelink-config/src/cas_server.rs | 2 +- .../remote_execution/worker_api.proto | 4 +- ..._machina.nativelink.remote_execution.pb.rs | 123 ++- nativelink-worker/BUILD.bazel | 6 +- nativelink-worker/src/action_stub.rs | 369 +++++++++ nativelink-worker/src/lib.rs | 4 + nativelink-worker/src/namespace_utils.rs | 194 +---- .../src/persistent_worker/live_worker.rs | 17 +- nativelink-worker/src/process_tree.rs | 266 +++++++ .../src/running_actions_manager.rs | 239 +++--- .../tests/namespace_utils_test.rs | 31 +- .../tests/running_actions_manager_test.rs | 707 +++++++++++++++++- .../reference/nativelink-config/index.mdx | 2 +- .../docs/reference/nativelink-config/main.mdx | 2 +- 14 files changed, 1545 insertions(+), 421 deletions(-) create mode 100644 nativelink-worker/src/action_stub.rs create mode 100644 nativelink-worker/src/process_tree.rs diff --git a/nativelink-config/src/cas_server.rs b/nativelink-config/src/cas_server.rs index 6fddb93bf70..b24af61b455 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 0fd3d61660b..b98361657c9 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 62926da037e..0b37087082b 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 @@ -217,17 +217,7 @@ pub struct ExecuteDeclined { } /// Nested message and enum types in `ExecuteDeclined`. pub mod execute_declined { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] #[repr(i32)] pub enum Reason { Unspecified = 0, @@ -451,10 +441,10 @@ pub mod worker_api_client { dead_code, missing_docs, clippy::wildcard_imports, - clippy::let_unit_value, + clippy::let_unit_value )] - use tonic::codegen::*; use tonic::codegen::http::Uri; + use tonic::codegen::*; /// / This API describes how schedulers communicate with Worker nodes. /// / /// / When a worker node comes online it must be pre-configured with the @@ -489,14 +479,13 @@ pub mod worker_api_client { F: tonic::service::Interceptor, T::ResponseBody: Default, T: tonic::codegen::Service< - http::Request, - Response = http::Response< - >::ResponseBody, + http::Request, + Response = http::Response< + >::ResponseBody, + >, >, - >, - , - >>::Error: Into + std::marker::Send + std::marker::Sync, + >>::Error: + Into + std::marker::Send + std::marker::Sync, { WorkerApiClient::new(InterceptedService::new(inner, interceptor)) } @@ -542,33 +531,23 @@ pub mod worker_api_client { /// / the worker what worker ID to place in action results. pub async fn connect_worker( &mut self, - request: impl tonic::IntoStreamingRequest< - Message = super::UpdateForScheduler, - >, + request: impl tonic::IntoStreamingRequest, ) -> std::result::Result< tonic::Response>, tonic::Status, > { - self.inner - .ready() - .await - .map_err(|e| { - tonic::Status::unknown( - format!("Service was not ready: {}", e.into()), - ) - })?; + self.inner.ready().await.map_err(|e| { + tonic::Status::unknown(format!("Service was not ready: {}", e.into())) + })?; let codec = tonic_prost::ProstCodec::default(); let path = http::uri::PathAndQuery::from_static( "/com.github.trace_machina.nativelink.remote_execution.WorkerApi/ConnectWorker", ); let mut req = request.into_streaming_request(); - req.extensions_mut() - .insert( - GrpcMethod::new( - "com.github.trace_machina.nativelink.remote_execution.WorkerApi", - "ConnectWorker", - ), - ); + req.extensions_mut().insert(GrpcMethod::new( + "com.github.trace_machina.nativelink.remote_execution.WorkerApi", + "ConnectWorker", + )); self.inner.streaming(req, path, codec).await } } @@ -580,7 +559,7 @@ pub mod worker_api_server { dead_code, missing_docs, clippy::wildcard_imports, - clippy::let_unit_value, + clippy::let_unit_value )] use tonic::codegen::*; /// Generated trait containing gRPC methods that should be implemented for use with WorkerApiServer. @@ -589,8 +568,7 @@ pub mod worker_api_server { /// Server streaming response type for the ConnectWorker method. type ConnectWorkerStream: tonic::codegen::tokio_stream::Stream< Item = std::result::Result, - > - + std::marker::Send + > + std::marker::Send + 'static; /// / Registers this worker and informs the scheduler what properties /// / this worker supports. The response must be listened on the client @@ -604,10 +582,7 @@ pub mod worker_api_server { async fn connect_worker( &self, request: tonic::Request>, - ) -> std::result::Result< - tonic::Response, - tonic::Status, - >; + ) -> std::result::Result, tonic::Status>; } /// / This API describes how schedulers communicate with Worker nodes. /// / @@ -637,10 +612,7 @@ pub mod worker_api_server { max_encoding_message_size: None, } } - pub fn with_interceptor( - inner: T, - interceptor: F, - ) -> InterceptedService + pub fn with_interceptor(inner: T, interceptor: F) -> InterceptedService where F: tonic::service::Interceptor, { @@ -695,21 +667,16 @@ pub mod worker_api_server { "/com.github.trace_machina.nativelink.remote_execution.WorkerApi/ConnectWorker" => { #[allow(non_camel_case_types)] struct ConnectWorkerSvc(pub Arc); - impl< - T: WorkerApi, - > tonic::server::StreamingService - for ConnectWorkerSvc { + impl tonic::server::StreamingService + for ConnectWorkerSvc + { type Response = super::UpdateForWorker; type ResponseStream = T::ConnectWorkerStream; - type Future = BoxFuture< - tonic::Response, - tonic::Status, - >; + type Future = + BoxFuture, tonic::Status>; fn call( &mut self, - request: tonic::Request< - tonic::Streaming, - >, + request: tonic::Request>, ) -> Self::Future { let inner = Arc::clone(&self.0); let fut = async move { @@ -740,25 +707,19 @@ pub mod worker_api_server { }; Box::pin(fut) } - _ => { - Box::pin(async move { - let mut response = http::Response::new( - tonic::body::Body::default(), - ); - let headers = response.headers_mut(); - headers - .insert( - tonic::Status::GRPC_STATUS, - (tonic::Code::Unimplemented as i32).into(), - ); - headers - .insert( - http::header::CONTENT_TYPE, - tonic::metadata::GRPC_CONTENT_TYPE, - ); - Ok(response) - }) - } + _ => Box::pin(async move { + let mut response = http::Response::new(tonic::body::Body::default()); + let headers = response.headers_mut(); + headers.insert( + tonic::Status::GRPC_STATUS, + (tonic::Code::Unimplemented as i32).into(), + ); + headers.insert( + http::header::CONTENT_TYPE, + tonic::metadata::GRPC_CONTENT_TYPE, + ); + Ok(response) + }), } } } diff --git a/nativelink-worker/BUILD.bazel b/nativelink-worker/BUILD.bazel index b32036527eb..b58eb0db40e 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 00000000000..7bbdb22a5ee --- /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 48997ca77a8..42b693ce14a 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 206759a92ee..0cf81b375ef 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 b8501191cff..12dc8189226 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 00000000000..9a67034979a --- /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 9c76c6e17d6..431c42d65f5 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 5648f7f8460..3316c2f23a2 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 9a2de33764c..072b86f8a0e 100644 --- a/nativelink-worker/tests/running_actions_manager_test.rs +++ b/nativelink-worker/tests/running_actions_manager_test.rs @@ -6853,14 +6853,386 @@ 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 { @@ -6901,13 +7273,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 +7333,256 @@ 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) + } + + 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 +9217,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 +9311,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 82d3d394d63..d416846aad2 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 1e18d69b7a9..c7c4390ad4b 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 From 266aea63cb5c3122ee1a16a61de42597fc25c720 Mon Sep 17 00:00:00 2001 From: Johnny Date: Fri, 16 Oct 2026 18:50:26 +0000 Subject: [PATCH 2/3] chore(proto): regenerate the remote_execution bindings `bazel run //nativelink-proto:update_protos` on this branch; the checked-in file had been formatted by hand and `//nativelink-proto:update_protos_test` refused it. --- ..._machina.nativelink.remote_execution.pb.rs | 119 ++++++++++++------ 1 file changed, 79 insertions(+), 40 deletions(-) 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 0b37087082b..745a638dbf4 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 @@ -217,7 +217,17 @@ pub struct ExecuteDeclined { } /// Nested message and enum types in `ExecuteDeclined`. pub mod execute_declined { - #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[derive( + Clone, + Copy, + Debug, + PartialEq, + Eq, + Hash, + PartialOrd, + Ord, + ::prost::Enumeration + )] #[repr(i32)] pub enum Reason { Unspecified = 0, @@ -441,10 +451,10 @@ pub mod worker_api_client { dead_code, missing_docs, clippy::wildcard_imports, - clippy::let_unit_value + clippy::let_unit_value, )] - use tonic::codegen::http::Uri; use tonic::codegen::*; + use tonic::codegen::http::Uri; /// / This API describes how schedulers communicate with Worker nodes. /// / /// / When a worker node comes online it must be pre-configured with the @@ -479,13 +489,14 @@ pub mod worker_api_client { F: tonic::service::Interceptor, T::ResponseBody: Default, T: tonic::codegen::Service< - http::Request, - Response = http::Response< - >::ResponseBody, - >, + http::Request, + Response = http::Response< + >::ResponseBody, >, - >>::Error: - Into + std::marker::Send + std::marker::Sync, + >, + , + >>::Error: Into + std::marker::Send + std::marker::Sync, { WorkerApiClient::new(InterceptedService::new(inner, interceptor)) } @@ -531,23 +542,33 @@ pub mod worker_api_client { /// / the worker what worker ID to place in action results. pub async fn connect_worker( &mut self, - request: impl tonic::IntoStreamingRequest, + request: impl tonic::IntoStreamingRequest< + Message = super::UpdateForScheduler, + >, ) -> std::result::Result< tonic::Response>, tonic::Status, > { - self.inner.ready().await.map_err(|e| { - tonic::Status::unknown(format!("Service was not ready: {}", e.into())) - })?; + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic_prost::ProstCodec::default(); let path = http::uri::PathAndQuery::from_static( "/com.github.trace_machina.nativelink.remote_execution.WorkerApi/ConnectWorker", ); let mut req = request.into_streaming_request(); - req.extensions_mut().insert(GrpcMethod::new( - "com.github.trace_machina.nativelink.remote_execution.WorkerApi", - "ConnectWorker", - )); + req.extensions_mut() + .insert( + GrpcMethod::new( + "com.github.trace_machina.nativelink.remote_execution.WorkerApi", + "ConnectWorker", + ), + ); self.inner.streaming(req, path, codec).await } } @@ -559,7 +580,7 @@ pub mod worker_api_server { dead_code, missing_docs, clippy::wildcard_imports, - clippy::let_unit_value + clippy::let_unit_value, )] use tonic::codegen::*; /// Generated trait containing gRPC methods that should be implemented for use with WorkerApiServer. @@ -568,7 +589,8 @@ pub mod worker_api_server { /// Server streaming response type for the ConnectWorker method. type ConnectWorkerStream: tonic::codegen::tokio_stream::Stream< Item = std::result::Result, - > + std::marker::Send + > + + std::marker::Send + 'static; /// / Registers this worker and informs the scheduler what properties /// / this worker supports. The response must be listened on the client @@ -582,7 +604,10 @@ pub mod worker_api_server { async fn connect_worker( &self, request: tonic::Request>, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; } /// / This API describes how schedulers communicate with Worker nodes. /// / @@ -612,7 +637,10 @@ pub mod worker_api_server { max_encoding_message_size: None, } } - pub fn with_interceptor(inner: T, interceptor: F) -> InterceptedService + pub fn with_interceptor( + inner: T, + interceptor: F, + ) -> InterceptedService where F: tonic::service::Interceptor, { @@ -667,16 +695,21 @@ pub mod worker_api_server { "/com.github.trace_machina.nativelink.remote_execution.WorkerApi/ConnectWorker" => { #[allow(non_camel_case_types)] struct ConnectWorkerSvc(pub Arc); - impl tonic::server::StreamingService - for ConnectWorkerSvc - { + impl< + T: WorkerApi, + > tonic::server::StreamingService + for ConnectWorkerSvc { type Response = super::UpdateForWorker; type ResponseStream = T::ConnectWorkerStream; - type Future = - BoxFuture, tonic::Status>; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; fn call( &mut self, - request: tonic::Request>, + request: tonic::Request< + tonic::Streaming, + >, ) -> Self::Future { let inner = Arc::clone(&self.0); let fut = async move { @@ -707,19 +740,25 @@ pub mod worker_api_server { }; Box::pin(fut) } - _ => Box::pin(async move { - let mut response = http::Response::new(tonic::body::Body::default()); - let headers = response.headers_mut(); - headers.insert( - tonic::Status::GRPC_STATUS, - (tonic::Code::Unimplemented as i32).into(), - ); - headers.insert( - http::header::CONTENT_TYPE, - tonic::metadata::GRPC_CONTENT_TYPE, - ); - Ok(response) - }), + _ => { + Box::pin(async move { + let mut response = http::Response::new( + tonic::body::Body::default(), + ); + let headers = response.headers_mut(); + headers + .insert( + tonic::Status::GRPC_STATUS, + (tonic::Code::Unimplemented as i32).into(), + ); + headers + .insert( + http::header::CONTENT_TYPE, + tonic::metadata::GRPC_CONTENT_TYPE, + ); + Ok(response) + }) + } } } } From 13a449423b8a8c240e7f87ecc2c9326b671e63b5 Mon Sep 17 00:00:00 2001 From: Johnny Date: Fri, 16 Oct 2026 20:19:52 +0000 Subject: [PATCH 3/3] test(worker): skip the detached-job tests without setsid or python3 The two tests spawn their job with util-linux's `setsid` and hold its memory from python3. The coverage build runs the tests in a nix sandbox with neither on PATH, so the job never started and the tests measured an action that finished in two milliseconds. Skip them where the tools are missing, as the other tests here do with `which`. --- .../tests/running_actions_manager_test.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/nativelink-worker/tests/running_actions_manager_test.rs b/nativelink-worker/tests/running_actions_manager_test.rs index 072b86f8a0e..0adcfa3f76f 100644 --- a/nativelink-worker/tests/running_actions_manager_test.rs +++ b/nativelink-worker/tests/running_actions_manager_test.rs @@ -7239,6 +7239,12 @@ exit 1 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"); @@ -7354,6 +7360,13 @@ exit 1 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");