diff --git a/Cargo.lock b/Cargo.lock index e5689e0e2f..a5664eab5a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -373,6 +373,7 @@ dependencies = [ "byteorder", "cfg-if", "crc-any", + "event_monitor", "flate2", "io-uring", "libc", diff --git a/block/Cargo.toml b/block/Cargo.toml index 22ea206de2..0b268888f5 100644 --- a/block/Cargo.toml +++ b/block/Cargo.toml @@ -13,6 +13,7 @@ io_uring = ["dep:io-uring"] bitflags = { workspace = true } byteorder = { workspace = true } crc-any = "2.5.0" +event_monitor = { path = "../event_monitor" } flate2 = "1.1" io-uring = { version = "0.7.12", optional = true } libc = { workspace = true } diff --git a/block/src/disk_file.rs b/block/src/disk_file.rs index 7f044ea7e3..49d54ea528 100644 --- a/block/src/disk_file.rs +++ b/block/src/disk_file.rs @@ -36,7 +36,7 @@ use std::fmt::Debug; use crate::async_io::{AsyncIo, BorrowedDiskFd}; -use crate::{BlockResult, DiskTopology}; +use crate::{BlockError, BlockErrorKind, BlockResult, DiskTopology}; /// Reported capacity of a disk image. pub trait DiskSize: Send + Debug { @@ -96,7 +96,12 @@ pub trait Resizable: Send + Debug { /// Every disk format implements `DiskSize` and `Geometry`. /// `Sync` is required so that `Arc` can be shared /// across threads for concurrent readonly access. -pub trait DiskFile: DiskSize + Geometry + Sync {} +pub trait DiskFile: DiskSize + Geometry + Sync { + /// Returns an error if this disk image cannot participate in block mirroring. + fn supports_mirroring(&self) -> BlockResult<()> { + Err(BlockError::from_kind(BlockErrorKind::UnsupportedFeature)) + } +} /// Full capability disk file trait. /// diff --git a/block/src/lib.rs b/block/src/lib.rs index 9d688f5ff4..5e58f69cfb 100644 --- a/block/src/lib.rs +++ b/block/src/lib.rs @@ -20,6 +20,7 @@ pub mod fixed_vhd; pub mod fixed_vhd_async; pub mod fixed_vhd_disk; pub mod fixed_vhd_sync; +pub mod mirror; pub mod qcow; #[cfg(feature = "io_uring")] pub(crate) mod qcow_async; diff --git a/block/src/mirror.rs b/block/src/mirror.rs new file mode 100644 index 0000000000..6bdaaa192f --- /dev/null +++ b/block/src/mirror.rs @@ -0,0 +1,1597 @@ +// Copyright © 2026 Cyberus Technology GmbH +// +// SPDX-License-Identifier: Apache-2.0 + +//! Blockdev-mirroring for virtio-blk devices. +//! +//! Mirrors guest writes to a destination disk while a background +//! worker copies existing data from source to destination. Once +//! both sides are in sync the device manager can complete the mirror, +//! switching the device to serve I/O from the destination. + +use std::collections::{BTreeMap, VecDeque}; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::JoinHandle; +use std::{io, mem, thread}; + +use event_monitor::event; +use libc::{iovec, off_t}; +use thiserror::Error; +use vmm_sys_util::eventfd::EventFd; +use vmm_sys_util::poll::PollContext; + +use crate::async_io::{AsyncIo, AsyncIoError, AsyncIoResult}; +use crate::disk_file::AsyncFullDiskFile; +use crate::error::BlockResult; +use crate::qcow_common::AlignedBuf; +use crate::{BatchRequest, RequestType}; + +/// Block size for the copy worker, in which it copies data from +/// source to destination and holds the range lock. +pub const MIRROR_BLOCK_SIZE: usize = 512 * 1024; // 512 KiB + +/// Serializes overlapping byte ranges between the copy worker and the +/// per-queue mirror writes. +/// +/// Each party calls [`Self::lock_range`] before submitting I/O and +/// holds the returned [`RangeGuard`] until completion. A conflicting +/// request blocks on a `Condvar` until the held guard is dropped. +struct RangeLockManager { + /// Held ranges as `start -> end_exclusive`. + /// + /// The mutex makes the overlap check and insert in [`Self::lock_range`] + /// atomic with respect to releases in [`Self::release`]. + ranges: Mutex>, + /// Notified when a range is released. + /// + /// Waiters re-check their range. + cv: Condvar, +} + +impl RangeLockManager { + pub fn new() -> Arc { + Arc::new(Self { + ranges: Mutex::new(BTreeMap::new()), + cv: Condvar::new(), + }) + } + + /// Returns true if `[start, end)` overlaps any range in `ranges`. + fn overlaps_any(ranges: &BTreeMap, start: u64, end: u64) -> bool { + ranges + .range(..end) + .next_back() + .is_some_and(|(_, &e)| e > start) + } + + /// Acquires an exclusive lock on `[offset, offset + length)`. + /// + /// Blocks while any held range overlaps. + fn lock_range(self: Arc, offset: u64, length: u64) -> io::Result { + if length == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Range length is zero", + )); + } + + let end = offset + .checked_add(length) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Range overflow"))?; + { + let mut ranges = self + .cv + .wait_while(self.ranges.lock().unwrap(), |ranges| { + RangeLockManager::overlaps_any(ranges, offset, end) + }) + .unwrap(); + ranges.insert(offset, end); + } + + Ok(RangeGuard { + manager: self, + start: offset, + }) + } + + /// Acquires a [`RangeGuard`] covering the contiguous bytes from + /// `offset` through the end of `iovecs`. + fn lock_iovecs(self: Arc, offset: off_t, iovecs: &[iovec]) -> io::Result { + let total_len = iovecs + .iter() + .try_fold(0u64, |acc, v| acc.checked_add(v.iov_len as u64)) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "iovec length overflow"))?; + + self.lock_range(offset as u64, total_len) + } + + /// Releases the range starting at `start` and wakes all waiters. + fn release(&self, start: u64) { + let mut ranges = self.ranges.lock().unwrap(); + ranges.remove(&start); + self.cv.notify_all(); + } +} + +/// RAII handle for a range held in a [`RangeLockManager`]. +/// +/// Dropping the handle releases the range and wakes all waiters. +struct RangeGuard { + manager: Arc, + start: u64, +} + +impl Drop for RangeGuard { + fn drop(&mut self) { + self.manager.release(self.start); + } +} + +/// Describes a failure recorded in a block mirror's shared state. +#[derive(Debug, Error)] +pub enum MirrorFailure { + /// The background copy worker failed. + #[error("Copy worker failed: {0}")] + CopyWorker(#[source] io::Error), + /// A destination completion returned an unexpected result. + #[error("Destination completion was {actual}, expected {expected}: user_data={user_data}")] + DestinationCompletion { + user_data: u64, + actual: i32, + expected: i32, + }, + /// Submitting an operation to the destination failed. + #[error("Destination request submission failed: {0}")] + DestinationSubmit(#[source] AsyncIoError), + /// Waiting for a destination completion failed. + #[error("Destination wait failed for user_data={user_data}: {source}")] + DestinationWait { + user_data: u64, + #[source] + source: io::Error, + }, + /// Installing the mirror backend failed. + #[error("Mirror installation failed")] + Installation, + /// A source completion returned an unexpected result. + #[error("Source completion was {actual}, expected {expected}: user_data={user_data}")] + SourceCompletion { + user_data: u64, + actual: i32, + expected: i32, + }, +} + +/// Phase of a mirror. +#[derive(Debug, Clone)] +pub enum MirrorPhase { + /// Background copy is in progress. + Running, + /// All blocks copied. Source and destination are in sync. + Ready, + /// Switch-over to the destination is in progress. + Completing, + /// All virtqueues switched to the destination. + Completed, + /// Mirror cancellation is in progress. + Cancelling, + /// The mirror has failed. + Failed(Arc), +} + +/// State shared by the copy worker and the per-queue mirroring +/// `AsyncIo` handles. +pub struct MirrorState { + /// Disk identifier of the block device being mirrored. + disk_id: String, + /// Current phase of the mirror. + phase: Mutex, + range_locks: Arc, + copied_bytes: AtomicU64, + total_bytes: u64, +} + +impl MirrorState { + pub fn new(logical_disk_size: u64, disk_id: String) -> Arc { + Arc::new(Self { + disk_id, + phase: Mutex::new(MirrorPhase::Running), + range_locks: RangeLockManager::new(), + copied_bytes: AtomicU64::new(0), + total_bytes: logical_disk_size, + }) + } + + /// Returns a snapshot of the current phase. + pub fn phase(&self) -> MirrorPhase { + self.phase.lock().unwrap().clone() + } + + /// Attempts a phase transition. + /// + /// Only documented transitions are applied. Invalid transitions panic. + /// + /// Reaching the `Ready` and `Failed(_)` outcomes emits the + /// `vm:disk-mirror-ready` and `vm:disk-mirror-failed` events. Exactly one + /// event fires per outcome because only the first transition applies. + /// + /// Allowed transitions: + /// ```text + /// Running -> Ready | Cancelling | Failed(_) + /// Ready -> Completing | Cancelling | Failed(_) + /// Completing -> Completed + /// Failed(_) -> Cancelling + /// ``` + /// + /// Plus idempotent self-transitions. `Completed` and `Cancelling` are + /// terminal: the mirror handle is dropped out of them, after which + /// `Block::mirror_status` reports no active mirror. + pub fn transition_to_phase(&self, target: MirrorPhase) { + use MirrorPhase::*; + let mut current = self.phase.lock().unwrap(); + + if mem::discriminant(&*current) == mem::discriminant(&target) { + return; + } + + let transition_allowed = matches!( + (&*current, &target), + (Running, Ready) + | (Running, Cancelling) + | (Running, Failed(_)) + | (Ready, Completing) + | (Ready, Cancelling) + | (Ready, Failed(_)) + | (Completing, Completed) + | (Failed(_), Cancelling) + ); + + if !transition_allowed { + // An invalid transition indicates a programming error. Reverting the + // virtqueue workers requires sending a `BlockQueueCommand` to each + // worker, which `MirrorState` cannot do. + panic!( + "Invalid mirror phase transition attempted: {:?} -> {:?}", + *current, target + ); + } + + *current = target; + + match *current { + Ready => event!("vm", "disk-mirror-ready", "id", &self.disk_id), + Failed(_) => event!("vm", "disk-mirror-failed", "id", &self.disk_id), + _ => {} + } + } + + /// Returns a snapshot of the mirror phase and copy progress. + pub fn status(&self) -> MirrorStatus { + MirrorStatus { + phase: self.phase(), + copied_bytes: self.copied_bytes.load(Ordering::Relaxed), + total_bytes: self.total_bytes, + } + } +} + +/// Snapshot of an active block mirror's phase and copy progress. +pub struct MirrorStatus { + /// Current lifecycle phase. + pub phase: MirrorPhase, + /// Number of source bytes copied by the background worker. + pub copied_bytes: u64, + /// Total logical number of bytes to copy. + pub total_bytes: u64, +} + +/// Per-virtqueue [`AsyncIo`] backend for an active block mirror. +/// +/// Reads use the source backend. Mutating requests use both the source and +/// destination backends. +/// +/// If the destination backend fails, all subsequent disk operations are passed +/// through to the source backend. +pub struct MirroringAsyncIo { + source: CompletionIo, + destination: CompletionIo, + state: Arc, + /// Queued completions `(user_data, result)` for + /// [`AsyncIo::next_completed_request`]. + /// + /// The `user_data` identifies the request it was submitted with. The + /// result is the number of bytes transferred or a negative errno. + inflight_completions: VecDeque<(u64, i32)>, + /// Set once this virtqueue worker observes a failure. + /// + /// While true, the worker forwards only to the source and ignores the + /// destination. + bypass_destination: bool, +} + +impl MirroringAsyncIo { + /// Creates a mirroring backend for one virtqueue. + /// + /// `state` must be shared with the copy worker and all other virtqueue + /// backends for the same mirror. + pub fn create( + source_disk: &dyn AsyncFullDiskFile, + destination_disk: &dyn AsyncFullDiskFile, + state: Arc, + ring_depth: u32, + ) -> BlockResult { + let source = CompletionIo::new(source_disk.create_async_io(ring_depth)?)?; + let destination = CompletionIo::new(destination_disk.create_async_io(ring_depth)?)?; + + Ok(Self { + source, + destination, + state, + inflight_completions: VecDeque::new(), + bypass_destination: false, + }) + } + + /// Flips the mirror to the `Failed` phase. + /// + /// The operator must cancel to clean up the destination and the copy worker. + fn fail(&mut self, failure: MirrorFailure) { + // Phase fails the mirror globally, passthrough is per worker, so other queues fail independently. + self.state + .transition_to_phase(MirrorPhase::Failed(Arc::new(failure))); + self.bypass_destination = true; + } + + /// Calls source and destination submissions with mirror-specific error handling. + /// + /// A source submission error is returned to the guest. A destination submission + /// error fails the mirror but is not returned, because `source` is the disk + /// visible to the guest. + fn mirror_request( + &mut self, + submit: impl Fn(&mut dyn AsyncIo) -> AsyncIoResult<()>, + ) -> AsyncIoResult<()> { + submit(self.source.io_mut())?; + if let Err(error) = submit(self.destination.io_mut()) { + self.fail(MirrorFailure::DestinationSubmit(error)); + } + Ok(()) + } + + /// Blocks until `user_data`'s source completion and, unless already in + /// passthrough mode, its destination completion arrive, then queues the + /// guest-visible `(user_data, src_result)`. + /// + /// Other completions seen while waiting are stashed for later delivery. + fn wait_for_completions(&mut self, user_data: u64, expected_result: i32) -> io::Result<()> { + let src_result = + Self::await_completion(&mut self.source, &mut self.inflight_completions, user_data)?; + + if !self.bypass_destination { + match Self::await_completion( + &mut self.destination, + &mut self.inflight_completions, + user_data, + ) { + Ok(dest_result) if dest_result != expected_result => { + self.fail(MirrorFailure::DestinationCompletion { + user_data, + actual: dest_result, + expected: expected_result, + }); + } + Ok(_) => {} + // The destination wait itself failed (broken notifier or epoll). + // Hide it from the guest like any other destination failure. + Err(source) => self.fail(MirrorFailure::DestinationWait { user_data, source }), + } + } + + if src_result != expected_result { + self.fail(MirrorFailure::SourceCompletion { + user_data, + actual: src_result, + expected: expected_result, + }); + } + + self.inflight_completions.push_back((user_data, src_result)); + let _ = self.source.io().notifier().write(1); + Ok(()) + } + + /// Drains `completion_io` until `user_data`'s completion appears and pushes + /// additional ones to `inflight_completions`. + fn await_completion( + completion_io: &mut CompletionIo, + inflight_completions: &mut VecDeque<(u64, i32)>, + user_data: u64, + ) -> io::Result { + loop { + let (id, res) = completion_io.next_completion()?; + if id == user_data { + return Ok(res); + } + inflight_completions.push_back((id, res)); + } + } +} + +impl AsyncIo for MirroringAsyncIo { + /// Returns the source notifier. + /// + /// The destination notifier is consumed internally and is not exposed to + /// the virtqueue worker. + fn notifier(&self) -> &EventFd { + self.source.io().notifier() + } + + fn read_vectored( + &mut self, + offset: off_t, + iovecs: &[iovec], + user_data: u64, + ) -> AsyncIoResult<()> { + self.source + .io_mut() + .read_vectored(offset, iovecs, user_data) + } + + fn write_vectored( + &mut self, + offset: off_t, + iovecs: &[iovec], + user_data: u64, + ) -> AsyncIoResult<()> { + if self.bypass_destination { + return self + .source + .io_mut() + .write_vectored(offset, iovecs, user_data); + } + + let expected_result = iovecs + .iter() + .map(|iov| iov.iov_len) + .sum::() + .try_into() + .map_err(|_| AsyncIoError::WriteVectored(io::Error::other("write is too large")))?; + + let _guard = self + .state + .range_locks + .clone() + .lock_iovecs(offset, iovecs) + .map_err(AsyncIoError::WriteVectored)?; + + self.mirror_request(|backend| backend.write_vectored(offset, iovecs, user_data))?; + + self.wait_for_completions(user_data, expected_result) + .map_err(AsyncIoError::WriteVectored)?; + Ok(()) + } + + fn fsync(&mut self, user_data: Option) -> AsyncIoResult<()> { + if self.bypass_destination { + return self.source.io_mut().fsync(user_data); + } + + self.mirror_request(|backend| backend.fsync(user_data))?; + + // A tracked fsync (Some) waits for its completion. A barrier fsync (None) does not. + if let Some(user_data) = user_data { + self.wait_for_completions(user_data, 0) + .map_err(AsyncIoError::Fsync)?; + } + Ok(()) + } + + fn punch_hole(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> { + if self.bypass_destination { + return self.source.io_mut().punch_hole(offset, length, user_data); + } + + let _guard = self + .state + .range_locks + .clone() + .lock_range(offset, length) + .map_err(AsyncIoError::PunchHole)?; + self.mirror_request(|backend| backend.punch_hole(offset, length, user_data))?; + + self.wait_for_completions(user_data, 0) + .map_err(AsyncIoError::PunchHole)?; + Ok(()) + } + + fn write_zeroes(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> { + if self.bypass_destination { + return self.source.io_mut().write_zeroes(offset, length, user_data); + } + + let _guard = self + .state + .range_locks + .clone() + .lock_range(offset, length) + .map_err(AsyncIoError::WriteZeroes)?; + self.mirror_request(|backend| backend.write_zeroes(offset, length, user_data))?; + + self.wait_for_completions(user_data, 0) + .map_err(AsyncIoError::WriteZeroes)?; + Ok(()) + } + + fn next_completed_request(&mut self) -> Option<(u64, i32)> { + // Mirrored writes are awaited synchronously, only reads and post-failure passthrough writes surface here. + while let Some((id, res)) = self.source.io_mut().next_completed_request() { + self.inflight_completions.push_back((id, res)); + } + self.inflight_completions.pop_front() + } + + fn batch_requests_enabled(&self) -> bool { + if self.bypass_destination { + return self.source.io().batch_requests_enabled(); + } + + true + } + + fn submit_batch_requests(&mut self, batch_request: &[BatchRequest]) -> AsyncIoResult<()> { + if self.bypass_destination { + return self.source.io_mut().submit_batch_requests(batch_request); + } + + for req in batch_request { + let result = match req.request_type { + RequestType::In => self.read_vectored(req.offset, &req.iovecs, req.user_data), + RequestType::Out => self.write_vectored(req.offset, &req.iovecs, req.user_data), + // Only In and Out are batched, see request.rs. + _ => unreachable!("Unexpected batch request type: {:?}", req.request_type), + }; + + // Push partial batch error to completions, vectored op has not + // pushed it to the inflight_completions queue. + if result.is_err() { + self.inflight_completions + .push_back((req.user_data, -libc::EIO)); + let _ = self.source.io().notifier().write(1); + } + } + Ok(()) + } + + fn alignment(&self) -> u64 { + if self.bypass_destination { + return self.source.io().alignment(); + } + + // Stricter alignment wins. Same iovec goes to both backends. + self.source + .io() + .alignment() + .max(self.destination.io().alignment()) + } +} + +/// Owns the copy worker thread's [`JoinHandle`]. +pub struct CopyWorkerHandle { + join: JoinHandle<()>, +} + +impl CopyWorkerHandle { + /// Returns whether the copy worker has finished. + pub fn is_finished(&self) -> bool { + self.join.is_finished() + } + + /// Waits for the copy worker thread to finish. + pub fn join(self) -> thread::Result<()> { + self.join.join() + } +} + +/// Background thread that copies existing source bytes to destination +/// in fixed-size blocks. +/// +/// The worker holds a [`RangeGuard`] across each block so virtqueue mirror +/// writes cannot race the copy. +pub struct CopyWorker { + source_io: CompletionIo, + dest_io: CompletionIo, + dest_is_sparse: bool, + state: Arc, + block_size_bytes: usize, + /// Tracks the next user_data for request and completion notifications. + next_user_data: u64, +} + +impl CopyWorker { + /// Builds and spawns the copy worker on a named thread. + /// + /// Queue depth 1 is enough because the worker is sequential. The caller + /// must initialize the destination disk. + pub fn spawn( + source_disk: &dyn AsyncFullDiskFile, + destination_disk: &dyn AsyncFullDiskFile, + state: Arc, + block_size_bytes: usize, + ) -> BlockResult { + let source_io = CompletionIo::new(source_disk.create_async_io(1)?)?; + let dest_io = CompletionIo::new(destination_disk.create_async_io(1)?)?; + + let worker = Self { + source_io, + dest_io, + dest_is_sparse: destination_disk.supports_sparse_operations(), + state, + block_size_bytes, + next_user_data: 0, + }; + let state = worker.state.clone(); + let join = thread::Builder::new() + .name("blockdev-mirror-copy-worker".into()) + .spawn(move || { + let mut worker = worker; + if let Err(error) = worker.run() { + state.transition_to_phase(MirrorPhase::Failed(Arc::new( + MirrorFailure::CopyWorker(error), + ))); + } + })?; + + Ok(CopyWorkerHandle { join }) + } + + /// Drives the block-by-block copy for predefined [`MirrorState::total_bytes`], + /// then transitions the migration phase to [`MirrorPhase::Ready`]. + fn run(&mut self) -> io::Result<()> { + let alignment = self + .source_io + .io() + .alignment() + .max(self.dest_io.io().alignment()); + let mut buf = AlignedBuf::new(self.block_size_bytes, alignment as usize)?; + let total_size = self.state.total_bytes; + let max_length = self.block_size_bytes as u64; + let mut offset = 0; + + while offset < total_size { + if !matches!(self.state.phase(), MirrorPhase::Running) { + return Ok(()); + } + + let length = max_length.min(total_size - offset) as usize; + self.copy_block(offset, length, &mut buf)?; + offset += length as u64; + } + + let user_data = self.generate_user_data(); + self.dest_io.flush(user_data)?; + self.state.transition_to_phase(MirrorPhase::Ready); + Ok(()) + } + + /// Copies `length` bytes at `offset` from source to destination. + /// + /// Holds a range lock for the duration so virtqueue mirror writes cannot race + /// the copy. + fn copy_block(&mut self, offset: u64, length: usize, buf: &mut AlignedBuf) -> io::Result<()> { + let _guard = self + .state + .range_locks + .clone() + .lock_range(offset, length as u64)?; + + let iovecs = [iovec { + iov_base: buf.as_mut_slice(length).as_mut_ptr().cast(), + iov_len: length, + }]; + let expected_result = + i32::try_from(length).map_err(|_| io::Error::other("copy block is too large"))?; + + // Read from source into buf. + buf.as_mut_slice(length).fill(0); + let read_id = self.generate_user_data(); + self.source_io + .io_mut() + .read_vectored(offset as off_t, &iovecs, read_id) + .map_err(|error| io::Error::other(format!("async io read_vectored failed: {error}")))?; + let (user_data, result) = self.source_io.next_completion()?; + if result < 0 { + return Err(io::Error::from_raw_os_error(-result)); + } + if result != expected_result { + return Err(io::Error::other(format!( + "source read completed {result} bytes, expected {expected_result}" + ))); + } + debug_assert_eq!(user_data, read_id); + + let write_id = self.generate_user_data(); + let punch_hole = self.dest_is_sparse && buf.as_slice(length).iter().all(|&byte| byte == 0); + if punch_hole { + // Source block is all zeros: punch a hole to keep the destination sparse. + self.dest_io + .io_mut() + .punch_hole(offset, length as u64, write_id) + .map_err(|error| { + io::Error::other(format!("async io punch_hole failed: {error}")) + })?; + } else { + // Write buf to destination. + self.dest_io + .io_mut() + .write_vectored(offset as off_t, &iovecs, write_id) + .map_err(|error| { + io::Error::other(format!("async io write_vectored failed: {error}")) + })?; + } + + let (user_data, result) = self.dest_io.next_completion()?; + if result < 0 { + return Err(io::Error::from_raw_os_error(-result)); + } + let expected_result = if punch_hole { 0 } else { expected_result }; + if result != expected_result { + return Err(io::Error::other(format!( + "destination write completed {result} bytes, expected {expected_result}" + ))); + } + debug_assert_eq!(user_data, write_id); + + self.state + .copied_bytes + .fetch_add(length as u64, Ordering::Relaxed); + + Ok(()) + } + + /// Returns the current [`Self::next_user_data`] and increments it, wrapping on overflow. + fn generate_user_data(&mut self) -> u64 { + let user_data = self.next_user_data; + self.next_user_data = self.next_user_data.wrapping_add(1); + + user_data + } +} + +/// Represents an active block mirror operation. +/// +/// The operation must be completed or cancelled before the handle is dropped. +/// Dropping it does not stop or join the background copy worker. +pub struct BlockMirrorHandle { + /// Shared lifecycle state and copy progress. + pub state: Arc, + /// Handle for joining the background copy worker. + pub copy_worker: CopyWorkerHandle, + /// Destination backend of the mirror. + pub destination: Box, + /// Host path backing the destination. + pub destination_path: PathBuf, +} + +/// Owns an [`AsyncIo`] backend and waits for its completions. +/// +/// Waiting uses a poller because the backend's notifier is created +/// non-blocking and therefore never blocks on read. +struct CompletionIo { + poll: PollContext<()>, + io: Box, +} + +impl CompletionIo { + fn new(io: Box) -> io::Result { + let poll = PollContext::new()?; + poll.add(io.notifier(), ())?; + Ok(Self { poll, io }) + } + + fn io(&self) -> &dyn AsyncIo { + self.io.as_ref() + } + + fn io_mut(&mut self) -> &mut dyn AsyncIo { + self.io.as_mut() + } + + /// Blocks until the owned backend reports a completion, then returns it. + fn next_completion(&mut self) -> io::Result<(u64, i32)> { + loop { + if let Some(completion) = self.io.next_completed_request() { + return Ok(completion); + } + // EINTR is retried inside `wait`. + self.poll.wait()?; + // Drain the eventfd so the next wait does not fire on a stale signal. + self.io.notifier().read()?; + } + } + + /// Submits a tracked flush and waits for its matching successful completion. + /// + /// The completion must carry `user_data` and report zero, as required by + /// [`AsyncIo::fsync`]. + fn flush(&mut self, user_data: u64) -> io::Result<()> { + self.io + .fsync(Some(user_data)) + .map_err(|error| io::Error::other(format!("async io fsync failed: {error}")))?; + + let (completed_user_data, result) = self.next_completion()?; + if completed_user_data != user_data { + return Err(io::Error::other(format!( + "fsync completed with unexpected user data {completed_user_data}, expected {user_data}" + ))); + } + if result != 0 { + return Err(if result < 0 { + io::Error::from_raw_os_error(-result) + } else { + io::Error::other(format!("fsync completed with unexpected result {result}")) + }); + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::AtomicUsize; + + use super::*; + + /// Overlap is detected whether the held range precedes the query or starts + /// inside it. + #[test] + fn overlaps_detects_overlap() { + let mut preceding = BTreeMap::new(); + preceding.insert(10u64, 25u64); + assert!(RangeLockManager::overlaps_any(&preceding, 20, 30)); + + let mut starts_inside = BTreeMap::new(); + starts_inside.insert(10u64, 20u64); + starts_inside.insert(25u64, 30u64); + assert!(RangeLockManager::overlaps_any(&starts_inside, 21, 26)); + } + + #[test] + fn overlaps_disjoint_returns_false() { + let mut locked = BTreeMap::new(); + locked.insert(10u64, 20u64); + locked.insert(30u64, 40u64); + assert!(!RangeLockManager::overlaps_any(&locked, 22, 28)); + } + + #[test] + fn overlaps_touching_boundary_is_not_overlap() { + let mut locked = BTreeMap::new(); + locked.insert(10u64, 20u64); + assert!(!RangeLockManager::overlaps_any(&locked, 20, 30)); + } + + /// Verifies that empty and overflowing ranges are rejected as invalid input. + #[test] + fn range_lock_rejects_empty_and_overflowing_ranges() { + let manager = RangeLockManager::new(); + + let empty = manager.clone().lock_range(0, 0).err().unwrap(); + assert_eq!(empty.kind(), io::ErrorKind::InvalidInput); + assert_eq!(empty.to_string(), "Range length is zero"); + + let overflow = manager.lock_range(u64::MAX, 1).err().unwrap(); + assert_eq!(overflow.kind(), io::ErrorKind::InvalidInput); + assert_eq!(overflow.to_string(), "Range overflow"); + } + + use std::collections::VecDeque; + use std::sync::mpsc; + use std::time::Duration; + + /// Submission counters shared between a [`MockAsyncIo`] and the test. + #[derive(Default)] + struct MockStats { + read_submissions: AtomicUsize, + write_submissions: AtomicUsize, + } + + /// In-memory [`AsyncIo`] backend that completes submissions immediately. + struct MockAsyncIo { + evt: EventFd, + completions: VecDeque<(u64, i32)>, + completion_result: Option, + /// When set, the write submission at this 0-based index fails instead + /// of completing. + fail_on_nth_write: Option, + /// Byte written into read buffers. + fill_byte: u8, + /// Submission counters shared with the test. + stats: Arc, + } + + impl MockStats { + #[track_caller] + fn assert_submissions(&self, reads: usize, writes: usize, message: &str) { + assert_eq!( + ( + self.read_submissions.load(Ordering::SeqCst), + self.write_submissions.load(Ordering::SeqCst) + ), + (reads, writes), + "{message}" + ); + } + } + + impl MockAsyncIo { + fn new() -> Self { + Self::with_fill_byte(0) + } + + /// Creates a mock that fills read buffers with `fill_byte` so a test + /// can tell which backend served a read. + fn with_fill_byte(fill_byte: u8) -> Self { + Self { + evt: EventFd::new(libc::EFD_NONBLOCK).unwrap(), + completions: VecDeque::new(), + completion_result: None, + fail_on_nth_write: None, + fill_byte, + stats: Arc::new(MockStats::default()), + } + } + + /// Returns the shared submission counters. + fn stats(&self) -> Arc { + Arc::clone(&self.stats) + } + + /// Fails the write submission at `index`, counting from zero. + fn fail_on_nth_write_submission(&mut self, index: usize) { + self.fail_on_nth_write = Some(index); + } + + /// Records a completion and signals the notifier. + fn complete(&mut self, user_data: u64, result: i32) { + let result = self.completion_result.take().unwrap_or(result); + self.completions.push_back((user_data, result)); + self.evt.write(1).unwrap(); + } + } + + impl AsyncIo for MockAsyncIo { + fn notifier(&self) -> &EventFd { + &self.evt + } + fn read_vectored(&mut self, _o: off_t, iovecs: &[iovec], ud: u64) -> AsyncIoResult<()> { + self.stats.read_submissions.fetch_add(1, Ordering::SeqCst); + for iov in iovecs { + // SAFETY: the mirror passes writable buffers that stay valid + // for the duration of the submission. + unsafe { + std::ptr::write_bytes(iov.iov_base.cast::(), self.fill_byte, iov.iov_len); + } + } + self.complete( + ud, + iovecs.iter().map(|iov| iov.iov_len).sum::() as i32, + ); + Ok(()) + } + fn write_vectored(&mut self, _o: off_t, iovecs: &[iovec], ud: u64) -> AsyncIoResult<()> { + let index = self.stats.write_submissions.fetch_add(1, Ordering::SeqCst); + if self.fail_on_nth_write == Some(index) { + return Err(AsyncIoError::WriteVectored(io::Error::other( + "injected write submit failure", + ))); + } + self.complete( + ud, + iovecs.iter().map(|iov| iov.iov_len).sum::() as i32, + ); + Ok(()) + } + fn fsync(&mut self, ud: Option) -> AsyncIoResult<()> { + if let Some(ud) = ud { + self.complete(ud, 0); + } + Ok(()) + } + fn punch_hole(&mut self, _o: u64, _l: u64, ud: u64) -> AsyncIoResult<()> { + self.complete(ud, 0); + Ok(()) + } + fn write_zeroes(&mut self, _o: u64, _l: u64, ud: u64) -> AsyncIoResult<()> { + self.complete(ud, 0); + Ok(()) + } + fn next_completed_request(&mut self) -> Option<(u64, i32)> { + self.completions.pop_front() + } + } + + fn mirror_with_mocks() -> MirroringAsyncIo { + mirror_from( + MockAsyncIo::new(), + MockAsyncIo::new(), + MirrorState::new(1 << 20, "test-disk".into()), + ) + } + + /// The one place to update when `MirroringAsyncIo`'s fields change. + fn mirror_from( + source: S, + destination: D, + state: Arc, + ) -> MirroringAsyncIo { + MirroringAsyncIo { + source: CompletionIo::new(Box::new(source)).unwrap(), + destination: CompletionIo::new(Box::new(destination)).unwrap(), + state, + inflight_completions: VecDeque::new(), + bypass_destination: false, + } + } + + /// One iovec over `buf` for a disk write. + fn iov_of(buf: &[u8]) -> [iovec; 1] { + [iovec { + iov_base: buf.as_ptr() as *mut libc::c_void, + iov_len: buf.len(), + }] + } + + /// One iovec over `buf` for a disk read, which the mock fills. + fn iov_of_mut(buf: &mut [u8]) -> [iovec; 1] { + [iovec { + iov_base: buf.as_mut_ptr().cast(), + iov_len: buf.len(), + }] + } + + /// Runs `f` on a worker thread and fails the test if it does not finish + /// within `timeout`. + /// + /// This turns a submit-path deadlock into a clean failure instead of a hung + /// suite: the worker stays blocked, but the test thread resumes after the + /// timeout and panics. + fn run_with_watchdog(timeout: Duration, f: impl FnOnce() + Send + 'static) { + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + f(); + let _ = tx.send(()); + }); + if rx.recv_timeout(timeout).is_err() { + panic!("scenario did not finish within {timeout:?} (deadlock)"); + } + } + + /// Drains completions until `n` have arrived (or the budget is exhausted). + fn drain_n(mirror: &mut MirroringAsyncIo, n: usize) -> Vec { + let mut acked = Vec::new(); + for _ in 0..64 { + while let Some((user_data, result)) = mirror.next_completed_request() { + assert!(result >= 0, "unexpected error completion: {result}"); + acked.push(user_data); + } + if acked.len() >= n { + break; + } + } + acked + } + + /// Returns the stored failure or panics if the mirror has not failed. + fn failure_reason(state: &MirrorState) -> Arc { + let MirrorPhase::Failed(reason) = state.phase() else { + panic!("mirror did not enter the failed phase"); + }; + reason + } + + /// Two overlapping guest writes submitted before either is reaped must both + /// complete in submission order without deadlocking. + #[test] + fn overlapping_writes_complete_in_order() { + run_with_watchdog(Duration::from_secs(5), || { + let mut mirror = mirror_with_mocks(); + let buf = [0u8; 4096]; + let iov = iov_of(&buf); + + mirror.write_vectored(0, &iov, 1).unwrap(); + mirror.write_vectored(0, &iov, 2).unwrap(); + + assert_eq!( + drain_n(&mut mirror, 2), + vec![1, 2], + "both overlapping writes complete in submission order" + ); + }); + } + + /// While the copy worker holds a range (simulated by holding a `RangeGuard` + /// on the shared lock manager), an overlapping guest write must block and + /// proceed only once the range is released. + #[test] + fn copy_worker_hold_serializes_overlapping_guest_write() { + let state = MirrorState::new(1 << 20, "test-disk".into()); + // The "copy worker" holds [0, 4096). + let guard = state.range_locks.clone().lock_range(0, 4096).unwrap(); + + let mut mirror = mirror_from(MockAsyncIo::new(), MockAsyncIo::new(), state.clone()); + + let (tx, rx) = mpsc::channel(); + let handle = thread::spawn(move || { + let buf = [0u8; 4096]; + let iov = iov_of(&buf); + mirror.write_vectored(0, &iov, 1).unwrap(); + tx.send(()).unwrap(); + }); + + // The held range must block the overlapping guest write. + assert!( + rx.recv_timeout(Duration::from_millis(200)).is_err(), + "guest write proceeded while the copy worker held the range" + ); + + // Releasing the range lets the write through. + drop(guard); + assert!( + rx.recv_timeout(Duration::from_secs(5)).is_ok(), + "guest write did not proceed after the range was released" + ); + handle.join().unwrap(); + } + + /// Reads are source-only passthrough (no range lock) and still complete. + #[test] + fn read_passes_through_to_source() { + run_with_watchdog(Duration::from_secs(5), || { + let source = MockAsyncIo::with_fill_byte(0xaa); + let source_stats = source.stats(); + let destination = MockAsyncIo::with_fill_byte(0xbb); + let destination_stats = destination.stats(); + let state = MirrorState::new(1 << 20, "test-disk".into()); + let mut mirror = mirror_from(source, destination, state.clone()); + + // The copy worker holds the range while it copies the block. A read + // that took the range lock would stall here until the watchdog fires. + let _guard = state.range_locks.clone().lock_range(0, 4096).unwrap(); + + let mut buf = [0u8; 4096]; + let iov = iov_of_mut(&mut buf); + mirror.read_vectored(0, &iov, 7).unwrap(); + + assert_eq!( + mirror.next_completed_request(), + Some((7, 4096)), + "read completes via the source" + ); + assert!( + buf.iter().all(|byte| *byte == 0xaa), + "read must return source data, not destination data" + ); + source_stats.assert_submissions(1, 0, "the read is submitted to the source"); + destination_stats.assert_submissions(0, 0, "no read is submitted to the destination"); + }); + } + + /// A destination submit failure degrades the mirror to source passthrough: + /// the phase goes `Failed`, and a subsequent write is submitted to the + /// source only. + #[test] + fn destination_submit_failure_degrades_to_passthrough() { + run_with_watchdog(Duration::from_secs(5), || { + let source = MockAsyncIo::new(); + let source_stats = source.stats(); + let mut dest = MockAsyncIo::new(); + let dest_stats = dest.stats(); + dest.fail_on_nth_write_submission(0); + let mut mirror = + mirror_from(source, dest, MirrorState::new(1 << 20, "test-disk".into())); + let buf = [0u8; 4096]; + let iov = iov_of(&buf); + + mirror.write_vectored(0, &iov, 1).unwrap(); + assert!( + matches!(mirror.state.phase(), MirrorPhase::Failed(_)), + "destination failure transitions the mirror to Failed" + ); + + // Subsequent write goes to the source only. + mirror.write_vectored(0, &iov, 2).unwrap(); + + let mut acked = drain_n(&mut mirror, 2); + acked.sort(); + assert_eq!(acked, vec![1, 2], "both writes complete off the source"); + source_stats.assert_submissions(0, 2, "both writes are submitted to the source"); + dest_stats.assert_submissions(0, 1, "the destination sees only the mirrored write"); + }); + } + + /// Verifies that a short destination completion fails the mirror while the guest + /// receives the successful source completion. + #[test] + fn short_destination_completion_uses_source_result() { + let mut destination = MockAsyncIo::new(); + destination.completion_result = Some(2048); + let mut mirror = mirror_from( + MockAsyncIo::new(), + destination, + MirrorState::new(4096, "test-disk".into()), + ); + let buf = [0u8; 4096]; + + mirror.write_vectored(0, &iov_of(&buf), 7).unwrap(); + + assert_eq!(mirror.next_completed_request(), Some((7, 4096))); + assert!(mirror.bypass_destination); + assert!(matches!( + failure_reason(&mirror.state).as_ref(), + MirrorFailure::DestinationCompletion { + user_data: 7, + actual: 2048, + expected: 4096, + } + )); + } + + /// Verifies that a source I/O error fails the mirror and is reported to the guest. + #[test] + fn source_io_error_reaches_guest() { + let mut source = MockAsyncIo::new(); + source.completion_result = Some(-libc::EIO); + let mut mirror = mirror_from( + source, + MockAsyncIo::new(), + MirrorState::new(4096, "test-disk".into()), + ); + let buf = [0u8; 4096]; + + mirror.write_vectored(0, &iov_of(&buf), 9).unwrap(); + + assert_eq!(mirror.next_completed_request(), Some((9, -libc::EIO))); + assert!(matches!( + failure_reason(&mirror.state).as_ref(), + MirrorFailure::SourceCompletion { + user_data: 9, + actual, + expected: 4096, + } if *actual == -libc::EIO + )); + } + + /// Mock backend whose completions are withheld until [`Gate::release`], so a + /// test can hold a write parked in `wait_for_completions`. + struct GatedMockAsyncIo { + evt: EventFd, + inner: Arc>, + /// Notified on each submit, so a test can wait until the in-flight write + /// has reached this backend (and so already holds its range guard). + on_submit: mpsc::Sender<()>, + } + + struct GatedInner { + /// Submitted, not yet released. + pending: VecDeque<(u64, i32)>, + /// Released, deliverable via `next_completed_request`. + ready: VecDeque<(u64, i32)>, + } + + /// Releases a [`GatedMockAsyncIo`]'s withheld completions from another thread. + struct Gate { + evt: EventFd, + inner: Arc>, + } + + impl Gate { + fn release(&self) { + let mut inner = self.inner.lock().unwrap(); + while let Some(completion) = inner.pending.pop_front() { + inner.ready.push_back(completion); + } + self.evt.write(1).unwrap(); + } + } + + impl GatedMockAsyncIo { + fn new(on_submit: mpsc::Sender<()>) -> Self { + Self { + evt: EventFd::new(libc::EFD_NONBLOCK).unwrap(), + inner: Arc::new(Mutex::new(GatedInner { + pending: VecDeque::new(), + ready: VecDeque::new(), + })), + on_submit, + } + } + + fn gate(&self) -> Gate { + Gate { + evt: self.evt.try_clone().unwrap(), + inner: Arc::clone(&self.inner), + } + } + + fn submit(&self, user_data: u64, result: i32) { + self.inner + .lock() + .unwrap() + .pending + .push_back((user_data, result)); + let _ = self.on_submit.send(()); + } + } + + impl AsyncIo for GatedMockAsyncIo { + fn notifier(&self) -> &EventFd { + &self.evt + } + fn read_vectored(&mut self, _o: off_t, iovecs: &[iovec], ud: u64) -> AsyncIoResult<()> { + self.submit( + ud, + iovecs.iter().map(|iov| iov.iov_len).sum::() as i32, + ); + Ok(()) + } + fn write_vectored(&mut self, _o: off_t, iovecs: &[iovec], ud: u64) -> AsyncIoResult<()> { + self.submit( + ud, + iovecs.iter().map(|iov| iov.iov_len).sum::() as i32, + ); + Ok(()) + } + fn fsync(&mut self, ud: Option) -> AsyncIoResult<()> { + if let Some(ud) = ud { + self.submit(ud, 0); + } + Ok(()) + } + fn punch_hole(&mut self, _o: u64, _l: u64, ud: u64) -> AsyncIoResult<()> { + self.submit(ud, 0); + Ok(()) + } + fn write_zeroes(&mut self, _o: u64, _l: u64, ud: u64) -> AsyncIoResult<()> { + self.submit(ud, 0); + Ok(()) + } + fn next_completed_request(&mut self) -> Option<(u64, i32)> { + self.inner.lock().unwrap().ready.pop_front() + } + } + + /// The range guard must stay held across the whole synchronous submit+wait, + /// not just acquisition. + /// + /// A regression to `let _ =` drops it early and lets an overlapping + /// `lock_range` acquire while the write is still in flight. + #[test] + fn guard_is_held_across_submit_and_wait() { + let state = MirrorState::new(1 << 20, "test-disk".into()); + + // Source completes immediately; destination is gated, so the write parks + // waiting on the destination completion while holding the range lock. + let (submitted_tx, submitted_rx) = mpsc::channel(); + let dest = GatedMockAsyncIo::new(submitted_tx); + let gate = dest.gate(); + let mut mirror = mirror_from(MockAsyncIo::new(), dest, state.clone()); + + let writer = thread::spawn(move || { + let buf = [0u8; 4096]; + let iov = iov_of(&buf); + mirror.write_vectored(0, &iov, 1).unwrap(); + }); + + // The write reached the destination submit, so its range guard is held. + submitted_rx.recv().unwrap(); + + // An overlapping lock_range must block while the in-flight write holds it. + let locker_state = state.clone(); + let (locked_tx, locked_rx) = mpsc::channel(); + let locker = thread::spawn(move || { + let _g = locker_state + .range_locks + .clone() + .lock_range(0, 4096) + .unwrap(); + locked_tx.send(()).unwrap(); + }); + assert!( + locked_rx.recv_timeout(Duration::from_millis(200)).is_err(), + "lock_range acquired while the in-flight write still held the range" + ); + + // Releasing the destination completion lets the write finish and drop its + // guard, which unblocks the overlapping lock_range. + gate.release(); + writer.join().unwrap(); + assert!( + locked_rx.recv_timeout(Duration::from_secs(5)).is_ok(), + "lock_range did not acquire after the write released the range" + ); + locker.join().unwrap(); + } + + /// A single-iovec `Out` batch entry backed by `buf`. + fn batch_write(offset: off_t, buf: &[u8], user_data: u64) -> BatchRequest { + BatchRequest { + offset, + iovecs: [iovec { + iov_base: buf.as_ptr() as *mut libc::c_void, + iov_len: buf.len(), + }] + .into_iter() + .collect(), + user_data, + request_type: RequestType::Out, + } + } + + /// A mid-batch submit failure must still return `Ok` with one completion per + /// entry, including an error completion for the failed one. + /// + /// The worker records the batch as in-flight only on `Ok`, so aborting with + /// `Err` strands the completions already queued for earlier entries. + #[test] + fn failed_batch_submit_accounts_every_request() { + // The second write fails at the source submit. The first one goes through. + let mut source = MockAsyncIo::new(); + source.fail_on_nth_write_submission(1); + let mut mirror = mirror_from( + source, + MockAsyncIo::new(), + MirrorState::new(1 << 20, "test-disk".into()), + ); + let buf = [0u8; 4096]; + + let batch = [batch_write(0, &buf, 1), batch_write(4096, &buf, 2)]; + + mirror + .submit_batch_requests(&batch) + .expect("a mid-batch submit failure must not fail the whole batch"); + + let mut completions = Vec::new(); + while let Some(completion) = mirror.next_completed_request() { + completions.push(completion); + } + completions.sort_by_key(|(user_data, _)| *user_data); + + assert_eq!( + completions.len(), + 2, + "every batch entry owes exactly one completion" + ); + assert_eq!( + completions[0], + (1, 4096), + "first write completes successfully" + ); + assert_eq!(completions[1].0, 2, "second entry is still accounted"); + assert!( + completions[1].1 < 0, + "second entry carries an error result (reported IOERR), not an orphan" + ); + } + + /// The lifecycle advances Running -> Ready -> Completing -> Completed, each + /// state reached only from its documented predecessor. + #[test] + fn phase_advances_through_the_lifecycle() { + let state = MirrorState::new(1 << 20, "test-disk".into()); + assert!(matches!(state.phase(), MirrorPhase::Running)); + state.transition_to_phase(MirrorPhase::Ready); + state.transition_to_phase(MirrorPhase::Completing); + state.transition_to_phase(MirrorPhase::Completed); + assert!(matches!(state.phase(), MirrorPhase::Completed)); + } + + /// An invalid phase transition panics. + #[test] + #[should_panic(expected = "Invalid mirror phase transition attempted")] + fn invalid_phase_transition_panics() { + let state = MirrorState::new(1 << 20, "test-disk".into()); + // Running -> Completed skips Ready and Completing, so it is rejected. + state.transition_to_phase(MirrorPhase::Completed); + } + + /// `Completed` is terminal: no later transition is accepted. + #[test] + #[should_panic(expected = "Invalid mirror phase transition attempted")] + fn completed_phase_is_terminal() { + let state = MirrorState::new(1 << 20, "test-disk".into()); + state.transition_to_phase(MirrorPhase::Ready); + state.transition_to_phase(MirrorPhase::Completing); + state.transition_to_phase(MirrorPhase::Completed); + state.transition_to_phase(MirrorPhase::Cancelling); + } + + /// A failure keeps its first reason (transitions compare only the variant) + /// and can still move to `Cancelling` for cleanup. + #[test] + fn failed_keeps_first_reason_then_cancels() { + let state = MirrorState::new(1 << 20, "test-disk".into()); + let first = Arc::new(MirrorFailure::SourceCompletion { + user_data: 1, + actual: -libc::EIO, + expected: 0, + }); + state.transition_to_phase(MirrorPhase::Failed(first.clone())); + state.transition_to_phase(MirrorPhase::Failed(Arc::new( + MirrorFailure::SourceCompletion { + user_data: 2, + actual: -libc::EIO, + expected: 0, + }, + ))); + let MirrorPhase::Failed(reason) = state.phase() else { + panic!("mirror did not enter the failed phase"); + }; + assert!(Arc::ptr_eq(&reason, &first)); + state.transition_to_phase(MirrorPhase::Cancelling); + assert!(matches!(state.phase(), MirrorPhase::Cancelling)); + } + + /// A tracked fsync (`Some`) flushes both backends and surfaces one guest + /// completion for its user_data. + #[test] + fn tracked_fsync_completes_to_guest() { + run_with_watchdog(Duration::from_secs(5), || { + let mut mirror = mirror_with_mocks(); + mirror.fsync(Some(5)).unwrap(); + assert_eq!(drain_n(&mut mirror, 1), vec![5]); + }); + } + + /// A barrier fsync (`None`) flushes both backends but owes the guest no + /// completion, so nothing surfaces. + #[test] + fn barrier_fsync_surfaces_no_completion() { + let mut mirror = mirror_with_mocks(); + mirror.fsync(None).unwrap(); + assert!(mirror.next_completed_request().is_none()); + } + + /// `write_zeroes` mirrors to both backends under the range lock and + /// surfaces one guest completion, like a write. + #[test] + fn write_zeroes_mirrors_and_completes() { + run_with_watchdog(Duration::from_secs(5), || { + let mut mirror = mirror_with_mocks(); + mirror.write_zeroes(0, 4096, 3).unwrap(); + assert_eq!(drain_n(&mut mirror, 1), vec![3]); + }); + } + + /// Once degraded to passthrough, every mutating op forwards to the source + /// alone and still completes, with no destination and no range lock. + #[test] + fn degraded_mirror_passes_all_ops_through_to_source() { + run_with_watchdog(Duration::from_secs(5), || { + let mut dest = MockAsyncIo::new(); + dest.fail_on_nth_write_submission(0); + let mut mirror = mirror_from( + MockAsyncIo::new(), + dest, + MirrorState::new(1 << 20, "test-disk".into()), + ); + let buf = [0u8; 4096]; + let iov = iov_of(&buf); + + // The first write fails on the destination and flips to passthrough. + mirror.write_vectored(0, &iov, 1).unwrap(); + assert!(matches!(mirror.state.phase(), MirrorPhase::Failed(_))); + + // Subsequent ops take the source-only passthrough branch. + mirror.fsync(Some(2)).unwrap(); + mirror.punch_hole(0, 4096, 3).unwrap(); + mirror.write_zeroes(0, 4096, 4).unwrap(); + + let mut acked = drain_n(&mut mirror, 4); + acked.sort(); + assert_eq!(acked, vec![1, 2, 3, 4]); + }); + } +} diff --git a/block/src/qcow_disk.rs b/block/src/qcow_disk.rs index ef27305264..dcf547b34a 100644 --- a/block/src/qcow_disk.rs +++ b/block/src/qcow_disk.rs @@ -3,6 +3,7 @@ // SPDX-License-Identifier: Apache-2.0 use std::fs::File; +use std::io::Error as IoError; use std::os::unix::io::AsRawFd; use std::sync::Arc; use std::{fmt, io}; @@ -142,7 +143,18 @@ impl disk_file::Resizable for QcowDisk { } } -impl disk_file::DiskFile for QcowDisk {} +impl disk_file::DiskFile for QcowDisk { + fn supports_mirroring(&self) -> BlockResult<()> { + if self.backing_file.is_some() { + return Err(BlockError::new( + BlockErrorKind::UnsupportedFeature, + IoError::other("block mirroring does not support backing files"), + )); + } + + Ok(()) + } +} impl disk_file::AsyncDiskFile for QcowDisk { fn try_clone(&self) -> BlockResult> { diff --git a/block/src/raw_disk.rs b/block/src/raw_disk.rs index 82dd3c5302..2d6bc1d7f7 100644 --- a/block/src/raw_disk.rs +++ b/block/src/raw_disk.rs @@ -113,7 +113,11 @@ impl disk_file::Resizable for RawDisk { } } -impl disk_file::DiskFile for RawDisk {} +impl disk_file::DiskFile for RawDisk { + fn supports_mirroring(&self) -> BlockResult<()> { + Ok(()) + } +} impl disk_file::AsyncDiskFile for RawDisk { fn try_clone(&self) -> BlockResult> { diff --git a/docs/disk_mirroring.md b/docs/disk_mirroring.md new file mode 100644 index 0000000000..023b336de6 --- /dev/null +++ b/docs/disk_mirroring.md @@ -0,0 +1,198 @@ +# Disk Mirroring + +Disk mirroring copies a running VM's disk to another file on the host and +keeps the two in sync, so the disk image can be moved to a different backing +store without stopping the guest. It is the Cloud Hypervisor counterpart of +QEMU's `blockdev-mirror`. Once started, the mirror keeps source and +destination in sync until it is completed or cancelled. + +A typical use is rebalancing storage: when the share backing a disk image +fills up, the operator mirrors that disk onto a file on another share and +switches the VM over to it. + +## Overview + +Mirroring runs as a sequence of phases driven by four API calls: + +- `/vm.disk-mirror-start` begins mirroring a disk onto a destination path. +- `/vm.disk-mirror-status` reports the current phase and copy progress. +- `/vm.disk-mirror-complete` switches the VM over to the destination. +- `/vm.disk-mirror-cancel` aborts and keeps the VM on the source. + +```mermaid +stateDiagram-v2 + [*] --> running: disk-mirror-start + running --> ready: background copy finished + ready --> completing: disk-mirror-complete + completing --> completed: all queues switched + completed --> [*] + running --> cancelling: disk-mirror-cancel + ready --> cancelling: disk-mirror-cancel + failed --> cancelling: disk-mirror-cancel + running --> failed: destination I/O error + ready --> failed: destination I/O error + cancelling --> [*] +``` + +While `running`, a background worker copies the existing data block by block. +At the same time every guest write is forwarded to both disks, so once the +copy finishes the two are identical. Reaching `ready` means the two disks are +in sync and stay so until the operator completes or cancels. + +## Operator usage + +The examples use `curl` against the VMM's API socket. Replace the socket path +and the disk identifier with your own. The disk identifier is the device `id` +shown by `vm.info` (the same `id` used when the disk was configured or hot +added). + +### Start a mirror + +The destination image must already exist with the same image format and +logical size as the source. Cloud Hypervisor does not create or resize it. +Care must be taken as all data on the destination is overwritten. Cloud +Hypervisor does not ask for confirmation. + +```console +curl --unix-socket /tmp/cloud-hypervisor.sock -i \ + -X PUT 'http://localhost/api/v1/vm.disk-mirror-start' \ + -H 'Content-Type: application/json' \ + -d '{"id": "_disk0", "destination_path": "/new/store/disk0.raw"}' +``` + +This switches the disk to a mirroring backend and starts the background copy. +The VM keeps serving I/O throughout. A `204` response means mirroring started. + +Mirroring supports standalone QCOW2 images. QCOW2 sources and destinations +with backing files are rejected because mirroring copies the full logical +contents and would flatten the image. + +### Check progress + +```console +curl --unix-socket /tmp/cloud-hypervisor.sock \ + -X PUT 'http://localhost/api/v1/vm.disk-mirror-status' \ + -H 'Content-Type: application/json' \ + -d '{"id": "_disk0"}' +``` + +The response reports the phase and how far the copy has progressed: + +```json +{"phase": "running", "copied_bytes": 1073741824, "total_bytes": 4294967296} +``` + +`phase` is one of `running`, `ready`, `completing`, `completed`, +`cancelling`, or `failed`. A `failed` status also carries a `failure` field +describing what went wrong. Poll this endpoint until the phase becomes +`ready`. + +### Complete the mirror + +Once the phase is `ready`, switch the VM over to the destination: + +```console +curl --unix-socket /tmp/cloud-hypervisor.sock -i \ + -X PUT 'http://localhost/api/v1/vm.disk-mirror-complete' \ + -H 'Content-Type: application/json' \ + -d '{"id": "_disk0"}' +``` + +The call blocks until the switch-over finishes. On success (`204`) the VM +serves all I/O from the destination disk and the source disk can be removed. +Completion is only accepted from the `ready` phase. A `404` or `400` leaves the +mirror active, so you can fix the cause and retry. + +### Cancel the mirror + +At any time before completion the operator can abort and keep the VM on the +source disk: + +```console +curl --unix-socket /tmp/cloud-hypervisor.sock -i \ + -X PUT 'http://localhost/api/v1/vm.disk-mirror-cancel' \ + -H 'Content-Type: application/json' \ + -d '{"id": "_disk0"}' +``` + +The destination disk is released and the VM continues on the source. Cancel is +refused once completion has been requested, because by then a queue may +already be writing only to the destination. + +### Failure handling + +If the destination disk fails (for example its backing store becomes +unreachable), the mirror moves to `failed` and the affected queues fall back +to serving the guest from the source disk, so the guest keeps running on +intact data. The operator then cancels the failed mirror to release the +destination. + +### Events + +The VMM emits one `vm` event for each mirror outcome via `--event-monitor`. +Each event includes an `id` property containing the disk identifier. + +- `vm:disk-mirror-ready` when the background copy finishes. +- `vm:disk-mirror-failed` when the mirror fails. +- `vm:disk-mirror-completed` when switching to the destination finishes. +- `vm:disk-mirror-cancelled` when the mirror is cancelled. + +### Unrecoverable errors + +Completing a mirror cannot be undone. Once the switch to the destination +begins, some virtqueues may already be writing only to the destination, so +there is no consistent state to roll back to. If a queue cannot be switched +over during completion, the VMM aborts. The alternative would leave the disk +half on the source and half on the destination and could lose acknowledged +writes. This is rare: it needs a queue worker to fail mid-swap (for example an +epoll registration error), or its switch-over command to be lost or +unacknowledged. + +### Conflicting operations + +While a mirror is active, the VMM rejects operations that would disturb it: +snapshotting, live migration, resizing the disk, removing the device, and +API requests to reboot, shut down, or delete the VM. Complete or cancel the +mirror first. If the guest requests a reboot or shutdown, the VMM stops the +guest but keeps the mirror and API available. The requested lifecycle operation +continues after the operator completes or cancels every active mirror. Pausing +the VM is allowed, but a mirror cannot be started, completed, or cancelled while +the device is paused. + +## Implementation details + +Mirroring is built from two cooperating pieces and a range lock that keeps +them from corrupting each other: + +```mermaid +flowchart LR + guest[Guest] -->|read / write| mio[MirroringAsyncIo] + mio -->|reads, all writes| src[(Source disk)] + mio -->|writes only| dst[(Destination disk)] + cw[CopyWorker] -->|read block| src + cw -->|write block| dst + mio -.range lock.- rl((RangeLockManager)) + cw -.range lock.- rl +``` + +**CopyWorker.** A background thread copies the source disk to the destination +in 512 KiB blocks. A block that reads back as all zeros is punched as a hole +on the destination instead of being written, so sparse images stay sparse. The +worker updates the copied-byte counter that `vm.disk-mirror-status` reports, +and stops early once the phase becomes terminal. + +**MirroringAsyncIo.** When a mirror starts, each virtqueue worker's `AsyncIo` +backend is swapped for a `MirroringAsyncIo`. It forwards reads to the source +and forwards every mutating operation (`write_vectored`, `fsync`, +`punch_hole`, `write_zeroes`) to both the source and the destination. The +completions of the two sides are awaited inside the write call, so an error on +the destination can be handled before the guest sees the write as done. On a +destination error that queue degrades to source passthrough and the mirror +fails, rather than letting the guest diverge from intact data. + +**Range lock.** The CopyWorker and the guest writes can target overlapping +byte ranges at the same time. Each side takes an exclusive lock on the range +it is about to touch and holds it until its I/O completes, so a copy and a +guest write to the same region cannot interleave into an inconsistent result. +Lookups are over a small set of held ranges, so the lock is cheap in the +common non-overlapping case. diff --git a/fuzz/fuzz_targets/http_api.rs b/fuzz/fuzz_targets/http_api.rs index aa3841243d..c2b2fa023c 100644 --- a/fuzz/fuzz_targets/http_api.rs +++ b/fuzz/fuzz_targets/http_api.rs @@ -113,6 +113,22 @@ impl RequestHandler for StubApiRequestHandler { Ok(()) } + fn vm_disk_mirror_start(&mut self, _: String, _: PathBuf) -> Result<(), VmError> { + Ok(()) + } + + fn vm_disk_mirror_status(&mut self, _: String) -> Result>, VmError> { + Ok(None) + } + + fn vm_disk_mirror_complete(&mut self, _: String) -> Result<(), VmError> { + Ok(()) + } + + fn vm_disk_mirror_cancel(&mut self, _: String) -> Result<(), VmError> { + Ok(()) + } + #[cfg(target_arch = "x86_64")] fn vm_coredump(&mut self, _: &str) -> Result<(), VmError> { Ok(()) diff --git a/virtio-devices/src/block.rs b/virtio-devices/src/block.rs index 44e8dd1d14..077f9d7ac0 100644 --- a/virtio-devices/src/block.rs +++ b/virtio-devices/src/block.rs @@ -15,7 +15,8 @@ use std::ops::Deref; use std::os::unix::io::AsRawFd; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, AtomicUsize, Ordering}; -use std::sync::{Arc, Barrier}; +use std::sync::mpsc::{Receiver, Sender}; +use std::sync::{Arc, Barrier, Mutex, mpsc}; use std::time::{Duration, Instant}; use std::{io, result, thread}; @@ -24,6 +25,10 @@ use block::async_io::{AsyncIo, AsyncIoError}; use block::disk_file::AsyncFullDiskFile; use block::error::BlockError; use block::fcntl::{LockError, LockGranularity, LockGranularityChoice, LockType}; +use block::mirror::{ + BlockMirrorHandle, CopyWorker, CopyWorkerHandle, MIRROR_BLOCK_SIZE, MirrorFailure, MirrorPhase, + MirrorState, MirrorStatus, MirroringAsyncIo, +}; use block::{ ExecuteAsync, ExecuteError, MAX_DISCARD_WRITE_ZEROES_SEG, Request, RequestType, VirtioBlockConfig, build_serial, @@ -63,6 +68,12 @@ const COMPLETION_EVENT: u16 = EPOLL_HELPER_EVENT_LAST + 2; // New 'wake up' event from the rate limiter const RATE_LIMITER_EVENT: u16 = EPOLL_HELPER_EVENT_LAST + 3; +// A `BlockQueueCommand` has been queued for this worker to apply (e.g. swap disk_image). +const BLOCK_COMMAND_EVENT: u16 = EPOLL_HELPER_EVENT_LAST + 4; + +// Maximum duration to wait for a command to be acknowledged by the virtqueue worker. +const MIRROR_COMMAND_ACK_TIMEOUT: Duration = Duration::from_secs(5); + // latency scale, for reduce precision loss in calculate. const LATENCY_SCALE: u64 = 10000; @@ -112,10 +123,145 @@ pub enum Error { ConfigChange(#[source] io::Error), #[error("Disk resize failed")] DiskResize(#[source] BlockError), + #[error("Mirror is currently active")] + MirrorActive, } pub type Result = result::Result; +/// Describes errors reported by synchronous block mirror operations. +#[derive(Error, Debug)] +pub enum MirrorError { + /// Reports an underlying block backend operation failure. + #[error("Block mirror backend operation failed")] + Backend(#[source] BlockError), + /// Indicates that the source and destination have different logical sizes. + #[error( + "Mirror destination logical size ({destination_size} bytes) differs from source logical size ({source_size} bytes)" + )] + DestinationSizeMismatch { + source_size: u64, + destination_size: u64, + }, + /// Reports a failure to acquire the mirror destination advisory lock. + #[error("Failed to acquire {lock_type:?} lock for mirror destination: {path}")] + DestinationLock { + path: PathBuf, + lock_type: LockType, + #[source] + error: LockError, + }, + /// Indicates that a mirror operation was requested before device activation. + #[error("Mirror operation rejected: the device is not active")] + DeviceNotActive, + /// Indicates that a mirror operation was requested while the device is paused. + #[error("Mirror operation rejected: the device is paused")] + DevicePaused, + /// Indicates that a mirror operation was requested without an active mirror. + #[error("No active mirror for the device")] + NotActive, + /// Indicates that completion was requested before the mirror became ready. + #[error("Mirror is not yet ready, cannot complete")] + NotReady, + /// Indicates that the source or destination does not support mirroring. + #[error("Block mirroring is not supported")] + Unsupported(#[source] BlockError), + /// Indicates that cancellation was requested after completion started. + #[error("Mirror completion already in progress")] + CompletionInProgress, + /// Reports a failure to register the new disk notifier. + #[error("Failed to register new disk notifier")] + RegisterNotifier(#[source] EpollHelperError), + /// Reports a failure to deregister the old disk notifier. + #[error("Failed to deregister old disk notifier")] + DeregisterNotifier(#[source] EpollHelperError), + /// Indicates that a queue already has a pending mirror command. + #[error("Mirror command slot is occupied")] + CommandSlotOccupied, + /// Reports a failure to notify a virtqueue worker about a mirror command. + #[error("Failed to notify mirror queue worker")] + NotifyWorker(#[source] io::Error), + /// Reports a missing or late acknowledgement from a virtqueue worker. + #[error("Failed waiting for mirror command acknowledgement")] + Ack(#[source] mpsc::RecvTimeoutError), +} + +/// Represents the result of a synchronous block mirror operation. +pub type MirrorResult = result::Result; + +/// Lifecycle command kind for a virtqueue worker. +#[derive(Debug, Clone, Copy)] +pub enum BlockQueueCommandKind { + /// Replaces the plain source backend with a mirroring backend. + InstallMirror, + /// Drains in-flight guest requests and stops new requests until the next queue command. + DrainAndStallQueue, + /// Replaces the mirroring backend with a destination backend. + CompleteToDestination, + /// Replaces the mirroring backend with a source backend. + CancelToSource, +} + +/// Acknowledgement sent by the corresponding virtqueue worker after handling +/// its command. +pub struct BlockQueueAck { + /// Result of applying the command inside the worker. + pub result: MirrorResult<()>, +} + +/// Command sent from `Block` to a virtqueue worker to change the worker's +/// active block I/O backend. +pub struct BlockQueueCommand { + /// Lifecycle action the worker should apply. + pub kind: BlockQueueCommandKind, + /// New async I/O backend that will replace the worker's current + /// `disk_image` after the old backend has drained. + /// + /// For start this is a `MirroringAsyncIo`. For cancel this is a plain + /// source `AsyncIo`. For completion this is a plain destination `AsyncIo`. + /// A drain command does not swap the backend and passes `None`. + pub async_io: Option>, + + /// Channel used by the worker to report that the command was applied or + /// failed. + pub ack: Sender, +} + +/// One command per virtqueue, each paired with the sender of its queue. +type QueueCommands<'a> = Vec<(&'a BlockQueueCommandSender, BlockQueueCommand)>; + +/// Worker side of the per-virtqueue command channel that receives commands +/// to swap the `disk_image` at runtime. +/// +/// `cmd` and `evt` are shared with the API thread, which puts a +/// [`BlockQueueCommand`] into `cmd` (from [`Block::start_mirror`], +/// `complete_mirror`, or `cancel_mirror`) and writes to `evt` to wake the +/// worker. The worker takes the command and applies it. +pub struct BlockQueueCommandReceiver { + /// Stores this worker's reference to the command slot held by `Block`. + /// + /// Each virtqueue worker has its own slot. `Block` writes a command to each + /// slot and signals the matching `evt` after the write. + pub cmd: Arc>>, + /// Wakes the worker after `cmd` is filled. + /// + /// Fires `BLOCK_COMMAND_EVENT` on the worker's epoll set. + pub evt: EventFd, + /// Command taken from `cmd` and held until `disk_image` reports no + /// in-flight requests. + pending_block_queue_command: Option, +} + +/// API-thread handles used to stage and signal commands for one virtqueue. +struct BlockQueueCommandSender { + /// Single command slot shared with the virtqueue worker. + cmd: Arc>>, + /// Eventfd used to wake the virtqueue worker. + evt: EventFd, + /// Virtqueue size used as the replacement backend's ring depth. + queue_size: u16, +} + // latency will be records as microseconds, average latency // will be save as scaled value. #[derive(Clone)] @@ -193,6 +339,10 @@ struct BlockEpollHandler { host_cpus: Option>, acked_features: u64, disable_sector0_writes: bool, + /// Receives mirror lifecycle commands for this virtqueue worker. + mirror_cmd_receiver: Option, + /// True while a drained queue stops submitting new guest requests. + submissions_stopped: bool, } fn has_feature(features: u64, feature_flag: u64) -> bool { @@ -250,6 +400,17 @@ impl BlockEpollHandler { return Ok(()); } + // Defer new descriptors while a mirror command is pending or guest + // requests are stopped. The queue_evt is kicked at the end of the swap. + if self.submissions_stopped + || self + .mirror_cmd_receiver + .as_ref() + .is_some_and(|receiver| receiver.pending_block_queue_command.is_some()) + { + return Ok(()); + } + let queue = &mut self.queue; let queue_size = queue.size(); let mut batch_requests = Vec::new(); @@ -466,6 +627,94 @@ impl BlockEpollHandler { self.try_signal_used_queue() } + /// Replaces the active [`AsyncIo`] backend and updates its completion-event + /// registration. + fn replace_disk_image( + &mut self, + new_disk_image: Box, + helper: &mut EpollHelper, + ) -> MirrorResult<()> { + let new_disk_fd = new_disk_image.notifier().as_raw_fd(); + let old_disk_fd = self.disk_image.notifier().as_raw_fd(); + + // Register the new backend's completion eventFd. + helper + .add_event(new_disk_fd, COMPLETION_EVENT) + .map_err(MirrorError::RegisterNotifier)?; + + // Deregister the old backend's completion eventFd. + if let Err(error) = + helper.del_event_custom(old_disk_fd, COMPLETION_EVENT, epoll::Events::EPOLLIN) + { + // Rollback the new disk_image registration. + let _ = helper.del_event_custom(new_disk_fd, COMPLETION_EVENT, epoll::Events::EPOLLIN); + return Err(MirrorError::DeregisterNotifier(error)); + } + + // Commit the swap. + self.disk_image = new_disk_image; + + Ok(()) + } + + /// Applies a pending mirror update if one is staged and the current + /// `disk_image` has no in-flight requests. + /// + /// Returns `Ok(())` without changes when either condition is not met. The + /// next completion event triggers another attempt. + fn try_apply_pending_block_queue_command( + &mut self, + helper: &mut EpollHelper, + ) -> result::Result<(), EpollHelperError> { + // If any disk requests are in flight, we can't apply the pending command. + if !self.inflight_requests.is_empty() { + return Ok(()); + } + + let Some(cmd_receiver) = self.mirror_cmd_receiver.as_mut() else { + return Ok(()); + }; + + let Some(command) = cmd_receiver.pending_block_queue_command.take() else { + return Ok(()); + }; + + let BlockQueueCommand { + kind, + async_io, + ack, + } = command; + + let result = if matches!(kind, BlockQueueCommandKind::DrainAndStallQueue) { + // Stop new guest requests until the next command swaps the backend. + self.submissions_stopped = true; + Ok(()) + } else { + let result = self.replace_disk_image( + async_io.expect("mirror swap command without a backend"), + helper, + ); + if result.is_ok() { + self.submissions_stopped = false; + } + result + }; + + let _ = ack.send(BlockQueueAck { result }); + + // While the command was pending, QUEUE_AVAIL_EVENT handling consumed the + // guest's kicks without submitting (see the guard in process_queue_submit). + // The guest won't kick again for descriptors it already queued, so process + // the avail ring now, whether the command succeeded or failed, or those + // requests stall until unrelated guest I/O arrives. + let rate_limit_reached = self.rate_limiter.as_ref().is_some_and(|r| r.is_blocked()); + if !rate_limit_reached { + self.process_queue_submit_and_signal()?; + } + + Ok(()) + } + #[inline] fn find_inflight_request(&mut self, completed_head: u16) -> Result { // This loop neatly handles the fast path where the completions are @@ -682,6 +931,9 @@ impl BlockEpollHandler { if let Some(rate_limiter) = &self.rate_limiter { helper.add_event(rate_limiter.as_raw_fd(), RATE_LIMITER_EVENT)?; } + if let Some(cmd_receiver) = &self.mirror_cmd_receiver { + helper.add_event(cmd_receiver.evt.as_raw_fd(), BLOCK_COMMAND_EVENT)?; + } self.set_queue_thread_affinity(); helper.run(paused, paused_sync, self)?; @@ -692,7 +944,7 @@ impl BlockEpollHandler { impl EpollHelperHandler for BlockEpollHandler { fn handle_event( &mut self, - _helper: &mut EpollHelper, + helper: &mut EpollHelper, event: &epoll::Event, ) -> result::Result<(), EpollHelperError> { let ev_type = event.data as u16; @@ -726,6 +978,7 @@ impl EpollHelperHandler for BlockEpollHandler { if !rate_limit_reached { self.process_queue_submit_and_signal()?; } + self.try_apply_pending_block_queue_command(helper)?; } RATE_LIMITER_EVENT => { if let Some(rate_limiter) = &mut self.rate_limiter { @@ -744,6 +997,25 @@ impl EpollHelperHandler for BlockEpollHandler { ))); } } + BLOCK_COMMAND_EVENT => { + if let Some(cmd_receiver) = self.mirror_cmd_receiver.as_mut() { + cmd_receiver.evt.read().map_err(|error| { + EpollHelperError::HandleEvent(anyhow!( + "Failed to read block command event: {error:?}" + )) + })?; + if let Some(update) = cmd_receiver.cmd.lock().unwrap().take() + && let Some(stale) = + cmd_receiver.pending_block_queue_command.replace(update) + { + warn!( + "Replacing pending block queue command {:?} before it was applied", + stale.kind + ); + } + } + self.try_apply_pending_block_queue_command(helper)?; + } _ => { return Err(EpollHelperError::HandleEvent(anyhow!( "Unexpected event: {ev_type}" @@ -777,6 +1049,13 @@ pub struct Block { device_status: Arc, active_request_count: Arc, draining_active_requests: Arc, + /// Per-virtqueue mirror writer-side handles, populated at + /// activation. + /// + /// `Block::start_mirror` fills each slot with a [`BlockQueueCommand`] and + /// writes the corresponding eventfd. + queue_cmd_senders: Vec, + mirror_handle: Option, } #[derive(Serialize, Deserialize)] @@ -944,6 +1223,8 @@ impl Block { device_status: Arc::new(AtomicU8::new(0)), active_request_count: Arc::new(AtomicUsize::new(0)), draining_active_requests: Arc::new(AtomicBool::new(false)), + queue_cmd_senders: Vec::new(), + mirror_handle: None, }) } @@ -1022,7 +1303,7 @@ impl Block { disk_path: &Path, lock_type: LockType, current_lock: LockType, - ) -> Result<()> { + ) -> result::Result<(), LockError> { let granularity = self.lock_granularity(disk_image, disk_path); debug!( "Attempting to acquire {lock_type:?} lock for disk image: id={},path={},granularity={granularity:?}", @@ -1032,18 +1313,12 @@ impl Block { let fd = disk_image.fd(); granularity .try_acquire_lock(&fd, lock_type, current_lock) - .map_err(|error| { + .inspect_err(|_| { error!( "Cannot acquire {lock_type:?} lock for disk image: id={},path={},granularity={granularity:?}", self.id, disk_path.display() ); - - Error::LockDiskImage { - path: disk_path.to_path_buf(), - error, - lock_type, - } })?; info!( "Acquired {lock_type:?} lock for disk image id={},path={}", @@ -1064,7 +1339,12 @@ impl Block { &self.disk_path, lock_type, self.held_lock, - )?; + ) + .map_err(|error| Error::LockDiskImage { + path: self.disk_path.clone(), + error, + lock_type, + })?; self.held_lock = lock_type; Ok(()) } @@ -1123,6 +1403,10 @@ impl Block { return Err(Error::InvalidSize); } + if self.mirror_handle.is_some() { + return Err(Error::MirrorActive); + } + self.disk_image .resize(new_size) .map_err(Error::DiskResize)?; @@ -1142,6 +1426,462 @@ impl Block { .map_err(Error::ConfigChange) } + /// Starts mirroring the device's disk to `destination`. + /// + /// `destination` is an already-opened disk backend whose file lives in + /// the host filesystem, typically on a different mount than the source + /// (e.g. another host mounted NFS share). + /// `destination_path` is the host path backing it. + /// + /// Each virtqueue worker swaps its `disk_image` to a new + /// [`MirroringAsyncIo`] that fans every mutating request out to both + /// backends. A background [`CopyWorker`] copies existing source bytes + /// to destination until all initial bytes are copied. + /// The [`MirroringAsyncIo`] stays in place until completion, keeping the device's + /// disk and `destination` in sync. + /// + /// The destination is write-locked before queue installation. Its open file + /// description retains that lock until completion transfers the backend to + /// the device or cancellation drops the final destination descriptor. + pub fn start_mirror( + &mut self, + destination: Box, + destination_path: PathBuf, + ) -> MirrorResult<()> { + self.supports_mirroring()?; + destination + .supports_mirroring() + .map_err(MirrorError::Unsupported)?; + + // Mirroring requires activation to have installed at least one live queue worker. + if self.common.epoll_threads.is_none() || self.queue_cmd_senders.is_empty() { + return Err(MirrorError::DeviceNotActive); + } + self.ensure_not_paused_for_mirror()?; + let source_size = self + .disk_image + .logical_size() + .map_err(MirrorError::Backend)?; + let destination_size = destination.logical_size().map_err(MirrorError::Backend)?; + if destination_size != source_size { + return Err(MirrorError::DestinationSizeMismatch { + source_size, + destination_size, + }); + } + + self.try_lock_disk_image( + destination.as_ref(), + &destination_path, + LockType::Write, + LockType::Unlock, + ) + .map_err(|error| MirrorError::DestinationLock { + path: destination_path.clone(), + lock_type: LockType::Write, + error, + })?; + + let (state, copy_worker) = self.initialize_mirror(destination.as_ref(), source_size)?; + + self.mirror_handle = Some(BlockMirrorHandle { + state, + copy_worker, + destination, + destination_path, + }); + Ok(()) + } + + /// Returns an error if this disk image cannot participate in block mirroring. + pub fn supports_mirroring(&self) -> MirrorResult<()> { + self.disk_image + .supports_mirroring() + .map_err(MirrorError::Unsupported) + } + + /// Switch the device's mirroring wrapper to the destination disk. + /// + /// Before the switch-over the queues are drained and stop accepting new + /// requests, which freezes the mirror phase until the swap completes. + /// Each virtqueue worker swaps its [`MirroringAsyncIo`] for a plain + /// [`AsyncIo`] on the destination through the same slot and eventfd + /// mechanism used to install the mirror. After this call the source + /// disk is no longer used by the VM and the operator can detach or + /// remove it. + /// + /// `readonly_destination` is the read-only backend the caller opened for a + /// read-only device, so it does not keep the writable fd of the mirror. + /// With `None` the mirror destination becomes the backend. + /// + /// Returns [`MirrorError::NotActive`] when no mirror is active for the + /// device, and [`MirrorError::NotReady`] when the copy worker has not yet + /// reported the ready phase or the mirror has since failed. Both errors + /// return before any queue command is sent, so the mirror handle is left in + /// place and the caller can poll the state and retry. + /// + /// # Panics + /// + /// Panics if a queue command cannot be sent or acknowledged after the + /// switch-over has started. At that point some queues may already write + /// to the destination only, and there is no revert that keeps + /// acknowledged writes, so aborting is preferred over data loss. + pub fn complete_mirror( + &mut self, + readonly_destination: Option>, + ) -> MirrorResult { + self.ensure_not_paused_for_mirror()?; + + let handle = self.mirror_handle.as_ref().ok_or(MirrorError::NotActive)?; + + if !matches!(handle.state.phase(), MirrorPhase::Ready) { + return Err(MirrorError::NotReady); + } + + // A read-only device keeps the backend the caller reopened read-only, + // so the destination lock moves from the mirror to that backend. + let mut destination_lock = LockType::Write; + if let Some(readonly_destination) = readonly_destination.as_deref() { + destination_lock = LockType::Read; + + let lock_error = |error| MirrorError::DestinationLock { + path: handle.destination_path.clone(), + lock_type: destination_lock, + error, + }; + + // The destination's write lock would reject the read lock of the + // new backend. Downgrade it first to keep the destination locked. + self.try_lock_disk_image( + handle.destination.as_ref(), + &handle.destination_path, + destination_lock, + LockType::Write, + ) + .map_err(lock_error)?; + + self.try_lock_disk_image( + readonly_destination, + &handle.destination_path, + destination_lock, + LockType::Unlock, + ) + .map_err(lock_error)?; + } + let swap_disk: &dyn AsyncFullDiskFile = readonly_destination + .as_deref() + .unwrap_or(handle.destination.as_ref()); + + let (commands, ack_rx) = self.create_mirror_queue_commands( + BlockQueueCommandKind::CompleteToDestination, + |ring_depth| { + swap_disk + .create_async_io(ring_depth) + .map(Some) + .map_err(MirrorError::Backend) + }, + )?; + + // Drain queues to avoid racing failures during completion. + self.drain_and_stall_queues()?; + + if matches!(handle.state.phase(), MirrorPhase::Failed(_)) { + // A destination write failed while draining. Put the workers back + // on the source and let the operator cancel the mirror. + self.revert_queues_to_source()?; + return Err(MirrorError::NotReady); + } + + handle.state.transition_to_phase(MirrorPhase::Completing); + + // Once the first command is sent a queue may write to the destination + // only, so a partial switch-over has no safe revert. We panic rather + // than risk losing acknowledged writes. + Self::send_mirror_queue_commands(commands).expect("mirror queue commands sent"); + self.wait_for_mirror_queue_command_acks(&ack_rx) + .expect("mirror queue command acks received"); + handle.state.transition_to_phase(MirrorPhase::Completed); + + let BlockMirrorHandle { + destination, + destination_path, + copy_worker, + state: _, + } = self.mirror_handle.take().unwrap(); + if let Err(error) = copy_worker.join() { + error!("copy worker thread panicked: {error:?}"); + } + + self.disk_image = readonly_destination.unwrap_or(destination); + self.disk_path = destination_path.clone(); + self.held_lock = destination_lock; + event!("vm", "disk-mirror-completed", "id", &self.id); + Ok(destination_path) + } + + /// Fails with [`MirrorError::DevicePaused`] when the device is paused, since a + /// parked worker cannot apply a staged mirror command. + fn ensure_not_paused_for_mirror(&self) -> MirrorResult<()> { + if self.common.paused.load(Ordering::SeqCst) { + return Err(MirrorError::DevicePaused); + } + Ok(()) + } + + /// Creates the backend for a queue worker. + /// + /// When the block device is reset and reactivated during an active mirror, + /// the worker reattaches to the mirror. + fn create_queue_async_io(&self, ring_depth: u32) -> MirrorResult> { + let mirror = self.mirror_handle.as_ref().filter(|handle| { + matches!( + handle.state.phase(), + MirrorPhase::Running | MirrorPhase::Ready + ) + }); + + if let Some(handle) = mirror { + return MirroringAsyncIo::create( + self.disk_image.as_ref(), + handle.destination.as_ref(), + handle.state.clone(), + ring_depth, + ) + .map(|io| Box::new(io) as Box) + .map_err(MirrorError::Backend); + } + + self.disk_image + .create_async_io(ring_depth) + .map_err(MirrorError::Backend) + } + + /// Installs the mirror backends and starts the copy worker. + /// + /// On success, returns after every virtqueue has acknowledged the new + /// backend. If installation fails after commands are created, the queues + /// are reverted to the source backend. + fn initialize_mirror( + &mut self, + destination: &dyn AsyncFullDiskFile, + source_size: u64, + ) -> MirrorResult<(Arc, CopyWorkerHandle)> { + let state = MirrorState::new(source_size, self.id.clone()); + let (commands, ack_rx) = self.create_mirror_queue_commands( + BlockQueueCommandKind::InstallMirror, + |ring_depth| { + Ok(Some(Box::new( + MirroringAsyncIo::create( + self.disk_image.as_ref(), + destination, + state.clone(), + ring_depth, + ) + .map_err(MirrorError::Backend)?, + ))) + }, + )?; + + Self::send_mirror_queue_commands(commands) + .inspect_err(|_| self.rollback_mirror_installation(&state))?; + + self.wait_for_mirror_queue_command_acks(&ack_rx) + .inspect_err(|_| self.rollback_mirror_installation(&state))?; + + let copy_worker = CopyWorker::spawn( + self.disk_image.as_ref(), + destination, + state.clone(), + MIRROR_BLOCK_SIZE, + ) + .map_err(MirrorError::Backend) + .inspect_err(|_| self.rollback_mirror_installation(&state))?; + + Ok((state, copy_worker)) + } + + /// Marks a mirror installation as failed and reverts to the source. + fn rollback_mirror_installation(&mut self, state: &Arc) { + state.transition_to_phase(MirrorPhase::Failed(Arc::new(MirrorFailure::Installation))); + + if let Err(revert_error) = self.revert_queues_to_source() { + error!( + "failed to revert virtqueues to source after mirror install failure: {revert_error}" + ); + } + } + + /// Creates one command per virtqueue, all sharing one ack channel. + /// + /// Returns each command paired with the sender of its queue, plus + /// the receiving end of the channel. + /// + /// `new_async_io` is called once per queue with the ring depth of + /// that queue and returns the backend the worker swaps to, or `None` + /// for commands that do not swap the backend. + /// + /// The ack sender lives only inside the returned commands. Once + /// every worker has consumed or dropped its command, a lost ack + /// shows up as `Disconnected` on the receiver instead of costing + /// the full ack timeout, and only the workers can ack this op. + fn create_mirror_queue_commands( + &self, + kind: BlockQueueCommandKind, + mut new_async_io: impl FnMut(u32) -> MirrorResult>>, + ) -> MirrorResult<(QueueCommands<'_>, Receiver)> { + let (ack_tx, ack_rx) = mpsc::channel(); + let commands = self + .queue_cmd_senders + .iter() + .map(|sender| { + Ok(( + sender, + BlockQueueCommand { + kind, + async_io: new_async_io(u32::from(sender.queue_size))?, + ack: ack_tx.clone(), + }, + )) + }) + .collect::>()?; + Ok((commands, ack_rx)) + } + + /// Drains every virtqueue and stalls it until the next queue command. + /// + /// Returns once no queue can report a destination failure anymore. + fn drain_and_stall_queues(&self) -> MirrorResult<()> { + let (commands, ack_rx) = self + .create_mirror_queue_commands(BlockQueueCommandKind::DrainAndStallQueue, |_| { + Ok(None) + })?; + Self::send_mirror_queue_commands(commands)?; + self.wait_for_mirror_queue_command_acks(&ack_rx) + } + + /// Sends one staged mirror command to each virtqueue worker. + fn send_mirror_queue_commands(commands: QueueCommands<'_>) -> MirrorResult<()> { + for (sender, command) in commands { + let mut slot = sender.cmd.lock().unwrap(); + + if slot.is_some() { + return Err(MirrorError::CommandSlotOccupied); + } + + *slot = Some(command); + sender.evt.write(1).map_err(MirrorError::NotifyWorker)?; + } + + Ok(()) + } + + /// Waits for all mirror-command acknowledgements. + /// + /// Returns an error when the shared deadline expires or an acknowledgement + /// reports an error. + fn wait_for_mirror_queue_command_acks( + &self, + ack_rx: &Receiver, + ) -> MirrorResult<()> { + let deadline = Instant::now() + MIRROR_COMMAND_ACK_TIMEOUT; + + for _ in 0..self.queue_cmd_senders.len() { + let remaining = deadline.saturating_duration_since(Instant::now()); + let ack = ack_rx.recv_timeout(remaining).map_err(MirrorError::Ack)?; + + ack.result?; + } + + Ok(()) + } + + /// Swaps every virtqueue worker back to a plain `AsyncIo` on the source disk. + fn revert_queues_to_source(&mut self) -> MirrorResult<()> { + // Discard any non-consumed worker command, avoiding to fail with MirrorError::CommandSlotOccupied. + for sender in &self.queue_cmd_senders { + let _ = sender.cmd.lock().unwrap().take(); + } + + let (commands, ack_rx) = self.create_mirror_queue_commands( + BlockQueueCommandKind::CancelToSource, + |ring_depth| { + self.disk_image + .create_async_io(ring_depth) + .map(Some) + .map_err(MirrorError::Backend) + }, + )?; + Self::send_mirror_queue_commands(commands)?; + self.wait_for_mirror_queue_command_acks(&ack_rx) + } + + /// Cancels an active mirror and reverts the device to the source disk. + /// + /// Drains the queues and stops new requests, transitions the mirror to + /// [`MirrorPhase::Cancelling`] to mark that cancellation has started, + /// reverts every virtqueue worker to a plain [`AsyncIo`] on the source, + /// then joins the copy worker and releases the destination. + /// + /// Returns [`MirrorError::NotActive`] when no mirror is active, and + /// [`MirrorError::CompletionInProgress`] once a completion has been + /// attempted, because a queue may already write to the destination only + /// and reverting would lose acknowledged guest writes. + /// + /// If the revert fails the mirror stays in [`MirrorPhase::Cancelling`] + /// with the handle held, so calling this again retries the revert and + /// finishes the cancellation. + /// + /// Blocks until the copy worker finishes its current block and joins, + /// which can stall on a slow or hung destination. + pub fn cancel_mirror(&mut self) -> MirrorResult<()> { + self.ensure_not_paused_for_mirror()?; + let state = self + .mirror_handle + .as_ref() + .ok_or(MirrorError::NotActive)? + .state + .clone(); + + match state.phase() { + MirrorPhase::Running + | MirrorPhase::Ready + | MirrorPhase::Failed(_) + | MirrorPhase::Cancelling => {} + MirrorPhase::Completing | MirrorPhase::Completed => { + return Err(MirrorError::CompletionInProgress); + } + } + + // Drain queues to avoid racing failures during cancellation. + self.drain_and_stall_queues()?; + + state.transition_to_phase(MirrorPhase::Cancelling); + self.revert_queues_to_source()?; + + if let Some(handle) = self.mirror_handle.take() + && let Err(e) = handle.copy_worker.join() + { + error!("copy worker thread panicked: {e:?}"); + } + + event!("vm", "disk-mirror-cancelled", "id", &self.id); + + Ok(()) + } + + /// Returns the destination path of the active mirror. + pub fn mirror_destination_path(&self) -> Option { + let handle = self.mirror_handle.as_ref()?; + Some(handle.destination_path.clone()) + } + + /// Returns a snapshot of the current mirror progress. + pub fn mirror_status(&self) -> Option { + self.mirror_handle + .as_ref() + .map(|handle| handle.state.status()) + } + #[cfg(fuzzing)] pub fn wait_for_epoll_threads(&mut self) { self.common.wait_for_epoll_threads(); @@ -1150,11 +1890,35 @@ impl Block { impl Drop for Block { fn drop(&mut self) { + let mirror_handle = self.mirror_handle.take(); + if let Some(handle) = mirror_handle.as_ref() { + // Cancelling is not possible once completion has started. + if matches!( + handle.state.phase(), + MirrorPhase::Running | MirrorPhase::Ready | MirrorPhase::Failed(_) + ) { + handle.state.transition_to_phase(MirrorPhase::Cancelling); + } + } + if let Some(kill_evt) = self.common.kill_evt.take() { // Ignore the result because there is nothing we can do about it. let _ = kill_evt.write(1); } self.common.wait_for_epoll_threads(); + + let Some(handle) = mirror_handle else { + return; + }; + + if !handle.copy_worker.is_finished() { + warn!("copy worker is still running during block teardown"); + return; + } + + if let Err(error) = handle.copy_worker.join() { + error!("copy worker thread panicked: {error:?}"); + } } } @@ -1223,6 +1987,9 @@ impl VirtioDevice for Block { let mut epoll_threads = Vec::new(); let event_idx = self.common.feature_acked(VIRTIO_RING_F_EVENT_IDX.into()); + // Discard command handles from a previous activation before rebuilding them. + self.queue_cmd_senders.clear(); + for i in 0..queues.len() { let (_, mut queue, queue_evt) = queues.remove(0); queue.set_event_idx(event_idx); @@ -1231,17 +1998,34 @@ impl VirtioDevice for Block { let (kill_evt, pause_evt) = self.common.dup_eventfds(); let queue_idx = i as u16; + let queue_command: Arc>> = Arc::new(Mutex::new(None)); + let queue_command_evt = EventFd::new(libc::EFD_NONBLOCK).map_err(|error| { + error!("failed to create mirror eventfd: {error}"); + ActivateError::BadActivate + })?; + let mirror_handler_evt = queue_command_evt.try_clone().map_err(|error| { + error!("failed to clone mirror eventfd: {error}"); + ActivateError::BadActivate + })?; + let cmd_receiver = BlockQueueCommandReceiver { + cmd: queue_command.clone(), + evt: mirror_handler_evt, + pending_block_queue_command: None, + }; + self.queue_cmd_senders.push(BlockQueueCommandSender { + cmd: queue_command, + evt: queue_command_evt, + queue_size, + }); + let mut handler = BlockEpollHandler { queue_index: queue_idx, queue, mem: mem.clone(), - disk_image: self - .disk_image - .create_async_io(queue_size as u32) - .map_err(|e| { - error!("failed to create new AsyncIo: {e}"); - ActivateError::BadActivate - })?, + disk_image: self.create_queue_async_io(queue_size as u32).map_err(|e| { + error!("failed to create new AsyncIo: {e}"); + ActivateError::BadActivate + })?, disk_nsectors: self.disk_nsectors.clone(), interrupt_cb: interrupt_cb.clone(), serial: self.serial.clone(), @@ -1266,6 +2050,8 @@ impl VirtioDevice for Block { disable_sector0_writes: self.disable_sector0_writes, active_request_count: self.active_request_count.clone(), draining_active_requests: self.draining_active_requests.clone(), + mirror_cmd_receiver: Some(cmd_receiver), + submissions_stopped: false, }; let paused = self.common.paused.clone(); @@ -1291,6 +2077,7 @@ impl VirtioDevice for Block { fn reset(&mut self) { self.common.reset(); + self.queue_cmd_senders.clear(); self.draining_active_requests.store(false, Ordering::SeqCst); self.active_request_count.store(0, Ordering::SeqCst); self.set_writeback_mode(true); @@ -1381,8 +2168,237 @@ impl Snapshottable for Block { } fn snapshot(&mut self) -> std::result::Result { + if self.mirror_handle.is_some() { + return Err(MigratableError::Snapshot(anyhow!( + "Cannot snapshot while mirror is active" + ))); + } + Snapshot::new_from_state(&self.state()) } } impl Transportable for Block {} impl Migratable for Block {} + +#[cfg(test)] +mod unit_tests { + use block::error::BlockErrorKind; + use block::factory::{DiskOpenOptions, open_disk}; + use block::qcow::{BackingFileConfig, ImageType, QcowFile, RawFile}; + use block::qcow_disk::QcowDisk; + use vmm_sys_util::tempfile::TempFile; + + use super::*; + + const TEST_DISK_SIZE: u64 = 1 << 20; + + fn qcow2_disk(with_backing_file: bool) -> (TempFile, Box) { + let image = TempFile::new().unwrap(); + let backing = with_backing_file.then(|| TempFile::new().unwrap()); + + if let Some(backing) = &backing { + backing.as_file().set_len(TEST_DISK_SIZE).unwrap(); + let backing_config = BackingFileConfig { + path: backing.as_path().to_string_lossy().into_owned(), + format: Some(ImageType::Raw), + }; + let raw = RawFile::new(image.as_file().try_clone().unwrap(), false); + QcowFile::new_from_backing(raw, 3, TEST_DISK_SIZE, &backing_config, true).unwrap(); + } else { + let raw = RawFile::new(image.as_file().try_clone().unwrap(), false); + QcowFile::new(raw, 3, TEST_DISK_SIZE, true).unwrap(); + } + + let disk = QcowDisk::new( + image.as_file().try_clone().unwrap(), + false, + with_backing_file, + true, + false, + ) + .unwrap(); + + (image, Box::new(disk)) + } + + fn block_with_disk( + disk_path: &Path, + disk: Box, + read_only: bool, + ) -> Block { + Block::new( + "test".to_string(), + disk, + disk_path.to_path_buf(), + read_only, + false, + 1, + 128, + None, + SeccompAction::Allow, + None, + EventFd::new(libc::EFD_NONBLOCK).unwrap(), + None, + BTreeMap::new(), + true, + false, + LockGranularityChoice::QemuCompatible, + ) + .unwrap() + } + + fn raw_disk(path: &Path, read_only: bool) -> Box { + open_disk(&DiskOpenOptions { + path, + readonly: read_only, + direct: false, + sparse: false, + backing_files: false, + disable_io_uring: true, + disable_aio: true, + }) + .unwrap() + .disk + } + + /// A temporary disk image of [`TEST_DISK_SIZE`] bytes. + fn temp_disk() -> TempFile { + let file = TempFile::new().unwrap(); + file.as_file().set_len(TEST_DISK_SIZE).unwrap(); + file + } + + /// Returns the access mode of a disk's backing file descriptor. + fn fd_access_mode(disk: &dyn AsyncFullDiskFile) -> libc::c_int { + let fd = disk.fd(); + // SAFETY: F_GETFL only reads the file status flags. + unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) & libc::O_ACCMODE } + } + + /// Returns a block device with an active mirror that copied everything and + /// is ready to switch over to `destination`. + fn block_with_ready_mirror(source: &Path, destination: &Path, read_only: bool) -> Block { + let mut block = block_with_disk(source, raw_disk(source, read_only), read_only); + let destination_disk = raw_disk(destination, false); + + let state = MirrorState::new(TEST_DISK_SIZE, "test".to_string()); + state.transition_to_phase(MirrorPhase::Ready); + let copy_worker = CopyWorker::spawn( + block.disk_image.as_ref(), + destination_disk.as_ref(), + state.clone(), + MIRROR_BLOCK_SIZE, + ) + .unwrap(); + block.mirror_handle = Some(BlockMirrorHandle { + state, + copy_worker, + destination: destination_disk, + destination_path: destination.to_path_buf(), + }); + block + } + + #[test] + fn mirror_rejects_qcow2_backing_source() { + let (source_file, source) = qcow2_disk(true); + let (destination_file, destination) = qcow2_disk(false); + let mut block = block_with_disk(source_file.as_path(), source, false); + + let error = block + .start_mirror(destination, destination_file.as_path().to_path_buf()) + .unwrap_err(); + + assert!(matches!( + error, + MirrorError::Unsupported(error) + if error.kind() == BlockErrorKind::UnsupportedFeature + )); + } + + #[test] + fn mirror_rejects_qcow2_backing_destination() { + let (source_file, source) = qcow2_disk(false); + let (destination_file, destination) = qcow2_disk(true); + let mut block = block_with_disk(source_file.as_path(), source, false); + + let error = block + .start_mirror(destination, destination_file.as_path().to_path_buf()) + .unwrap_err(); + + assert!(matches!( + error, + MirrorError::Unsupported(error) + if error.kind() == BlockErrorKind::UnsupportedFeature + )); + } + + /// A completed mirror must not leave the writable destination fd in place + /// for a read-only disk. + #[test] + fn completed_mirror_uses_a_read_only_fd_for_a_read_only_disk() { + let source = temp_disk(); + let destination = temp_disk(); + let final_disk = temp_disk(); + let mut block = block_with_ready_mirror(source.as_path(), destination.as_path(), true); + + block + .complete_mirror(Some(raw_disk(final_disk.as_path(), true))) + .unwrap(); + + assert_eq!( + fd_access_mode(block.disk_image.as_ref()), + libc::O_RDONLY, + "a read-only disk must not keep a writable fd after the mirror" + ); + } + + /// Registers a fake queue worker on the block's active mirror. + /// + /// It acknowledges every command, and reports a destination failure when a + /// command arrives while the mirror is still ready. It blocks until the + /// block stages a command. + fn register_failing_mirror_worker(block: &mut Block) { + let state = block.mirror_handle.as_ref().unwrap().state.clone(); + let cmd: Arc>> = Arc::new(Mutex::new(None)); + let evt = EventFd::new(0).unwrap(); + block.queue_cmd_senders.push(BlockQueueCommandSender { + cmd: Arc::clone(&cmd), + evt: evt.try_clone().unwrap(), + queue_size: 8, + }); + + thread::spawn(move || { + while evt.read().is_ok() { + let Some(command) = cmd.lock().unwrap().take() else { + return; + }; + if matches!(state.phase(), MirrorPhase::Ready) { + state.transition_to_phase(MirrorPhase::Failed(Arc::new( + MirrorFailure::DestinationSubmit(AsyncIoError::WriteVectored( + io::Error::other("destination write failed"), + )), + ))); + } + let _ = command.ack.send(BlockQueueAck { result: Ok(()) }); + } + }); + } + + /// A destination failure reported while the completion is in progress must + /// surface as an error instead of an illegal phase transition. + #[test] + fn completion_survives_a_destination_failure() { + let source = temp_disk(); + let destination = temp_disk(); + let mut block = block_with_ready_mirror(source.as_path(), destination.as_path(), false); + register_failing_mirror_worker(&mut block); + + let result = block.complete_mirror(None); + + assert!( + matches!(result, Err(MirrorError::NotReady)), + "a destination failure during completion must fail the completion: {result:?}" + ); + } +} diff --git a/vmm/src/api/http/http_endpoint.rs b/vmm/src/api/http/http_endpoint.rs index 57aa6c4469..96ce174392 100644 --- a/vmm/src/api/http/http_endpoint.rs +++ b/vmm/src/api/http/http_endpoint.rs @@ -39,6 +39,7 @@ use std::sync::mpsc::Sender; use log::info; use micro_http::{Body, Method, Request, Response, StatusCode, Version}; +use virtio_devices::block::MirrorError; use vmm_sys_util::eventfd::EventFd; #[cfg(all(target_arch = "x86_64", feature = "guest_debug"))] @@ -48,13 +49,15 @@ use crate::api::http::{EndpointHandler, HttpError, error_response}; use crate::api::{ AddDisk, ApiAction, ApiError, ApiRequest, NetConfig, VmAddDevice, VmAddFs, VmAddGenericVhostUser, VmAddNet, VmAddPmem, VmAddUserDevice, VmAddVdpa, VmAddVsock, VmBoot, - VmCancelMigration, VmConfig, VmCounters, VmDelete, VmMigrationProgress, VmNmi, VmPause, + VmCancelMigration, VmConfig, VmCounters, VmDelete, VmDiskMirrorCancel, VmDiskMirrorComplete, + VmDiskMirrorStart, VmDiskMirrorStatus, VmMigrationProgress, VmNmi, VmPause, VmPostMigrationAnnounce, VmPowerButton, VmReboot, VmReceiveMigration, VmReceiveMigrationData, VmRemoveDevice, VmResize, VmResizeDisk, VmResizeZone, VmRestore, VmResume, VmSendMigration, VmShutdown, VmSnapshot, }; use crate::config::RestoreConfig; use crate::cpu::Error as CpuError; +use crate::device_manager::DeviceManagerError; use crate::vm::Error as VmError; /// Helper module for attaching externally opened FDs to config objects. @@ -383,6 +386,11 @@ macro_rules! vm_action_put_handler { macro_rules! vm_action_put_handler_body { ($action:ty) => { + vm_action_put_handler_body!($action, HttpError::ApiError); + }; + // The two-argument form takes an error mapper for actions that + // want their own `ApiError` to HTTP status translation. + ($action:ty, $map_err:expr) => { impl PutHandler for $action { fn handle_request( &'static self, @@ -397,7 +405,7 @@ macro_rules! vm_action_put_handler_body { api_sender, serde_json::from_slice(body.raw())?, ) - .map_err(HttpError::ApiError) + .map_err($map_err) } else { Err(HttpError::BadRequest) } @@ -518,6 +526,69 @@ impl PutHandler for VmSendMigration { impl GetHandler for VmSendMigration {} +vm_action_put_handler_body!(VmDiskMirrorStart, |error| { + if let ApiError::VmDiskMirrorStart(VmError::DeviceManager(device_error)) = &error { + match device_error { + DeviceManagerError::UnknownDeviceId(_) => { + return HttpError::NotFoundWithApiError(error); + } + DeviceManagerError::DiskImageTypeMismatch { .. } + | DeviceManagerError::BlockMirrorAlreadyActive(_) + | DeviceManagerError::BlockMirrorDestinationInUse(_) + | DeviceManagerError::BlockMirrorStart( + MirrorError::DeviceNotActive + | MirrorError::DevicePaused + | MirrorError::DestinationSizeMismatch { .. } + | MirrorError::DestinationLock { .. } + | MirrorError::Unsupported(_), + ) => return HttpError::BadRequestWithApiError(error), + _ => {} + } + } + + HttpError::ApiError(error) +}); + +vm_action_put_handler_body!(VmDiskMirrorStatus, |error| match &error { + ApiError::VmDiskMirrorStatus(VmError::DeviceManager(DeviceManagerError::UnknownDeviceId( + _, + ))) => HttpError::NotFoundWithApiError(error), + ApiError::VmDiskMirrorStatus(VmError::DeviceManager( + DeviceManagerError::BlockMirrorNotActive(_), + )) => HttpError::NotFoundWithApiError(error), + _ => HttpError::ApiError(error), +}); + +vm_action_put_handler_body!(VmDiskMirrorComplete, |error| match &error { + ApiError::VmDiskMirrorComplete(VmError::DeviceManager( + DeviceManagerError::UnknownDeviceId(_), + )) => HttpError::NotFoundWithApiError(error), + ApiError::VmDiskMirrorComplete(VmError::DeviceManager( + DeviceManagerError::BlockMirrorComplete(MirrorError::NotActive), + )) => HttpError::NotFoundWithApiError(error), + ApiError::VmDiskMirrorComplete(VmError::DeviceManager( + DeviceManagerError::BlockMirrorComplete( + MirrorError::DestinationLock { .. } | MirrorError::DevicePaused | MirrorError::NotReady, + ), + )) => HttpError::BadRequestWithApiError(error), + _ => HttpError::ApiError(error), +}); + +vm_action_put_handler_body!(VmDiskMirrorCancel, |error| match &error { + ApiError::VmDiskMirrorCancel(VmError::DeviceManager(DeviceManagerError::UnknownDeviceId( + _, + ))) => HttpError::NotFoundWithApiError(error), + ApiError::VmDiskMirrorCancel(VmError::DeviceManager( + DeviceManagerError::BlockMirrorCancel(MirrorError::NotActive), + )) => HttpError::NotFoundWithApiError(error), + ApiError::VmDiskMirrorCancel(VmError::DeviceManager( + DeviceManagerError::BlockMirrorCancel( + MirrorError::CompletionInProgress | MirrorError::DevicePaused, + ), + )) => HttpError::BadRequestWithApiError(error), + _ => HttpError::ApiError(error), +}); + impl PutHandler for VmResize { fn handle_request( &'static self, diff --git a/vmm/src/api/http/mod.rs b/vmm/src/api/http/mod.rs index 5464ca87ab..8cdd6dcb4d 100644 --- a/vmm/src/api/http/mod.rs +++ b/vmm/src/api/http/mod.rs @@ -30,9 +30,10 @@ use crate::api::VmCoredump; use crate::api::{ AddDisk, ApiError, ApiRequest, VmAddDevice, VmAddFs, VmAddGenericVhostUser, VmAddNet, VmAddPmem, VmAddUserDevice, VmAddVdpa, VmAddVsock, VmBoot, VmCancelMigration, VmCounters, - VmDelete, VmMigrationProgress, VmNmi, VmPause, VmPostMigrationAnnounce, VmPowerButton, - VmReboot, VmReceiveMigration, VmRemoveDevice, VmResize, VmResizeDisk, VmResizeZone, VmRestore, - VmResume, VmSendMigration, VmShutdown, VmSnapshot, + VmDelete, VmDiskMirrorCancel, VmDiskMirrorComplete, VmDiskMirrorStart, VmDiskMirrorStatus, + VmMigrationProgress, VmNmi, VmPause, VmPostMigrationAnnounce, VmPowerButton, VmReboot, + VmReceiveMigration, VmRemoveDevice, VmResize, VmResizeDisk, VmResizeZone, VmRestore, VmResume, + VmSendMigration, VmShutdown, VmSnapshot, }; use crate::landlock::Landlock; use crate::seccomp_filters::{Thread, get_seccomp_filter}; @@ -53,10 +54,21 @@ pub enum HttpError { #[error("Bad Request")] BadRequest, + /// A bad request caused by an internal API error. The source is retained + /// for logging and for the response body while the HTTP status remains 400. + #[error("Bad Request")] + BadRequestWithApiError(#[source] ApiError), + /// Undefined endpoints #[error("Not Found")] NotFound, + /// A not-found response caused by an internal API error. The source is + /// retained for logging and for the response body while the HTTP status + /// remains 404. + #[error("Not Found")] + NotFoundWithApiError(#[source] ApiError), + /// Too many requests #[error("Too Many Requests")] TooManyRequests, @@ -137,10 +149,15 @@ pub trait EndpointHandler { } } Err(e @ HttpError::BadRequest) => error_response(e, StatusCode::BadRequest), + Err(e @ HttpError::BadRequestWithApiError(_)) => { + error_response(e, StatusCode::BadRequest) + } Err(e @ HttpError::SerdeJsonDeserialize(_)) => { error_response(e, StatusCode::BadRequest) } Err(e @ HttpError::TooManyRequests) => error_response(e, StatusCode::TooManyRequests), + Err(e @ HttpError::NotFound) => error_response(e, StatusCode::NotFound), + Err(e @ HttpError::NotFoundWithApiError(_)) => error_response(e, StatusCode::NotFound), Err(e) => error_response(e, StatusCode::InternalServerError), } } @@ -233,6 +250,22 @@ pub static HTTP_ROUTES: LazyLock = LazyLock::new(|| { endpoint!("/vm.delete"), Box::new(VmActionHandler::new(&VmDelete)), ); + r.routes.insert( + endpoint!("/vm.disk-mirror-start"), + Box::new(VmActionHandler::new(&VmDiskMirrorStart)), + ); + r.routes.insert( + endpoint!("/vm.disk-mirror-status"), + Box::new(VmActionHandler::new(&VmDiskMirrorStatus)), + ); + r.routes.insert( + endpoint!("/vm.disk-mirror-complete"), + Box::new(VmActionHandler::new(&VmDiskMirrorComplete)), + ); + r.routes.insert( + endpoint!("/vm.disk-mirror-cancel"), + Box::new(VmActionHandler::new(&VmDiskMirrorCancel)), + ); r.routes.insert(endpoint!("/vm.info"), Box::new(VmInfo {})); r.routes.insert( endpoint!("/vm.pause"), diff --git a/vmm/src/api/mod.rs b/vmm/src/api/mod.rs index 6d1f7c6d7e..99a0c6a20c 100644 --- a/vmm/src/api/mod.rs +++ b/vmm/src/api/mod.rs @@ -40,6 +40,7 @@ use std::str::FromStr; use std::sync::mpsc::{RecvError, SendError, Sender, channel}; use std::time::Duration; +use block::mirror::{MirrorPhase, MirrorStatus}; use log::{info, trace}; use micro_http::Body; use option_parser::{OptionParser, OptionParserError, Toggle}; @@ -149,6 +150,20 @@ pub enum ApiError { #[error("The disk could not be resized")] VmResizeDisk(#[source] VmError), + /// Error starting disk mirror + #[error("Error starting disk mirror")] + VmDiskMirrorStart(#[source] VmError), + + #[error("Error reading disk mirror state")] + VmDiskMirrorStatus(#[source] VmError), + + #[error("Error completing disk mirror")] + VmDiskMirrorComplete(#[source] VmError), + + /// Error cancelling disk mirror + #[error("Error cancelling disk mirror")] + VmDiskMirrorCancel(#[source] VmError), + /// The memory zone could not be resized. #[error("The memory zone could not be resized")] VmResizeZone(#[source] VmError), @@ -235,6 +250,68 @@ pub struct VmInfoResponse { pub device_tree: Option, } +#[derive(Clone, Deserialize, Serialize, Default, Debug)] +pub struct VmDiskMirrorStartData { + pub id: String, + pub destination_path: PathBuf, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct VmDiskMirrorStatusData { + pub id: String, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct VmDiskMirrorCompleteData { + pub id: String, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct VmDiskMirrorCancelData { + pub id: String, +} + +/// Wire form of [`MirrorPhase`], without the failure reason payload, +/// which the response carries separately in `failure`. +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum VmDiskMirrorPhase { + Running, + Ready, + Completing, + Completed, + Cancelling, + Failed, +} + +#[derive(Clone, Debug, Serialize)] +pub struct VmDiskMirrorStatusResponse { + pub phase: VmDiskMirrorPhase, + pub copied_bytes: u64, + pub total_bytes: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub failure: Option, +} + +impl From for VmDiskMirrorStatusResponse { + fn from(status: MirrorStatus) -> Self { + let (phase, failure) = match status.phase { + MirrorPhase::Running => (VmDiskMirrorPhase::Running, None), + MirrorPhase::Ready => (VmDiskMirrorPhase::Ready, None), + MirrorPhase::Cancelling => (VmDiskMirrorPhase::Cancelling, None), + MirrorPhase::Failed(reason) => (VmDiskMirrorPhase::Failed, Some(reason.to_string())), + MirrorPhase::Completing => (VmDiskMirrorPhase::Completing, None), + MirrorPhase::Completed => (VmDiskMirrorPhase::Completed, None), + }; + Self { + phase, + copied_bytes: status.copied_bytes, + total_bytes: status.total_bytes, + failure, + } + } +} + #[derive(Clone, Deserialize, Serialize)] pub struct VmmPingResponse { pub build_version: String, @@ -753,6 +830,18 @@ pub trait RequestHandler { fn vm_resize_disk(&mut self, id: String, desired_size: u64) -> Result<(), VmError>; + fn vm_disk_mirror_start( + &mut self, + id: String, + destination_path: PathBuf, + ) -> Result<(), VmError>; + + fn vm_disk_mirror_status(&mut self, id: String) -> Result>, VmError>; + + fn vm_disk_mirror_complete(&mut self, id: String) -> Result<(), VmError>; + + fn vm_disk_mirror_cancel(&mut self, id: String) -> Result<(), VmError>; + fn vm_add_device(&mut self, device_cfg: DeviceConfig) -> Result>, VmError>; fn vm_add_user_device( @@ -1380,6 +1469,126 @@ impl ApiAction for VmDelete { } } +pub struct VmDiskMirrorStart; + +impl ApiAction for VmDiskMirrorStart { + type RequestBody = VmDiskMirrorStartData; + type ResponseBody = Option; + + fn request(&self, data: Self::RequestBody, response_sender: Sender) -> ApiRequest { + Box::new(move |vmm| { + let response = vmm + .vm_disk_mirror_start(data.id, data.destination_path) + .map_err(ApiError::VmDiskMirrorStart) + .map(|_| ApiResponsePayload::Empty); + + response_sender + .send(response) + .map_err(VmmError::ApiResponseSend)?; + + Ok(false) + }) + } + + fn send( + &self, + api_evt: EventFd, + api_sender: Sender, + data: Self::RequestBody, + ) -> ApiResult { + get_response_body(self, api_evt, api_sender, data) + } +} + +pub struct VmDiskMirrorStatus; +impl ApiAction for VmDiskMirrorStatus { + type RequestBody = VmDiskMirrorStatusData; + type ResponseBody = Option; + + fn request(&self, data: Self::RequestBody, response_sender: Sender) -> ApiRequest { + Box::new(move |vmm| { + let response = vmm + .vm_disk_mirror_status(data.id) + .map_err(ApiError::VmDiskMirrorStatus) + .map(ApiResponsePayload::VmAction); + + response_sender + .send(response) + .map_err(VmmError::ApiResponseSend)?; + Ok(false) + }) + } + + fn send( + &self, + api_evt: EventFd, + api_sender: Sender, + data: Self::RequestBody, + ) -> ApiResult { + get_response_body(self, api_evt, api_sender, data) + } +} + +pub struct VmDiskMirrorComplete; + +impl ApiAction for VmDiskMirrorComplete { + type RequestBody = VmDiskMirrorCompleteData; + type ResponseBody = Option; + + fn request(&self, data: Self::RequestBody, response_sender: Sender) -> ApiRequest { + Box::new(move |vmm| { + let response = vmm + .vm_disk_mirror_complete(data.id) + .map_err(ApiError::VmDiskMirrorComplete) + .map(|_| ApiResponsePayload::Empty); + + response_sender + .send(response) + .map_err(VmmError::ApiResponseSend)?; + Ok(false) + }) + } + + fn send( + &self, + api_evt: EventFd, + api_sender: Sender, + data: Self::RequestBody, + ) -> ApiResult { + get_response_body(self, api_evt, api_sender, data) + } +} + +pub struct VmDiskMirrorCancel; + +impl ApiAction for VmDiskMirrorCancel { + type RequestBody = VmDiskMirrorCancelData; + type ResponseBody = Option; + + fn request(&self, data: Self::RequestBody, response_sender: Sender) -> ApiRequest { + Box::new(move |vmm| { + let response = vmm + .vm_disk_mirror_cancel(data.id) + .map_err(ApiError::VmDiskMirrorCancel) + .map(|_| ApiResponsePayload::Empty); + + response_sender + .send(response) + .map_err(VmmError::ApiResponseSend)?; + Ok(false) + }) + } + + fn send( + &self, + api_evt: EventFd, + api_sender: Sender, + data: Self::RequestBody, + ) -> ApiResult { + get_response_body(self, api_evt, api_sender, data) + } +} + pub struct VmInfo; impl ApiAction for VmInfo { diff --git a/vmm/src/api/openapi/cloud-hypervisor.yaml b/vmm/src/api/openapi/cloud-hypervisor.yaml index ff3156139c..3539bd899c 100644 --- a/vmm/src/api/openapi/cloud-hypervisor.yaml +++ b/vmm/src/api/openapi/cloud-hypervisor.yaml @@ -510,6 +510,88 @@ paths: 500: description: The VM migration could not be sent. + /vm.disk-mirror-start: + put: + summary: Start mirroring a disk to a destination + requestBody: + description: The disk to mirror and the destination path + content: + application/json: + schema: + $ref: "#/components/schemas/VmDiskMirrorStartData" + required: true + responses: + 204: + description: Disk mirroring was successfully started. + 400: + description: A mirror is already active for the disk, or the destination is not usable. + 404: + description: No disk with the given identifier was found. + 500: + description: Disk mirroring could not be started. + + /vm.disk-mirror-status: + put: + summary: Query the status of a disk mirror + requestBody: + description: The identifier of the mirrored disk + content: + application/json: + schema: + $ref: "#/components/schemas/VmDiskMirrorStatusData" + required: true + responses: + 200: + description: The current status of the disk mirror. + content: + application/json: + schema: + $ref: "#/components/schemas/VmDiskMirrorStatusResponse" + 404: + description: No disk with the given identifier was found, or no mirror is active for it. + 500: + description: The disk mirror status could not be retrieved. + + /vm.disk-mirror-complete: + put: + summary: Complete a disk mirror and switch to the destination + requestBody: + description: The identifier of the mirrored disk + content: + application/json: + schema: + $ref: "#/components/schemas/VmDiskMirrorCompleteData" + required: true + responses: + 204: + description: The disk mirror was completed and the device now uses the destination. + 400: + description: The mirror is not ready to complete. + 404: + description: No disk with the given identifier was found, or no mirror is active for it. + 500: + description: The disk mirror could not be completed. + + /vm.disk-mirror-cancel: + put: + summary: Cancel a disk mirror and keep the source disk + requestBody: + description: The identifier of the mirrored disk + content: + application/json: + schema: + $ref: "#/components/schemas/VmDiskMirrorCancelData" + required: true + responses: + 204: + description: The disk mirror was cancelled and the device keeps using the source. + 400: + description: The mirror cannot be cancelled because completion is already in progress. + 404: + description: No disk with the given identifier was found, or no mirror is active for it. + 500: + description: The disk mirror could not be cancelled. + components: schemas: VmmPingResponse: @@ -1571,3 +1653,63 @@ components: type: string access: type: string + + VmDiskMirrorStartData: + required: + - id + - destination_path + type: object + properties: + id: + type: string + destination_path: + type: string + + VmDiskMirrorStatusData: + required: + - id + type: object + properties: + id: + type: string + + VmDiskMirrorStatusResponse: + required: + - phase + - copied_bytes + - total_bytes + type: object + properties: + phase: + type: string + enum: + - running + - ready + - completing + - completed + - cancelling + - failed + copied_bytes: + type: integer + format: int64 + total_bytes: + type: integer + format: int64 + failure: + type: string + + VmDiskMirrorCompleteData: + required: + - id + type: object + properties: + id: + type: string + + VmDiskMirrorCancelData: + required: + - id + type: object + properties: + id: + type: string diff --git a/vmm/src/device_manager.rs b/vmm/src/device_manager.rs index 80fb481819..b6fcde2046 100644 --- a/vmm/src/device_manager.rs +++ b/vmm/src/device_manager.rs @@ -18,7 +18,7 @@ use std::os::unix::io::{AsRawFd, FromRawFd}; #[cfg(not(target_arch = "riscv64"))] use std::path::Path; use std::path::PathBuf; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; #[cfg(not(target_arch = "riscv64"))] use std::time::Instant; @@ -34,8 +34,10 @@ use arch::layout::{APIC_START, IOAPIC_SIZE, IOAPIC_START}; use arch::{DeviceType, MmioDeviceInfo}; use arch::{NumaNodes, layout}; use block::ImageType; +use block::disk_file::AsyncFullDiskFile; use block::error::BlockError; use block::factory::{DiskOpenOptions, open_disk}; +use block::mirror::MirrorStatus; #[cfg(target_arch = "riscv64")] use devices::aia; #[cfg(target_arch = "x86_64")] @@ -86,6 +88,7 @@ use tracer::trace_scoped; #[cfg(feature = "kvm")] use vfio_ioctls::VfioIommufd; use vfio_ioctls::{VfioContainer, VfioDevice, VfioDeviceFd, VfioOps}; +use virtio_devices::block::MirrorError; use virtio_devices::transport::{VirtioPciDevice, VirtioPciDeviceActivator, VirtioTransport}; use virtio_devices::vhost_user::VhostUserConfig; use virtio_devices::{ @@ -690,6 +693,38 @@ pub enum DeviceManagerError { /// required to generate a device path for OVMF. #[error("No BDF assigned to boot entry")] BdfForBootDeviceMissing, + + /// Reports a failure to start block mirroring. + #[error("Failed to start block mirroring")] + BlockMirrorStart(#[source] MirrorError), + + /// Block mirroring is not active for the current device. + #[error("Block mirroring is not active for the current disk with identifier: {0}")] + BlockMirrorNotActive(String), + + /// Mirroring is already active for the current device. + #[error( + "Failed to start block mirroring for the disk with identifier: {0} as mirroring is already active" + )] + BlockMirrorAlreadyActive(String), + + /// The mirror destination path is already backing one of the VM's disks. + #[error("Cannot mirror to '{0}': it is already in use as a disk image by this VM")] + BlockMirrorDestinationInUse(String), + + /// Cannot perform given action, as the device is currently performing a block mirroring operation. + #[error( + "Failed to perform the requested action for the disk with identifier: {0} as it is currently performing a block mirroring operation" + )] + BlockMirrorActive(String), + + /// Reports a failure to complete block mirroring. + #[error("Failed to complete block mirroring")] + BlockMirrorComplete(#[source] MirrorError), + + /// Cancelling the block mirror failed. + #[error("Failed to cancel block mirroring")] + BlockMirrorCancel(#[source] MirrorError), } pub type DeviceManagerResult = result::Result; @@ -4864,16 +4899,20 @@ impl DeviceManager { // Release advisory locks by dropping all references. // Linux automatically releases all locks of that file if the last open FD is closed. { - let maybe_block_device_index = self + if let Some(index) = self .block_devices .iter() - .enumerate() - .find(|(_, dev)| { - let dev = dev.lock().unwrap(); - dev.id() == id - }) - .map(|(i, _)| i); - if let Some(index) = maybe_block_device_index { + .position(|dev| dev.lock().unwrap().id() == id) + { + // Deny removal of active mirroring block device. + if self.block_devices[index] + .lock() + .unwrap() + .mirror_status() + .is_some() + { + return Err(DeviceManagerError::BlockMirrorActive(id.to_string())); + } let _ = self.block_devices.swap_remove(index); } } @@ -5341,16 +5380,22 @@ impl DeviceManager { 0 } + /// Locks and returns the block device with the given id. + /// + /// Returns [`DeviceManagerError::UnknownDeviceId`] when no attached + /// block device matches. + fn find_block_device(&self, device_id: &str) -> DeviceManagerResult> { + self.block_devices + .iter() + .map(|dev| dev.lock().unwrap()) + .find(|disk| disk.id() == device_id) + .ok_or_else(|| DeviceManagerError::UnknownDeviceId(device_id.to_string())) + } + pub fn resize_disk(&mut self, device_id: &str, new_size: u64) -> DeviceManagerResult<()> { - for dev in &self.block_devices { - let mut disk = dev.lock().unwrap(); - if disk.id() == device_id { - return disk - .resize(new_size) - .map_err(DeviceManagerError::DiskResize); - } - } - Err(DeviceManagerError::UnknownDeviceId(device_id.to_string())) + self.find_block_device(device_id)? + .resize(new_size) + .map_err(DeviceManagerError::DiskResize) } pub fn device_tree(&self) -> Arc> { @@ -5423,6 +5468,160 @@ impl DeviceManager { } } + /// Returns a copy of the config of the disk with the given device id. + fn find_disk_config(&self, device_id: &str) -> DeviceManagerResult { + self.config + .lock() + .unwrap() + .disks + .iter() + .flatten() + .find(|disk_config| disk_config.pci_common.id.as_deref() == Some(device_id)) + .cloned() + .ok_or_else(|| DeviceManagerError::UnknownDeviceId(device_id.to_string())) + } + + /// Opens a disk image with `config`'s options and validates its type. + fn open_disk_with_config( + config: &DiskConfig, + path: &Path, + readonly: bool, + ) -> DeviceManagerResult> { + let options = DiskOpenOptions { + path, + readonly, + direct: config.direct, + sparse: config.sparse, + backing_files: config.backing_files, + disable_io_uring: config.disable_io_uring, + disable_aio: config.disable_aio, + }; + + let opened = open_disk(&options).map_err(DeviceManagerError::Disk)?; + if opened.image_type != config.image_type { + return Err(DeviceManagerError::DiskImageTypeMismatch { + specified: config.image_type, + detected: opened.image_type, + }); + } + Ok(opened.disk) + } + + /// Starts mirroring the disk identified by `device_id` to `dest_path`. + /// + /// The destination file must already exist and use the same image + /// format as the source disk. It is handed to the virtio block device, + /// which mirrors later guest writes out to both backends while a + /// background worker copies the existing source contents. + /// + /// Returns an error if no disk with the given identifier is attached + /// to the VM, or the destination cannot be opened. + pub fn mirror_disk(&self, device_id: &str, dest_path: &Path) -> DeviceManagerResult<()> { + let mut disk = self.find_block_device(device_id)?; + + if disk.mirror_status().is_some() { + return Err(DeviceManagerError::BlockMirrorAlreadyActive( + device_id.to_string(), + )); + } + + disk.supports_mirroring() + .map_err(DeviceManagerError::BlockMirrorStart)?; + + // Refuse a destination that already backs one of this VM's disks, comparing canonicalized paths. + let canon = + |path: &Path| std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); + let dest_canon = canon(dest_path); + let dest_in_use = self + .config + .lock() + .unwrap() + .disks + .iter() + .flatten() + .filter_map(|disk_config| disk_config.path.as_deref()) + .any(|source_path| canon(source_path) == dest_canon); + if dest_in_use { + return Err(DeviceManagerError::BlockMirrorDestinationInUse( + dest_path.display().to_string(), + )); + } + + let src = self.find_disk_config(device_id)?; + + let destination = Self::open_disk_with_config(&src, dest_path, false)?; + + disk.start_mirror(destination, dest_path.to_path_buf()) + .map_err(DeviceManagerError::BlockMirrorStart)?; + + Ok(()) + } + + /// Returns the current state of the active mirror for the disk + /// identified by `device_id`. + /// + /// Returns an error if no disk with the given identifier is + /// attached to the VM, or if the disk has no active mirror. + pub fn mirror_disk_status(&self, device_id: &str) -> DeviceManagerResult { + self.find_block_device(device_id)? + .mirror_status() + .ok_or_else(|| DeviceManagerError::BlockMirrorNotActive(device_id.to_string())) + } + + /// Completes the active block mirror for `device_id` and switches to its + /// destination disk. + /// + /// Returns an error if the disk is not attached, no mirror is active, or the + /// mirror is not ready. + pub fn mirror_disk_complete(&self, device_id: &str) -> DeviceManagerResult<()> { + let mut disk = self.find_block_device(device_id)?; + + let src = self.find_disk_config(device_id)?; + + // The mirror destination is opened writable. Reopen it read-only so a + // read-only disk does not keep write access to it. + let readonly_destination = match disk.mirror_destination_path() { + Some(destination_path) if src.readonly => { + Some(Self::open_disk_with_config(&src, &destination_path, true)?) + } + _ => None, + }; + + let new_path = disk + .complete_mirror(readonly_destination) + .map_err(DeviceManagerError::BlockMirrorComplete)?; + + // Repoint the config entry so a rebuild reopens the destination. + if let Some(cfg) = self + .config + .lock() + .unwrap() + .disks + .as_mut() + .and_then(|disks| { + disks + .iter_mut() + .find(|disk_config| disk_config.pci_common.id.as_deref() == Some(device_id)) + }) + { + cfg.path = Some(new_path); + } + + Ok(()) + } + + /// Cancels the active block mirror for the disk identified by + /// `device_id`, reverting all virtqueue workers to the source disk + /// and releasing the destination. + /// + /// Returns an error if the disk is not attached, no mirror is active, + /// mirror completion has started, or reverting a virtqueue worker fails. + pub fn mirror_disk_cancel(&self, device_id: &str) -> DeviceManagerResult<()> { + self.find_block_device(device_id)? + .cancel_mirror() + .map_err(DeviceManagerError::BlockMirrorCancel) + } + /// Helps the environment converge quickly after a live migration by /// prompting devices to advertise the VM from its new host. /// @@ -5471,6 +5670,13 @@ impl DeviceManager { pub fn boot_order_entries(&self) -> Vec { self.boot_order.clone().into_values().collect() } + + /// Returns true if there is an active mirror in any of the block devices, false otherwise. + pub fn any_active_block_mirrors(&self) -> bool { + self.block_devices + .iter() + .any(|dev| dev.lock().unwrap().mirror_status().is_some()) + } } /// Starts a thread that periodically performs the post-migration announcements. diff --git a/vmm/src/lib.rs b/vmm/src/lib.rs index ef7d16b690..3329c7df2f 100644 --- a/vmm/src/lib.rs +++ b/vmm/src/lib.rs @@ -17,7 +17,6 @@ use std::fs::File; use std::io::{Read, Write, stdout}; use std::os::unix::io::{AsRawFd, FromRawFd, RawFd}; use std::panic::AssertUnwindSafe; -#[cfg(feature = "guest_debug")] use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{Receiver, RecvError, SendError, Sender}; @@ -63,8 +62,8 @@ use vmm_sys_util::signal::unblock_signal; use vmm_sys_util::sock_ctrl_msg::ScmSocket; use crate::api::{ - ApiRequest, ApiResponse, RequestHandler, TimeoutStrategy, VmInfoResponse, - VmReceiveMigrationData, VmSendMigrationData, VmmPingResponse, + ApiRequest, ApiResponse, RequestHandler, TimeoutStrategy, VmDiskMirrorStatusResponse, + VmInfoResponse, VmReceiveMigrationData, VmSendMigrationData, VmmPingResponse, }; use crate::config::{MemoryRestoreMode, RestoreConfig, add_to_config}; #[cfg(all(target_arch = "x86_64", feature = "guest_debug"))] @@ -79,7 +78,7 @@ use crate::migration_transport::{ ReceiveAdditionalConnections, ReceiveListener, SendAdditionalConnections, SocketStream, }; use crate::seccomp_filters::{Thread, get_seccomp_filter}; -use crate::vm::{Error as VmError, PostMigrationLifecycleEvent, Vm, VmState}; +use crate::vm::{Error as VmError, PostponedLifecycleEvent, Vm, VmState}; use crate::vm_config::{ DeviceConfig, DiskConfig, FsConfig, GenericVhostUserConfig, MemoryZoneConfig, NetConfig, PmemConfig, UserDeviceConfig, VdpaConfig, VmConfig, VsockConfig, @@ -683,7 +682,7 @@ struct MigrationWorker { check_migration_evt: EventFd, config: VmSendMigrationData, // Shared with main VMM thread - postponed_lifecycle_event: Arc>>, + postponed_lifecycle_event: Arc>>, #[cfg(all(feature = "kvm", target_arch = "x86_64"))] hypervisor: Arc, cancel: Arc, @@ -725,7 +724,7 @@ impl MigrationWorker { vm: Vm, check_migration_evt: EventFd, config: VmSendMigrationData, - postponed_lifecycle_event: Arc>>, + postponed_lifecycle_event: Arc>>, #[cfg(all(feature = "kvm", target_arch = "x86_64"))] hypervisor: Arc< dyn hypervisor::Hypervisor, >, @@ -873,12 +872,24 @@ pub struct Vmm { console_info: Option, no_shutdown: bool, check_migration_evt: EventFd, - postponed_lifecycle_event: Arc>>, - received_postponed_lifecycle_event: Option, + postponed_lifecycle_event: Arc>>, + received_postponed_lifecycle_event: Option, /// Handle to the [`MigrationWorker`] thread. migration_thread_handle: Option, } +/// Replays a postponed guest lifecycle event. +fn replay_lifecycle_event( + event: PostponedLifecycleEvent, + reset_evt: &EventFd, + guest_exit_evt: &EventFd, +) -> io::Result<()> { + match event { + PostponedLifecycleEvent::VmReboot => reset_evt.write(1), + PostponedLifecycleEvent::VmShutdown => guest_exit_evt.write(1), + } +} + /// Just a wrapper for the data that goes into /// [`ReceiveMigrationState::Configured`] struct ReceiveMigrationConfiguredData { @@ -1109,16 +1120,30 @@ impl Vmm { }) } - fn postpone_lifecycle_event_during_migration(&self, event: PostMigrationLifecycleEvent) { - let mut postponed_event = self.postponed_lifecycle_event.lock().unwrap(); - if postponed_event.is_none() { - *postponed_event = Some(event); - info!("Postponed post-migration lifecycle event: {event:?}"); + /// Postpones a lifecycle event while migration or disk mirroring is active. + fn postpone_lifecycle_event( + &mut self, + event: PostponedLifecycleEvent, + ) -> result::Result { + let vm = match &mut self.vm { + MaybeVmOwnership::Migration(_) => None, + MaybeVmOwnership::Vmm(vm) if vm.any_active_block_mirrors() => Some(vm), + _ => return Ok(false), + }; + + { + let mut postponed_event = self.postponed_lifecycle_event.lock().unwrap(); + if postponed_event.is_none() { + *postponed_event = Some(event); + info!("Postponed guest lifecycle event: {event:?}"); + } + } + + if let Some(vm) = vm { + vm.shutdown()?; } - } - fn current_postponed_lifecycle_event(&self) -> Option { - *self.postponed_lifecycle_event.lock().unwrap() + Ok(true) } fn clear_postponed_lifecycle_event(&self) { @@ -1126,6 +1151,18 @@ impl Vmm { *postponed_event = None; } + /// Replays and clears the postponed lifecycle event. + fn replay_postponed_lifecycle_event(&self) { + let mut postponed_event = self.postponed_lifecycle_event.lock().unwrap(); + let Some(event) = *postponed_event else { + return; + }; + match replay_lifecycle_event(event, &self.reset_evt, &self.guest_exit_evt) { + Ok(()) => *postponed_event = None, + Err(e) => error!("Failed replaying postponed lifecycle event: {e}"), + } + } + /// Try to receive a file descriptor from a socket. Returns the slot number and the file descriptor. fn vm_receive_memory_fd( socket: &mut SocketStream, @@ -1307,16 +1344,9 @@ impl Vmm { resume_duration.as_millis() ); } - Some(PostMigrationLifecycleEvent::VmReboot) => { - self.reset_evt - .write(1) - .context("Failed writing reset eventfd after migration") - .map_err(MigratableError::MigrateReceive)?; - } - Some(PostMigrationLifecycleEvent::VmShutdown) => { - self.guest_exit_evt - .write(1) - .context("Failed writing guest exit eventfd after migration") + Some(event) => { + replay_lifecycle_event(event, &self.reset_evt, &self.guest_exit_evt) + .context("Failed replaying lifecycle event after migration") .map_err(MigratableError::MigrateReceive)?; } } @@ -1593,7 +1623,7 @@ impl Vmm { ctx: &mut MemoryMigrationContext, is_converged: impl Fn(&MemoryMigrationContext) -> result::Result, mem_send: &mut SendAdditionalConnections, - postponed_lifecycle_event: &Mutex>, + postponed_lifecycle_event: &Mutex>, return_if_cancelled_cb: &impl Fn(&mut SocketStream) -> result::Result<(), MigratableError>, ) -> result::Result { let total_memory_size_bytes = vm @@ -1804,7 +1834,7 @@ impl Vmm { send_data_migration: &VmSendMigrationData, mem_send: &mut SendAdditionalConnections, ctx: &mut OngoingMigrationContext, - postponed_lifecycle_event: &Mutex>, + postponed_lifecycle_event: &Mutex>, return_if_cancelled_cb: &impl Fn(&mut SocketStream) -> result::Result<(), MigratableError>, ) -> result::Result<(), MigratableError> { let mut mem_ctx = MemoryMigrationContext::new(); @@ -1866,7 +1896,7 @@ impl Vmm { #[cfg(all(feature = "kvm", target_arch = "x86_64"))] hypervisor: &dyn hypervisor::Hypervisor, send_data_migration: &VmSendMigrationData, - postponed_lifecycle_event: &Mutex>, + postponed_lifecycle_event: &Mutex>, cancel: Arc, ) -> result::Result<(), MigratableError> { // State machine that is updated with more context as we progress. @@ -2262,24 +2292,7 @@ impl Vmm { // Give VMM back control. self.vm = MaybeVmOwnership::Vmm(vm); - if let Some(event) = self.current_postponed_lifecycle_event() { - match event { - PostMigrationLifecycleEvent::VmReboot => { - self.reset_evt - .write(1) - .context("Failed replaying reset event after failed migration") - .inspect_err(|write_err| error!("{write_err}")) - .ok(); - } - PostMigrationLifecycleEvent::VmShutdown => { - self.guest_exit_evt - .write(1) - .context("Failed replaying guest exit event after failed migration") - .inspect_err(|write_err| error!("{write_err}")) - .ok(); - } - } - } + self.replay_postponed_lifecycle_event(); }; match migration_res { @@ -2383,11 +2396,10 @@ impl Vmm { info!("VM reset event"); // Consume the event. self.reset_evt.read().map_err(Error::EventFdRead)?; - // Workaround for guest-induced shutdown during a live-migration. - if matches!(self.vm, MaybeVmOwnership::Migration(_)) { - self.postpone_lifecycle_event_during_migration( - PostMigrationLifecycleEvent::VmReboot, - ); + if self + .postpone_lifecycle_event(PostponedLifecycleEvent::VmReboot) + .map_err(Error::VmReboot)? + { continue; } self.vm_reboot().map_err(Error::VmReboot)?; @@ -2395,11 +2407,10 @@ impl Vmm { EpollDispatch::GuestExit => { info!("VM guest exit event"); self.guest_exit_evt.read().map_err(Error::EventFdRead)?; - // Workaround for guest-induced shutdown during a live-migration. - if matches!(self.vm, MaybeVmOwnership::Migration(_)) { - self.postpone_lifecycle_event_during_migration( - PostMigrationLifecycleEvent::VmShutdown, - ); + if self + .postpone_lifecycle_event(PostponedLifecycleEvent::VmShutdown) + .map_err(Error::VmShutdown)? + { continue; } if self.no_shutdown { @@ -2612,6 +2623,10 @@ impl RequestHandler for Vmm { fn vm_snapshot(&mut self, destination_url: &str) -> result::Result<(), VmError> { match self.vm { MaybeVmOwnership::Vmm(ref mut vm) => { + if vm.any_active_block_mirrors() { + return Err(VmError::ActiveBlockMirror); + } + // Drain console_info so that FDs are not reused let _ = self.console_info.take(); vm.snapshot() @@ -2708,6 +2723,11 @@ impl RequestHandler for Vmm { MaybeVmOwnership::Migration(_) => return Err(VmError::VmMigrating), MaybeVmOwnership::None => return Err(VmError::VmNotRunning), }; + + if vm.any_active_block_mirrors() { + return Err(VmError::ActiveBlockMirror); + } + // Drain console_info so that the FDs are not reused let _ = self.console_info.take(); let r = vm.shutdown(); @@ -2729,6 +2749,11 @@ impl RequestHandler for Vmm { MaybeVmOwnership::Migration(_) => return Err(VmError::VmMigrating), MaybeVmOwnership::None => return Err(VmError::VmNotRunning), }; + + if vm.any_active_block_mirrors() { + return Err(VmError::ActiveBlockMirror); + } + let config = vm.get_config(); vm.shutdown()?; self.vm = MaybeVmOwnership::None; @@ -2850,7 +2875,11 @@ impl RequestHandler for Vmm { } match &self.vm { - MaybeVmOwnership::Vmm(_vm) => { + MaybeVmOwnership::Vmm(vm) => { + if vm.any_active_block_mirrors() { + return Err(VmError::ActiveBlockMirror); + } + event!("vm", "deleted"); // If a VM is booted, we first try to shut it down. @@ -3356,8 +3385,14 @@ impl RequestHandler for Vmm { .context("Invalid send migration configuration") .map_err(MigratableError::MigrateSend)?; - match self.vm { - MaybeVmOwnership::Vmm(_) => (), + match &self.vm { + MaybeVmOwnership::Vmm(vm) => { + if vm.any_active_block_mirrors() { + return Err(MigratableError::MigrateSend(anyhow!( + "Cannot start migration with active disk mirrors" + ))); + } + } MaybeVmOwnership::Migration(_) => { return Err(MigratableError::MigrateSend(anyhow!( "There is already an ongoing migration" @@ -3494,6 +3529,67 @@ impl RequestHandler for Vmm { let lock = MIGRATION_PROGRESS_SNAPSHOT.lock().unwrap(); lock.clone() } + + fn vm_disk_mirror_start( + &mut self, + id: String, + destination_path: PathBuf, + ) -> result::Result<(), VmError> { + self.vm_config.as_ref().ok_or(VmError::VmNotCreated)?; + + match self.vm { + MaybeVmOwnership::Vmm(ref mut vm) => vm.mirror_disk(&id, &destination_path), + MaybeVmOwnership::Migration(_) => Err(VmError::VmMigrating), + MaybeVmOwnership::None => Err(VmError::DiskMirrorStart), + } + } + + fn vm_disk_mirror_status(&mut self, id: String) -> result::Result>, VmError> { + self.vm_config.as_ref().ok_or(VmError::VmNotCreated)?; + + match self.vm { + MaybeVmOwnership::Vmm(ref vm) => { + let status = vm.mirror_disk_status(&id)?; + let response: VmDiskMirrorStatusResponse = status.into(); + let json = serde_json::to_vec(&response).map_err(|_| VmError::DiskMirrorStatus)?; + Ok(Some(json)) + } + MaybeVmOwnership::Migration(_) => Err(VmError::VmMigrating), + MaybeVmOwnership::None => Err(VmError::DiskMirrorStatus), + } + } + + fn vm_disk_mirror_complete(&mut self, id: String) -> result::Result<(), VmError> { + self.vm_config.as_ref().ok_or(VmError::VmNotCreated)?; + + let vm = match self.vm { + MaybeVmOwnership::Vmm(ref mut vm) => vm, + MaybeVmOwnership::Migration(_) => return Err(VmError::VmMigrating), + MaybeVmOwnership::None => return Err(VmError::DiskMirrorComplete), + }; + vm.mirror_disk_complete(&id)?; + + if !vm.any_active_block_mirrors() { + self.replay_postponed_lifecycle_event(); + } + Ok(()) + } + + fn vm_disk_mirror_cancel(&mut self, id: String) -> result::Result<(), VmError> { + self.vm_config.as_ref().ok_or(VmError::VmNotCreated)?; + + let vm = match self.vm { + MaybeVmOwnership::Vmm(ref mut vm) => vm, + MaybeVmOwnership::Migration(_) => return Err(VmError::VmMigrating), + MaybeVmOwnership::None => return Err(VmError::DiskMirrorCancel), + }; + vm.mirror_disk_cancel(&id)?; + + if !vm.any_active_block_mirrors() { + self.replay_postponed_lifecycle_event(); + } + Ok(()) + } } const CPU_MANAGER_SNAPSHOT_ID: &str = "cpu-manager"; diff --git a/vmm/src/vm.rs b/vmm/src/vm.rs index c7c17fb41d..9b34111cc4 100644 --- a/vmm/src/vm.rs +++ b/vmm/src/vm.rs @@ -21,6 +21,7 @@ use std::mem::size_of; use std::num::Wrapping; use std::ops::Deref; use std::os::unix::net::UnixStream; +use std::path::Path; use std::sync::{Arc, Mutex}; #[cfg(not(target_arch = "riscv64"))] use std::time::Instant; @@ -36,6 +37,7 @@ use arch::x86_64::MAX_SUPPORTED_CPUS_LEGACY; #[cfg(feature = "tdx")] use arch::x86_64::tdx::TdvfSection; use arch::{EntryPoint, NumaNode, NumaNodes, get_host_cpu_phys_bits}; +use block::mirror::MirrorStatus; use devices::AcpiNotificationFlags; #[cfg(target_arch = "aarch64")] use devices::interrupt_controller; @@ -277,6 +279,21 @@ pub enum Error { #[error("Failed resizing a disk image")] ResizeDisk, + #[error("Failed to start disk mirror")] + DiskMirrorStart, + + #[error("Failed to read disk mirror state")] + DiskMirrorStatus, + + #[error("Failed to complete disk mirror")] + DiskMirrorComplete, + + #[error("Failed to cancel disk mirror")] + DiskMirrorCancel, + + #[error("At least one disk mirror is active")] + ActiveBlockMirror, + #[error("Cannot activate virtio devices")] ActivateVirtioDevices(#[source] DeviceManagerError), @@ -581,11 +598,11 @@ pub struct Vm { stop_on_boot: bool, load_payload_handle: Option>>, vcpu_throttler: ThrottleThreadHandle, - post_migration_lifecycle_event: Option, + post_migration_lifecycle_event: Option, } #[derive(Copy, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub enum PostMigrationLifecycleEvent { +pub enum PostponedLifecycleEvent { VmReboot, VmShutdown, } @@ -1502,14 +1519,14 @@ impl Vm { self.vcpu_throttler.reset(); } - pub fn set_post_migration_lifecycle_event( - &mut self, - event: Option, - ) { + /// Stores a lifecycle event until migration and active disk mirrors permit + /// it to be replayed. + pub fn set_post_migration_lifecycle_event(&mut self, event: Option) { self.post_migration_lifecycle_event = event; } - pub fn post_migration_lifecycle_event(&self) -> Option { + /// Returns the lifecycle event currently waiting to be replayed. + pub fn post_migration_lifecycle_event(&self) -> Option { self.post_migration_lifecycle_event } @@ -2221,6 +2238,11 @@ impl Vm { pub fn shutdown(&mut self) -> Result<()> { let new_state = VmState::Shutdown; + // Shutting down an already shut down VM is a no-op + if self.state == new_state { + return Ok(()); + } + self.state.valid_transition(new_state)?; // Wake up the DeviceManager threads so they will get terminated cleanly @@ -3409,6 +3431,52 @@ impl Vm { .map_err(Error::ErrorNmi); } + pub fn mirror_disk(&self, id: &str, dest_path: &Path) -> Result<()> { + self.device_manager + .lock() + .unwrap() + .mirror_disk(id, dest_path) + .map_err(Error::DeviceManager)?; + + Ok(()) + } + + /// Returns the current mirror status for `id`. + pub fn mirror_disk_status(&self, id: &str) -> Result { + self.device_manager + .lock() + .unwrap() + .mirror_disk_status(id) + .map_err(Error::DeviceManager) + } + + /// Completes the mirror for `id` and switches to its destination. + pub fn mirror_disk_complete(&self, id: &str) -> Result<()> { + self.device_manager + .lock() + .unwrap() + .mirror_disk_complete(id) + .map_err(Error::DeviceManager)?; + Ok(()) + } + + /// Cancels the mirror for `id` and keeps its source backend. + pub fn mirror_disk_cancel(&self, id: &str) -> Result<()> { + self.device_manager + .lock() + .unwrap() + .mirror_disk_cancel(id) + .map_err(Error::DeviceManager) + } + + /// Returns true if there is an active mirror in any of the block devices, false otherwise. + pub fn any_active_block_mirrors(&self) -> bool { + self.device_manager + .lock() + .unwrap() + .any_active_block_mirrors() + } + /// Calls [`DeviceManager::post_migration_announce`]. pub fn post_migration_announce(&self) { self.device_manager @@ -3494,7 +3562,7 @@ impl Pausable for Vm { #[derive(Serialize, Deserialize)] pub struct VmSnapshot { #[serde(default)] - pub post_migration_lifecycle_event: Option, + pub post_migration_lifecycle_event: Option, #[cfg(target_arch = "x86_64")] pub clock: Option, #[cfg(all(feature = "kvm", target_arch = "x86_64"))]