aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorNikolay Govorov <me@govorov.online>2026-07-12 23:26:12 +0100
committerNikolay Govorov <me@govorov.online>2026-07-12 23:26:12 +0100
commit53f3149e1d632f044c28162244304fd3b6b8ea9b (patch)
tree75ec1859faf89f2a53cfbc031259e1f4730d3c22
parentaceebb27f0cd07c4852f8e927b1a41d307023536 (diff)
downloadtar
tar.gz
tar.bz2
tar.lz
tar.xz
tar.zst
zip
Use QMP for managing qemu machine lifecycle
Diffstat
-rw-r--r--Cargo.lock2+1 −1
-rw-r--r--crates/hule-vmm/Cargo.toml5+5 −0
-rw-r--r--crates/hule-vmm/src/backend/qemu/cli.rs11+11 −0
-rw-r--r--crates/hule-vmm/src/backend/qemu/mod.rs140+121 −19
-rw-r--r--crates/hule-vmm/src/backend/qemu/qmp.rs165+165 −0
5 files changed, 303 insertions, 20 deletions
diff --git a/Cargo.lock b/Cargo.lock
index 6e8c52a..08e65ee 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -599,8 +599,8 @@ name = "hule-vmm"
version = "0.1.0"
dependencies = [
"async-trait",
- "base64",
"hule-image",
+ "serde",
"serde_json",
"tokio",
"uuid",
diff --git a/crates/hule-vmm/Cargo.toml b/crates/hule-vmm/Cargo.toml
index ed32410..acfc1bf 100644
--- a/crates/hule-vmm/Cargo.toml
+++ b/crates/hule-vmm/Cargo.toml
@@ -17,3 +17,8 @@ hule-image.workspace = true
uuid = { version = "1", features = ["v7"] }
tokio = { workspace = true, features = ["process", "fs", "io-util", "sync", "net", "time"] }
async-trait = "0.1"
+serde.workspace = true
+serde_json.workspace = true
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
diff --git a/crates/hule-vmm/src/backend/qemu/cli.rs b/crates/hule-vmm/src/backend/qemu/cli.rs
index b17f8e9..077c111 100644
--- a/crates/hule-vmm/src/backend/qemu/cli.rs
+++ b/crates/hule-vmm/src/backend/qemu/cli.rs
@@ -147,6 +147,11 @@ pub struct Command {
/// `-pidfile`: qemu writes its own pid here. The only reliable way to
/// signal/check it later from a process that isn't its parent.
pub pidfile: Option<PathBuf>,
+ /// `-qmp tcp:127.0.0.1:<port>,server=on,wait=off`: the control channel.
+ /// TCP loopback, not a unix socket -- the latter doesn't exist on
+ /// Windows at all (`tokio::net::UnixStream` is unix-only), and this way
+ /// there's nothing to `#[cfg]` on the transport.
+ pub qmp_port: Option<u16>,
}
impl Command {
@@ -187,6 +192,12 @@ impl Command {
if let Some(pidfile) = &self.pidfile {
argv.extend(["-pidfile".into(), pidfile.display().to_string()]);
}
+ if let Some(port) = self.qmp_port {
+ argv.extend([
+ "-qmp".into(),
+ format!("tcp:127.0.0.1:{port},server=on,wait=off"),
+ ]);
+ }
argv
}
}
diff --git a/crates/hule-vmm/src/backend/qemu/mod.rs b/crates/hule-vmm/src/backend/qemu/mod.rs
index 0ae4cd8..75f51c3 100644
--- a/crates/hule-vmm/src/backend/qemu/mod.rs
+++ b/crates/hule-vmm/src/backend/qemu/mod.rs
@@ -1,17 +1,17 @@
// SPDX-FileCopyrightText: 2026 Nikolay Govorov
// SPDX-License-Identifier: Apache-2.0
-//! qemu backend. Basic extraction from the old `hule` binary's boot code:
-//! `start`/`kill`/`wait`/`state`/`delete` work, everything needing QMP or a
-//! console channel (`stop`, `restart`, `pause`, `exec`, `attach`, ...) is
-//! `todo!()` -- QMP is opportunistic, for operations signals can't do, not
-//! load-bearing for the base lifecycle.
+//! QMP is an unconditional implementation detail of this backend and backs
+//! lifecycle operations that cannot be expressed through process signals.
+//! Guest access and console channels are separate concerns and are not wired
+//! up here yet.
//!
//! `cli` is the typed qemu CLI wrapper; `supervise` gets an exit code out of
//! a process even if we stop being its parent; this module only translates
//! Hule's image/machine types into both.
mod cli;
+mod qmp;
mod supervise;
use std::path::{Path, PathBuf};
@@ -54,6 +54,7 @@ pub struct QemuMachine {
disk: Disk,
boot: QemuBoot,
child: Mutex<ChildState>,
+ qmp_client: Mutex<Option<qmp::Qmp>>,
}
impl QemuMachine {
@@ -68,6 +69,45 @@ impl QemuMachine {
self.dir.join("qemu.exit-code")
}
+ /// Where the port `start()` picked for `-qmp tcp:127.0.0.1:<port>` is
+ /// recorded -- chosen by us, not qemu, so (unlike `pid_path`) nothing
+ /// needs to wait for this file; it exists before qemu is spawned.
+ fn qmp_port_path(&self) -> PathBuf {
+ self.dir.join("qemu.qmp-port")
+ }
+
+ async fn qmp_port(&self) -> R<u16> {
+ let data = tokio::fs::read_to_string(self.qmp_port_path()).await?;
+ data.trim()
+ .parse()
+ .map_err(|_| Error::InvalidState("invalid qemu.qmp-port contents".into()))
+ }
+
+ /// Connects on first use (there's a short window after spawn before
+ /// qemu's TCP listener is actually up, so a couple of retries) and
+ /// reuses the connection after that.
+ async fn qmp_client(&self) -> R<tokio::sync::MutexGuard<'_, Option<qmp::Qmp>>> {
+ let mut client = self.qmp_client.lock().await;
+ if client.is_none() {
+ let port = self.qmp_port().await?;
+ let mut connected = None;
+ for _ in 0..50 {
+ match qmp::Qmp::connect(port).await {
+ Ok(qmp) => {
+ connected = Some(qmp);
+ break;
+ }
+ Err(_) => tokio::time::sleep(Duration::from_millis(20)).await,
+ }
+ }
+ *client = Some(
+ connected
+ .ok_or_else(|| Error::InvalidState("qemu's qmp port never came up".into()))?,
+ );
+ }
+ Ok(client)
+ }
+
/// Reads qemu's pid, waiting briefly for `-pidfile` to actually appear
/// (there's a short window between spawn and qemu writing it).
async fn qemu_pid(&self) -> R<supervise::Pid> {
@@ -127,15 +167,30 @@ impl Machine for QemuMachine {
if let Some(status) = self.read_exit_code().await {
return State::Exited(status.code().unwrap_or(-1));
}
- match self.qemu_pid().await {
- Ok(pid) if supervise::pid_alive(pid).await => State::Running,
- Ok(_) => State::Dead,
- Err(_) => match &*self.child.lock().await {
- ChildState::NotStarted => State::Created,
- ChildState::Exited(status) => State::Exited(status.code().unwrap_or(-1)),
- ChildState::Running(_) => State::Running,
- },
+ let pid = match self.qemu_pid().await {
+ Ok(pid) => pid,
+ Err(_) => {
+ return match &*self.child.lock().await {
+ ChildState::NotStarted => State::Created,
+ ChildState::Exited(status) => State::Exited(status.code().unwrap_or(-1)),
+ ChildState::Running(_) => State::Running,
+ };
+ }
+ };
+ if !supervise::pid_alive(pid).await {
+ return State::Dead;
+ }
+ // A live pid alone can't distinguish running from paused; ask QMP.
+ if let Ok(mut client) = self.qmp_client().await
+ && let Ok(qmp::Status::Paused) = client
+ .as_mut()
+ .expect("initialized by qmp_client")
+ .status()
+ .await
+ {
+ return State::Paused;
}
+ State::Running
}
async fn stats(&self) -> R<Stats> {
@@ -153,6 +208,11 @@ impl Machine for QemuMachine {
}
let _ = tokio::fs::remove_file(self.pid_path()).await;
let _ = tokio::fs::remove_file(self.exit_code_path()).await;
+ let _ = tokio::fs::remove_file(self.qmp_port_path()).await;
+ *self.qmp_client.lock().await = None;
+
+ let qmp_port = qmp::free_port().await?;
+ tokio::fs::write(self.qmp_port_path(), qmp_port.to_string()).await?;
let kvm = Architecture::host() == Some(self.arch) && cli::kvm_available().await;
let sys_arch = match self.arch {
@@ -213,6 +273,7 @@ impl Machine for QemuMachine {
}
}
cmd.pidfile = Some(self.pid_path());
+ cmd.qmp_port = Some(qmp_port);
let argv = cmd.to_argv();
let wrapper = supervise::spawn(&argv, &self.exit_code_path())?;
@@ -221,7 +282,12 @@ impl Machine for QemuMachine {
}
async fn stop(&mut self) -> R<()> {
- todo!("needs QMP system_powerdown, not wired up yet")
+ self.qmp_client()
+ .await?
+ .as_mut()
+ .expect("initialized by qmp_client")
+ .power_down()
+ .await
}
async fn kill(&mut self) -> R<()> {
@@ -240,7 +306,12 @@ impl Machine for QemuMachine {
}
async fn restart(&mut self) -> R<()> {
- todo!("needs QMP system_reset, not wired up yet")
+ self.qmp_client()
+ .await?
+ .as_mut()
+ .expect("initialized by qmp_client")
+ .reset()
+ .await
}
async fn delete(&mut self) -> R<()> {
@@ -249,19 +320,49 @@ impl Machine for QemuMachine {
}
let _ = tokio::fs::remove_file(self.pid_path()).await;
let _ = tokio::fs::remove_file(self.exit_code_path()).await;
+ let _ = tokio::fs::remove_file(self.qmp_port_path()).await;
Ok(())
}
async fn pause(&mut self) -> R<()> {
- todo!("needs QMP stop, not wired up yet")
+ self.qmp_client()
+ .await?
+ .as_mut()
+ .expect("initialized by qmp_client")
+ .pause()
+ .await
}
async fn unpause(&mut self) -> R<()> {
- todo!("needs QMP cont, not wired up yet")
+ self.qmp_client()
+ .await?
+ .as_mut()
+ .expect("initialized by qmp_client")
+ .resume()
+ .await
}
- async fn update(&mut self, _settings: Settings) -> R<()> {
- todo!("needs a diff against current settings and QMP/hotplug to apply it")
+ async fn update(&mut self, settings: Settings) -> R<()> {
+ if settings.cpu != self.settings.cpu {
+ return Err(Error::Unsupported(
+ "qemu backend can't hot-change cpu count (needs maxcpus reserved at boot)".into(),
+ ));
+ }
+ if settings.port_forwards != self.settings.port_forwards {
+ return Err(Error::Unsupported(
+ "qemu backend can't hot-change port forwards (fixed at boot via -net user)".into(),
+ ));
+ }
+ if settings.ram != self.settings.ram {
+ self.qmp_client()
+ .await?
+ .as_mut()
+ .expect("initialized by qmp_client")
+ .set_balloon_size(settings.ram * 1024 * 1024)
+ .await?;
+ }
+ self.settings = settings;
+ Ok(())
}
async fn rename(&mut self, name: Option<&str>) -> R<()> {
@@ -360,6 +461,7 @@ impl Hypervisor for QemuHypervisor {
disk,
boot: qemu_boot,
child: Mutex::new(ChildState::NotStarted),
+ qmp_client: Mutex::new(None),
}))
}
diff --git a/crates/hule-vmm/src/backend/qemu/qmp.rs b/crates/hule-vmm/src/backend/qemu/qmp.rs
new file mode 100644
--- /dev/null
+++ b/crates/hule-vmm/src/backend/qemu/qmp.rs
@@ -0,0 +1,165 @@
+// SPDX-FileCopyrightText: 2026 Nikolay Govorov
+// SPDX-License-Identifier: Apache-2.0
+
+//! Minimal QMP client: line-delimited JSON commands/responses over the TCP
+//! loopback channel qemu exposes via
+//! `-qmp tcp:127.0.0.1:<port>,server=on,wait=off`. Events are read and
+//! discarded, not surfaced.
+
+use serde::Deserialize;
+use serde_json::{Value, json};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::TcpStream;
+use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
+
+use crate::{Error, R};
+
+pub struct Qmp {
+ reader: BufReader<OwnedReadHalf>,
+ writer: OwnedWriteHalf,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Status {
+ Running,
+ Paused,
+ Transitional,
+}
+
+impl Status {
+ fn from_wire(status: &str) -> Self {
+ match status {
+ "running" => Self::Running,
+ "paused" => Self::Paused,
+ _ => Self::Transitional,
+ }
+ }
+}
+
+impl Qmp {
+ /// Connects and completes the capabilities handshake (greeting ->
+ /// `qmp_capabilities`), after which arbitrary commands are allowed.
+ pub async fn connect(port: u16) -> R<Self> {
+ let (read_half, write_half) = TcpStream::connect(("127.0.0.1", port)).await?.into_split();
+ let mut qmp = Self {
+ reader: BufReader::new(read_half),
+ writer: write_half,
+ };
+ qmp.read_message().await?; // greeting
+ qmp.execute("qmp_capabilities", None).await?;
+ Ok(qmp)
+ }
+
+ pub async fn status(&mut self) -> R<Status> {
+ #[derive(Deserialize)]
+ struct Response {
+ status: String,
+ }
+
+ let response: Response = self.execute_typed("query-status", None).await?;
+ Ok(Status::from_wire(&response.status))
+ }
+
+ pub async fn power_down(&mut self) -> R<()> {
+ self.execute_empty("system_powerdown", None).await
+ }
+
+ pub async fn reset(&mut self) -> R<()> {
+ self.execute_empty("system_reset", None).await
+ }
+
+ pub async fn pause(&mut self) -> R<()> {
+ self.execute_empty("stop", None).await
+ }
+
+ pub async fn resume(&mut self) -> R<()> {
+ self.execute_empty("cont", None).await
+ }
+
+ pub async fn set_balloon_size(&mut self, bytes: u64) -> R<()> {
+ self.execute_empty("balloon", Some(json!({ "value": bytes })))
+ .await
+ }
+
+ async fn read_message(&mut self) -> R<Value> {
+ loop {
+ let mut line = String::new();
+ if self.reader.read_line(&mut line).await? == 0 {
+ return Err(Error::InvalidState("qmp connection closed".into()));
+ }
+
+ let line = line.trim();
+ if line.is_empty() {
+ continue;
+ }
+
+ let value: Value = serde_json::from_str(line)
+ .map_err(|e| Error::InvalidState(format!("invalid qmp message: {e}")))?;
+
+ // Events arrive unprompted between a command and its response;
+ // we don't surface them (yet), so skip past them here.
+ if value.get("event").is_some() {
+ continue;
+ }
+
+ return Ok(value);
+ }
+ }
+
+ async fn execute(&mut self, command: &str, arguments: Option<Value>) -> R<Value> {
+ let mut request = json!({ "execute": command });
+ if let Some(arguments) = arguments {
+ request["arguments"] = arguments;
+ }
+ let mut bytes = serde_json::to_vec(&request).expect("serializable request");
+ bytes.push(b'\n');
+ self.writer.write_all(&bytes).await?;
+ let response = self.read_message().await?;
+ if let Some(error) = response.get("error") {
+ return Err(Error::InvalidState(format!(
+ "qmp '{command}' failed: {error}"
+ )));
+ }
+ response.get("return").cloned().ok_or_else(|| {
+ Error::InvalidState(format!("qmp '{command}' response has no return value"))
+ })
+ }
+
+ async fn execute_empty(&mut self, command: &str, arguments: Option<Value>) -> R<()> {
+ self.execute(command, arguments).await.map(|_| ())
+ }
+
+ async fn execute_typed<T: for<'de> Deserialize<'de>>(
+ &mut self,
+ command: &str,
+ arguments: Option<Value>,
+ ) -> R<T> {
+ let value = self.execute(command, arguments).await?;
+ serde_json::from_value(value).map_err(|error| {
+ Error::InvalidState(format!("invalid qmp '{command}' result: {error}"))
+ })
+ }
+}
+
+/// Picks a free TCP port on 127.0.0.1: binds to port 0 (OS assigns one),
+/// reads it back, then drops the listener before qemu binds it for real.
+/// Small TOCTOU race in principle; standard practice in the absence of a
+/// way to ask qemu what port it actually bound.
+pub async fn free_port() -> std::io::Result<u16> {
+ Ok(tokio::net::TcpListener::bind(("127.0.0.1", 0))
+ .await?
+ .local_addr()?
+ .port())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn wire_status_does_not_escape_the_qmp_boundary() {
+ assert_eq!(Status::from_wire("running"), Status::Running);
+ assert_eq!(Status::from_wire("paused"), Status::Paused);
+ assert_eq!(Status::from_wire("inmigrate"), Status::Transitional);
+ }
+}
generated by Gilti