From f649d670dcd0c49c0e65decf585e55bd49d71d08 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Mon, 6 Jul 2026 09:29:09 +0000 Subject: [PATCH] add per-task CPU utilisation This allows calculating average vs. median CPU utilisation, which helps identifying load imbalances. This new piece of information is now printed in the per-VM periodic message (when enabled). Specifically, a new field is added in the 'cpu' field containing the median, before: INFO ... cpu=average per-thread/total worker CPU percent after: INFO ... cpu=average per-thread/median/total worker CPU percent Signed-off-by: Thanos Makatos --- Cargo.lock | 227 +++++++++++++++++++++++++++++++++++++++++++++- Cargo.toml | 1 + README.md | 4 +- src/controller.rs | 16 +++- src/instance.rs | 153 ++++++++++++++++++++++++++----- 5 files changed, 371 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b6c759b..c58d84a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -107,6 +107,15 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "autocfg" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0dde43e75fd43e8a1bf86103336bc699aa8d17ad1be60c76c0bdfd4828e19b78" +dependencies = [ + "autocfg 1.5.1", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -204,6 +213,15 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e78417baa3b3114dc0e95e7357389a249c4da97c3c2b540700079db6171bfd7" +[[package]] +name = "cloudabi" +version = "0.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddfc5b9aa5d4507acaf872de71051dfd0e309860e88966e1051e462a077aac4f" +dependencies = [ + "bitflags 1.3.2", +] + [[package]] name = "colorchoice" version = "1.0.5" @@ -486,6 +504,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "fuchsia-cprng" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a06f77d526c1a601b7c4cdd98f54b5eaabffc14d5f2f0296febdc7f357c6d3ba" + [[package]] name = "futures-core" version = "0.3.34" @@ -683,6 +707,7 @@ dependencies = [ "sanitize-filename", "serde", "serde_json", + "statistical", "tempfile", "test-log", "thiserror", @@ -831,7 +856,7 @@ version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" dependencies = [ - "autocfg", + "autocfg 1.5.1", ] [[package]] @@ -868,6 +893,81 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8536030f9fea7127f841b45bb6243b27255787fb4eb83958aa1ef9d2fdc0c36" +dependencies = [ + "num-bigint", + "num-complex", + "num-integer", + "num-iter", + "num-rational", + "num-traits", +] + +[[package]] +name = "num-bigint" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "090c7f9998ee0ff65aa5b723e4009f7b217707f1fb5ea551329cc4d6231fb304" +dependencies = [ + "autocfg 1.5.1", + "num-integer", + "num-traits", +] + +[[package]] +name = "num-complex" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6b19411a9719e753aff12e5187b74d60d3dc449ec3f4dc21e3989c3f554bc95" +dependencies = [ + "autocfg 1.5.1", + "num-traits", +] + +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + +[[package]] +name = "num-iter" +version = "0.1.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c92800bd69a1eac91786bcfe9da64a897eb72911b8dc3095decbd07429e8048b" +dependencies = [ + "num-integer", + "num-traits", +] + +[[package]] +name = "num-rational" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c000134b5dbf44adc5cb772486d335293351644b801551abe8f75c84cfa4aef" +dependencies = [ + "autocfg 1.5.1", + "num-bigint", + "num-integer", + "num-traits", +] + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg 1.5.1", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -992,6 +1092,112 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "rand" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d71dacdc3c88c1fde3885a3be3fbab9f35724e6ce99467f7d9c5026132184ca" +dependencies = [ + "autocfg 0.1.8", + "libc", + "rand_chacha", + "rand_core 0.4.3", + "rand_hc", + "rand_isaac", + "rand_jitter", + "rand_os", + "rand_pcg", + "rand_xorshift", + "winapi", +] + +[[package]] +name = "rand_chacha" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "556d3a1ca6600bfcbab7c7c91ccb085ac7fbbcd70e008a98742e7847f4f7bcef" +dependencies = [ + "autocfg 0.1.8", + "rand_core 0.3.2", +] + +[[package]] +name = "rand_core" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96f815e01bbd9678b50d927f79aa1cf3ffdfdb1b9787317c1284dadb894ad0e8" +dependencies = [ + "rand_core 0.4.3", +] + +[[package]] +name = "rand_core" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e5937858e6fd18cd595d558f90bb5de3b72ae23f9e3763af0e805949b04ef60" + +[[package]] +name = "rand_hc" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b40677c7be09ae76218dc623efbf7b18e34bced3f38883af07bb75630a21bc4" +dependencies = [ + "rand_core 0.3.2", +] + +[[package]] +name = "rand_isaac" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ded997c9d5f13925be2a6fd7e66bf1872597f759fd9dd93513dd7e92e5a5ee08" +dependencies = [ + "rand_core 0.3.2", +] + +[[package]] +name = "rand_jitter" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1166d5c91dc97b88d1decc3285bb0a99ed84b05cfd0bc2341bdf2d43fc41e39b" +dependencies = [ + "libc", + "rand_core 0.4.3", + "winapi", +] + +[[package]] +name = "rand_os" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b75f676a1e053fc562eafbb47838d67c84801e38fc1ba459e8f180deabd5071" +dependencies = [ + "cloudabi", + "fuchsia-cprng", + "libc", + "rand_core 0.4.3", + "rdrand", + "winapi", +] + +[[package]] +name = "rand_pcg" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "abf9b09b01790cfe0364f52bf32995ea3c39f4d2dd011eac241d2914146d0b44" +dependencies = [ + "autocfg 0.1.8", + "rand_core 0.4.3", +] + +[[package]] +name = "rand_xorshift" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cbf7e9e623549b0e21f6e97cf8ecf247c1a8fd2e8a992ae265314300b2455d5c" +dependencies = [ + "rand_core 0.3.2", +] + [[package]] name = "ratatui" version = "0.29.0" @@ -1013,6 +1219,15 @@ dependencies = [ "unicode-width 0.2.0", ] +[[package]] +name = "rdrand" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "678054eb77286b51581ba43620cc911abf02758c91f93f479767aed0f90458b2" +dependencies = [ + "rand_core 0.3.2", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1276,6 +1491,16 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "statistical" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49d57902bb128e5e38b5218d3681215ae3e322d99f65d5420e9849730d2ea372" +dependencies = [ + "num", + "rand", +] + [[package]] name = "strsim" version = "0.11.1" diff --git a/Cargo.toml b/Cargo.toml index ada8ac9..01766e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -104,6 +104,7 @@ instability = "=0.3.10" # Only for the TUI: anyhow = "1.0.104" +statistical = "1.0.0" [dev-dependencies] tempfile = "=3.14.0" diff --git a/README.md b/README.md index 63445a0..ca56d95 100644 --- a/README.md +++ b/README.md @@ -145,7 +145,7 @@ occupies one line. ``` INFO vm=vm-a thr=4 iops=155593/0/0 bw_mb_s=20394/0 \ - cpu=90/358 iops_1_5_15m=154995/153840/151220 \ + cpu=90/87/358 iops_1_5_15m=154995/153840/151220 \ cpu_us_per_io_1_5_15m=23/16/19 ``` @@ -154,7 +154,7 @@ INFO vm=vm-a thr=4 iops=155593/0/0 bw_mb_s=20394/0 \ - `iops`: current read / write / other operations per second. - `iops_1_5_15m`: average total IOPS over rolling 1m / 5m / 15m windows. - `bw_mb_s`: current read / write bandwidth in MB/s. -- `cpu`: average per-thread / total pool CPU percentage. +- `cpu`: average / median / total worker CPU percentage. - `cpu_us_per_io_1_5_15m`: CPU microseconds per completed I/O over the same windows. diff --git a/src/controller.rs b/src/controller.rs index b70746b..e6b83a2 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -14,6 +14,7 @@ use std::{ use futures_util::future::join_all; use procfs::{CurrentSI, ProcError}; use serde::{Deserialize, Serialize}; +use statistical::median; use thiserror::Error; use crate::{ @@ -122,6 +123,16 @@ struct CpuStats { /// Summarise per-worker utilisation, falling back to aggregate samples. fn compute_cpu_stats(s: &crate::instance::InstanceStatus) -> CpuStats { + if !s.per_worker_util.is_empty() { + let total = s.per_worker_util.iter().sum::(); + let median = median(&s.per_worker_util); + return CpuStats { + avg_pct: (total / s.per_worker_util.len() as f64 * 100.0).round() as u64, + median_pct: (median * 100.0).round() as u64, + total_pct: (total * 100.0).round() as u64, + }; + } + CpuStats { avg_pct: (s.per_thread_util * 100.0).round() as u64, median_pct: (s.per_thread_util * 100.0).round() as u64, @@ -730,8 +741,7 @@ impl Controller { status: &InstanceStatus, iops_windows: [Option; 3], ) { - let cpu_average = status.per_thread_util.clamp(0.0, 1.0) * 100.0; - let cpu_total = cpu_average * f64::from(status.thread_count); + let cpu = compute_cpu_stats(status); tracing::info!( target: "status", vm = instance.to_string(), @@ -748,7 +758,7 @@ impl Controller { status.read_bytes_per_second / 1_000_000, status.write_bytes_per_second / 1_000_000 ), - cpu = %format!("{cpu_average:.0}/{cpu_total:.0}"), + cpu = %format!("{}/{}/{}", cpu.avg_pct, cpu.median_pct, cpu.total_pct), cpu_us_per_io_1_5_15m = %format_1_5_15(&status.rolling, |rolling, window| { rolling.cpu_us_per_io_over(window) diff --git a/src/instance.rs b/src/instance.rs index c3c4e3b..ec2a3c3 100644 --- a/src/instance.rs +++ b/src/instance.rs @@ -10,7 +10,7 @@ //! A [`crate::backends::Backend`] is the fleet-level adapter that owns //! backend-wide configuration and discovers zero or more such records. -use std::{fmt, sync::LazyLock, time::Instant}; +use std::{collections::HashMap, fmt, sync::LazyLock, time::Instant}; use async_trait::async_trait; use procfs::process::Process; @@ -221,28 +221,25 @@ impl Instance { status.vcpu_count = snapshot.vcpu_count; status.perf = snapshot.perf; status.alive = true; - if let Some(per_thread_util) = snapshot.per_thread_util { - status.per_thread_util = per_thread_util.clamp(0.0, 1.0); - } else if let (Some(prev), Some(cur)) = (status.last_cpu_sample.as_ref(), cpu.as_ref()) - && prev.thread_count == cur.thread_count - && cur.thread_count > 0 - && let Some(d_ticks) = cur.cpu_ticks.checked_sub(prev.cpu_ticks) + let backend_util = snapshot.per_thread_util.map(|util| util.clamp(0.0, 1.0)); + if let Some(backend_util) = backend_util { + status.per_worker_util.clear(); + status.last_worker_names = None; + status.per_thread_util = backend_util; + } else if let (Some(previous), Some(current)) = + (status.last_cpu_sample.as_ref(), cpu.as_ref()) { - let d_time_ticks = cur + let wall_ticks = current .sampled_at - .saturating_duration_since(prev.sampled_at) - .as_secs_f64() - * *TICKS_PER_SECOND; - if d_time_ticks > 0.0 { - let measured = d_ticks as f64 / d_time_ticks / f64::from(cur.thread_count); - // Scheduler-tick quantisation can make a short - // interval appear fractionally above a fully - // occupied CPU; bound that sampling artifact. - status.per_thread_util = measured.clamp(0.0, 1.0); + .checked_duration_since(previous.sampled_at) + .map(|elapsed| elapsed.as_secs_f64() * *TICKS_PER_SECOND) + .filter(|ticks| *ticks > 0.0); + if let Some(wall_ticks) = wall_ticks { + let worker_util = + compute_per_worker_util(&previous.per_worker, ¤t.per_worker, wall_ticks); + update_worker_utilisation(&mut status, worker_util); } } - // A changed task count invalidates the delta. Keep the last - // utilisation until the next like-for-like sample. let io_ops_total = match snapshot.perf { Some(perf) => perf .read_io_count @@ -250,7 +247,7 @@ impl Instance { .saturating_add(perf.other_io_count), None => 0, }; - if let Some(per_thread_util) = snapshot.per_thread_util { + if let Some(per_thread_util) = backend_util { status.rolling.push_from_backend_util( now, io_ops_total, @@ -265,7 +262,7 @@ impl Instance { *TICKS_PER_SECOND, ); } - status.last_cpu_sample = cpu; + status.last_cpu_sample = if backend_util.is_none() { cpu } else { None }; true } @@ -374,6 +371,10 @@ pub struct InstanceStatus { pub per_thread_util: f64, /// Previous cumulative CPU sample used to compute a delta. pub last_cpu_sample: Option, + /// Latest comparable utilisation fraction for each sampled worker. + pub per_worker_util: Vec, + /// Sorted names from the latest worker sample, for roster-change logging. + pub last_worker_names: Option>, /// Bounded 1m/5m/15m I/O and CPU history. pub rolling: RollingMetrics, /// Time of the previous backend performance snapshot. @@ -407,7 +408,7 @@ impl InstanceStatus { } /// Cumulative CPU counters sampled across one backend process. -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone)] pub struct CpuSample { /// Sum of user and system CPU ticks across sampled tasks. pub cpu_ticks: u64, @@ -415,6 +416,19 @@ pub struct CpuSample { pub sampled_at: Instant, /// Number of tasks included in the sample. pub thread_count: u32, + /// Cumulative CPU counters for individual sampled tasks. + pub per_worker: Vec, +} + +/// One task's identity, name, and cumulative CPU counter. +#[derive(Debug, Clone)] +pub struct TaskCpuSample { + /// Linux task identifier. + pub tid: i32, + /// Task name read from `/proc/.../comm`. + pub name: String, + /// Cumulative user and system CPU ticks. + pub cpu_ticks: u64, } /// Backend-neutral performance counters from one snapshot. @@ -456,6 +470,56 @@ pub struct ThreadPoolSnapshot { pub per_thread_util: Option, } +/// Store one set of per-worker utilisation samples. +fn update_worker_utilisation(status: &mut InstanceStatus, worker_util: Vec<(String, f64)>) { + let mut names = worker_util + .iter() + .map(|(name, _)| name.clone()) + .collect::>(); + names.sort(); + if status.last_worker_names.as_ref() != Some(&names) { + tracing::debug!( + target: "controller", + thread_count = names.len(), + threads = %names.join(","), + "worker thread set changed" + ); + status.last_worker_names = Some(names); + } + + status.per_worker_util = worker_util.into_iter().map(|(_, util)| util).collect(); + status.per_thread_util = if status.per_worker_util.is_empty() { + 0.0 + } else { + status.per_worker_util.iter().sum::() / status.per_worker_util.len() as f64 + }; +} + +/// Match task counters by TID and convert deltas to utilisation fractions. +fn compute_per_worker_util( + previous: &[TaskCpuSample], + current: &[TaskCpuSample], + wall_ticks: f64, +) -> Vec<(String, f64)> { + let previous_by_tid: HashMap = previous + .iter() + .map(|task| (task.tid, (task.name.as_str(), task.cpu_ticks))) + .collect(); + current + .iter() + .filter_map(|task| { + let (_, previous_ticks) = previous_by_tid + .get(&task.tid) + .filter(|(name, _)| *name == task.name.as_str())?; + let delta = task.cpu_ticks.checked_sub(*previous_ticks)?; + // Scheduler-tick quantisation can make a short interval appear + // fractionally above one fully occupied CPU; bound that artifact. + let util = (delta as f64 / wall_ticks).clamp(0.0, 1.0); + Some((task.name.clone(), util)) + }) + .collect() +} + /// Read cumulative CPU time across matching `/proc//task/*/stat` files. #[tracing::instrument(skip(filter), fields(pid))] fn read_cpu_sample( @@ -466,6 +530,7 @@ fn read_cpu_sample( let tasks = process.tasks()?; let mut cpu_ticks = 0u64; let mut thread_count = 0u32; + let mut per_worker = Vec::new(); for task in tasks { let task = match task { Ok(task) => task, @@ -495,6 +560,11 @@ fn read_cpu_sample( let task_ticks = stat.utime.saturating_add(stat.stime); cpu_ticks = cpu_ticks.saturating_add(task_ticks); thread_count += 1; + per_worker.push(TaskCpuSample { + tid: stat.pid, + name: stat.comm, + cpu_ticks: task_ticks, + }); } if thread_count == 0 { return Err(CpuSampleError::NoCpuSamples); @@ -503,6 +573,7 @@ fn read_cpu_sample( cpu_ticks, sampled_at: Instant::now(), thread_count, + per_worker, }) } @@ -531,9 +602,8 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { thread_count: self.threads, - // FIXME these weren't required, looked like broken due to rebase + vcpu_count: 2, perf: None, - vcpu_count: 1, per_thread_util: None, }) } @@ -606,4 +676,39 @@ mod tests { ); assert_eq!(instance.to_string(), "vm-1"); } + + use super::{TaskCpuSample, compute_per_worker_util}; + + fn task(tid: i32, name: &str, cpu_ticks: u64) -> TaskCpuSample { + TaskCpuSample { + tid, + name: name.to_string(), + cpu_ticks, + } + } + + /// Test that two workers sharing a name still get independent + /// CPU-delta util samples. + #[test] + fn duplicate_worker_names_keep_independent_deltas() { + let previous = vec![task(10, "worker", 100), task(11, "worker", 200)]; + let current = vec![task(11, "worker", 400), task(10, "worker", 200)]; + + let util = compute_per_worker_util(&previous, ¤t, 500.0); + assert_eq!(util.len(), 2); + assert!((util[0].1 - 0.4).abs() < f64::EPSILON); + assert!((util[1].1 - 0.2).abs() < f64::EPSILON); + } + + /// Test that a newly appeared worker does not spike util from + /// lifetime counters. + #[test] + fn new_worker_starts_without_a_lifetime_spike() { + let previous = vec![task(10, "worker0", 100)]; + let current = vec![task(10, "worker0", 200), task(11, "worker1", 900_000)]; + + let util = compute_per_worker_util(&previous, ¤t, 500.0); + assert_eq!(util.len(), 1); + assert!((util[0].1 - 0.2).abs() < f64::EPSILON); + } }