Tmp

README.md

@@ -90,6 +90,20 @@ Controller, engine, and backend JSON files may contain an `include` field. Its
 path is resolved relative to the including file, loaded recursively, and merged
 before the including file's keys are applied.
 
+## Durations
+
+Durations are integer seconds in settings ending in `_secs`. Every duration,
+including engine ones, must be a whole multiple of `scale_poll_secs`, and zero
+is allowed. Otherwise the daemon refuses to start and lists each offending
+setting with the nearest valid values. Changing only `scale_poll_secs` can
+therefore invalidate other settings.
+
+To try a new poll period without rewriting the other settings, set
+`round_up_durations` or pass `--round-up-durations`. Each offending duration is
+then rounded up to the next multiple with a warning. At startup the daemon logs
+the effective controller, engine, and backend configuration, which shows the
+rounded values.
+
 ## Thread-name filtering
 
 Backends may select fallback `/proc` CPU samples with regular expressions and
@@ -131,21 +145,21 @@ queries and thread-count changes need an independently stateful handle for each
 VM. This keeps the controller independent of libvirt, QMP, control sockets, and
 other transport details.
 
-After a successful action, the engine waits
-`scale_validation_sample_polls` complete samples. The optional
-`scale_validation_sample_polls_per_thread` setting adds that many samples for
-each thread in the resulting pool, allowing larger pools more time to settle.
+After a successful action, the engine samples for
+`scale_validation_delay_secs`. The optional
+`scale_validation_delay_per_thread_secs` setting adds that much time for each
+thread in the resulting pool, allowing larger pools more time to settle.
 A scale-up is reverted unless IOPS gained at least
 `scale_up_min_gain_percent`; a scale-down is reverted when IOPS lost more than
 `scale_down_revert_drop_percent`. Setting either percentage to zero disables
 that direction's validation.
 
 Repeated failures back off per VM. After a reverted scale-up, scale-up is
-blocked for `post_revert_up_grace_polls` plus `post_revert_up_recovery_polls`;
+blocked for `post_revert_up_grace_secs` plus `post_revert_up_recovery_secs`;
 that hold doubles with each consecutive failed scale-up, up to
-`post_revert_up_backoff_max_polls`. After a reverted scale-down, the next
-scale-down needs `scale_down_sustain_polls` low-utilisation polls doubled once
-per consecutive failure, up to `scale_down_backoff_max_polls`. A kept action
+`post_revert_up_backoff_max_secs`. After a reverted scale-down, the next
+scale-down needs `scale_down_sustain_secs` of low utilisation doubled once per
+consecutive failure, up to `scale_down_backoff_max_secs`. A kept action
 resets its direction's backoff, as does an applied action in the opposite
 direction. Setting a maximum to zero disables that direction's backoff.
 
@@ -179,33 +193,40 @@ on the `status` tracing target. With `--print-status-header`, it also emits a
 
 By default each VM's line is appended to
 `/var/log/io-thread-controller/vms/<vm>.log`, so the line itself does not repeat
-the VM identifier. Fields are positional, without keys. The examples are
-wrapped with backslashes for readability; each emitted record occupies one
-line.
+the VM identifier. Fields are positional, without keys, and separated by single
+spaces. The examples are wrapped with backslashes for readability; each emitted
+record occupies one line.
 
 ```
-2026-01-01T00:00:00.000Z status=4 155593/0/0 154995/153840/151220 \
-  20394/0 90/87/358 40/80/120/45 - 23/16/19
+2026-01-01T00:00:00.000Z status=4 155593 0 0 154995 153840 151220 \
+  20394 0 90 87 358 40 80 120 45 - - - - 23 16 19
 ```
 
 After the leading timestamp, the `status=` prefix distinguishes status records
 from `event=` records, and the fields are:
 
 1. matched worker threads.
