perf(runtime): size the Tokio runtime to the cgroup CPU quota + bound blocking pool
Replace the bare #[tokio::main] with an explicit runtime::Builder so both
pools are sized, logged at startup, and operator-tunable.
#[tokio::main] hides two defaults that misbehave under container limits:
- worker_threads = available_parallelism(), which honours CPU affinity
(sched_getaffinity: cpuset, taskset) but IGNORES the CFS bandwidth quota
(docker --cpus, cgroup v2 cpu.max, v1 cpu.cfs_quota_us). On a 2-core-quota
container on a 64-core host it spawns 64 workers time-slicing 2 cores.
- max_blocking_threads = 512, a multi-GB RSS blast radius for this heavy
spawn_blocking user (thumbnails, transcode, zip, PDF/text extraction,
Argon2 ~19 MB/hash).
New common::runtime module folds the CFS quota back in: effective_parallelism()
= min(available_parallelism, cgroup quota). runtime_pool_sizes() defaults
workers to that and caps the blocking pool at max(32, 8*workers), both
overridable via OXICLOUD_WORKER_THREADS (or TOKIO_WORKER_THREADS) and
OXICLOUD_MAX_BLOCKING_THREADS. The cgroup v1/v2 parsers are pure + unit-tested.
Note: this corrects the premise that "rayon respects the quota, tokio doesn't"
— the image/rayon pools also use available_parallelism(), so they over-spawn
under a CFS quota too; effective_parallelism() is the reusable fix. Unset env
on an uncontended host reproduces the previous worker count exactly.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JG5yYZ9s868mJwqT2Qz7ez
This commit is contained in:
@@ -3,4 +3,5 @@ pub mod di;
|
|||||||
pub mod errors;
|
pub mod errors;
|
||||||
pub mod locale;
|
pub mod locale;
|
||||||
pub mod mime_detect;
|
pub mod mime_detect;
|
||||||
|
pub mod runtime;
|
||||||
pub mod stubs;
|
pub mod stubs;
|
||||||
|
|||||||
@@ -0,0 +1,144 @@
|
|||||||
|
//! Tokio runtime pool sizing — CFS-quota-aware worker / blocking thread counts.
|
||||||
|
//!
|
||||||
|
//! `#[tokio::main]` and the existing rayon / image pools all size themselves from
|
||||||
|
//! [`std::thread::available_parallelism`], which reflects CPU **affinity**
|
||||||
|
//! (`sched_getaffinity`: cpuset cgroup, `taskset`) but **not** the CFS bandwidth
|
||||||
|
//! quota (`docker --cpus`, cgroup v2 `cpu.max`, v1 `cpu.cfs_quota_us`). On a
|
||||||
|
//! quota-limited container on a many-core host it therefore over-reports the
|
||||||
|
//! usable core count — tokio then spawns one worker per *host* core that
|
||||||
|
//! time-slice across the *quota's* cores. These helpers fold the CFS quota back
|
||||||
|
//! in so the runtime (and any caller that wants it) sizes to the real budget.
|
||||||
|
//!
|
||||||
|
//! All parsing is split into pure functions so the cgroup formats are unit-tested
|
||||||
|
//! without touching `/sys`.
|
||||||
|
|
||||||
|
/// Parse cgroup **v2** `cpu.max` contents — `"<quota_us> <period_us>"`, or
|
||||||
|
/// `"max <period_us>"` when unlimited. Returns the quota in whole cores (rounded
|
||||||
|
/// up), or `None` when unlimited / unparseable.
|
||||||
|
fn parse_cpu_max_v2(s: &str) -> Option<usize> {
|
||||||
|
let mut it = s.split_whitespace();
|
||||||
|
let quota = it.next()?;
|
||||||
|
let period = it.next()?;
|
||||||
|
if quota == "max" {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
let quota: f64 = quota.parse().ok()?;
|
||||||
|
let period: f64 = period.parse().ok()?;
|
||||||
|
if quota > 0.0 && period > 0.0 {
|
||||||
|
Some((quota / period).ceil() as usize)
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Parse cgroup **v1** `cpu.cfs_quota_us` / `cpu.cfs_period_us`. A quota of `-1`
|
||||||
|
/// (or any non-positive value) means unlimited → `None`. Otherwise whole cores,
|
||||||
|
/// rounded up.
|
||||||
|
fn parse_cpu_quota_v1(quota: &str, period: &str) -> Option<usize> {
|
||||||
|
let quota: i64 = quota.trim().parse().ok()?;
|
||||||
|
let period: i64 = period.trim().parse().ok()?;
|
||||||
|
if quota > 0 && period > 0 {
|
||||||
|
Some(((quota as f64) / (period as f64)).ceil() as usize)
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The cgroup CPU quota in whole cores (v2 first, then v1), or `None` when there
|
||||||
|
/// is no quota (unlimited) or it can't be read.
|
||||||
|
pub fn cgroup_cpu_quota() -> Option<usize> {
|
||||||
|
// cgroup v2 unified hierarchy.
|
||||||
|
if let Ok(s) = std::fs::read_to_string("/sys/fs/cgroup/cpu.max")
|
||||||
|
&& let Some(n) = parse_cpu_max_v2(&s)
|
||||||
|
{
|
||||||
|
return Some(n);
|
||||||
|
}
|
||||||
|
// cgroup v1.
|
||||||
|
let quota = std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").ok()?;
|
||||||
|
let period = std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.cfs_period_us").ok()?;
|
||||||
|
parse_cpu_quota_v1("a, &period)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Effective CPU parallelism: affinity-parallelism capped by the CFS quota.
|
||||||
|
///
|
||||||
|
/// `available_parallelism()` alone over-reports under a CFS quota (see module
|
||||||
|
/// docs); we take the min of it and the quota, floored at 1.
|
||||||
|
pub fn effective_parallelism() -> usize {
|
||||||
|
let affinity = std::thread::available_parallelism()
|
||||||
|
.map(|n| n.get())
|
||||||
|
.unwrap_or(1);
|
||||||
|
match cgroup_cpu_quota() {
|
||||||
|
Some(q) => affinity.min(q.max(1)),
|
||||||
|
None => affinity,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `(worker_threads, max_blocking_threads)` for the Tokio runtime, from env with
|
||||||
|
/// CFS-aware defaults.
|
||||||
|
///
|
||||||
|
/// - `OXICLOUD_WORKER_THREADS` (or tokio's native `TOKIO_WORKER_THREADS`) sets
|
||||||
|
/// the worker count; default [`effective_parallelism`].
|
||||||
|
/// - `OXICLOUD_MAX_BLOCKING_THREADS` sets the blocking-pool cap; default
|
||||||
|
/// `max(32, 8 × workers)` — vs tokio's flat **512**, which for this heavy
|
||||||
|
/// `spawn_blocking` user (thumbnails, transcode, zip, PDF/text extraction,
|
||||||
|
/// Argon2 ≈19 MB/hash) is a multi-GB RSS blast radius with no ceiling.
|
||||||
|
///
|
||||||
|
/// Both are clamped to ≥1. Unset env on an uncontended host yields the same
|
||||||
|
/// worker count as the previous `#[tokio::main]` default.
|
||||||
|
pub fn runtime_pool_sizes() -> (usize, usize) {
|
||||||
|
let workers = std::env::var("OXICLOUD_WORKER_THREADS")
|
||||||
|
.or_else(|_| std::env::var("TOKIO_WORKER_THREADS"))
|
||||||
|
.ok()
|
||||||
|
.and_then(|v| v.parse::<usize>().ok())
|
||||||
|
.filter(|&n| n > 0)
|
||||||
|
.unwrap_or_else(effective_parallelism)
|
||||||
|
.max(1);
|
||||||
|
let max_blocking = std::env::var("OXICLOUD_MAX_BLOCKING_THREADS")
|
||||||
|
.ok()
|
||||||
|
.and_then(|v| v.parse::<usize>().ok())
|
||||||
|
.filter(|&n| n > 0)
|
||||||
|
.unwrap_or_else(|| (workers * 8).max(32));
|
||||||
|
(workers, max_blocking)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn v2_exact_two_cores() {
|
||||||
|
assert_eq!(parse_cpu_max_v2("200000 100000"), Some(2));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn v2_rounds_up_fractional() {
|
||||||
|
// 1.5 cores of quota → 2 whole worker threads.
|
||||||
|
assert_eq!(parse_cpu_max_v2("150000 100000"), Some(2));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn v2_unlimited_is_none() {
|
||||||
|
assert_eq!(parse_cpu_max_v2("max 100000"), None);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn v2_garbage_is_none() {
|
||||||
|
assert_eq!(parse_cpu_max_v2("not a quota"), None);
|
||||||
|
assert_eq!(parse_cpu_max_v2(""), None);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn v1_two_cores() {
|
||||||
|
assert_eq!(parse_cpu_quota_v1("200000", "100000"), Some(2));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn v1_unlimited_sentinel_is_none() {
|
||||||
|
assert_eq!(parse_cpu_quota_v1("-1", "100000"), None);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn v1_rounds_up() {
|
||||||
|
assert_eq!(parse_cpu_quota_v1("250000", "100000"), Some(3));
|
||||||
|
}
|
||||||
|
}
|
||||||
+48
-2
@@ -126,8 +126,7 @@ fn make_socket(addr: &SocketAddr, reuse_port: bool) -> std::io::Result<Socket> {
|
|||||||
Ok(socket)
|
Ok(socket)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::main]
|
fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|
||||||
// Minimal CLI:
|
// Minimal CLI:
|
||||||
// --version Print version + branch + commit hash and exit.
|
// --version Print version + branch + commit hash and exit.
|
||||||
// --config <path> Load env from this file. When given, the default
|
// --config <path> Load env from this file. When given, the default
|
||||||
@@ -186,6 +185,39 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Build the Tokio runtime explicitly (not via `#[tokio::main]`) so the
|
||||||
|
// worker + blocking pools are sized from the cgroup CPU quota and bounded
|
||||||
|
// — with the `.env` loaded above already in scope. See `build_runtime`.
|
||||||
|
let runtime = build_runtime()?;
|
||||||
|
runtime.block_on(run())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Construct the multi-threaded Tokio runtime with explicit, CFS-quota-aware
|
||||||
|
/// pool sizes.
|
||||||
|
///
|
||||||
|
/// `#[tokio::main]` hides two defaults that misbehave under container limits:
|
||||||
|
/// • worker threads default to `available_parallelism()`, which honours CPU
|
||||||
|
/// affinity but **ignores the CFS quota** (`--cpus` / `cpu.max`) — so on a
|
||||||
|
/// 2-core-quota container on a 64-core host it spawns 64 workers that
|
||||||
|
/// time-slice across 2 cores.
|
||||||
|
/// • the blocking pool defaults to a flat **512** threads — a multi-GB RSS
|
||||||
|
/// blast radius for this heavy `spawn_blocking` user.
|
||||||
|
///
|
||||||
|
/// Both come from [`common::runtime::runtime_pool_sizes`] (env-overridable via
|
||||||
|
/// `OXICLOUD_WORKER_THREADS` / `OXICLOUD_MAX_BLOCKING_THREADS`). Unset env on an
|
||||||
|
/// uncontended host reproduces the previous behaviour.
|
||||||
|
fn build_runtime() -> std::io::Result<tokio::runtime::Runtime> {
|
||||||
|
let (workers, max_blocking) = common::runtime::runtime_pool_sizes();
|
||||||
|
tokio::runtime::Builder::new_multi_thread()
|
||||||
|
.worker_threads(workers)
|
||||||
|
.max_blocking_threads(max_blocking)
|
||||||
|
.thread_name("oxicloud-worker")
|
||||||
|
.enable_all()
|
||||||
|
.build()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Async entrypoint, driven by the runtime built in [`main`].
|
||||||
|
async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||||
// Initialize tracing.
|
// Initialize tracing.
|
||||||
//
|
//
|
||||||
// Default access-log policy — two independent directives are
|
// Default access-log policy — two independent directives are
|
||||||
@@ -236,6 +268,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|||||||
env!("GIT_HASH")
|
env!("GIT_HASH")
|
||||||
);
|
);
|
||||||
|
|
||||||
|
// Surface the runtime pool sizing chosen in `build_runtime`. `available`
|
||||||
|
// is what tokio's default would have used; `cgroup_cpu_quota` is the CFS
|
||||||
|
// limit it ignores. When the two diverge, the worker count tracks the
|
||||||
|
// smaller (effective) value — the whole point of the explicit builder.
|
||||||
|
let (rt_workers, rt_max_blocking) = common::runtime::runtime_pool_sizes();
|
||||||
|
tracing::info!(
|
||||||
|
worker_threads = rt_workers,
|
||||||
|
max_blocking_threads = rt_max_blocking,
|
||||||
|
available_parallelism =
|
||||||
|
std::thread::available_parallelism().map(|n| n.get()).unwrap_or(0),
|
||||||
|
cgroup_cpu_quota = ?common::runtime::cgroup_cpu_quota(),
|
||||||
|
"Tokio runtime pools sized"
|
||||||
|
);
|
||||||
|
|
||||||
// Load configuration from environment variables
|
// Load configuration from environment variables
|
||||||
let config = common::config::AppConfig::from_env();
|
let config = common::config::AppConfig::from_env();
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user