diff --git a/crates/codegen/kigi-agent/src/error.rs b/crates/codegen/kigi-agent/src/error.rs index 707a232..ef8c2a1 100644 --- a/crates/codegen/kigi-agent/src/error.rs +++ b/crates/codegen/kigi-agent/src/error.rs @@ -17,6 +17,11 @@ pub enum AgentBuildError { #[error("IO error during agent construction: {0}")] IoError(#[from] std::io::Error), + /// The per-session runtime could not be built — fd or thread + /// exhaustion (EMFILE/EAGAIN), not a definition problem. + #[error("failed to build the session runtime: {0}")] + RuntimeBuild(std::io::Error), + /// Carries template line numbers and surrounding context. #[error("template rendering error: {0}")] MiniJinjaError(#[from] minijinja::Error), diff --git a/crates/codegen/kigi-bin/src/main.rs b/crates/codegen/kigi-bin/src/main.rs index c55b96c..2ca9ba7 100644 --- a/crates/codegen/kigi-bin/src/main.rs +++ b/crates/codegen/kigi-bin/src/main.rs @@ -1221,10 +1221,26 @@ fn main() { "Found crashed sessions from a previous run" ); } - let runtime = tokio::runtime::Builder::new_multi_thread() + // Capped workers: tokio's per-core default pins 100+ thread slots on + // many-core shared hosts and later spawns die with EAGAIN. + let runtime = match tokio::runtime::Builder::new_multi_thread() + .worker_threads(kigi_tty_utils::runtime::capped_worker_threads().get()) .enable_all() .build() - .unwrap_or_else(|e| panic!("failed to start tokio runtime: {e}")); + { + Ok(runtime) => runtime, + Err(e) => { + // A host out of threads (RLIMIT_NPROC / pids.max) is an + // environment problem, not a bug — report it, don't panic. + kigi_tty_utils::restore_native_stderr(); + eprintln!("Error: failed to start the async runtime: {e}"); + eprintln!( + "The host may be out of threads (RLIMIT_NPROC or the cgroup pids ceiling); \ + close other processes or raise the limit, then retry." + ); + std::process::exit(1); + } + }; let result = run_and_shutdown(runtime, async_main(), RUNTIME_SHUTDOWN_GRACE); kigi_log::debug_log::flush(); if let Err(e) = result { diff --git a/crates/codegen/kigi-fsnotify/src/watcher.rs b/crates/codegen/kigi-fsnotify/src/watcher.rs index 2e6c2bd..bd09cc3 100644 --- a/crates/codegen/kigi-fsnotify/src/watcher.rs +++ b/crates/codegen/kigi-fsnotify/src/watcher.rs @@ -971,7 +971,7 @@ pub(crate) fn start_with_timeout( let progress_for_thread = progress.clone(); - let thread = std::thread::spawn(move || { + let watcher_loop = move || { let update_stage = |stage: &'static str| { if let Ok(mut p) = progress_for_thread.lock() { p.set_stage(stage); @@ -1351,7 +1351,13 @@ pub(crate) fn start_with_timeout( let _ = ready_tx.send(Err(Box::new(e))); } } - }); + }; + // Recoverable spawn: a host out of threads must surface as an error the + // caller can degrade on, not a `std::thread::spawn` panic. + let thread = std::thread::Builder::new() + .name("fsnotify-watcher".into()) + .spawn(watcher_loop) + .map_err(|e| crate::FsNotifyError::WatcherStart(Box::new(e)))?; // Wait for watcher to be ready (with timeout) if let Ok(mut p) = progress.lock() { diff --git a/crates/codegen/kigi-shell/src/agent/models.rs b/crates/codegen/kigi-shell/src/agent/models.rs index 9e760b8..5db3997 100644 --- a/crates/codegen/kigi-shell/src/agent/models.rs +++ b/crates/codegen/kigi-shell/src/agent/models.rs @@ -1938,7 +1938,7 @@ fn resolve_prefetch_env(kimi_code_config: Option) -> Option) -> Option { let env = resolve_prefetch_env_with_auth(auth)?; - Some(spawn_prefetch_thread(env)) + spawn_prefetch_thread(env) } /// Start the model-catalog prefetch on a background thread. @@ -1949,11 +1949,16 @@ pub fn start_early_prefetch( kimi_code_config: Option, ) -> Option { let env = resolve_prefetch_env(kimi_code_config)?; - Some(spawn_prefetch_thread(env)) + spawn_prefetch_thread(env) } -fn spawn_prefetch_thread(env: PrefetchEnv) -> EarlyPrefetchHandle { - std::thread::spawn(move || { +/// `None` when the OS refuses a thread: the prefetch is an optimization and +/// runs before the runtime builds, so a spawn panic here would preempt the +/// runtime's own out-of-threads diagnostic. +fn spawn_prefetch_thread(env: PrefetchEnv) -> Option { + let spawned = std::thread::Builder::new() + .name("early-prefetch".into()) + .spawn(move || { let mut timer = crate::instrumentation_timer!("startup.early_prefetch"); let proxy_endpoint = env.endpoints.proxy_url(); timer.with_field("endpoint", proxy_endpoint.as_str()); @@ -1993,7 +1998,14 @@ fn spawn_prefetch_thread(env: PrefetchEnv) -> EarlyPrefetchHandle { } EarlyPrefetchResult { models } - }) + }); + match spawned { + Ok(handle) => Some(handle), + Err(e) => { + tracing::error!(error = %e, "early prefetch thread spawn failed; skipping prefetch"); + None + } + } } /// Map a model id (catalog key or routing slug) to its catalog key. diff --git a/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs b/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs index 54ba2a2..c8d3cd7 100644 --- a/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs +++ b/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs @@ -726,10 +726,18 @@ pub struct MvpAgent { /// sampling client is discarded, but the TLS root certificates it loads /// are cached at the process level by `rustls-native-certs`. pub fn warm_async_http_client() { - std::thread::spawn(|| { - let _timer = crate::instrumentation_timer!("startup.async_http_warmup"); - let _ = crate::http::shared_client(); - }); + // Recoverable spawn: the warmup is an optimization that runs before the + // runtime builds — an out-of-threads host must reach the runtime's own + // diagnostic, not panic here. + let spawned = std::thread::Builder::new() + .name("http-warmup".into()) + .spawn(|| { + let _timer = crate::instrumentation_timer!("startup.async_http_warmup"); + let _ = crate::http::shared_client(); + }); + if let Err(e) = spawned { + tracing::warn!(error = %e, "async HTTP warmup thread spawn failed; first request warms instead"); + } } pub(crate) fn resolve_required_agent_type( model_agent_type: Option<&str>, diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs index ece7898..f125cc5 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs +++ b/crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs @@ -1639,10 +1639,23 @@ pub(crate) async fn spawn_session_on_thread( }; (initial_last_compaction, initial_prompt_texts) }; - let rt = tokio::runtime::Builder::new_current_thread() + // Runtime construction acquires fds (epoll/kqueue, waker) and + // fails with EMFILE/EAGAIN under resource pressure — report + // through the init channel instead of panicking the thread. + let rt = match tokio::runtime::Builder::new_current_thread() .enable_all() .build() - .expect("session runtime"); + { + Ok(rt) => rt, + Err(e) => { + tracing::error!( + error = %e, + "failed to build session runtime (resource exhaustion?)" + ); + let _ = init_tx.send(Err(kigi_agent::AgentBuildError::RuntimeBuild(e))); + return; + } + }; let local = tokio::task::LocalSet::new(); local.block_on(&rt, async move { let _trace_span = parent_traceparent.as_ref().map(|tp| { @@ -1763,8 +1776,19 @@ pub(crate) async fn spawn_session_on_thread( })); let _ = session_done_rx.await; }); - }) - .expect("spawn session thread"); + }); + let join_handle = match join_handle { + Ok(handle) => handle, + Err(e) => { + tracing::error!( + error = %e, + "failed to spawn session thread (thread/PID limit or memory pressure?)" + ); + return Err( + acp::Error::internal_error().data(format!("failed to spawn session thread: {e}")) + ); + } + }; let init = init_rx .await .map_err(|_| { diff --git a/crates/codegen/kigi-tty-utils/src/lib.rs b/crates/codegen/kigi-tty-utils/src/lib.rs index 4839af2..c1b0979 100644 --- a/crates/codegen/kigi-tty-utils/src/lib.rs +++ b/crates/codegen/kigi-tty-utils/src/lib.rs @@ -40,6 +40,7 @@ use std::collections::HashMap; use std::io; mod process_scope; +pub mod runtime; pub use process_scope::{ProcessScope, global_process_scope}; // TTY detach — pre_exec building block diff --git a/crates/codegen/kigi-tty-utils/src/runtime.rs b/crates/codegen/kigi-tty-utils/src/runtime.rs new file mode 100644 index 0000000..98cf1c5 --- /dev/null +++ b/crates/codegen/kigi-tty-utils/src/runtime.rs @@ -0,0 +1,55 @@ +//! Worker-thread policy for multi-thread tokio runtimes. +//! +//! Tokio defaults to one worker per core. On many-core shared hosts (100+-core +//! HPC login nodes) that pins 100+ thread slots per kigi process against +//! per-user ceilings — systemd user-slice `pids.max` (commonly 1000) or +//! `RLIMIT_NPROC` — and later thread spawns die with EAGAIN. Kigi's runtimes +//! are I/O-bound, so throughput does not scale with workers past a small +//! count. +//! +//! This is the single home for the cap policy; every multi-thread runtime in +//! the workspace derives its worker count from here so the policy cannot +//! drift across crates. + +use std::num::NonZeroUsize; + +/// Maximum runtime worker threads for any kigi process. +pub const MAX_WORKER_THREADS: NonZeroUsize = NonZeroUsize::new(8).unwrap(); + +/// Pure, testable: `min(cores, MAX_WORKER_THREADS)`. +pub fn cap_worker_threads(cores: NonZeroUsize) -> NonZeroUsize { + cores.min(MAX_WORKER_THREADS) +} + +/// Reads the host: `min(available_parallelism, MAX_WORKER_THREADS)`. +pub fn capped_worker_threads() -> NonZeroUsize { + cap_worker_threads(std::thread::available_parallelism().unwrap_or(NonZeroUsize::MIN)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn nz(n: usize) -> NonZeroUsize { + NonZeroUsize::new(n).unwrap() + } + + #[test] + fn cap_is_identity_at_or_below_max() { + assert_eq!(cap_worker_threads(nz(1)), nz(1)); + assert_eq!(cap_worker_threads(nz(4)), nz(4)); + assert_eq!(cap_worker_threads(nz(8)), nz(8)); + } + + #[test] + fn cap_clamps_many_core_hosts() { + assert_eq!(cap_worker_threads(nz(9)), nz(8)); + assert_eq!(cap_worker_threads(nz(360)), nz(8)); + } + + #[test] + fn capped_worker_threads_stays_in_bounds() { + let n = capped_worker_threads(); + assert!(n <= MAX_WORKER_THREADS, "got {n}"); + } +}