Merge pull request 'Feat: Introduce resource profiling and monitoring APIs' (#75) from push-sktolxuyskpp into main

Reviewed-on: #75

src/Cargo.lock

@@ -1480,7 +1480,7 @@ dependencies = [
 
 [[package]]
 name = "obikmer"
-version = "1.3.3"
+version = "1.3.4"
 dependencies = [
  "clap",
  "csv",

src/obikmer/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "obikmer"
-version = "1.3.3"
+version = "1.3.4"
 edition = "2024"
 
 [[bin]]

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),

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,

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");
+}

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<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
+            return None;
-            // 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);
+        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();
+        let Some(rate) = self.sample_rate() else {
-        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 {
@@ -374,8 +398,6 @@ impl IoSample {
             self.previous_rate, rate, activate
         );
         self.previous_rate = rate;
-        self.bytes = n;
-        self.wall = Instant::now();
 
         activate
     }