diff --git a/crates/animus-runtime-shared/src/lib.rs b/crates/animus-runtime-shared/src/lib.rs index 1e60d8fe..0f6c3b48 100644 --- a/crates/animus-runtime-shared/src/lib.rs +++ b/crates/animus-runtime-shared/src/lib.rs @@ -13,6 +13,7 @@ //! `direct_exec`) is plugin-private and intentionally NOT here. pub mod actor_env; +pub use orchestrator_plugin_host::cgroup_threads; pub mod agent_state; pub mod config_context; pub mod ensure_execution_cwd; diff --git a/crates/orchestrator-cli/src/bin/animus_mcp_proxy.rs b/crates/orchestrator-cli/src/bin/animus_mcp_proxy.rs index 95daf58a..2bd059b8 100644 --- a/crates/orchestrator-cli/src/bin/animus_mcp_proxy.rs +++ b/crates/orchestrator-cli/src/bin/animus_mcp_proxy.rs @@ -63,8 +63,17 @@ impl animus_mcp_oauth::proxy::BearerTokenSource for BrokerBearerSource { } } -#[tokio::main] -async fn main() -> Result<()> { +fn main() -> Result<()> { + let worker_threads = animus_runtime_shared::cgroup_threads::tokio_worker_threads(); + tokio::runtime::Builder::new_multi_thread() + .worker_threads(worker_threads) + .enable_all() + .build() + .expect("failed to build tokio runtime") + .block_on(async_main()) +} + +async fn async_main() -> Result<()> { let args = Args::parse(); let project_root = args .project_root diff --git a/crates/orchestrator-cli/src/main.rs b/crates/orchestrator-cli/src/main.rs index 7db057a5..f4e31052 100644 --- a/crates/orchestrator-cli/src/main.rs +++ b/crates/orchestrator-cli/src/main.rs @@ -12,8 +12,21 @@ mod shared; pub(crate) use cli_types::*; pub(crate) use shared::*; -#[tokio::main] -async fn main() { +fn main() { + // Use a cgroup-aware worker thread count so the daemon doesn't default to + // the host CPU count (e.g. 48 on Railway) inside a resource-capped + // container, exhausting the cgroup PID limit with ~480 Tokio threads. + // `TOKIO_WORKER_THREADS` still overrides (operator escape hatch). + let worker_threads = animus_runtime_shared::cgroup_threads::tokio_worker_threads(); + tokio::runtime::Builder::new_multi_thread() + .worker_threads(worker_threads) + .enable_all() + .build() + .expect("failed to build tokio runtime") + .block_on(async_main()); +} + +async fn async_main() { // Pre-scan argv for `--json` so that clap argparse failures (unknown // subcommands, bad flag values) still emit the `animus.cli.v1` error // envelope when the caller asked for machine-readable output. `Cli::parse` diff --git a/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs b/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs index e148a5fb..d1e8979c 100644 --- a/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs +++ b/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs @@ -324,9 +324,13 @@ impl ProcessManager { // Bound the workflow-runner's tokio pool for the same reason we cap plugins // (orchestrator-plugin-host host.rs): a bare `#[tokio::main]` otherwise sizes // the worker pool to all CPU cores, compounding the PID/thread pressure. This - // path inherits the daemon env, so only impose the default when unset. + // path inherits the daemon env, so only impose the default when unset, using + // the cgroup-aware count rather than a hard-coded value. if std::env::var_os("TOKIO_WORKER_THREADS").is_none() { - command.env("TOKIO_WORKER_THREADS", "2"); + command.env( + "TOKIO_WORKER_THREADS", + animus_runtime_shared::cgroup_threads::tokio_worker_threads().to_string(), + ); } // Phase skills pass-down: resolve the union of phase-level `skills:` // and the executing agent profile's `skills:` daemon-side (scoped diff --git a/crates/orchestrator-plugin-host/src/cgroup_threads.rs b/crates/orchestrator-plugin-host/src/cgroup_threads.rs new file mode 100644 index 00000000..f73acc86 --- /dev/null +++ b/crates/orchestrator-plugin-host/src/cgroup_threads.rs @@ -0,0 +1,154 @@ +//! Compute the Tokio worker-thread count from the cgroup CPU quota so +//! binaries don't default to the host CPU count when running inside a +//! resource-capped container. +//! +//! Priority: +//! 1. `TOKIO_WORKER_THREADS` env var (explicit operator override) +//! 2. cgroup v2 `cpu.max` quota (`/sys/fs/cgroup/cpu.max`) +//! 3. cgroup v1 quota (`/sys/fs/cgroup/cpu/cpu.cfs_quota_us`) +//! 4. `std::thread::available_parallelism()` (OS default) +//! +//! The result is always >= 1. +//! +//! This module lives in `orchestrator-plugin-host` (a leaf crate) so that +//! both `orchestrator-plugin-host` itself and `animus-runtime-shared` (which +//! depends on this crate) can use it without creating a dependency cycle. + +/// Return the number of Tokio worker threads appropriate for this process, +/// honouring the cgroup CPU quota when present. +/// +/// Pass the return value to `tokio::runtime::Builder::worker_threads` instead +/// of relying on `#[tokio::main]`'s default (which uses the host CPU count, +/// not the container quota). +pub fn tokio_worker_threads() -> usize { + read_env_override() + .or_else(cgroup_v2_threads) + .or_else(cgroup_v1_threads) + .unwrap_or_else(system_threads) + .max(1) +} + +fn read_env_override() -> Option { + let s = std::env::var("TOKIO_WORKER_THREADS").ok()?; + parse_env_override(s.trim()) +} + +/// Parse a `TOKIO_WORKER_THREADS` value; returns `None` for zero or non-numeric. +pub(crate) fn parse_env_override(s: &str) -> Option { + s.parse::().ok().filter(|&n| n > 0) +} + +fn cgroup_v2_threads() -> Option { + let content = std::fs::read_to_string("/sys/fs/cgroup/cpu.max").ok()?; + parse_cgroup_v2(&content) +} + +/// Parse a cgroup v2 `cpu.max` file content (` `). +/// +/// Returns `None` when the quota is unlimited (`"max"`) or the file is malformed. +pub(crate) fn parse_cgroup_v2(content: &str) -> Option { + let mut parts = content.split_whitespace(); + let quota_str = parts.next()?; + let period_str = parts.next()?; + if quota_str == "max" { + return None; + } + let quota: u64 = quota_str.parse().ok()?; + let period: u64 = period_str.parse().ok()?; + if period == 0 { + return None; + } + Some(((quota / period) as usize).max(1)) +} + +fn cgroup_v1_threads() -> Option { + let quota_str = std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").ok()?; + let period_str = std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.cfs_period_us").ok()?; + parse_cgroup_v1(quota_str.trim(), period_str.trim()) +} + +/// Parse cgroup v1 quota/period strings from `cpu.cfs_quota_us` / `cpu.cfs_period_us`. +/// +/// Returns `None` when the quota is unlimited (`-1`) or either value is malformed. +pub(crate) fn parse_cgroup_v1(quota_str: &str, period_str: &str) -> Option { + let quota: i64 = quota_str.parse().ok()?; + if quota < 0 { + return None; // -1 = unlimited + } + let period: u64 = period_str.parse().ok()?; + if period == 0 { + return None; + } + Some(((quota as u64 / period) as usize).max(1)) +} + +fn system_threads() -> usize { + std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn system_threads_is_at_least_one() { + assert!(system_threads() >= 1); + } + + #[test] + fn result_is_at_least_one() { + assert!(tokio_worker_threads() >= 1); + } + + #[test] + fn env_override_zero_is_rejected() { + assert!(parse_env_override("0").is_none()); + } + + #[test] + fn env_override_positive_is_accepted() { + assert_eq!(parse_env_override("4"), Some(4)); + } + + #[test] + fn env_override_invalid_is_rejected() { + assert!(parse_env_override("abc").is_none()); + } + + #[test] + fn cgroup_v2_unlimited_returns_none() { + assert!(parse_cgroup_v2("max 100000").is_none()); + } + + #[test] + fn cgroup_v2_two_cpus() { + // 200_000 quota / 100_000 period = 2 CPUs + assert_eq!(parse_cgroup_v2("200000 100000"), Some(2)); + } + + #[test] + fn cgroup_v2_fractional_rounds_down_to_one() { + // 50_000 / 100_000 = 0.5 CPUs → clamped to 1 + assert_eq!(parse_cgroup_v2("50000 100000"), Some(1)); + } + + #[test] + fn cgroup_v2_zero_period_returns_none() { + assert!(parse_cgroup_v2("200000 0").is_none()); + } + + #[test] + fn cgroup_v1_unlimited_returns_none() { + assert!(parse_cgroup_v1("-1", "100000").is_none()); + } + + #[test] + fn cgroup_v1_two_cpus() { + assert_eq!(parse_cgroup_v1("200000", "100000"), Some(2)); + } + + #[test] + fn cgroup_v1_zero_period_returns_none() { + assert!(parse_cgroup_v1("200000", "0").is_none()); + } +} diff --git a/crates/orchestrator-plugin-host/src/host.rs b/crates/orchestrator-plugin-host/src/host.rs index 2f69e8ee..39636bf0 100644 --- a/crates/orchestrator-plugin-host/src/host.rs +++ b/crates/orchestrator-plugin-host/src/host.rs @@ -762,19 +762,16 @@ impl PluginHost { } // Bound each plugin's tokio runtime. A bare `#[tokio::main]` (every Animus - // stdio plugin) sizes its multi-thread worker pool to - // `available_parallelism()` — all CPU cores. With v0.6's resident-plugin - // fleet (config_source + subject backends + queue + workflow_runner + - // providers + transport) that is hundreds of threads on a many-core host, - // exhausting the PID/thread budget so new forks — including the provider CLI - // an agent phase spawns — fail with EAGAIN and the run hangs. Plugins are - // I/O-bound stdio RPC servers, so a tiny pool is sufficient. `env_clear()` - // above dropped any inherited value (TOKIO_WORKER_THREADS is not in the base - // allowlist), so set it explicitly here, honoring an operator override on the - // daemon env so a deploy can still tune it up or down. + // stdio plugin) sizes its multi-thread worker pool to the host CPU count. + // With v0.6's resident-plugin fleet that is hundreds of threads on a many-core + // host, exhausting the PID/thread budget. Use the cgroup CPU quota instead. + // `env_clear()` above dropped any inherited value (TOKIO_WORKER_THREADS is + // not in the base allowlist), so set it explicitly here, honoring an operator + // override on the daemon env so a deploy can still tune it up or down. command.env( "TOKIO_WORKER_THREADS", - std::env::var_os("TOKIO_WORKER_THREADS").unwrap_or_else(|| std::ffi::OsString::from("2")), + std::env::var_os("TOKIO_WORKER_THREADS") + .unwrap_or_else(|| std::ffi::OsString::from(crate::cgroup_threads::tokio_worker_threads().to_string())), ); for missing in &options.missing_required_env { diff --git a/crates/orchestrator-plugin-host/src/lib.rs b/crates/orchestrator-plugin-host/src/lib.rs index d7b5f6c8..f9b1fa3a 100644 --- a/crates/orchestrator-plugin-host/src/lib.rs +++ b/crates/orchestrator-plugin-host/src/lib.rs @@ -2,6 +2,7 @@ mod discovery; mod host; +pub mod cgroup_threads; pub mod lockfile; pub mod manifest_cache; mod registry;