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.
This commit is contained in:
Generated
+1
-1
@@ -1480,7 +1480,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "obikmer"
|
||||
version = "1.3.3"
|
||||
version = "1.3.4"
|
||||
dependencies = [
|
||||
"clap",
|
||||
"csv",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "obikmer"
|
||||
version = "1.3.3"
|
||||
version = "1.3.4"
|
||||
edition = "2024"
|
||||
|
||||
[[bin]]
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
+38
-16
@@ -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<f64> {
|
||||
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<f64> {
|
||||
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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user