From 5f95e866f8e0d74ae2e6f64baed1a24d9d68b2bb Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Tue, 11 Aug 2026 12:17:26 +0200 Subject: [PATCH] refactor: centralize CPU core detection using cgroup-aware utility Introduce `obisys::effective_parallelism()` to read Linux cgroup v1/v2 CPU quotas from sysfs, preventing thread pool oversubscription in containerized environments. Replace direct `std::thread::available_parallelism()` calls across `obikindex` and `obikmer` with this centralized function. Bump `obikmer` version to 1.1.41. --- .gitea/workflows/ci.yml | 2 +- src/Cargo.lock | 2 +- src/obikindex/src/numa.rs | 8 +-- src/obikindex/src/siblings.rs | 2 +- src/obikmer/Cargo.toml | 2 +- src/obikmer/src/cli.rs | 4 +- src/obikmer/src/cmd/query.rs | 4 +- src/obisys/src/lib.rs | 92 +++++++++++++++++++++++++++++++++-- 8 files changed, 97 insertions(+), 19 deletions(-) diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 64ab1af..1f8e17d 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -1,4 +1,4 @@ -name: CI +pname: CI on: pull_request: diff --git a/src/Cargo.lock b/src/Cargo.lock index 25955a3..a4703b9 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -1715,7 +1715,7 @@ dependencies = [ [[package]] name = "obikmer" -version = "1.1.40" +version = "1.1.41" dependencies = [ "clap", "csv", diff --git a/src/obikindex/src/numa.rs b/src/obikindex/src/numa.rs index 0a9d243..8dd76b0 100644 --- a/src/obikindex/src/numa.rs +++ b/src/obikindex/src/numa.rs @@ -79,9 +79,7 @@ pub fn build() -> NumaSetup { } // UMA fallback: single synthetic node, all cores, no pool, no pinning. - let n_cores = std::thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(1); + let n_cores = obisys::effective_parallelism(); debug!("UMA: single synthetic node, {} core(s)", n_cores); NumaSetup { pools: vec![None], @@ -91,9 +89,7 @@ pub fn build() -> NumaSetup { #[cfg(not(feature = "numa"))] pub fn build() -> NumaSetup { - let n_cores = std::thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(1); + let n_cores = obisys::effective_parallelism(); debug!("UMA: single synthetic node, {} core(s)", n_cores); NumaSetup { pools: vec![None], diff --git a/src/obikindex/src/siblings.rs b/src/obikindex/src/siblings.rs index da51270..f255ed3 100644 --- a/src/obikindex/src/siblings.rs +++ b/src/obikindex/src/siblings.rs @@ -350,7 +350,7 @@ impl KmerIndex { // every single variant — fine at the scale of a handful of test // k-mers, but with billions of lookups against a real index this // manifested as ~90% system time, observed in practice. ─────────── - let n_workers = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); + let n_workers = obisys::effective_parallelism(); let capacity = 256; // Throttling is not optional once a `Flat` stage is in the pipeline diff --git a/src/obikmer/Cargo.toml b/src/obikmer/Cargo.toml index 8c0e42b..49e2485 100644 --- a/src/obikmer/Cargo.toml +++ b/src/obikmer/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "obikmer" -version = "1.1.40" +version = "1.1.41" edition = "2024" [[bin]] diff --git a/src/obikmer/src/cli.rs b/src/obikmer/src/cli.rs index 928264e..6018ad3 100644 --- a/src/obikmer/src/cli.rs +++ b/src/obikmer/src/cli.rs @@ -38,9 +38,7 @@ pub struct CommonArgs { #[arg( short = 'T', long, - default_value_t = std::thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(1) + default_value_t = obisys::effective_parallelism() )] pub threads: usize, diff --git a/src/obikmer/src/cmd/query.rs b/src/obikmer/src/cmd/query.rs index 900aeb7..b63f828 100644 --- a/src/obikmer/src/cmd/query.rs +++ b/src/obikmer/src/cmd/query.rs @@ -70,9 +70,7 @@ pub struct QueryArgs { #[arg( short = 'T', long, - default_value_t = std::thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(1) + default_value_t = obisys::effective_parallelism() )] pub threads: usize, diff --git a/src/obisys/src/lib.rs b/src/obisys/src/lib.rs index 0c539a8..21d4987 100644 --- a/src/obisys/src/lib.rs +++ b/src/obisys/src/lib.rs @@ -202,6 +202,94 @@ fn cgroup_v1_available() -> Option { Some(limit.saturating_sub(used)) } +// ── CPU parallelism query ──────────────────────────────────────────────────── + +/// Returns the number of cores this process can actually use concurrently. +/// +/// `std::thread::available_parallelism()` reads CPU affinity +/// (`sched_getaffinity`), not the container's CPU quota — a Docker/cgroup +/// container commonly reports the *host's* full core count this way while +/// actually being throttled (via `cpu.max`/`cpu.cfs_quota_us`) to a fraction +/// of a core. Sizing a thread/worker pool off the unthrottled count causes +/// severe oversubscription: dozens of threads contending for a sliver of +/// real CPU time, which can look indistinguishable from a hang for minutes +/// or hours (observed in CI). On Linux, this reads the cgroup CPU quota +/// first and returns `min(cgroup_quota, host_parallelism)` when a finite +/// quota is found; falls back to `available_parallelism()` otherwise (same +/// convention as [`available_memory_bytes`]). +pub fn effective_parallelism() -> usize { + let host = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); + #[cfg(target_os = "linux")] + { + if let Some(quota) = cgroup_v2_cpu_quota() { + let effective = quota.clamp(1, host); + tracing::debug!(host, quota, effective, source = "cgroup v2", "effective_parallelism"); + return effective; + } + if let Some(quota) = cgroup_v1_cpu_quota() { + let effective = quota.clamp(1, host); + tracing::debug!(host, quota, effective, source = "cgroup v1", "effective_parallelism"); + return effective; + } + } + tracing::debug!(host, effective = host, source = "available_parallelism (no cgroup quota found)", "effective_parallelism"); + host +} + +/// cgroup v2 (unified hierarchy): reads `cpu.max` (" ", or +/// "max " when unlimited) for the current process's cgroup, rounded +/// up to whole cores. Returns `None` if unlimited or on any parse error. +#[cfg(target_os = "linux")] +fn cgroup_v2_cpu_quota() -> Option { + let cgroup = std::fs::read_to_string("/proc/self/cgroup").ok()?; + let rel = cgroup + .lines() + .find(|l| l.starts_with("0::"))? + .strip_prefix("0::")? + .trim(); + let base = format!("/sys/fs/cgroup{rel}"); + let raw = std::fs::read_to_string(format!("{base}/cpu.max")).ok()?; + let mut parts = raw.split_whitespace(); + let quota_str = parts.next()?; + let period: f64 = parts.next()?.parse().ok()?; + if quota_str == "max" { + return None; // unlimited + } + let quota: f64 = quota_str.parse().ok()?; + Some((quota / period).ceil().max(1.0) as usize) +} + +/// cgroup v1 (cpu subsystem): reads `cpu.cfs_quota_us`/`cpu.cfs_period_us`, +/// rounded up to whole cores. Returns `None` if unlimited (quota <= 0) or on +/// any parse error. +#[cfg(target_os = "linux")] +fn cgroup_v1_cpu_quota() -> Option { + let cgroup = std::fs::read_to_string("/proc/self/cgroup").ok()?; + let path = cgroup + .lines() + .find(|l| l.contains(":cpu:") || l.contains(":cpu,cpuacct:"))? + .split(':') + .nth(2)?; + let base = format!("/sys/fs/cgroup/cpu{path}"); + let quota: i64 = std::fs::read_to_string(format!("{base}/cpu.cfs_quota_us")) + .ok()? + .trim() + .parse() + .ok()?; + if quota <= 0 { + return None; // unlimited + } + let period: i64 = std::fs::read_to_string(format!("{base}/cpu.cfs_period_us")) + .ok()? + .trim() + .parse() + .ok()?; + if period <= 0 { + return None; + } + Some(((quota as f64) / (period as f64)).ceil().max(1.0) as usize) +} + // ── raw helpers ─────────────────────────────────────────────────────────────── fn get_rusage() -> rusage { @@ -654,9 +742,7 @@ impl fmt::Display for Reporter { return Ok(()); } - let n_cores = std::thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(1); + let n_cores = effective_parallelism(); // column widths let nw = self