Count I/O threads in /proc instead of asking backends

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 <iothread id>`. 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

src/backends/qemu/client.rs

@@ -150,13 +150,11 @@ impl QemuInstanceClient {
     async fn snapshot_from_cached_topology(
         &self,
     ) -> Result<ThreadPoolSnapshot, BackendClientError> {
-        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,

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 <id>`, provided threads are named
+/// (`-name debug-threads=on`, which libvirt sets).
+const IOTHREAD_NAME_PATTERN: &str = "^IO ";
+
 fn build_qemu_backend(dir: &Path) -> Result<Box<dyn Backend>, 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
     }

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<ThreadPoolSnapshot, BackendClientError> {
             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<ThreadPoolSnapshot, BackendClientError> {
             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<AtomicUsize>,
     }
 
@@ -1569,8 +1569,7 @@ mod tests {
 
         async fn get_thread_pool_snapshot(&self) -> Result<ThreadPoolSnapshot, BackendClientError> {
             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<AtomicUsize>) -> Arc<Instance> {
+    fn instance(id: &str, closed: Arc<AtomicUsize>) -> Arc<Instance> {
         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<Instance>,
+            _context: &EngineTickContext,
+        ) -> ScaleAction {
+            ScaleAction::Up(2)
+        }
+
+        async fn on_instance_added(&self, _instance: &Arc<Instance>) {}
+
+        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);
+    }
 }

src/engines/mod.rs

@@ -342,7 +342,7 @@ mod tests {
 
         async fn get_thread_pool_snapshot(&self) -> Result<ThreadPoolSnapshot, BackendClientError> {
             Ok(ThreadPoolSnapshot {
-                thread_count: 1,
+                thread_count: None,
                 vcpu_count: 2,
                 perf: None,
                 per_thread_util: Some(0.0),

src/engines/threshold.rs

@@ -896,7 +896,7 @@ mod tests {
 
         async fn get_thread_pool_snapshot(&self) -> Result<ThreadPoolSnapshot, BackendClientError> {
             Ok(ThreadPoolSnapshot {
-                thread_count: 1,
+                thread_count: None,
                 vcpu_count: 8,
                 perf: None,
                 per_thread_util: Some(0.0),