Eric Coissac
5f95e866f8
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.
825 lines
28 KiB
Rust
825 lines
28 KiB
Rust
use std::fmt;
|
||
use std::sync::atomic::{AtomicU64, Ordering};
|
||
use std::sync::{Condvar, Mutex};
|
||
use std::time::{Duration, Instant};
|
||
|
||
use indicatif::{ProgressBar, ProgressStyle};
|
||
use tracing::{debug, info, warn};
|
||
|
||
const BRAILLE: &[&str] = &["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"];
|
||
|
||
// ── TracedBar ──────────────────────────────────────────────────────────────────
|
||
|
||
/// Wrapper around `ProgressBar` that emits `tracing` events when stderr is not
|
||
/// a TTY (e.g. HPC job logs): every 10% for bounded bars, every ~10 s for
|
||
/// spinners (throttled on `set_message`).
|
||
pub struct TracedBar {
|
||
pb: ProgressBar,
|
||
label: String,
|
||
unit: String,
|
||
total: u64, // 0 for spinners
|
||
start: Instant, // creation time, for spinner throttling
|
||
last_pct: AtomicU64, // last emitted 10%-bucket (1..=10), 0 = none yet
|
||
last_log_ms: AtomicU64, // ms since `start` at last spinner log
|
||
}
|
||
|
||
impl TracedBar {
|
||
pub fn inc(&self, delta: u64) {
|
||
self.pb.inc(delta);
|
||
if self.pb.is_hidden() && self.total > 0 {
|
||
let pos = self.pb.position();
|
||
let pct10 = (pos * 10) / self.total; // 0..=10
|
||
let last = self.last_pct.load(Ordering::Relaxed);
|
||
if pct10 > last
|
||
&& self
|
||
.last_pct
|
||
.compare_exchange(last, pct10, Ordering::Relaxed, Ordering::Relaxed)
|
||
.is_ok()
|
||
{
|
||
info!(
|
||
stage = %self.label,
|
||
progress = format_args!("{}%", pct10 * 10),
|
||
"{}/{} {}",
|
||
pos, self.total, self.unit
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
pub fn set_message(&self, msg: impl Into<String>) {
|
||
let msg = msg.into();
|
||
if self.pb.is_hidden() {
|
||
if self.total > 0 {
|
||
debug!(stage = %self.label, "{msg}");
|
||
} else {
|
||
// spinner: throttle to ~10 s
|
||
let now_ms = self.start.elapsed().as_millis() as u64;
|
||
let last = self.last_log_ms.load(Ordering::Relaxed);
|
||
if now_ms >= last + 10_000
|
||
&& self
|
||
.last_log_ms
|
||
.compare_exchange(last, now_ms, Ordering::Relaxed, Ordering::Relaxed)
|
||
.is_ok()
|
||
{
|
||
info!(stage = %self.label, "{msg}");
|
||
}
|
||
}
|
||
}
|
||
self.pb.set_message(msg);
|
||
}
|
||
|
||
pub fn finish_and_clear(&self) {
|
||
self.pb.finish_and_clear();
|
||
}
|
||
}
|
||
|
||
/// Spinner with the standard project look: `⠋ label — msg 0s`.
|
||
/// Caller updates the message with `pb.set_message(...)`.
|
||
pub fn spinner(label: &str) -> TracedBar {
|
||
let pb = ProgressBar::new_spinner();
|
||
pb.set_style(
|
||
ProgressStyle::with_template(&format!("{{spinner}} {label} — {{msg}} {{elapsed}}"))
|
||
.unwrap()
|
||
.tick_strings(BRAILLE),
|
||
);
|
||
pb.enable_steady_tick(Duration::from_millis(100));
|
||
TracedBar {
|
||
pb,
|
||
label: label.to_string(),
|
||
unit: String::new(),
|
||
total: 0,
|
||
start: Instant::now(),
|
||
last_pct: AtomicU64::new(0),
|
||
last_log_ms: AtomicU64::new(0),
|
||
}
|
||
}
|
||
|
||
/// Progress bar with the standard project look:
|
||
/// `⠋ label — [████░░░░] pos/len unit elapsed`.
|
||
pub fn progress_bar(label: &str, n: u64, unit: &str) -> TracedBar {
|
||
let pb = ProgressBar::new(n);
|
||
pb.set_style(
|
||
ProgressStyle::with_template(&format!(
|
||
"{{spinner}} {label} — {{bar:40.cyan/blue}} {{pos}}/{{len}} {unit} {{elapsed}}"
|
||
))
|
||
.unwrap()
|
||
.tick_strings(BRAILLE),
|
||
);
|
||
pb.enable_steady_tick(Duration::from_millis(100));
|
||
TracedBar {
|
||
pb,
|
||
label: label.to_string(),
|
||
unit: unit.to_string(),
|
||
total: n,
|
||
start: Instant::now(),
|
||
last_pct: AtomicU64::new(0),
|
||
last_log_ms: AtomicU64::new(0),
|
||
}
|
||
}
|
||
|
||
use libc::{RUSAGE_SELF, getrusage, rusage, timeval};
|
||
use sysinfo::System;
|
||
|
||
// ── Memory query ──────────────────────────────────────────────────────────────
|
||
|
||
/// Returns the number of bytes available for allocation in the current process context.
|
||
///
|
||
/// On Linux, cgroup memory limits (SLURM, containers) are checked first: the
|
||
/// process may be constrained to far less than the host's available RAM.
|
||
/// Returns `min(cgroup_available, host_available)` when a finite limit is found.
|
||
///
|
||
/// On macOS, `available_memory()` can return 0 when the memory compressor
|
||
/// inflates the page count; in that case we fall back to half of total memory.
|
||
/// Returns the process peak RSS (high-water mark since process start).
|
||
/// Monotonically increasing — use delta before/after a phase to measure its RAM cost.
|
||
pub fn peak_rss_bytes() -> u64 {
|
||
rss_to_bytes(&get_rusage())
|
||
}
|
||
|
||
pub fn available_memory_bytes() -> u64 {
|
||
let sys = System::new_all();
|
||
let host_avail = match sys.available_memory() {
|
||
0 => sys.total_memory() / 2,
|
||
n => n,
|
||
};
|
||
#[cfg(target_os = "linux")]
|
||
if let Some(cg) = cgroup_v2_available().or_else(cgroup_v1_available) {
|
||
return cg.min(host_avail);
|
||
}
|
||
host_avail
|
||
}
|
||
|
||
/// cgroup v2 (unified hierarchy): reads memory.max and memory.current for the
|
||
/// current process's cgroup. Returns None if unlimited or on any parse error.
|
||
#[cfg(target_os = "linux")]
|
||
fn cgroup_v2_available() -> Option<u64> {
|
||
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}");
|
||
// "max" means no limit → parse::<u64>() fails → None
|
||
let limit: u64 = std::fs::read_to_string(format!("{base}/memory.max"))
|
||
.ok()?
|
||
.trim()
|
||
.parse()
|
||
.ok()?;
|
||
let used: u64 = std::fs::read_to_string(format!("{base}/memory.current"))
|
||
.ok()?
|
||
.trim()
|
||
.parse()
|
||
.ok()?;
|
||
Some(limit.saturating_sub(used))
|
||
}
|
||
|
||
/// cgroup v1 (memory subsystem): reads memory.limit_in_bytes and
|
||
/// memory.usage_in_bytes. Returns None if unlimited or on any parse error.
|
||
#[cfg(target_os = "linux")]
|
||
fn cgroup_v1_available() -> Option<u64> {
|
||
let cgroup = std::fs::read_to_string("/proc/self/cgroup").ok()?;
|
||
let path = cgroup
|
||
.lines()
|
||
.find(|l| l.contains(":memory:"))?
|
||
.split(':')
|
||
.nth(2)?;
|
||
let base = format!("/sys/fs/cgroup/memory{path}");
|
||
let limit: u64 = std::fs::read_to_string(format!("{base}/memory.limit_in_bytes"))
|
||
.ok()?
|
||
.trim()
|
||
.parse()
|
||
.ok()?;
|
||
// Kernel uses 2^63 (rounded to page) as "no limit" sentinel
|
||
if limit > (1u64 << 62) {
|
||
return None;
|
||
}
|
||
let used: u64 = std::fs::read_to_string(format!("{base}/memory.usage_in_bytes"))
|
||
.ok()?
|
||
.trim()
|
||
.parse()
|
||
.ok()?;
|
||
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` ("<quota> <period>", or
|
||
/// "max <period>" 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<usize> {
|
||
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<usize> {
|
||
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 {
|
||
let mut ru = unsafe { std::mem::zeroed::<rusage>() };
|
||
unsafe { getrusage(RUSAGE_SELF, &mut ru) };
|
||
ru
|
||
}
|
||
|
||
fn tv_to_secs(tv: timeval) -> f64 {
|
||
tv.tv_sec as f64 + tv.tv_usec as f64 * 1e-6
|
||
}
|
||
|
||
#[cfg(target_os = "macos")]
|
||
fn rss_to_bytes(ru: &rusage) -> u64 {
|
||
ru.ru_maxrss as u64
|
||
}
|
||
|
||
#[cfg(not(target_os = "macos"))]
|
||
fn rss_to_bytes(ru: &rusage) -> u64 {
|
||
ru.ru_maxrss as u64 * 1024
|
||
}
|
||
|
||
// Monotonically increasing counters — negative delta would be a kernel bug.
|
||
fn delta(end: i64, start: i64) -> u64 {
|
||
(end - start).max(0) as u64
|
||
}
|
||
|
||
// ── CpuSample ─────────────────────────────────────────────────────────────────
|
||
|
||
/// Snapshot of process-wide CPU time + wall clock at a point in time.
|
||
/// Use [`cpu_efficiency`](Self::cpu_efficiency) to measure the fraction of
|
||
/// available cores used since the snapshot was taken.
|
||
pub struct CpuSample {
|
||
wall: Instant,
|
||
user_secs: f64,
|
||
sys_secs: f64,
|
||
previous: f64,
|
||
}
|
||
|
||
impl CpuSample {
|
||
pub fn now() -> Self {
|
||
let ru = get_rusage();
|
||
Self {
|
||
wall: Instant::now(),
|
||
user_secs: tv_to_secs(ru.ru_utime),
|
||
sys_secs: tv_to_secs(ru.ru_stime),
|
||
previous: 0.0,
|
||
}
|
||
}
|
||
|
||
/// (user_delta + sys_delta) / (wall_delta × n_cores) since this snapshot.
|
||
/// Returns 0.0 if less than 100 ms have elapsed (too noisy).
|
||
pub fn cpu_efficiency(&self, n_cores: usize) -> f64 {
|
||
let ru = get_rusage();
|
||
let wall = self.wall.elapsed().as_secs_f64();
|
||
if wall < 0.1 {
|
||
return 0.0;
|
||
}
|
||
let cpu =
|
||
(tv_to_secs(ru.ru_utime) - self.user_secs) + (tv_to_secs(ru.ru_stime) - self.sys_secs);
|
||
cpu / (wall * n_cores as f64)
|
||
}
|
||
|
||
pub fn do_i_activate(&mut self, threshold: f64) -> bool {
|
||
let delta_wall = self.wall.elapsed().as_secs_f64();
|
||
if delta_wall < 0.1 {
|
||
// Window too short to be meaningful — leave state untouched so it
|
||
// keeps accumulating until a real sample can be taken.
|
||
return false;
|
||
}
|
||
|
||
let n = CpuSample::now();
|
||
let delta_ru = (n.user_secs - self.user_secs) + (n.sys_secs - self.sys_secs);
|
||
|
||
let efficiency = delta_ru / delta_wall;
|
||
let activate = 0f64.max(efficiency - self.previous) >= threshold;
|
||
|
||
debug!(
|
||
"Do I activate : {} -> {} = {} Activate: {}",
|
||
self.previous,
|
||
efficiency,
|
||
0f64.max(efficiency - self.previous),
|
||
activate
|
||
);
|
||
self.previous = efficiency;
|
||
self.user_secs = n.user_secs;
|
||
self.sys_secs = n.sys_secs;
|
||
self.wall = n.wall;
|
||
|
||
activate
|
||
}
|
||
}
|
||
|
||
// ── IoSample ──────────────────────────────────────────────────────────────────
|
||
|
||
/// Snapshot of process-wide block I/O (bytes read + written) + wall clock.
|
||
///
|
||
/// Same activation protocol as [`CpuSample`], but the growth check in
|
||
/// [`do_i_activate`](Self::do_i_activate) is *relative* rather than absolute:
|
||
/// raw I/O throughput has no portable scale across storage devices, unlike a
|
||
/// core count.
|
||
pub struct IoSample {
|
||
wall: Instant,
|
||
bytes: u64,
|
||
previous_rate: f64,
|
||
}
|
||
|
||
impl IoSample {
|
||
pub fn now() -> Self {
|
||
Self {
|
||
wall: Instant::now(),
|
||
bytes: Self::read_bytes(),
|
||
previous_rate: 0.0,
|
||
}
|
||
}
|
||
|
||
/// Bytes actually submitted to the block layer (read + write), summed
|
||
/// process-wide. Returns 0 if unavailable — degrades gracefully to a
|
||
/// signal that never triggers activation (CPU-only heuristic).
|
||
#[cfg(target_os = "linux")]
|
||
fn read_bytes() -> u64 {
|
||
let Ok(io) = std::fs::read_to_string("/proc/self/io") else {
|
||
return 0;
|
||
};
|
||
io.lines()
|
||
.filter_map(|l| {
|
||
l.strip_prefix("read_bytes: ")
|
||
.or_else(|| l.strip_prefix("write_bytes: "))
|
||
})
|
||
.filter_map(|v| v.trim().parse::<u64>().ok())
|
||
.sum()
|
||
}
|
||
|
||
#[cfg(target_os = "macos")]
|
||
fn read_bytes() -> u64 {
|
||
use libc::{RUSAGE_INFO_V4, getpid, proc_pid_rusage, rusage_info_v4};
|
||
let mut info: rusage_info_v4 = unsafe { std::mem::zeroed() };
|
||
let ret =
|
||
unsafe { proc_pid_rusage(getpid(), RUSAGE_INFO_V4, &mut info as *mut _ as *mut _) };
|
||
if ret != 0 {
|
||
return 0;
|
||
}
|
||
info.ri_diskio_bytesread + info.ri_diskio_byteswritten
|
||
}
|
||
|
||
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
|
||
fn read_bytes() -> u64 {
|
||
0
|
||
}
|
||
|
||
/// Same protocol as [`CpuSample::do_i_activate`] (0.1 s minimum window,
|
||
/// state untouched on early return), but growth is measured relative to
|
||
/// the previous rate. `threshold` is a fraction, e.g. `0.2` for a 20 %
|
||
/// increase in throughput since the last real sample.
|
||
pub fn do_i_activate(&mut self, threshold: f64) -> bool {
|
||
let elapsed = self.wall.elapsed().as_secs_f64();
|
||
if elapsed < 0.1 {
|
||
return false;
|
||
}
|
||
|
||
let n = Self::read_bytes();
|
||
let rate = n.saturating_sub(self.bytes) as f64 / elapsed;
|
||
let activate = if self.previous_rate == 0.0 {
|
||
rate > 0.0 // bootstrap: any measured throughput is signal enough
|
||
} else {
|
||
(rate - self.previous_rate) / self.previous_rate >= threshold
|
||
};
|
||
|
||
debug!(
|
||
"Do I activate (I/O) : {} -> {} Activate: {}",
|
||
self.previous_rate, rate, activate
|
||
);
|
||
self.previous_rate = rate;
|
||
self.bytes = n;
|
||
self.wall = Instant::now();
|
||
|
||
activate
|
||
}
|
||
}
|
||
|
||
// ── public API ────────────────────────────────────────────────────────────────
|
||
|
||
/// Snapshot taken at the start of a pipeline stage.
|
||
#[must_use = "call .stop() to record the stage"]
|
||
pub struct Stage {
|
||
label: String,
|
||
wall: Instant,
|
||
ru: rusage,
|
||
}
|
||
|
||
impl Stage {
|
||
pub fn start(label: impl Into<String>) -> Self {
|
||
let label = label.into();
|
||
info!(stage = %label, "started");
|
||
Self {
|
||
label,
|
||
wall: Instant::now(),
|
||
ru: get_rusage(),
|
||
}
|
||
}
|
||
|
||
pub fn stop(self) -> StageStats {
|
||
let wall_secs = self.wall.elapsed().as_secs_f64();
|
||
let end = get_rusage();
|
||
let stats = StageStats {
|
||
label: self.label,
|
||
wall_secs,
|
||
user_secs: tv_to_secs(end.ru_utime) - tv_to_secs(self.ru.ru_utime),
|
||
sys_secs: tv_to_secs(end.ru_stime) - tv_to_secs(self.ru.ru_stime),
|
||
max_rss_bytes: rss_to_bytes(&end),
|
||
minor_faults: delta(end.ru_minflt as i64, self.ru.ru_minflt as i64),
|
||
major_faults: delta(end.ru_majflt as i64, self.ru.ru_majflt as i64),
|
||
vol_ctx: delta(end.ru_nvcsw as i64, self.ru.ru_nvcsw as i64),
|
||
invol_ctx: delta(end.ru_nivcsw as i64, self.ru.ru_nivcsw as i64),
|
||
in_blocks: delta(end.ru_inblock as i64, self.ru.ru_inblock as i64),
|
||
out_blocks: delta(end.ru_oublock as i64, self.ru.ru_oublock as i64),
|
||
swaps: delta(end.ru_nswap as i64, self.ru.ru_nswap as i64),
|
||
};
|
||
info!(
|
||
stage = %stats.label,
|
||
wall_secs = format_args!("{:.3}", stats.wall_secs),
|
||
rss = %fmt_bytes(stats.max_rss_bytes),
|
||
swaps = stats.swaps,
|
||
"done"
|
||
);
|
||
if stats.swaps > 0 {
|
||
warn!(
|
||
stage = %stats.label,
|
||
swaps = stats.swaps,
|
||
"working set exceeds available RAM"
|
||
);
|
||
}
|
||
stats
|
||
}
|
||
}
|
||
|
||
/// Per-stage efficiency metrics collected from `getrusage(RUSAGE_SELF)` deltas.
|
||
pub struct StageStats {
|
||
pub label: String,
|
||
pub wall_secs: f64,
|
||
pub user_secs: f64,
|
||
pub sys_secs: f64,
|
||
/// Peak RSS at end of stage (bytes). ru_maxrss is a process-lifetime maximum,
|
||
/// so this reflects the high-water mark up to and including this stage.
|
||
pub max_rss_bytes: u64,
|
||
pub minor_faults: u64,
|
||
pub major_faults: u64,
|
||
pub vol_ctx: u64, // voluntary context switches
|
||
pub invol_ctx: u64, // involuntary context switches
|
||
pub in_blocks: u64, // filesystem block reads (after page cache)
|
||
pub out_blocks: u64, // filesystem block writes
|
||
pub swaps: u64,
|
||
}
|
||
|
||
impl StageStats {
|
||
/// (user + sys) / wall — effective thread count utilisation.
|
||
pub fn parallelism(&self) -> f64 {
|
||
if self.wall_secs > 1e-9 {
|
||
(self.user_secs + self.sys_secs) / self.wall_secs
|
||
} else {
|
||
0.0
|
||
}
|
||
}
|
||
|
||
/// parallelism / n_cores — fraction of available CPU power used (0..1+).
|
||
pub fn efficiency(&self, n_cores: usize) -> f64 {
|
||
self.parallelism() / n_cores as f64
|
||
}
|
||
}
|
||
|
||
/// Accumulates stage stats and prints a human-readable summary table.
|
||
#[derive(Default)]
|
||
pub struct Reporter {
|
||
stages: Vec<StageStats>,
|
||
}
|
||
|
||
impl Reporter {
|
||
pub fn new() -> Self {
|
||
Self::default()
|
||
}
|
||
pub fn push(&mut self, stats: StageStats) {
|
||
self.stages.push(stats);
|
||
}
|
||
pub fn stages(&self) -> &[StageStats] {
|
||
&self.stages
|
||
}
|
||
/// Print the summary to stderr.
|
||
pub fn print(&self) {
|
||
eprint!("{self}");
|
||
}
|
||
}
|
||
|
||
// ── diagnosis ─────────────────────────────────────────────────────────────────
|
||
|
||
struct Diagnosis {
|
||
tag: &'static str,
|
||
detail: Option<String>,
|
||
}
|
||
|
||
// Thresholds are intentionally conservative to avoid false positives.
|
||
fn diagnose(s: &StageStats, n_cores: usize) -> Diagnosis {
|
||
let eff = s.efficiency(n_cores);
|
||
let cpu_pct = eff * 100.0;
|
||
let io_ops = s.in_blocks + s.out_blocks;
|
||
|
||
// swaps > 0 is the only reliable cross-platform indicator of true RAM exhaustion.
|
||
// ru_majflt is intentionally excluded: on macOS it counts all file-backed mmap
|
||
// page-ins (even from page cache), making it useless as a memory-pressure signal
|
||
// for mmap-heavy code. On Linux it is more meaningful, but swaps covers the
|
||
// severe case on both platforms.
|
||
if s.swaps > 0 {
|
||
return Diagnosis {
|
||
tag: "swapping",
|
||
detail: Some(format!(
|
||
"swapped {} time(s) — working set exceeds available RAM",
|
||
s.swaps,
|
||
)),
|
||
};
|
||
}
|
||
if eff < 0.3 && io_ops > 100 {
|
||
return Diagnosis {
|
||
tag: "disk I/O",
|
||
detail: Some(format!(
|
||
"{} block reads + {} writes — CPU at {:.0}%, stage is I/O-bound",
|
||
s.in_blocks, s.out_blocks, cpu_pct,
|
||
)),
|
||
};
|
||
}
|
||
if eff < 0.3 && s.vol_ctx > 200 {
|
||
return Diagnosis {
|
||
tag: "contention",
|
||
detail: Some(format!(
|
||
"{} voluntary context switches — CPU at {:.0}%, possible lock contention or I/O wait",
|
||
s.vol_ctx, cpu_pct,
|
||
)),
|
||
};
|
||
}
|
||
Diagnosis {
|
||
tag: "—",
|
||
detail: None,
|
||
}
|
||
}
|
||
|
||
// ── display helpers ───────────────────────────────────────────────────────────
|
||
|
||
fn fmt_secs(s: f64) -> String {
|
||
if s >= 100.0 {
|
||
format!("{:.0}s", s)
|
||
} else if s >= 10.0 {
|
||
format!("{:.1}s", s)
|
||
} else if s >= 1.0 {
|
||
format!("{:.2}s", s)
|
||
} else {
|
||
format!("{:.0}ms", s * 1000.0)
|
||
}
|
||
}
|
||
|
||
fn fmt_bytes(b: u64) -> String {
|
||
if b >= 1 << 30 {
|
||
format!("{:.1} GB", b as f64 / (1u64 << 30) as f64)
|
||
} else if b >= 1 << 20 {
|
||
format!("{:.0} MB", b as f64 / (1u64 << 20) as f64)
|
||
} else {
|
||
format!("{:.0} KB", b as f64 / 1024.0)
|
||
}
|
||
}
|
||
|
||
fn fmt_efficiency(par: f64, n_cores: usize) -> String {
|
||
format!(
|
||
"{:.1}×/{} ({:.0}%)",
|
||
par,
|
||
n_cores,
|
||
par / n_cores as f64 * 100.0
|
||
)
|
||
}
|
||
|
||
// ── Display ───────────────────────────────────────────────────────────────────
|
||
|
||
// ── MemoryBudget ──────────────────────────────────────────────────────────────
|
||
|
||
struct BudgetInner {
|
||
remaining: u64,
|
||
active: usize,
|
||
peak_active: usize,
|
||
}
|
||
|
||
/// Counting semaphore that limits total concurrent estimated memory usage.
|
||
///
|
||
/// Each worker acquires a cost (bytes) before starting and releases it on
|
||
/// completion. Non-deadlock guarantee: when no worker is active the next
|
||
/// acquire always succeeds regardless of cost vs. remaining budget.
|
||
pub struct MemoryBudget {
|
||
total: u64,
|
||
inner: Mutex<BudgetInner>,
|
||
condvar: Condvar,
|
||
}
|
||
|
||
impl MemoryBudget {
|
||
pub fn new(total: u64) -> Self {
|
||
Self {
|
||
total,
|
||
inner: Mutex::new(BudgetInner {
|
||
remaining: total,
|
||
active: 0,
|
||
peak_active: 0,
|
||
}),
|
||
condvar: Condvar::new(),
|
||
}
|
||
}
|
||
|
||
pub fn acquire(&self, cost: u64) {
|
||
let mut g = self.inner.lock().unwrap();
|
||
loop {
|
||
if g.active == 0 || g.remaining >= cost {
|
||
g.remaining = g.remaining.saturating_sub(cost);
|
||
g.active += 1;
|
||
g.peak_active = g.peak_active.max(g.active);
|
||
return;
|
||
}
|
||
g = self.condvar.wait(g).unwrap();
|
||
}
|
||
}
|
||
|
||
pub fn release(&self, cost: u64) {
|
||
let mut g = self.inner.lock().unwrap();
|
||
g.remaining = (g.remaining + cost).min(self.total);
|
||
g.active -= 1;
|
||
self.condvar.notify_all();
|
||
}
|
||
|
||
pub fn total(&self) -> u64 {
|
||
self.total
|
||
}
|
||
pub fn active(&self) -> usize {
|
||
self.inner.lock().unwrap().active
|
||
}
|
||
pub fn remaining(&self) -> u64 {
|
||
self.inner.lock().unwrap().remaining
|
||
}
|
||
pub fn peak_active(&self) -> usize {
|
||
self.inner.lock().unwrap().peak_active
|
||
}
|
||
}
|
||
|
||
// ── Display ───────────────────────────────────────────────────────────────────
|
||
|
||
impl fmt::Display for Reporter {
|
||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||
if self.stages.is_empty() {
|
||
return Ok(());
|
||
}
|
||
|
||
let n_cores = effective_parallelism();
|
||
|
||
// column widths
|
||
let nw = self
|
||
.stages
|
||
.iter()
|
||
.map(|s| s.label.len())
|
||
.max()
|
||
.unwrap_or(5)
|
||
.max(5);
|
||
// efficiency col: worst-case width for this run's n_cores value
|
||
let ew = format!("{:.1}×/{} (100%)", 99.9f64, n_cores).len();
|
||
|
||
let sep_w = nw + 2 + 7 + 2 + ew + 2 + 8 + 2 + 12;
|
||
let sep = "─".repeat(sep_w);
|
||
|
||
// header
|
||
writeln!(
|
||
f,
|
||
"{:<nw$} {:>7} {:>ew$} {:>8} status",
|
||
"stage", "wall", "efficiency", "peak RSS"
|
||
)?;
|
||
writeln!(f, "{sep}")?;
|
||
|
||
// compute all diagnoses up front (needed for both table and footnotes)
|
||
let diagnoses: Vec<Diagnosis> = self.stages.iter().map(|s| diagnose(s, n_cores)).collect();
|
||
|
||
// per-stage rows
|
||
for (s, d) in self.stages.iter().zip(diagnoses.iter()) {
|
||
writeln!(
|
||
f,
|
||
"{:<nw$} {:>7} {:>ew$} {:>8} {}",
|
||
s.label,
|
||
fmt_secs(s.wall_secs),
|
||
fmt_efficiency(s.parallelism(), n_cores),
|
||
fmt_bytes(s.max_rss_bytes),
|
||
d.tag,
|
||
)?;
|
||
}
|
||
|
||
// totals
|
||
let tw = self.stages.iter().map(|s| s.wall_secs).sum::<f64>();
|
||
let tu = self.stages.iter().map(|s| s.user_secs).sum::<f64>();
|
||
let ts = self.stages.iter().map(|s| s.sys_secs).sum::<f64>();
|
||
let trss = self
|
||
.stages
|
||
.iter()
|
||
.map(|s| s.max_rss_bytes)
|
||
.max()
|
||
.unwrap_or(0);
|
||
let tpar = if tw > 1e-9 { (tu + ts) / tw } else { 0.0 };
|
||
|
||
writeln!(f, "{sep}")?;
|
||
writeln!(
|
||
f,
|
||
"{:<nw$} {:>7} {:>ew$} {:>8}",
|
||
"TOTAL",
|
||
fmt_secs(tw),
|
||
fmt_efficiency(tpar, n_cores),
|
||
fmt_bytes(trss),
|
||
)?;
|
||
|
||
// bottleneck footnotes (only if at least one anomaly detected)
|
||
let bottlenecks: Vec<(&str, &str)> = self
|
||
.stages
|
||
.iter()
|
||
.zip(diagnoses.iter())
|
||
.filter_map(|(s, d)| d.detail.as_deref().map(|det| (s.label.as_str(), det)))
|
||
.collect();
|
||
|
||
if !bottlenecks.is_empty() {
|
||
writeln!(f, "\nBottlenecks:")?;
|
||
for (label, detail) in &bottlenecks {
|
||
writeln!(f, " {label} — {detail}")?;
|
||
}
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
}
|