-2. current read / write / other operations per second.
-3. average total IOPS over rolling 1m / 5m / 15m windows.
-4. current read / write bandwidth in MB/s.
-5. average / median / total worker CPU percentage.
-6. read latency p50 / p95 / p99 / average in microseconds.
-7. write latency p50 / p95 / p99 / average in microseconds.
-8. CPU microseconds per completed I/O over rolling 1m / 5m / 15m windows.
-
-A dash (`-`) means no complete sample is available for that window.
+2. current read, write, and other operations per second (3 fields).
+3. average total IOPS over rolling 1m, 5m, and 15m windows (3 fields).
+4. current read and write bandwidth in MB/s (2 fields).
+5. average, median, and total worker CPU percentage (3 fields).
+6. read latency p50, p95, p99, and average in microseconds (4 fields).
+7. write latency p50, p95, p99, and average in microseconds (4 fields).
+8. CPU microseconds per completed I/O over rolling 1m, 5m, and 15m windows (3
+   fields).
+
+A dash (`-`) means no sample is available for that field.
+
+The per-VM log also records `event=vm_added` when the daemon starts tracking
+the VM and `event=vm_removed reason=<not_discovered|refresh_failed>` when it
+stops. Every other log event that names the VM in a `vm` field, such as scaling
+decisions, blocked or failed scale operations, and backend errors, is copied to
+the VM's file as its message followed by its remaining `key=value` fields.
 
 ### Aggregate line
 
 ```
-INFO status: 1 4 154995/153840/151220
+INFO status: 1 4 154995 153840 151220
 ```
 
 `status:` is the tracing target, which separates status lines from the
@@ -226,11 +247,11 @@ you get one INFO line on the `engine` target using the same
 `vm=<id>` convention as the status lines:
 
 ```
-INFO vm=vm-1615a59c-... util=0.8955 action="up" thr=4->5
+INFO vm=vm-1615a59c-... util=89.55% action="up" thr=4->5
 ```
 
 `vm` is the same identifier used by status lines, `util` is
-the per-thread utilisation that triggered the decision, `action`
+the per-thread utilisation percentage that triggered the decision, `action`
 is `up`, `down`, or `revert`, and `thr` is the requested
 `<old>-><new>` transition.
 

STYLE.md

@@ -0,0 +1,11 @@
+# Commit messages
+
+- Commit title should start with a capital letter.
+- If you used AI, state so in the commit message, e.g.:
+  "AI-generated: everything except commit message".
+
+# Unit Tests
+
+- Unit test name should have the `test_` prefix.
+- Unit test documentation should explain what this does test tests, e.g.:
+  "Tests that foo does such and such."

io-thread-controller.d/engines/threshold.json

@@ -1,14 +1,14 @@
 {
   "scale_up_threshold_percent": 95,
-  "scale_down_sustain_polls": 12,
+  "scale_down_sustain_secs": 120,
   "max_scale_down_step": 0,
   "scale_up_min_gain_percent": 5,
   "scale_down_revert_drop_percent": 3,
-  "scale_validation_sample_polls": 3,
-  "scale_validation_sample_polls_per_thread": 0,
+  "scale_validation_delay_secs": 30,
+  "scale_validation_delay_per_thread_secs": 0,
   "block_scale_up_when_cgroup_throttled": true,
-  "post_revert_up_grace_polls": 1,
-  "post_revert_up_recovery_polls": 2,
-  "post_revert_up_backoff_max_polls": 30,
-  "scale_down_backoff_max_polls": 120
+  "post_revert_up_grace_secs": 10,
+  "post_revert_up_recovery_secs": 20,
+  "post_revert_up_backoff_max_secs": 300,
+  "scale_down_backoff_max_secs": 1200
 }

io-thread-controller.json

