From c0e1968654415c5e447ceff69948885ae8ec901c Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Fri, 11 Sep 2026 09:44:41 +0200 Subject: [PATCH] Feat: Introduce resource profiling and monitoring APIs This change introduces a comprehensive profiling system, allowing for periodic resource usage monitoring (CPU, I/O, memory) via a background thread. It also exposes new public APIs for budget management, locking, NUMA partitioning, and progress tracking. --- src/Cargo.lock | 2 +- src/obikmer/Cargo.toml | 2 +- src/obikmer/src/main.rs | 4 +++ src/obisys/src/lib.rs | 2 ++ src/obisys/src/profiler.rs | 50 ++++++++++++++++++++++++++++++++++ src/obisys/src/resources.rs | 54 ++++++++++++++++++++++++++----------- 6 files changed, 96 insertions(+), 18 deletions(-) create mode 100644 src/obisys/src/profiler.rs diff --git a/src/Cargo.lock b/src/Cargo.lock index 947c0a9e..ed71e1af 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -1480,7 +1480,7 @@ dependencies = [ [[package]] name = "obikmer" -version = "1.3.3" +version = "1.3.4" dependencies = [ "clap", "csv", diff --git a/src/obikmer/Cargo.toml b/src/obikmer/Cargo.toml index 78b45a85..5ad6428a 100644 --- a/src/obikmer/Cargo.toml +++ b/src/obikmer/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "obikmer" -version = "1.3.3" +version = "1.3.4" edition = "2024" [[bin]] diff --git a/src/obikmer/src/main.rs b/src/obikmer/src/main.rs index ae68c132..422af2d5 100644 --- a/src/obikmer/src/main.rs +++ b/src/obikmer/src/main.rs @@ -73,6 +73,10 @@ fn main() { warn!("could not configure the global rayon pool: {e}"); } + // Always-on, near-zero-cost when disabled: RUST_LOG=obisys::profiler=trace + // to see periodic CPU/IO/memory snapshots on a production run. + obisys::start_profiler(std::time::Duration::from_secs(5)); + match cli.command { Commands::Index(args) => cmd::index::run(args), Commands::Superkmer(args) => cmd::superkmer::run(args), diff --git a/src/obisys/src/lib.rs b/src/obisys/src/lib.rs index 63da76f8..da45400f 100644 --- a/src/obisys/src/lib.rs +++ b/src/obisys/src/lib.rs @@ -4,6 +4,7 @@ mod budget; mod lock; mod numa; +mod profiler; mod progress; mod resources; mod stage; @@ -11,6 +12,7 @@ mod stage; pub use budget::MemoryBudget; pub use lock::DirLock; pub use numa::PartitionRunner; +pub use profiler::start as start_profiler; pub use progress::{Progress, TracedBar, progress_bar, spinner}; pub use resources::{ CpuSample, IoSample, available_memory_bytes, cpu_budget, effective_parallelism, peak_rss_bytes, diff --git a/src/obisys/src/profiler.rs b/src/obisys/src/profiler.rs new file mode 100644 index 00000000..031d6784 --- /dev/null +++ b/src/obisys/src/profiler.rs @@ -0,0 +1,50 @@ +use std::thread; +use std::time::Duration; + +use tracing::{Level, trace}; + +use crate::{CpuSample, IoSample, available_memory_bytes, cpu_budget, peak_rss_bytes}; + +/// Spawn a detached background thread that logs a periodic snapshot of +/// process-wide resource usage — CPU cores used (and as a % of the process's +/// `--cpu-max` budget), I/O throughput, peak RSS, and remaining host/cgroup +/// memory — at `trace!` level, every `interval`. +/// +/// Meant to run for the whole process lifetime as always-on instrumentation +/// installed once at startup (no stop handle). The sampling itself (a +/// `getrusage` call and a `/proc/self/io` read) is cheap, but +/// `available_memory_bytes()` builds a fresh `sysinfo::System` snapshot each +/// time, so the whole body is skipped unless the `obisys::profiler` target +/// is actually enabled at `TRACE`. +/// +/// Enable with `RUST_LOG=obisys::profiler=trace` (combine with +/// `RUST_LOG=debug,obisys::profiler=trace` to also get the per-partition +/// `debug!` logs that already exist elsewhere) to profile a run directly on +/// production hardware without attaching a separate profiler. +pub fn start(interval: Duration) { + let budget = cpu_budget().max(1) as f64; + thread::Builder::new() + .name("obisys-profiler".into()) + .spawn(move || { + let mut cpu = CpuSample::now(); + let mut io = IoSample::now(); + loop { + thread::sleep(interval); + if !tracing::enabled!(target: "obisys::profiler", Level::TRACE) { + continue; + } + let cores = cpu.sample_cores().unwrap_or(0.0); + let io_mb_s = io.sample_rate().unwrap_or(0.0) / 1_000_000.0; + trace!( + target: "obisys::profiler", + cpu_pct = format_args!("{:.0}%", cores / budget * 100.0), + cpu_cores = format_args!("{cores:.1}"), + io_mb_s = format_args!("{io_mb_s:.1}"), + rss_peak_mb = peak_rss_bytes() / 1_000_000, + mem_avail_mb = available_memory_bytes() / 1_000_000, + "profiler snapshot", + ); + } + }) + .expect("failed to spawn profiler thread"); +} diff --git a/src/obisys/src/resources.rs b/src/obisys/src/resources.rs index 6f8d4ad0..989468bf 100644 --- a/src/obisys/src/resources.rs +++ b/src/obisys/src/resources.rs @@ -264,18 +264,31 @@ impl CpuSample { cpu / (wall * n_cores as f64) } - pub fn do_i_activate(&mut self, threshold: f64) -> bool { + /// Cores of CPU time consumed per wall-clock second since the last call + /// (or since [`now`](Self::now) for the first call) — an absolute core + /// count, unlike [`cpu_efficiency`](Self::cpu_efficiency) which + /// normalizes to `n_cores` and never resets its baseline. Resets the + /// window on every call that returns `Some`; returns `None` (state left + /// untouched, so the window keeps accumulating) when less than 100 ms + /// have elapsed — too short to be meaningful. + pub fn sample_cores(&mut self) -> Option { 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; + return None; } - let n = CpuSample::now(); let delta_ru = (n.user_secs - self.user_secs) + (n.sys_secs - self.sys_secs); + self.user_secs = n.user_secs; + self.sys_secs = n.sys_secs; + self.wall = n.wall; + Some(delta_ru / delta_wall) + } + + pub fn do_i_activate(&mut self, threshold: f64) -> bool { + let Some(efficiency) = self.sample_cores() else { + return false; + }; - let efficiency = delta_ru / delta_wall; let activate = 0f64.max(efficiency - self.previous) >= threshold; debug!( @@ -286,9 +299,6 @@ impl CpuSample { activate ); self.previous = efficiency; - self.user_secs = n.user_secs; - self.sys_secs = n.sys_secs; - self.wall = n.wall; activate } @@ -351,18 +361,32 @@ impl IoSample { 0 } + /// Bytes/second submitted to the block layer since the last call (or + /// since [`now`](Self::now) for the first call). Resets the window on + /// every call that returns `Some`; returns `None` (state left + /// untouched) when less than 100 ms have elapsed — too short to be + /// meaningful. + pub fn sample_rate(&mut self) -> Option { + let elapsed = self.wall.elapsed().as_secs_f64(); + if elapsed < 0.1 { + return None; + } + let n = Self::read_bytes(); + let rate = n.saturating_sub(self.bytes) as f64 / elapsed; + self.bytes = n; + self.wall = Instant::now(); + Some(rate) + } + /// 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 { + let Some(rate) = self.sample_rate() else { 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 { @@ -374,8 +398,6 @@ impl IoSample { self.previous_rate, rate, activate ); self.previous_rate = rate; - self.bytes = n; - self.wall = Instant::now(); activate }