From 4992fabd158f5208f134ed12efaed770016e94da Mon Sep 17 00:00:00 2001 From: Sebastien Boeuf Date: Thu, 21 May 2026 02:49:16 -0700 Subject: [PATCH] offload_daemon: Introduce a reference implementation Adding a new dedicated binary that is meant to be used as a reference implementation for validating that offloaded snapshot/restore works and meant to be used through tests in general. Signed-off-by: Sebastien Boeuf Assisted-by: Claude:claude-opus-4-7 --- Cargo.lock | 28 + Cargo.toml | 1 + offload_daemon/Cargo.toml | 25 + offload_daemon/src/main.rs | 543 ++++++++++++++++++ .../gitlint/rules/TitleStartsWithComponent.py | 1 + 5 files changed, 598 insertions(+) create mode 100644 offload_daemon/Cargo.toml create mode 100644 offload_daemon/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 039aed9b6..0aa03fe38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -453,6 +453,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" dependencies = [ "clap_builder", + "clap_derive", ] [[package]] @@ -468,6 +469,18 @@ dependencies = [ "terminal_size", ] +[[package]] +name = "clap_derive" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "clap_lex" version = "1.1.0" @@ -1554,6 +1567,21 @@ dependencies = [ "memchr", ] +[[package]] +name = "offload_daemon" +version = "0.1.0" +dependencies = [ + "clap", + "env_logger", + "libc", + "log", + "serde_json", + "thiserror", + "vm-migration", + "vmm", + "vmm-sys-util", +] + [[package]] name = "once_cell" version = "1.21.4" diff --git a/Cargo.toml b/Cargo.toml index 5f829c930..d684e3e1c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,6 +23,7 @@ members = [ "event_monitor", "hypervisor", "net_util", + "offload_daemon", "option_parser", "pci", "performance-metrics", diff --git a/offload_daemon/Cargo.toml b/offload_daemon/Cargo.toml new file mode 100644 index 000000000..5da6499e2 --- /dev/null +++ b/offload_daemon/Cargo.toml @@ -0,0 +1,25 @@ +[package] +authors = ["The Cloud Hypervisor Authors"] +edition.workspace = true +name = "offload_daemon" +rust-version.workspace = true +version = "0.1.0" + +[dependencies] +clap = { workspace = true, features = ["derive"] } +env_logger = { workspace = true } +libc = { workspace = true } +log = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +vm-migration = { path = "../vm-migration" } +vmm = { path = "../vmm" } +vmm-sys-util = { workspace = true } + +[features] +default = ["kvm"] +kvm = ["vmm/kvm"] +mshv = ["vmm/mshv"] + +[lints] +workspace = true diff --git a/offload_daemon/src/main.rs b/offload_daemon/src/main.rs new file mode 100644 index 000000000..5ca8ad2ce --- /dev/null +++ b/offload_daemon/src/main.rs @@ -0,0 +1,543 @@ +// Copyright © 2026 The Cloud Hypervisor Authors +// +// SPDX-License-Identifier: Apache-2.0 + +//! Reference offload daemon for Cloud Hypervisor snapshot/restore. +//! +//! It acts as the local live-migration peer of CH's existing +//! `vm.send-migration` and `vm.receive-migration` endpoints, persisting the +//! migration stream to a directory and replaying it later. +//! +//! Snapshot (daemon receives): writes each guest RAM slot to `memory-`, +//! the `VmMigrationConfig` to `migration_config.json`, and the device state to +//! `state.json`. +//! +//! Restore (daemon sends): replays those files back to CH. `--resume` resumes +//! the VM instead of leaving it paused. + +use std::ffi::{CString, NulError}; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Read, Seek, SeekFrom, Write}; +use std::os::fd::{AsRawFd, FromRawFd}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::result; + +use clap::{Parser, Subcommand}; +use log::{debug, info}; +use thiserror::Error; +use vm_migration::MigratableError; +use vm_migration::protocol::{Command, Request, Response, Status}; +use vmm::VmMigrationConfig; +use vmm::migration::SNAPSHOT_STATE_FILE; +use vmm_sys_util::errno; +use vmm_sys_util::sock_ctrl_msg::ScmSocket; + +const MIGRATION_CONFIG_FILENAME: &str = "migration_config.json"; + +#[derive(Debug, Error)] +enum Error { + #[error("Creating the output directory")] + CreateOutputDir(#[source] io::Error), + #[error("Opening the socket lock file")] + OpenLockFile(#[source] io::Error), + #[error("Socket {0:?} is already in use by another daemon")] + SocketInUse(PathBuf), + #[error("Locking the socket")] + LockSocket(#[source] io::Error), + #[error("Binding the UNIX socket")] + BindSocket(#[source] io::Error), + #[error("Connecting to the CH socket")] + Connect(#[source] io::Error), + #[error("Accepting CH connection")] + Accept(#[source] io::Error), + #[error("Migration protocol")] + Protocol(#[source] MigratableError), + #[error("Receiving memory fd")] + RecvMemoryFd(#[source] errno::Error), + #[error("MemoryFd command carried no file descriptor")] + MissingMemoryFd, + #[error("Sending memory fd")] + SendMemoryFd(#[source] errno::Error), + #[error("Reading a snapshot artifact")] + ReadFile(#[source] io::Error), + #[error("Writing a snapshot artifact")] + WriteFile(#[source] io::Error), + #[error("Reading migration payload")] + ReadPayload(#[source] io::Error), + #[error("Sending migration payload")] + WritePayload(#[source] io::Error), + #[error("(De)serializing VmMigrationConfig")] + Config(#[from] serde_json::Error), + #[error("Unexpected command {0:?} while expecting {1}")] + UnexpectedCommand(Command, &'static str), + #[error("CH abandoned the snapshot")] + Abandoned, + #[error("CH rejected the {0} command")] + Rejected(&'static str), + #[error("Completion received before {0}")] + PrematureCompletion(&'static str), + #[error("Creating memfd")] + MemfdCreate(#[source] io::Error), + #[error("Invalid memfd name")] + MemfdName(#[source] NulError), + #[error("Sizing memfd")] + MemfdSetLen(#[source] io::Error), + #[error("Copying snapshot memory")] + CopyMemory(#[source] io::Error), + #[error("No MemoryFd received for slot {0}")] + MissingSlot(u32), + #[error("Field {0:?} missing from memory_manager_data")] + MissingField(&'static str), +} + +type Result = result::Result; + +fn memory_slot_filename(slot: u32) -> String { + format!("memory-{slot}") +} + +#[derive(Parser, Debug)] +#[command(name = "offload_daemon")] +struct Cli { + #[command(subcommand)] + mode: Mode, +} + +#[derive(Subcommand, Debug)] +enum Mode { + /// Receive a snapshot from CH and persist it to disk. + Snapshot { + /// Path to a UNIX socket to bind and listen on. + #[arg(long)] + socket: PathBuf, + /// Directory to write snapshot artifacts into. + #[arg(long)] + output_dir: PathBuf, + }, + /// Read a snapshot from disk and stream it to a listening CH instance. + Restore { + /// Path of the UNIX socket that CH is listening on. + #[arg(long)] + socket: PathBuf, + /// Directory to read snapshot artifacts from. + #[arg(long)] + input_dir: PathBuf, + /// If set, the restored VM is resumed (Complete) instead of left + /// paused (CompletePaused). + #[arg(long)] + resume: bool, + }, +} + +fn main() -> Result<()> { + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init(); + + let cli = Cli::parse(); + match cli.mode { + Mode::Snapshot { socket, output_dir } => run_snapshot(&socket, &output_dir), + Mode::Restore { + socket, + input_dir, + resume, + } => run_restore(&socket, &input_dir, resume), + } +} + +/// Take an exclusive lock on `.lock` so two daemons can't bind the same +/// socket, and so removing a stale socket file is safe. +fn acquire_socket_lock(socket_path: &Path) -> Result { + let lock_path: PathBuf = { + let mut p = socket_path.as_os_str().to_os_string(); + p.push(".lock"); + p.into() + }; + let lock = OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&lock_path) + .map_err(Error::OpenLockFile)?; + + let flock = libc::flock { + l_type: libc::F_WRLCK as libc::c_short, + l_whence: libc::SEEK_SET as libc::c_short, + l_start: 0, + l_len: 0, // 0 means the whole file. + l_pid: 0, + }; + loop { + // SAFETY: fcntl() with F_OFD_SETLK and a valid flock pointer on an owned fd. + if unsafe { libc::fcntl(lock.as_raw_fd(), libc::F_OFD_SETLK, &flock) } == 0 { + break; + } + let err = io::Error::last_os_error(); + match err.raw_os_error() { + // Interrupted by a signal before the lock was taken: retry. + Some(libc::EINTR) => continue, + // The lock is held by another daemon. + Some(libc::EACCES) | Some(libc::EAGAIN) => { + return Err(Error::SocketInUse(socket_path.to_path_buf())); + } + _ => return Err(Error::LockSocket(err)), + } + } + Ok(lock) +} + +// Snapshot mode (migration receiver). +fn run_snapshot(socket_path: &Path, output_dir: &Path) -> Result<()> { + fs::create_dir_all(output_dir).map_err(Error::CreateOutputDir)?; + + // Hold the lock for the daemon's lifetime. While we hold it, any socket at + // this path is stale from a crashed run, so removing it before bind is safe. + let _lock = acquire_socket_lock(socket_path)?; + let _ = fs::remove_file(socket_path); + let listener = UnixListener::bind(socket_path).map_err(Error::BindSocket)?; + info!("Offload daemon listening at {socket_path:?}"); + + let (mut stream, _) = listener.accept().map_err(Error::Accept)?; + info!("CH connected; starting snapshot receive"); + + expect_command(&mut stream, Command::Start, "Start")?; + Response::ok() + .write_to(&mut stream) + .map_err(Error::Protocol)?; + + let mut memory_slots: Vec<(u32, File)> = Vec::new(); + let mut migration_config: Option = None; + let mut state_bytes: Option> = None; + + loop { + let req = Request::read_from(&mut stream).map_err(Error::Protocol)?; + debug!("snapshot: received command {:?}", req.command()); + match req.command() { + Command::MemoryFd => { + let (slot, file) = recv_memory_fd(&stream)?; + debug!("snapshot: received memory fd for slot {slot}"); + memory_slots.push((slot, file)); + Response::ok() + .write_to(&mut stream) + .map_err(Error::Protocol)?; + } + Command::Config => { + let mut buf = vec![0u8; req.length() as usize]; + stream.read_exact(&mut buf).map_err(Error::ReadPayload)?; + migration_config = Some(serde_json::from_slice(&buf)?); + fs::write(output_dir.join(MIGRATION_CONFIG_FILENAME), &buf) + .map_err(Error::WriteFile)?; + Response::ok() + .write_to(&mut stream) + .map_err(Error::Protocol)?; + } + Command::State => { + let mut buf = vec![0u8; req.length() as usize]; + stream.read_exact(&mut buf).map_err(Error::ReadPayload)?; + fs::write(output_dir.join(SNAPSHOT_STATE_FILE), &buf).map_err(Error::WriteFile)?; + state_bytes = Some(buf); + Response::ok() + .write_to(&mut stream) + .map_err(Error::Protocol)?; + } + Command::CompletePaused | Command::Complete => { + // Invariant: drain + fsync every memory fd BEFORE ACKing — + // CH may exit right after and these fds are our only copy. + let mm = migration_config + .as_ref() + .ok_or(Error::PrematureCompletion("Config"))?; + let _ = state_bytes + .as_ref() + .ok_or(Error::PrematureCompletion("State"))?; + dump_memory_slots(&memory_slots, mm, output_dir)?; + Response::ok() + .write_to(&mut stream) + .map_err(Error::Protocol)?; + info!("Snapshot persisted to {output_dir:?}"); + break; + } + Command::Abandon => { + // ACK before bailing so CH's ok_or_abandon read returns + // cleanly instead of hitting EOF. + Response::ok().write_to(&mut stream).ok(); + return Err(Error::Abandoned); + } + c => return Err(Error::UnexpectedCommand(c, "a snapshot command")), + } + } + + Ok(()) +} + +fn expect_command(stream: &mut UnixStream, want: Command, name: &'static str) -> Result { + let req = Request::read_from(stream).map_err(Error::Protocol)?; + if req.command() != want { + return Err(Error::UnexpectedCommand(req.command(), name)); + } + Ok(req) +} + +fn recv_memory_fd(stream: &UnixStream) -> Result<(u32, File)> { + let mut buf = [0u8; 4]; + let (_n, file) = stream.recv_with_fd(&mut buf).map_err(Error::RecvMemoryFd)?; + let file = file.ok_or(Error::MissingMemoryFd)?; + Ok((u32::from_le_bytes(buf), file)) +} + +fn dump_memory_slots( + slots: &[(u32, File)], + config: &VmMigrationConfig, + output_dir: &Path, +) -> Result<()> { + let sizes = slot_sizes(config)?; + for (expected_slot, _, _) in &sizes { + if !slots.iter().any(|(s, _)| s == expected_slot) { + return Err(Error::MissingSlot(*expected_slot)); + } + } + for (slot, file) in slots { + let (size, file_offset) = sizes + .iter() + .find(|(s, _, _)| s == slot) + .map(|(_, sz, fo)| (*sz, *fo)) + .ok_or(Error::MissingSlot(*slot))?; + let path = output_dir.join(memory_slot_filename(*slot)); + dump_fd_to_path(file, file_offset, size, &path)?; + debug!("dumped {size} bytes from slot {slot} (fd offset {file_offset:#x}) to {path:?}"); + } + Ok(()) +} + +fn dump_fd_to_path(file: &File, src_offset: u64, size: u64, path: &Path) -> Result<()> { + let mut out = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .open(path) + .map_err(Error::WriteFile)?; + out.set_len(size).map_err(Error::WriteFile)?; + // dup so our own offset doesn't disturb CH's view of the shared fd. + // SAFETY: dup() has no preconditions; result is checked. + let raw = unsafe { libc::dup(file.as_raw_fd()) }; + if raw < 0 { + return Err(Error::CopyMemory(io::Error::last_os_error())); + } + // SAFETY: `raw` is a fresh fd we now own. + let mut src = unsafe { File::from_raw_fd(raw) }; + src.seek(SeekFrom::Start(src_offset)) + .map_err(Error::CopyMemory)?; + let mut remaining = size; + let mut buf = vec![0u8; 1 << 20]; + while remaining > 0 { + let want = (remaining as usize).min(buf.len()); + src.read_exact(&mut buf[..want]) + .map_err(Error::CopyMemory)?; + out.write_all(&buf[..want]).map_err(Error::CopyMemory)?; + remaining -= want as u64; + } + out.sync_all().map_err(Error::WriteFile)?; + Ok(()) +} + +/// (slot, size, file_offset) per memory slot, parsed via JSON to avoid +/// depending on the private fields of `MemoryManagerSnapshotData`. +/// +/// `file_offset` is non-zero when a zone spans multiple regions sharing one +/// backing memfd. +fn slot_sizes(config: &VmMigrationConfig) -> Result> { + parse_guest_ram_mappings(&serde_json::to_value(config.memory_manager_data())?) +} + +/// Parse the `guest_ram_mappings` array out of the serialized +/// `MemoryManagerSnapshotData`. +fn parse_guest_ram_mappings(value: &serde_json::Value) -> Result> { + let mappings = value + .get("guest_ram_mappings") + .and_then(|v| v.as_array()) + .ok_or(Error::MissingField("guest_ram_mappings"))?; + let mut out = Vec::new(); + for m in mappings { + let slot = m + .get("slot") + .and_then(|v| v.as_u64()) + .ok_or(Error::MissingField("slot"))? as u32; + let size = m + .get("size") + .and_then(|v| v.as_u64()) + .ok_or(Error::MissingField("size"))?; + let file_offset = m + .get("file_offset") + .and_then(|v| v.as_u64()) + .ok_or(Error::MissingField("file_offset"))?; + // CH allocates one fresh memslot per GuestRegionMmap, so each + // (slot, size, file_offset) appears at most once here. + out.push((slot, size, file_offset)); + } + Ok(out) +} + +// Restore mode (migration sender). +fn run_restore(socket_path: &Path, input_dir: &Path, resume: bool) -> Result<()> { + let migration_config_bytes = + fs::read(input_dir.join(MIGRATION_CONFIG_FILENAME)).map_err(Error::ReadFile)?; + let migration_config: VmMigrationConfig = serde_json::from_slice(&migration_config_bytes)?; + let state_bytes = fs::read(input_dir.join(SNAPSHOT_STATE_FILE)).map_err(Error::ReadFile)?; + let sizes = slot_sizes(&migration_config)?; + + let mut stream = UnixStream::connect(socket_path).map_err(Error::Connect)?; + info!("Offload daemon connected to {socket_path:?}"); + + send_request_expect_ok(&mut stream, Request::start(), "Start")?; + + for (slot, size, file_offset) in &sizes { + let memfd = create_memfd_with_contents( + &input_dir.join(memory_slot_filename(*slot)), + *file_offset, + *size, + &format!("offload-slot-{slot}"), + )?; + send_memory_fd(&mut stream, *slot, &memfd)?; + debug!( + "restore: sent memory fd for slot {slot} ({size} bytes at fd offset {file_offset:#x})" + ); + } + + send_payload_expect_ok( + &mut stream, + Request::config(migration_config_bytes.len() as u64), + &migration_config_bytes, + "Config", + )?; + send_payload_expect_ok( + &mut stream, + Request::state(state_bytes.len() as u64), + &state_bytes, + "State", + )?; + + let final_req = if resume { + Request::complete() + } else { + Request::complete_paused() + }; + send_request_expect_ok(&mut stream, final_req, "Complete")?; + + info!("Restore replay finished"); + Ok(()) +} + +fn send_request_expect_ok(stream: &mut UnixStream, req: Request, name: &'static str) -> Result<()> { + req.write_to(stream).map_err(Error::Protocol)?; + expect_ok_response(stream, name) +} + +fn send_payload_expect_ok( + stream: &mut UnixStream, + req: Request, + payload: &[u8], + name: &'static str, +) -> Result<()> { + req.write_to(stream).map_err(Error::Protocol)?; + stream.write_all(payload).map_err(Error::WritePayload)?; + expect_ok_response(stream, name) +} + +fn expect_ok_response(stream: &mut UnixStream, name: &'static str) -> Result<()> { + let resp = Response::read_from(stream).map_err(Error::Protocol)?; + if resp.status() != Status::Ok { + return Err(Error::Rejected(name)); + } + Ok(()) +} + +fn send_memory_fd(stream: &mut UnixStream, slot: u32, memfd: &File) -> Result<()> { + Request::memory_fd(size_of::() as u64) + .write_to(stream) + .map_err(Error::Protocol)?; + stream + .send_with_fd(&slot.to_le_bytes()[..], memfd.as_raw_fd()) + .map_err(Error::SendMemoryFd)?; + expect_ok_response(stream, "MemoryFd") +} + +fn create_empty_memfd(size: u64, name: &str) -> Result { + let cname = CString::new(name).map_err(Error::MemfdName)?; + // SAFETY: memfd_create has no preconditions. We check the return value. + let raw = unsafe { libc::memfd_create(cname.as_ptr(), 0) }; + if raw < 0 { + return Err(Error::MemfdCreate(io::Error::last_os_error())); + } + // SAFETY: `raw` is a fresh fd we now own. + let memfd = unsafe { File::from_raw_fd(raw) }; + memfd.set_len(size).map_err(Error::MemfdSetLen)?; + Ok(memfd) +} + +fn create_memfd_with_contents( + src_path: &Path, + file_offset: u64, + size: u64, + name: &str, +) -> Result { + // Size the memfd to cover the range CH maps at `file_offset`. + let mut memfd = create_empty_memfd(file_offset + size, name)?; + memfd + .seek(SeekFrom::Start(file_offset)) + .map_err(Error::CopyMemory)?; + let mut src = File::open(src_path).map_err(Error::ReadFile)?; + let mut remaining = size; + let mut buf = vec![0u8; 1 << 20]; + while remaining > 0 { + let want = (remaining as usize).min(buf.len()); + src.read_exact(&mut buf[..want]) + .map_err(Error::CopyMemory)?; + memfd.write_all(&buf[..want]).map_err(Error::CopyMemory)?; + remaining -= want as u64; + } + memfd.seek(SeekFrom::Start(0)).map_err(Error::CopyMemory)?; + Ok(memfd) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn test_memory_slot_filename() { + assert_eq!(memory_slot_filename(0), "memory-0"); + assert_eq!(memory_slot_filename(13), "memory-13"); + } + + #[test] + fn test_parse_guest_ram_mappings() { + let value = json!({ + "guest_ram_mappings": [ + { "slot": 0, "size": 4096u64, "file_offset": 0u64, "virtio_mem": true }, + { "slot": 1, "size": 8192u64, "file_offset": 4096u64 }, + ] + }); + assert_eq!( + parse_guest_ram_mappings(&value).unwrap(), + vec![(0, 4096, 0), (1, 8192, 4096)] + ); + } + + #[test] + fn test_parse_guest_ram_mappings_missing_array() { + let value = json!({}); + assert!(matches!( + parse_guest_ram_mappings(&value), + Err(Error::MissingField("guest_ram_mappings")) + )); + } + + #[test] + fn test_parse_guest_ram_mappings_missing_field() { + let value = json!({ "guest_ram_mappings": [{ "slot": 0, "size": 4096u64 }] }); + assert!(matches!( + parse_guest_ram_mappings(&value), + Err(Error::MissingField("file_offset")) + )); + } +} diff --git a/scripts/gitlint/rules/TitleStartsWithComponent.py b/scripts/gitlint/rules/TitleStartsWithComponent.py index a25172a62..0f66ca2b0 100644 --- a/scripts/gitlint/rules/TitleStartsWithComponent.py +++ b/scripts/gitlint/rules/TitleStartsWithComponent.py @@ -47,6 +47,7 @@ class TitleStartsWithComponent(LineRule): 'main', 'misc', 'net_util', + 'offload_daemon', 'openapi', 'option_parser', 'pci',