@@ -1,11 +1,12 @@
 {
-  "scale_poll_secs": 10.0,
+  "scale_poll_secs": 10,
+  "round_up_durations": false,
   "enable_per_vm_status_line": true,
   "enable_aggregate_status_line": true,
   "min_thread_count": 1,
   "max_thread_count": 8,
   "host_cpu_scale_up_ceiling_percent": 90,
-  "cooldown_secs": 10.0,
+  "cooldown_secs": 10,
   "engine": "threshold",
   "engine_config_dir": "/etc/io-thread-controller.d/engines",
   "backend_config_dir": "/etc/io-thread-controller.d/backends",

src/backends/mod.rs

@@ -140,7 +140,13 @@ fn is_peer_disconnect_kind(kind: ErrorKind) -> bool {
     )
 }
 
-use std::{io, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    io,
+    os::unix::fs::MetadataExt,
+    path::PathBuf,
+    sync::{Arc, LazyLock, Mutex},
+};
 
 #[cfg(feature = "qemu-backend")]
 pub mod qemu;
@@ -240,6 +246,7 @@ where
         }
     };
     let mut instances = Vec::new();
+    let mut seen = HashSet::new();
     while let Some(entry) = entries.next_entry().await? {
         match entry.metadata().await {
             Ok(metadata) => {
@@ -256,9 +263,16 @@ where
             continue;
         };
         let sock_path = Path::new(&socket_dir.join(entry.path().join(socket_name)));
-        // FIXME add unit test for this function
-        let pid = match control_socket_peer_pid(&sock_path).await {
+        let pid = match cached_control_socket_peer_pid(&sock_path).await {
             Ok(pid) => pid,
+            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                tracing::debug!(
+                    target: "controller",
+                    vm = %id,
+                    "control socket busy; retrying on next discovery"
+                );
+                continue;
+            }
             Err(error) => {
                 if error.kind() != io::ErrorKind::NotFound {
                     // FIXME ignore ENOENT
@@ -272,12 +286,74 @@ where
                 continue;
             }
         };
+        seen.insert(sock_path.to_path_buf());
         let client = make_client(sock_path.clone());
         instances.push(Arc::new(Instance::new(id, sock_path, pid, client)));
     }
+    PEER_CACHE
+        .lock()
+        .expect("lock poisoned")
+        .retain(|path, _| !path.starts_with(socket_dir) || seen.contains(path));
     Ok(instances)
 }
 
+/// Peer process last identified behind one control socket.
+#[derive(Debug, Clone, Copy)]
+struct CachedPeer {
+    /// Socket file device; a restarted backend binds a new socket file.
+    dev: u64,
+    /// Socket file inode.
+    ino: u64,
+    /// Peer process ID.
+    pid: i32,
+    /// Peer process start time, guarding against PID reuse.
+    start_time: u64,
+}
+
+/// Control-socket peers identified by earlier discovery passes, keyed by
+/// socket path.
+///
+/// Backends may listen with a minimal backlog, in which case connecting on
+/// every pass just to read the peer credentials fails with `EAGAIN` whenever
+/// an earlier connection has not been accepted yet.
+static PEER_CACHE: LazyLock<Mutex<HashMap<PathBuf, CachedPeer>>> = LazyLock::new(Default::default);
+
+/// Return the control-socket peer PID, connecting only when the socket file
+/// or the previously identified peer process changed.
+async fn cached_control_socket_peer_pid(sock_path: &Path) -> io::Result<i32> {
+    let key: &std::path::Path = sock_path;
+    let metadata = tokio::fs::metadata(sock_path).await?;
+    let cached = PEER_CACHE.lock().expect("lock poisoned").get(key).copied();
+    if let Some(peer) = cached
+        && peer.dev == metadata.dev()
+        && peer.ino == metadata.ino()
+        && process_start_time(peer.pid) == Some(peer.start_time)
+    {
+        return Ok(peer.pid);
+    }
+    let pid = control_socket_peer_pid(sock_path).await?;
+    if let Some(start_time) = process_start_time(pid) {
+        PEER_CACHE.lock().expect("lock poisoned").insert(
+            key.to_path_buf(),
+            CachedPeer {
+                dev: metadata.dev(),
+                ino: metadata.ino(),
+                pid,
+                start_time,
+            },
+        );
+    }
+    Ok(pid)
+}
+
+/// Return the start time of `pid`, or `None` if it no longer exists.
+fn process_start_time(pid: i32) -> Option<u64> {
+    procfs::process::Process::new(pid)
+        .and_then(|process| process.stat())
+        .map(|stat| stat.starttime)
+        .ok()
+}
+
 /// Return the process ID at the far end of a UNIX control socket.
 async fn control_socket_peer_pid(sock_path: &Path) -> io::Result<i32> {
     let stream = tokio::net::UnixStream::connect(sock_path).await?;
@@ -288,3 +364,85 @@ async fn control_socket_peer_pid(sock_path: &Path) -> io::Result<i32> {
         )
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use std::os::unix::net::UnixListener;
+
+    use async_trait::async_trait;
+
+    use super::*;
+    use crate::instance::ThreadPoolSnapshot;
+
+    struct NullClient;
+
+    #[async_trait]
+    impl InstanceClient for NullClient {
+        async fn set_thread_count(&self, _count: u32) -> Result<(), BackendClientError> {
+            Ok(())
+        }
+
+        async fn get_thread_pool_snapshot(&self) -> Result<ThreadPoolSnapshot, BackendClientError> {
+            Err(BackendClientError::NotSupported("snapshot".to_string()))
+        }
+
+        async fn close(&self) {}
+    }
+
+    async fn discover(socket_dir: &std::path::Path) -> Vec<Arc<Instance>> {
+        discover_via_control_socket(Path::new(socket_dir), "ctl", |_| {
+            Box::new(NullClient) as Box<dyn InstanceClient>
+        })
+        .await
+        .unwrap()
+    }
+
+    fn bind(path: &std::path::Path) -> UnixListener {
+        let listener = UnixListener::bind(path).unwrap();
+        listener.set_nonblocking(true).unwrap();
+        listener
+    }
+
+    /// Return how many connections are pending on `listener`.
+    fn pending(listener: &UnixListener) -> usize {
+        std::iter::from_fn(|| listener.accept().ok()).count()
+    }
+
+    /// Test that discovery identifies a control-socket peer once and does
+    /// not reconnect on later passes until the socket is rebound.
+    #[tokio::test]
+    async fn test_discovery_reuses_identified_control_socket_peer() {
+        let socket_dir = tempfile::tempdir().unwrap();
+        std::fs::create_dir(socket_dir.path().join("vm-a")).unwrap();
+        let sock_path = socket_dir.path().join("vm-a").join("ctl");
+        let mut listener = bind(&sock_path);
+
+        let instances = discover(socket_dir.path()).await;
+        assert_eq!(instances.len(), 1);
+        assert_eq!(instances[0].id, "vm-a");
+        assert_eq!(instances[0].pid, std::process::id() as i32);
+        assert_eq!(pending(&listener), 1);
+
+        assert_eq!(discover(socket_dir.path()).await.len(), 1);
+        assert_eq!(pending(&listener), 0);
+
+        // Keep the old socket inode alive while rebinding so the filesystem
+        // cannot hand the same inode number to the new socket.
+        std::fs::remove_file(&sock_path).unwrap();
+        let old_listener = std::mem::replace(&mut listener, bind(&sock_path));
+        drop(old_listener);
+        assert_eq!(discover(socket_dir.path()).await.len(), 1);
+        assert_eq!(pending(&listener), 1);
+
+        drop(listener);
+        std::fs::remove_file(&sock_path).unwrap();
+        assert!(discover(socket_dir.path()).await.is_empty());
+        assert!(
+            !PEER_CACHE
+                .lock()
+                .unwrap()
+                .keys()
+                .any(|path| path.starts_with(socket_dir.path()))
+        );
+    }
+}