diff --git a/README.md b/README.md index 09107c48..6e2512b7 100644 --- a/README.md +++ b/README.md @@ -92,9 +92,15 @@ before the including file's keys are applied. ## Thread-name filtering -Backends may select fallback `/proc` CPU samples with regular expressions and -an exact-name ignore list. Exact ignores take precedence; when no match -expressions are configured, every nonignored task is sampled. +Each backend tells the controller which of its tasks are I/O workers, using +regular expressions and an exact-name ignore list. Exact ignores take +precedence; when no match expressions are configured, every nonignored task is +a worker. On every refresh the controller counts the matching tasks in `/proc` +to determine the current thread count, and samples their CPU usage unless the +backend reports per-thread utilisation itself. Backends may also report the +thread count, but the controller currently ignores it. If no task matches, the +controller keeps the previous count. Until a VM's threads have been counted, +its initial pool is not classified and no engine decision is applied to it. ## Scale-up state machine @@ -156,6 +162,10 @@ through `virDomainQemuMonitorCommand`. Its configuration is loaded from `backends.d/qemu.json`. Named IOThreads and virtqueue mappings can be inspected or changed through the backend CLI and D-Bus operations. +The I/O workers are the QEMU tasks named `IO `. QEMU names its +threads only when started with `-name debug-threads=on`, which libvirt sets; +without it no task matches and the VM is never scaled. + ## Cgroup throttling With `refresh_cgroup_on_each_read` enabled, each refresh reads the VM process's diff --git a/src/backends/qemu/client.rs b/src/backends/qemu/client.rs index cd72439a..eb8a3d36 100644 --- a/src/backends/qemu/client.rs +++ b/src/backends/qemu/client.rs @@ -150,13 +150,11 @@ impl QemuInstanceClient { async fn snapshot_from_cached_topology( &self, ) -> Result { - let thread_count = { - let s = self.inner.lock().await; - s.iothreads.len() as u32 - } - .max(1); - - if thread_count == 1 { + let thread_count = match self.inner.lock().await.iothreads.len() { + 0 => None, + count => Some(count as u32), + }; + if thread_count.is_none() { tracing::debug!( target: "qemu", uuid = %self.uuid, @@ -319,9 +317,8 @@ impl InstanceClient for QemuInstanceClient { { perf.per_vq_depth = Some(topo.vq_inflight.clone()); } - let thread_count = topo.iothreads.len() as u32; Ok(ThreadPoolSnapshot { - thread_count, + thread_count: Some(topo.iothreads.len() as u32), perf, per_thread_util, vcpu_count: self.vcpu_count, diff --git a/src/backends/qemu/mod.rs b/src/backends/qemu/mod.rs index 51265dc0..d616c606 100644 --- a/src/backends/qemu/mod.rs +++ b/src/backends/qemu/mod.rs @@ -32,7 +32,7 @@ use serde::{Deserialize, Serialize}; use crate::{ backends::{Backend, BackendClientError, BackendRegistration}, config::{ConfigError, load_json_with_includes}, - instance::{Instance, InstanceClient}, + instance::{Instance, InstanceClient, ThreadNameFilter}, util::Path, }; @@ -42,6 +42,10 @@ pub mod libvirt; const BACKEND_NAME: &str = "qemu"; +/// Task name QEMU gives each IOThread, `IO `, provided threads are named +/// (`-name debug-threads=on`, which libvirt sets). +const IOTHREAD_NAME_PATTERN: &str = "^IO "; + fn build_qemu_backend(dir: &Path) -> Result, BackendClientError> { Ok(Box::new(QemuBackend::from_config_dir(dir)?)) } @@ -178,6 +182,8 @@ impl Backend for QemuBackend { return Vec::new(); } }; + let iothread_filter = ThreadNameFilter::new(IOTHREAD_NAME_PATTERN, &[]) + .expect("IOTHREAD_NAME_PATTERN is a valid regex"); let mut out = Vec::with_capacity(uuids.len()); for uuid in uuids { let pid = conn.qemu_host_pid(&uuid).await.unwrap_or(0); @@ -212,12 +218,10 @@ impl Backend for QemuBackend { // it is purely informational (log context), so we // stuff the libvirt URI in there. let sock_path = Path::new(&self.cfg.libvirt_uri); - out.push(Arc::new(Instance::new( - uuid.to_string(), - sock_path, - pid, - client, - ))); + out.push(Arc::new( + Instance::new(uuid.to_string(), sock_path, pid, client) + .with_thread_name_filter(iothread_filter.clone()), + )); } out } diff --git a/src/controller.rs b/src/controller.rs index e5026061..2f562de9 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -604,7 +604,8 @@ impl Controller { }; let status = instance.status.read().await; let current = status.thread_count; - if target == current { + // Zero means no refresh has counted the I/O threads yet. + if current == 0 || target == current { continue; } plan.push((target > current, sequence, item.instance_id, action)); @@ -1215,7 +1216,7 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: self.target.load(Ordering::Relaxed), + thread_count: None, vcpu_count: 4, perf: None, per_thread_util: None, @@ -1503,7 +1504,7 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: self.target.load(Ordering::Relaxed), + thread_count: None, vcpu_count: 16, perf: None, per_thread_util: Some(0.0), @@ -1557,7 +1558,6 @@ mod tests { } struct SnapshotClient { - threads: u32, closed: Arc, } @@ -1569,8 +1569,7 @@ 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 + thread_count: None, vcpu_count: 2, perf: None, per_thread_util: None, @@ -1631,12 +1630,12 @@ mod tests { } } - fn instance(id: &str, threads: u32, closed: Arc) -> Arc { + fn instance(id: &str, closed: Arc) -> Arc { Arc::new(Instance::new( id.to_string(), Path::new(""), 1, - SnapshotClient { threads, closed }, + SnapshotClient { closed }, )) } @@ -1663,7 +1662,7 @@ mod tests { .unwrap(); let closed = Arc::new(AtomicUsize::new(0)); - let first = instance("vm-a", 1, Arc::clone(&closed)); + let first = instance("vm-a", Arc::clone(&closed)); let (n_added, n_removed) = controller .sync_instances(vec![Arc::clone(&first)]) .await @@ -1672,7 +1671,7 @@ mod tests { assert_eq!(controller.instances.len(), 1); assert_eq!(added.load(Ordering::Relaxed), 1); - let duplicate = instance("vm-a", 9, Arc::clone(&closed)); + let duplicate = instance("vm-a", Arc::clone(&closed)); let (n_added, n_removed) = controller.sync_instances(vec![duplicate]).await.unwrap(); assert_eq!((n_added, n_removed), (0, 0)); assert_eq!(controller.instances.len(), 1); @@ -1708,7 +1707,7 @@ mod tests { .unwrap(); let closed = Arc::new(AtomicUsize::new(0)); - let ok = instance("ok", 4, Arc::clone(&closed)); + let ok = instance("ok", Arc::clone(&closed)); let bad = Arc::new(Instance::new( "bad".to_string(), Path::new(""), @@ -1722,11 +1721,69 @@ mod tests { assert_eq!(controller.tick_index, 1); assert_eq!(controller.instances.len(), 1); assert!(controller.instances.contains_key("ok")); - assert_eq!( - controller.instances["ok"].status.read().await.thread_count, - 4 - ); + assert!(controller.instances["ok"].status.read().await.alive); assert_eq!(evaluated.load(Ordering::Relaxed), 1); assert_eq!(removed.load(Ordering::Relaxed), 1); } + + struct ProposingEngine; + + #[async_trait] + impl ScalingEngine for ProposingEngine { + fn name(&self) -> &'static str { + "proposing" + } + + fn dump_config(&self) -> serde_json::Value { + serde_json::Value::Null + } + + async fn evaluate( + &self, + _instance: &Arc, + _context: &EngineTickContext, + ) -> ScaleAction { + ScaleAction::Up(2) + } + + async fn on_instance_added(&self, _instance: &Arc) {} + + async fn on_instance_removed(&self, _instance_id: &str) {} + } + + /// Tests that no engine decision is applied to a VM whose I/O threads + /// haven't been counted in `/proc` yet. + #[test(tokio::test)] + async fn test_tick_skips_vm_with_uncounted_threads() { + let state_dir = tempfile::tempdir().unwrap(); + let cfg = Config { + vm_state_path: Path::new(&state_dir.path().join("ownership.json")), + cooldown_secs: 0.0, + host_cpu_scale_up_ceiling: 0.0, + ..Default::default() + }; + let mut controller = Controller::new(cfg, Box::new(ProposingEngine)).unwrap(); + let target = Arc::new(AtomicU32::new(7)); + // PID 0 has no `/proc` entry, so refreshing never counts its threads. + let instance = Arc::new(Instance::new( + "vm-uncounted".to_string(), + Path::new(""), + 0, + VcpuLimitedClient { + target: Arc::clone(&target), + }, + )); + controller + .sync_instances(vec![Arc::clone(&instance)]) + .await + .unwrap(); + instance.status.write().await.ownership_classification = Some(true); + + controller.tick().await.unwrap(); + assert_eq!(target.load(Ordering::Relaxed), 7); + + instance.status.write().await.thread_count = 1; + controller.tick().await.unwrap(); + assert_eq!(target.load(Ordering::Relaxed), 2); + } } diff --git a/src/engines/mod.rs b/src/engines/mod.rs index 15b8c4c9..60325e3a 100644 --- a/src/engines/mod.rs +++ b/src/engines/mod.rs @@ -342,7 +342,7 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: 1, + thread_count: None, vcpu_count: 2, perf: None, per_thread_util: Some(0.0), diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 65d358ae..d1539289 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -896,7 +896,7 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: 1, + thread_count: None, vcpu_count: 8, perf: None, per_thread_util: Some(0.0), diff --git a/src/instance.rs b/src/instance.rs index 8ce74b2f..bddc91af 100644 --- a/src/instance.rs +++ b/src/instance.rs @@ -125,7 +125,8 @@ pub struct Instance { pub sock_path: Path, /// Backend process ID. pub pid: i32, - /// Backend-selected task names included in fallback CPU sampling. + /// Backend-selected names of the I/O worker tasks, which the controller + /// counts and, unless the backend reports their utilisation, samples. pub thread_name_filter: ThreadNameFilter, /// Per-VM backend implementation. pub client: Box, @@ -196,7 +197,7 @@ impl Instance { } } - /// Replace the default all-task CPU sampling filter. + /// Replace the default filter, which matches every backend task. pub fn with_thread_name_filter(mut self, filter: ThreadNameFilter) -> Self { self.thread_name_filter = filter; self @@ -229,20 +230,16 @@ impl Instance { snapshot: ThreadPoolSnapshot, refresh_cgroup: bool, ) -> bool { - let cpu = if snapshot.per_thread_util.is_none() { - match read_cpu_sample(self.pid, &self.thread_name_filter) { - Ok(sample) => Some(sample), - Err(error) => { - tracing::warn!( - target: "controller", - %error, - "failed to sample backend task CPU" - ); - None - } + let cpu = match read_cpu_sample(self.pid, &self.thread_name_filter) { + Ok(sample) => Some(sample), + Err(error) => { + tracing::warn!( + target: "controller", + %error, + "failed to sample backend I/O threads" + ); + None } - } else { - None }; let previous_cgroup = self.status.read().await.cgroup; let cgroup = if refresh_cgroup || previous_cgroup.is_none() { @@ -250,16 +247,21 @@ impl Instance { } else { previous_cgroup }; + let thread_count = cpu.as_ref().map(|sample| sample.thread_count); let now = Instant::now(); let mut status = self.status.write().await; - if status.ownership_classification.is_none() { - status.ownership_classification = Some( - self.client - .initial_pool_is_managed(snapshot.thread_count, snapshot.vcpu_count), - ); + // Without a thread count, keep the previous one and classify the + // initial pool once a later poll observes it. + if let Some(thread_count) = thread_count { + if status.ownership_classification.is_none() { + status.ownership_classification = Some( + self.client + .initial_pool_is_managed(thread_count, snapshot.vcpu_count), + ); + } + status.thread_count = thread_count; } Self::update_perf_rates(&mut status, &snapshot.perf, now); - status.thread_count = snapshot.thread_count; status.vcpu_count = snapshot.vcpu_count; status.perf = snapshot.perf.clone(); status.read_latency_us = snapshot.perf.as_ref().and_then(|perf| perf.read_latency_us); @@ -299,12 +301,10 @@ impl Instance { None => 0, }; if let Some(per_thread_util) = backend_util { - status.rolling.push_from_backend_util( - now, - io_ops_total, - per_thread_util, - snapshot.thread_count, - ); + let thread_count = status.thread_count; + status + .rolling + .push_from_backend_util(now, io_ops_total, per_thread_util, thread_count); } else if let Some(current) = cpu.as_ref() { status .rolling @@ -576,8 +576,11 @@ impl InstancePerfSample { /// One backend snapshot consumed by the controller and engine. #[derive(Debug, Clone)] pub struct ThreadPoolSnapshot { - /// Number of active I/O workers. - pub thread_count: u32, + /// Number of active I/O workers, if the backend knows it. + /// + /// The controller ignores this: it counts the I/O workers itself, as the + /// backend tasks matching [`Instance::thread_name_filter`] in `/proc`. + pub thread_count: Option, /// Current backend-reported vCPU count. pub vcpu_count: u32, /// Performance counters captured with the current worker count. @@ -761,9 +764,7 @@ mod tests { use super::*; use crate::backends::BackendClientError; - struct SnapshotClient { - threads: u32, - } + struct SnapshotClient; #[async_trait] impl InstanceClient for SnapshotClient { @@ -773,7 +774,7 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: self.threads, + thread_count: None, vcpu_count: 2, perf: None, per_thread_util: None, @@ -802,20 +803,66 @@ mod tests { } } - /// Test that a successful client snapshot updates - /// alive/thread_count state. + /// Test that a successful client snapshot marks the instance alive + /// and updates its state. #[tokio::test] async fn refresh_state_applies_successful_snapshot() { - let instance = Instance::new( - "vm-ok".to_string(), - Path::new(""), - 7, - SnapshotClient { threads: 3 }, - ); + let instance = Instance::new("vm-ok".to_string(), Path::new(""), 7, SnapshotClient); assert!(instance.refresh_state(false).await); let status = instance.status.read().await; assert!(status.alive); - assert_eq!(status.thread_count, 3); + assert_eq!(status.vcpu_count, 2); + } + + /// Tests that refreshing counts the backend tasks matching the thread + /// name filter in `/proc`, and that the previous count is kept when no + /// task matches. + #[tokio::test] + async fn test_refresh_state_counts_threads_from_procfs() { + const WORKERS: usize = 2; + let barrier = Arc::new(std::sync::Barrier::new(WORKERS + 1)); + let workers = (0..WORKERS) + .map(|_| { + let barrier = Arc::clone(&barrier); + std::thread::Builder::new() + .name("itc-procfs-wkr".to_string()) + .spawn(move || { + barrier.wait(); + barrier.wait(); + }) + .unwrap() + }) + .collect::>(); + barrier.wait(); + + let pid = i32::try_from(std::process::id()).unwrap(); + let instance = Instance::new("vm-procfs".to_string(), Path::new(""), pid, SnapshotClient) + .with_thread_name_filter(ThreadNameFilter::new("^itc-procfs-wkr$", &[]).unwrap()); + assert!(instance.refresh_state(false).await); + { + let status = instance.status.read().await; + assert_eq!(status.thread_count, 2); + assert_eq!(status.vcpu_count, 2); + assert_eq!(status.ownership_classification, Some(true)); + } + + barrier.wait(); + for worker in workers { + worker.join().unwrap(); + } + + let unmatched = Instance::new( + "vm-unmatched".to_string(), + Path::new(""), + pid, + SnapshotClient, + ) + .with_thread_name_filter(ThreadNameFilter::new("^itc-no-such-thr$", &[]).unwrap()); + assert!(unmatched.refresh_state(false).await); + let status = unmatched.status.read().await; + assert!(status.alive); + assert_eq!(status.thread_count, 0); + assert_eq!(status.ownership_classification, None); } /// Test that client errors mark the instance dead and close the @@ -840,12 +887,7 @@ mod tests { /// Test that `Display` for an instance prints the bare VM id. #[test] fn display_trims_instance_id() { - let instance = Instance::new( - " vm-1 ".to_string(), - Path::new(""), - 1, - SnapshotClient { threads: 1 }, - ); + let instance = Instance::new(" vm-1 ".to_string(), Path::new(""), 1, SnapshotClient); assert_eq!(instance.to_string(), "vm-1"); }