block: Create io/ module for shared I/O infrastructure

Move async_io.rs, fcntl.rs, and request.rs into block/src/io/. These
files are generic I/O infrastructure shared by all formats rather
than format specific code.

The io/ directory name clashes with std::io in lib.rs, so the module
is declared as io_impl via #[path] and the submodules are re-exported
at the crate root to keep existing import paths working.

Signed-off-by: Anatol Belski <anbelski@linux.microsoft.com>

block/src/io/async_io.rs

@@ -0,0 +1,184 @@
+// Copyright © 2021 Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use std::marker::PhantomData;
+use std::os::fd::{AsRawFd, OwnedFd, RawFd};
+mod aio_data_io;
+mod common;
+mod completion;
+mod guest_memory_target;
+mod operation;
+mod owned_io_buffer;
+#[cfg(feature = "io_uring")]
+mod uring_data_io;
+
+use std::io;
+
+pub use aio_data_io::AioDataIo;
+pub use completion::AsyncIoCompletion;
+pub use guest_memory_target::GuestMemoryTarget;
+pub use operation::AsyncIoOperation;
+pub use owned_io_buffer::OwnedIoBuffer;
+use thiserror::Error;
+#[cfg(feature = "io_uring")]
+pub use uring_data_io::UringDataIo;
+use vmm_sys_util::eventfd::EventFd;
+
+use crate::SECTOR_SIZE;
+
+#[derive(Error, Debug)]
+pub enum DiskFileError {
+    /// Failed getting disk file size.
+    #[error("Failed getting disk file size")]
+    Size(#[source] std::io::Error),
+    /// Failed creating a new AsyncIo.
+    #[error("Failed creating a new AsyncIo")]
+    NewAsyncIo(#[source] std::io::Error),
+    /// Unsupported operation.
+    #[error("Unsupported operation")]
+    Unsupported,
+    /// Resize failed
+    #[error("Resize failed")]
+    ResizeError(#[source] std::io::Error),
+    #[error("Failed cloning disk file")]
+    Clone(#[source] std::io::Error),
+}
+
+pub type DiskFileResult<T> = std::result::Result<T, DiskFileError>;
+
+/// A wrapper for [`RawFd`] capturing the lifetime of a corresponding disk file.
+///
+/// This fulfills the same role as [`BorrowedFd`] but is tailored to the limitations
+/// by some disk implementations, which wrap the effective [`File`]
+/// in an `Arc<Mutex<T>>`, making the use of [`BorrowedFd`] impossible.
+///
+/// [`BorrowedFd`]: std::os::fd::BorrowedFd
+#[derive(Copy, Clone, Debug)]
+pub struct BorrowedDiskFd<'fd> {
+    raw_fd: RawFd,
+    _lifetime: PhantomData<&'fd OwnedFd>,
+}
+
+impl BorrowedDiskFd<'_> {
+    pub(crate) fn new(raw_fd: RawFd) -> Self {
+        Self {
+            raw_fd,
+            _lifetime: PhantomData,
+        }
+    }
+}
+
+impl AsRawFd for BorrowedDiskFd<'_> {
+    fn as_raw_fd(&self) -> RawFd {
+        self.raw_fd
+    }
+}
+
+#[derive(Error, Debug)]
+pub enum AsyncIoError {
+    /// Failed vectored reading from file.
+    #[error("Failed vectored reading from file")]
+    ReadVectored(#[source] std::io::Error),
+    /// Failed vectored writing to file.
+    #[error("Failed vectored writing to file")]
+    WriteVectored(#[source] std::io::Error),
+    /// Failed synchronizing file.
+    #[error("Failed synchronizing file")]
+    Fsync(#[source] std::io::Error),
+    /// Failed punching hole.
+    #[error("Failed punching hole")]
+    PunchHole(#[source] std::io::Error),
+    /// Failed writing zeroes.
+    #[error("Failed writing zeroes")]
+    WriteZeroes(#[source] std::io::Error),
+    /// Failed submitting batch requests.
+    #[error("Failed submitting batch requests")]
+    SubmitBatchRequests(#[source] std::io::Error),
+}
+
+pub type AsyncIoResult<T> = std::result::Result<T, AsyncIoError>;
+
+pub trait AsyncIo: Send {
+    fn notifier(&self) -> &EventFd;
+
+    /// Submits one owned data operation.
+    ///
+    /// Takes ownership of `op`.
+    /// Implementations that complete asynchronously must retain it until its
+    /// completion is returned.
+    fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()>;
+
+    /// Submits a read from `offset` into guest memory.
+    fn read_to_memory(
+        &mut self,
+        offset: libc::off_t,
+        target: GuestMemoryTarget,
+        user_data: u64,
+    ) -> AsyncIoResult<()> {
+        self.submit_data_operation(AsyncIoOperation::read_to_memory(offset, target, user_data))
+    }
+
+    /// Submits a write to `offset` from guest memory.
+    fn write_from_memory(
+        &mut self,
+        offset: libc::off_t,
+        target: GuestMemoryTarget,
+        user_data: u64,
+    ) -> AsyncIoResult<()> {
+        self.submit_data_operation(AsyncIoOperation::write_from_memory(
+            offset, target, user_data,
+        ))
+    }
+
+    /// Submits a read from `offset` into an owned host-memory buffer.
+    fn read_to_vec(
+        &mut self,
+        offset: libc::off_t,
+        buffer: OwnedIoBuffer,
+        user_data: u64,
+    ) -> AsyncIoResult<()> {
+        self.submit_data_operation(AsyncIoOperation::read_to_vec(offset, buffer, user_data))
+    }
+
+    /// Submits a write to `offset` from an owned host-memory buffer.
+    fn write_from_vec(
+        &mut self,
+        offset: libc::off_t,
+        buffer: OwnedIoBuffer,
+        user_data: u64,
+    ) -> AsyncIoResult<()> {
+        self.submit_data_operation(AsyncIoOperation::write_from_vec(offset, buffer, user_data))
+    }
+
+    fn fsync(&mut self, user_data: Option<u64>) -> AsyncIoResult<()>;
+    fn punch_hole(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()>;
+    fn write_zeroes(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()>;
+
+    /// Returns the next owned completion, if one is available.
+    ///
+    /// Read completions from owned host-memory buffers return that buffer here.
+    fn next_completed_request(&mut self) -> Option<AsyncIoCompletion>;
+
+    fn batch_requests_enabled(&self) -> bool {
+        false
+    }
+
+    /// Submits a batch of owned data operations.
+    ///
+    /// Backends either accept the whole batch for eventual completion or return
+    /// an error before taking ownership of any operation.
+    fn submit_batch_requests(&mut self, batch_request: Vec<AsyncIoOperation>) -> AsyncIoResult<()> {
+        if batch_request.is_empty() {
+            Ok(())
+        } else {
+            Err(AsyncIoError::SubmitBatchRequests(io::Error::other(
+                "batch requests are not supported by this backend",
+            )))
+        }
+    }
+
+    fn alignment(&self) -> u64 {
+        SECTOR_SIZE
+    }
+}

block/src/io/async_io/aio_data_io.rs

@@ -0,0 +1,235 @@
+// Copyright © 2023 Intel Corporation
+//
+// Copyright © 2023 Crusoe Energy Systems LLC
+//
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use std::collections::{HashMap, VecDeque};
+use std::io;
+use std::os::fd::{AsRawFd, RawFd};
+
+use log::warn;
+use vmm_sys_util::aio;
+use vmm_sys_util::eventfd::EventFd;
+
+use super::common::{duplicate_user_data_error, errno_result, validate_batch};
+use super::{AsyncIoCompletion, AsyncIoOperation};
+
+/// Retained Linux AIO queue for owned async data I/O operations.
+pub struct AioDataIo {
+    // Keep this before `in_flight`: Rust drops fields in declaration order, so
+    // dropping the context destroys kernel AIO state before retained
+    // operations release the buffers referenced by their iovecs.
+    ctx: aio::IoContext,
+    // The `EventFd` for completion signals.
+    eventfd: EventFd,
+    // `in_flight` tracks every user_data value accepted by the kernel. Owned
+    // data operations store `Some(op)` so their iovecs and backing buffers
+    // remain valid until completion; metadata operations store `None`.
+    in_flight: HashMap<u64, Option<AsyncIoOperation>>,
+    // `completions` holds locally produced completions and kernel events that
+    // have been fetched but not yet returned to the caller.
+    completions: VecDeque<AsyncIoCompletion>,
+}
+
+impl AioDataIo {
+    /// Creates a Linux AIO context and its completion eventfd.
+    pub fn new(queue_depth: u32) -> io::Result<Self> {
+        Ok(Self {
+            ctx: aio::IoContext::new(queue_depth)?,
+            eventfd: EventFd::new(libc::EFD_NONBLOCK)?,
+            in_flight: HashMap::new(),
+            completions: VecDeque::new(),
+        })
+    }
+
+    /// Returns the eventfd signaled when completions are available.
+    pub fn notifier(&self) -> &EventFd {
+        &self.eventfd
+    }
+
+    #[allow(unused_unsafe)]
+    fn submit_iocbs(ctx: &aio::IoContext, iocbs: &[&mut aio::IoControlBlock]) -> io::Result<usize> {
+        // SAFETY: vmm_sys_util currently marks IoContext::submit safe, but
+        // io_submit consumes raw pointers asynchronously. Callers must ensure
+        // all iovec and buffer memory referenced by each iocb remains valid
+        // until completion or failed submission.
+        unsafe { ctx.submit(iocbs) }
+    }
+
+    /// Submits one owned read or write operation to the queue.
+    ///
+    /// Submission failures are converted into injected completions so callers
+    /// can observe every accepted request through the normal completion path.
+    pub fn submit_operation(&mut self, fd: RawFd, op: AsyncIoOperation) -> io::Result<()> {
+        validate_batch(
+            |user_data| self.in_flight.contains_key(&user_data),
+            std::slice::from_ref(&op),
+        )?;
+
+        let user_data = op.user_data();
+        let iovecs = op.iovecs();
+        let opcode = if op.is_read() {
+            aio::IOCB_CMD_PREADV
+        } else {
+            aio::IOCB_CMD_PWRITEV
+        };
+        let mut iocb = aio::IoControlBlock {
+            aio_fildes: fd.as_raw_fd() as u32,
+            aio_lio_opcode: opcode as u16,
+            aio_buf: iovecs.as_ptr() as u64,
+            aio_nbytes: iovecs.len() as u64,
+            aio_offset: op.offset(),
+            aio_data: user_data,
+            aio_flags: aio::IOCB_FLAG_RESFD,
+            aio_resfd: self.eventfd.as_raw_fd() as u32,
+            ..Default::default()
+        };
+        self.in_flight.insert(user_data, Some(op));
+
+        let result = match Self::submit_iocbs(&self.ctx, &[&mut iocb]) {
+            Ok(1) => return Ok(()),
+            Ok(_) => -libc::EAGAIN,
+            Err(e) => errno_result(&e),
+        };
+
+        let buffer = self
+            .in_flight
+            .remove(&user_data)
+            .flatten()
+            .and_then(AsyncIoOperation::into_completion_buffer);
+        self.inject_completion(AsyncIoCompletion::new(user_data, result, buffer));
+        Ok(())
+    }
+
+    /// Submits an fsync operation carrying `user_data`.
+    pub fn submit_fsync(&mut self, fd: RawFd, user_data: u64) -> io::Result<()> {
+        if self.in_flight.contains_key(&user_data) {
+            return Err(duplicate_user_data_error(user_data));
+        }
+
+        let mut iocb = aio::IoControlBlock {
+            aio_fildes: fd.as_raw_fd() as u32,
+            aio_lio_opcode: aio::IOCB_CMD_FSYNC as u16,
+            aio_data: user_data,
+            aio_flags: aio::IOCB_FLAG_RESFD,
+            aio_resfd: self.eventfd.as_raw_fd() as u32,
+            ..Default::default()
+        };
+        self.in_flight.insert(user_data, None);
+        let result = match Self::submit_iocbs(&self.ctx, &[&mut iocb]) {
+            Ok(1) => return Ok(()),
+            Ok(_) => -libc::EAGAIN,
+            Err(e) => errno_result(&e),
+        };
+
+        self.in_flight.remove(&user_data);
+        self.inject_completion(AsyncIoCompletion::new(user_data, result, None));
+        Ok(())
+    }
+
+    /// Injects a completion that did not come from a kernel AIO event.
+    ///
+    /// The notifier is signaled so callers can drain it with
+    /// [`Self::next_completion`].
+    pub fn inject_completion(&mut self, completion: AsyncIoCompletion) {
+        self.completions.push_back(completion);
+        self.eventfd.write(1).unwrap();
+    }
+
+    /// Returns the next kernel or injected completion if one is available.
+    ///
+    /// Consuming a kernel completion returns ownership of any buffer retained
+    /// by the corresponding operation.
+    pub fn next_completion(&mut self) -> Option<AsyncIoCompletion> {
+        if self.completions.is_empty() {
+            let mut events = [aio::IoEvent::default(); 32];
+            let rc = match self.ctx.get_events(0, &mut events, None) {
+                Ok(rc) => rc,
+                Err(e) => {
+                    warn!("Linux AIO get_events failed: {e}");
+                    return None;
+                }
+            };
+            for event in &events[..rc] {
+                self.completions.push_back(AsyncIoCompletion::new(
+                    event.data,
+                    event.res as i32,
+                    self.in_flight
+                        .remove(&event.data)
+                        .flatten()
+                        .and_then(AsyncIoOperation::into_completion_buffer),
+                ));
+            }
+        }
+
+        self.completions.pop_front()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io::Write;
+    use std::os::fd::AsRawFd;
+    use std::thread::sleep;
+    use std::time::Duration;
+
+    use vmm_sys_util::tempfile::TempFile;
+
+    use super::AioDataIo;
+    use crate::async_io::{AsyncIoCompletion, AsyncIoOperation, OwnedIoBuffer};
+
+    fn wait_for_completion(data_io: &mut AioDataIo) -> AsyncIoCompletion {
+        for _ in 0..1000 {
+            if let Some(completion) = data_io.next_completion() {
+                return completion;
+            }
+            sleep(Duration::from_millis(1));
+        }
+
+        panic!("timed out waiting for Linux AIO completion");
+    }
+
+    #[test]
+    fn aio_rejects_duplicate_user_data_for_metadata_ops() {
+        let mut file = TempFile::new().unwrap().into_file();
+        file.write_all(&[0xa5; 512]).unwrap();
+        let fd = file.as_raw_fd();
+        let mut data_io = AioDataIo::new(8).unwrap();
+
+        data_io
+            .submit_operation(
+                fd,
+                AsyncIoOperation::read_to_vec(0, OwnedIoBuffer::from_vec(vec![0; 512]), 7),
+            )
+            .unwrap();
+
+        assert_eq!(
+            data_io.submit_fsync(fd, 7).unwrap_err().kind(),
+            std::io::ErrorKind::AlreadyExists
+        );
+
+        let completion = wait_for_completion(&mut data_io);
+        assert_eq!(completion.user_data, 7);
+        assert_eq!(completion.result, 512);
+        assert_eq!(
+            completion.buffer.unwrap().as_slice(),
+            [0xa5; 512].as_slice()
+        );
+    }
+
+    #[test]
+    fn aio_injected_completion_uses_completion_path() {
+        let mut data_io = AioDataIo::new(8).unwrap();
+
+        data_io.inject_completion(AsyncIoCompletion::new(9, -libc::EIO, None));
+
+        let completion = data_io.next_completion().unwrap();
+        assert_eq!(completion.user_data, 9);
+        assert_eq!(completion.result, -libc::EIO);
+        assert!(completion.buffer.is_none());
+        assert!(data_io.next_completion().is_none());
+    }
+}

block/src/io/async_io/common.rs

@@ -0,0 +1,40 @@
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+//! Helpers used by both aio and uring async io.
+
+use std::collections::HashSet;
+use std::io;
+
+use super::AsyncIoOperation;
+
+/// Converts an I/O error into the negative errno form used in completions.
+pub(super) fn errno_result(error: &io::Error) -> i32 {
+    -error.raw_os_error().unwrap_or(libc::EIO)
+}
+
+/// Builds the error returned when a new request reuses in-flight `user_data`.
+pub(super) fn duplicate_user_data_error(user_data: u64) -> io::Error {
+    io::Error::new(
+        io::ErrorKind::AlreadyExists,
+        format!("duplicate async I/O user_data {user_data}"),
+    )
+}
+
+/// Validates that a batch has unique `user_data` not already in flight.
+pub(super) fn validate_batch<F>(mut is_in_flight: F, batch: &[AsyncIoOperation]) -> io::Result<()>
+where
+    F: FnMut(u64) -> bool,
+{
+    let mut seen = HashSet::with_capacity(batch.len());
+
+    for op in batch {
+        let user_data = op.user_data();
+        if is_in_flight(user_data) || !seen.insert(user_data) {
+            return Err(duplicate_user_data_error(user_data));
+        }
+    }
+
+    Ok(())
+}

block/src/io/async_io/completion.rs

@@ -0,0 +1,42 @@
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use super::{AsyncIoOperation, OwnedIoBuffer};
+
+/// Completion returned by an owned async I/O backend.
+///
+/// The completion carries the caller provided `user_data`, the result,
+/// and any owned buffer that can now be dropped.
+#[derive(Debug)]
+pub struct AsyncIoCompletion {
+    /// Caller provided identifier associated with the submitted operation.
+    pub user_data: u64,
+    /// I/O result reported by the backend.
+    ///
+    /// Successful operations report a non-negative byte count. Failed
+    /// operations report a negative errno value.
+    pub result: i32,
+    /// The backing buffer that can now be dropped or re-used.
+    pub buffer: Option<OwnedIoBuffer>,
+}
+
+impl AsyncIoCompletion {
+    /// Creates a completion from its parts.
+    pub fn new(user_data: u64, result: i32, buffer: Option<OwnedIoBuffer>) -> Self {
+        Self {
+            user_data,
+            result,
+            buffer,
+        }
+    }
+
+    /// Creates a completion by consuming the operation that just completed.
+    ///
+    /// This returns ownership of any completion buffer carried by the
+    /// operation.
+    pub fn from_operation(op: AsyncIoOperation, result: i32) -> Self {
+        let user_data = op.user_data();
+        Self::new(user_data, result, op.into_completion_buffer())
+    }
+}

block/src/io/async_io/guest_memory_target.rs

@@ -0,0 +1,242 @@
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use std::cmp::min;
+use std::fmt;
+use std::sync::Arc;
+
+use smallvec::SmallVec;
+use vm_memory::bitmap::Bitmap;
+use vm_memory::{Address, Bytes, GuestAddress, GuestMemory, GuestMemoryError, GuestMemoryMmap};
+
+trait GuestMemoryTargetOwner: Send + Sync {
+    fn iovec_for_range(
+        &self,
+        addr: GuestAddress,
+        len: usize,
+    ) -> Result<libc::iovec, GuestMemoryError>;
+    fn write_guest_slice(&self, buf: &[u8], addr: GuestAddress) -> Result<(), GuestMemoryError>;
+    fn read_guest_slice(&self, buf: &mut [u8], addr: GuestAddress) -> Result<(), GuestMemoryError>;
+}
+
+impl<B> GuestMemoryTargetOwner for GuestMemoryMmap<B>
+where
+    B: Bitmap + Send + Sync + 'static,
+{
+    fn iovec_for_range(
+        &self,
+        addr: GuestAddress,
+        len: usize,
+    ) -> Result<libc::iovec, GuestMemoryError> {
+        let slice = self.get_slice(addr, len)?;
+        let guard = slice.ptr_guard_mut();
+
+        Ok(libc::iovec {
+            iov_base: guard.as_ptr().cast(),
+            iov_len: len,
+        })
+    }
+
+    fn write_guest_slice(&self, buf: &[u8], addr: GuestAddress) -> Result<(), GuestMemoryError> {
+        <Self as Bytes<GuestAddress>>::write_slice(self, buf, addr)
+    }
+
+    fn read_guest_slice(&self, buf: &mut [u8], addr: GuestAddress) -> Result<(), GuestMemoryError> {
+        <Self as Bytes<GuestAddress>>::read_slice(self, buf, addr)
+    }
+}
+
+/// Retains a guest-memory Arc and the validated ranges used for I/O.
+///
+/// Keeping the guest memory arc with the ranges guarantees that the iovecs
+/// remain valid for as long as Self is alive. The iovecs are also shared with
+/// the kernel and must be stable.
+pub struct GuestMemoryTarget {
+    owner: Arc<dyn GuestMemoryTargetOwner>,
+    ranges: SmallVec<[(GuestAddress, usize); 1]>,
+    iovecs: Vec<libc::iovec>,
+}
+
+// SAFETY: GuestMemoryTarget owns an Arc to the guest memory backing and
+// holds its iovecs in a heap allocation, so moving the target leaves the
+// iovec addresses (and the host pointers they reference) stable.
+unsafe impl Send for GuestMemoryTarget {}
+
+impl GuestMemoryTarget {
+    /// Creates a new `GuestMemoryTarget`.
+    ///
+    /// The memory Arc is retained for the life of `Self`, making this
+    /// appropriate for asynchronous I/O operations on the specified ranges.
+    pub fn new<B>(
+        mem: Arc<GuestMemoryMmap<B>>,
+        ranges: &[(GuestAddress, u32)],
+    ) -> Result<Self, GuestMemoryError>
+    where
+        B: Bitmap + Send + Sync + 'static,
+    {
+        let retained_ranges: SmallVec<[(GuestAddress, usize); 1]> = ranges
+            .iter()
+            .copied()
+            .filter(|&(_, len)| len != 0)
+            .map(|(addr, len)| {
+                let len = len as usize;
+                mem.get_slice(addr, len)?;
+                Ok((addr, len))
+            })
+            .collect::<Result<SmallVec<[_; 1]>, GuestMemoryError>>()?;
+
+        // iovec_for_range cannot fail: each range was just validated by
+        // get_slice above and the Arc keeps the mapping alive.
+        let iovecs: Vec<libc::iovec> = retained_ranges
+            .iter()
+            .map(|&(addr, len)| {
+                mem.iovec_for_range(addr, len)
+                    .expect("range validated above and retained by owner Arc")
+            })
+            .collect();
+
+        Ok(Self {
+            owner: mem,
+            ranges: retained_ranges,
+            iovecs,
+        })
+    }
+
+    /// Returns the raw iovecs to be passed to the kernel for asynchronous I/O.
+    #[allow(dead_code)]
+    pub(super) fn iovecs(&self) -> &[libc::iovec] {
+        &self.iovecs
+    }
+
+    /// Returns the total length of the ranges specified at creation.
+    pub fn total_len(&self) -> usize {
+        self.ranges.iter().map(|(_, len)| len).sum()
+    }
+
+    pub(crate) fn write_bytes_at(&self, start: usize, data: &[u8]) -> Result<(), GuestMemoryError> {
+        self.for_each_range(start, data.len(), |addr, offset, len| {
+            self.owner
+                .write_guest_slice(&data[offset..offset + len], addr)
+        })
+    }
+
+    pub(crate) fn read_bytes_at(
+        &self,
+        start: usize,
+        data: &mut [u8],
+    ) -> Result<(), GuestMemoryError> {
+        self.for_each_range(start, data.len(), |addr, offset, len| {
+            self.owner
+                .read_guest_slice(&mut data[offset..offset + len], addr)
+        })
+    }
+
+    pub(crate) fn fill_zeroes_at(&self, start: usize, len: usize) -> Result<(), GuestMemoryError> {
+        let zeroes = [0u8; 4096];
+        self.for_each_range(start, len, |addr, _, mut len| {
+            let mut offset = 0usize;
+            while len > 0 {
+                let count = min(len, zeroes.len());
+                let addr = addr
+                    .checked_add(offset as u64)
+                    .ok_or(GuestMemoryError::InvalidGuestAddress(addr))?;
+                self.owner.write_guest_slice(&zeroes[..count], addr)?;
+                offset += count;
+                len -= count;
+            }
+            Ok(())
+        })
+    }
+
+    fn for_each_range<F>(&self, start: usize, len: usize, mut f: F) -> Result<(), GuestMemoryError>
+    where
+        F: FnMut(GuestAddress, usize, usize) -> Result<(), GuestMemoryError>,
+    {
+        self.validate_range(start, len)?;
+
+        let mut copied = 0usize;
+        let mut pos = 0usize;
+        for &(addr, range_len) in self.ranges.iter() {
+            let range_end = pos + range_len;
+            if range_end <= start || copied == len {
+                pos = range_end;
+                continue;
+            }
+
+            let range_start = start.saturating_sub(pos);
+            let count = min(range_len - range_start, len - copied);
+            let addr = addr
+                .checked_add(range_start as u64)
+                .ok_or(GuestMemoryError::InvalidGuestAddress(addr))?;
+            f(addr, copied, count)?;
+
+            copied += count;
+            if copied == len {
+                break;
+            }
+            pos = range_end;
+        }
+
+        if copied != len {
+            return Err(GuestMemoryError::PartialBuffer {
+                expected: len,
+                completed: copied,
+            });
+        }
+
+        Ok(())
+    }
+
+    fn validate_range(&self, start: usize, len: usize) -> Result<(), GuestMemoryError> {
+        let total_len = self.total_len();
+        if start <= total_len
+            && let Some(end) = start.checked_add(len)
+            && end <= total_len
+        {
+            return Ok(());
+        }
+
+        Err(GuestMemoryError::PartialBuffer {
+            expected: len,
+            completed: total_len.saturating_sub(start).min(len),
+        })
+    }
+}
+
+impl fmt::Debug for GuestMemoryTarget {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut debug = f.debug_struct("GuestMemoryTarget");
+        debug.field("ranges", &self.ranges.len());
+        debug
+            .field("iovecs", &self.iovecs.len())
+            .finish_non_exhaustive()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use vm_memory::{GuestAddress, GuestMemoryMmap};
+
+    use super::GuestMemoryTarget;
+
+    #[test]
+    fn iovecs_survive_move() {
+        // The iovec array must live on the heap so its address stays valid
+        // after the GuestMemoryTarget (and the AsyncIoOperation that owns it)
+        // is moved into an in-flight map. Capture the addresses before the
+        // move and confirm they still match afterwards.
+        let mem = Arc::new(GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 4096)]).unwrap());
+        let target = GuestMemoryTarget::new(mem, &[(GuestAddress(0), 512)]).unwrap();
+        let iovec_ptr_before = target.iovecs().as_ptr() as usize;
+        let iov_base_before = target.iovecs()[0].iov_base as usize;
+
+        let moved = Box::new(target);
+
+        assert_eq!(moved.iovecs().as_ptr() as usize, iovec_ptr_before);
+        assert_eq!(moved.iovecs()[0].iov_base as usize, iov_base_before);
+        assert_eq!(moved.iovecs().len(), 1);
+    }
+}

block/src/io/async_io/operation.rs

@@ -0,0 +1,235 @@
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use std::io;
+use std::ops::Range;
+
+use super::{GuestMemoryTarget, OwnedIoBuffer};
+
+/// A single async IO operation.
+///
+/// Each operation owns or retains the memory target for the duration of the
+/// operation so backends can submit it to the kernel or copy through safe helper
+/// methods.
+#[derive(Debug)]
+pub enum AsyncIoOperation {
+    /// Read from disk into guest memory.
+    ReadToMemory {
+        /// Disk offset for the operation.
+        offset: libc::off_t,
+        /// Guest-memory destination.
+        target: GuestMemoryTarget,
+        /// Caller-provided completion identifier.
+        user_data: u64,
+    },
+    /// Write from guest memory to disk.
+    WriteFromMemory {
+        /// Disk offset for the operation.
+        offset: libc::off_t,
+        /// Guest-memory source.
+        target: GuestMemoryTarget,
+        /// Caller-provided completion identifier.
+        user_data: u64,
+    },
+    /// Read from disk into an owned host-memory buffer.
+    ReadToVec {
+        /// Disk offset for the operation.
+        offset: libc::off_t,
+        /// Owned destination buffer.
+        buffer: OwnedIoBuffer,
+        /// Caller-provided completion identifier.
+        user_data: u64,
+    },
+    /// Write from an owned host-memory buffer to disk.
+    WriteFromVec {
+        /// Disk offset for the operation.
+        offset: libc::off_t,
+        /// Owned source buffer.
+        buffer: OwnedIoBuffer,
+        /// Caller-provided completion identifier.
+        user_data: u64,
+    },
+}
+
+impl AsyncIoOperation {
+    /// Creates an operation that reads from disk into guest memory.
+    pub fn read_to_memory(offset: libc::off_t, target: GuestMemoryTarget, user_data: u64) -> Self {
+        Self::ReadToMemory {
+            offset,
+            target,
+            user_data,
+        }
+    }
+
+    /// Creates an operation that writes from guest memory to disk.
+    pub fn write_from_memory(
+        offset: libc::off_t,
+        target: GuestMemoryTarget,
+        user_data: u64,
+    ) -> Self {
+        Self::WriteFromMemory {
+            offset,
+            target,
+            user_data,
+        }
+    }
+
+    /// Creates an operation that reads from disk into an owned buffer.
+    pub fn read_to_vec(offset: libc::off_t, buffer: OwnedIoBuffer, user_data: u64) -> Self {
+        Self::ReadToVec {
+            offset,
+            buffer,
+            user_data,
+        }
+    }
+
+    /// Creates an operation that writes from an owned buffer to disk.
+    pub fn write_from_vec(offset: libc::off_t, buffer: OwnedIoBuffer, user_data: u64) -> Self {
+        Self::WriteFromVec {
+            offset,
+            buffer,
+            user_data,
+        }
+    }
+
+    /// Returns the value provided at construction.
+    pub fn user_data(&self) -> u64 {
+        match self {
+            Self::ReadToMemory { user_data, .. }
+            | Self::WriteFromMemory { user_data, .. }
+            | Self::ReadToVec { user_data, .. }
+            | Self::WriteFromVec { user_data, .. } => *user_data,
+        }
+    }
+
+    /// Returns the disk offset for this operation.
+    pub fn offset(&self) -> libc::off_t {
+        match self {
+            Self::ReadToMemory { offset, .. }
+            | Self::WriteFromMemory { offset, .. }
+            | Self::ReadToVec { offset, .. }
+            | Self::WriteFromVec { offset, .. } => *offset,
+        }
+    }
+
+    /// Updates the disk offset for this operation.
+    pub fn set_offset(&mut self, new_offset: libc::off_t) {
+        match self {
+            Self::ReadToMemory { offset, .. }
+            | Self::WriteFromMemory { offset, .. }
+            | Self::ReadToVec { offset, .. }
+            | Self::WriteFromVec { offset, .. } => *offset = new_offset,
+        }
+    }
+
+    /// Returns whether this operation reads from disk.
+    pub fn is_read(&self) -> bool {
+        matches!(self, Self::ReadToMemory { .. } | Self::ReadToVec { .. })
+    }
+
+    /// Returns the retained iovec array for kernel submission.
+    ///
+    /// The iovec pointers are valid while this operation is alive.
+    pub fn iovecs(&self) -> &[libc::iovec] {
+        match self {
+            Self::ReadToMemory { target, .. } | Self::WriteFromMemory { target, .. } => {
+                target.iovecs()
+            }
+            Self::ReadToVec { buffer, .. } | Self::WriteFromVec { buffer, .. } => buffer.iovecs(),
+        }
+    }
+
+    /// Returns the total number of bytes described by the operation iovecs.
+    pub fn total_len(&self) -> usize {
+        match self {
+            Self::ReadToMemory { target, .. } | Self::WriteFromMemory { target, .. } => {
+                target.total_len()
+            }
+            Self::ReadToVec { buffer, .. } | Self::WriteFromVec { buffer, .. } => {
+                buffer.total_len()
+            }
+        }
+    }
+
+    fn checked_range(total_len: usize, start: usize, len: usize) -> io::Result<Range<usize>> {
+        if start <= total_len
+            && let Some(end) = start.checked_add(len)
+            && end <= total_len
+        {
+            return Ok(start..end);
+        }
+
+        Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "async I/O buffer range out of bounds",
+        ))
+    }
+
+    /// Copies bytes into a read operation at `start`.
+    pub(crate) fn write_bytes_at(&mut self, start: usize, data: &[u8]) -> io::Result<()> {
+        match self {
+            Self::ReadToMemory { target, .. } => {
+                target.write_bytes_at(start, data).map_err(io::Error::other)
+            }
+            Self::ReadToVec { buffer, .. } => {
+                let range = Self::checked_range(buffer.total_len(), start, data.len())?;
+                buffer.as_mut_slice()[range].copy_from_slice(data);
+                Ok(())
+            }
+            Self::WriteFromMemory { .. } | Self::WriteFromVec { .. } => Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "cannot write into a write operation",
+            )),
+        }
+    }
+
+    /// Fills a read operation with zeroes at `start`.
+    pub(crate) fn fill_zeroes_at(&mut self, start: usize, len: usize) -> io::Result<()> {
+        match self {
+            Self::ReadToMemory { target, .. } => {
+                target.fill_zeroes_at(start, len).map_err(io::Error::other)
+            }
+            Self::ReadToVec { buffer, .. } => {
+                let range = Self::checked_range(buffer.total_len(), start, len)?;
+                buffer.as_mut_slice()[range].fill(0);
+                Ok(())
+            }
+            Self::WriteFromMemory { .. } | Self::WriteFromVec { .. } => Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "cannot write into a write operation",
+            )),
+        }
+    }
+
+    /// Copies bytes out of a write operation at `start`.
+    pub(crate) fn read_bytes_at(&self, start: usize, data: &mut [u8]) -> io::Result<()> {
+        match self {
+            Self::WriteFromMemory { target, .. } => {
+                target.read_bytes_at(start, data).map_err(io::Error::other)
+            }
+            Self::WriteFromVec { buffer, .. } => {
+                let range = Self::checked_range(buffer.total_len(), start, data.len())?;
+                data.copy_from_slice(&buffer.as_slice()[range]);
+                Ok(())
+            }
+            Self::ReadToMemory { .. } | Self::ReadToVec { .. } => Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "cannot read from a read operation",
+            )),
+        }
+    }
+
+    /// Consumes the operation and returns the buffer needed by its completion.
+    ///
+    /// Only `ReadToVec` operations return a buffer because callers need the
+    /// data they read.
+    pub fn into_completion_buffer(self) -> Option<OwnedIoBuffer> {
+        match self {
+            Self::ReadToVec { buffer, .. } => Some(buffer),
+            Self::ReadToMemory { .. }
+            | Self::WriteFromMemory { .. }
+            | Self::WriteFromVec { .. } => None,
+        }
+    }
+}

block/src/io/async_io/owned_io_buffer.rs

@@ -0,0 +1,174 @@
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use std::alloc::{Layout, alloc_zeroed, dealloc};
+use std::{fmt, io};
+
+// Storage owned by an async I/O request for host-memory buffers.
+//
+// `Vec` is used when ordinary vector storage is sufficient. `Aligned` is used
+// when the backend requires an alignment that a normal `Vec` cannot
+// guarantee.
+enum OwnedIoBufferStorage {
+    // Buffer backed by a standard `Vec<u8>`.
+    Vec(Vec<u8>),
+    // Buffer backed by an explicitly aligned allocation.
+    Aligned {
+        // Pointer returned by `alloc_zeroed` for `layout`.
+        ptr: *mut u8,
+        // Layout used to allocate and deallocate `ptr`.
+        layout: Layout,
+        // Logical buffer length exposed to I/O.
+        len: usize,
+    },
+}
+
+// SAFETY: OwnedIoBufferStorage owns its allocation exclusively. Moving it to
+// another thread transfers that ownership.
+unsafe impl Send for OwnedIoBufferStorage {}
+
+impl OwnedIoBufferStorage {
+    fn new(len: usize, alignment: usize) -> io::Result<Self> {
+        if alignment <= 1 {
+            return Ok(Self::Vec(vec![0; len]));
+        }
+
+        let alloc_len = len.max(1).next_multiple_of(alignment);
+        let layout = Layout::from_size_align(alloc_len, alignment)
+            .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
+        // SAFETY: layout has non-zero size because alloc_len is at least 1.
+        let ptr = unsafe { alloc_zeroed(layout) };
+        if ptr.is_null() {
+            return Err(io::Error::new(
+                io::ErrorKind::OutOfMemory,
+                "alloc_zeroed returned null",
+            ));
+        }
+
+        Ok(Self::Aligned { ptr, layout, len })
+    }
+
+    fn as_mut_ptr(&mut self) -> *mut u8 {
+        match self {
+            Self::Vec(buf) => buf.as_mut_ptr(),
+            Self::Aligned { ptr, .. } => *ptr,
+        }
+    }
+
+    fn as_slice(&self) -> &[u8] {
+        match self {
+            Self::Vec(buf) => buf.as_slice(),
+            Self::Aligned { ptr, len, .. } => {
+                // SAFETY: alloc_zeroed initialized `len` bytes at `ptr` and the
+                // allocation is owned by Self.
+                unsafe { std::slice::from_raw_parts(*ptr, *len) }
+            }
+        }
+    }
+
+    fn as_mut_slice(&mut self) -> &mut [u8] {
+        match self {
+            Self::Vec(buf) => buf.as_mut_slice(),
+            Self::Aligned { ptr, len, .. } => {
+                // SAFETY: alloc_zeroed initialized `len` bytes at `ptr`,
+                // &mut self ensures unique access, and the allocation is
+                // owned by Self.
+                unsafe { std::slice::from_raw_parts_mut(*ptr, *len) }
+            }
+        }
+    }
+}
+
+impl Drop for OwnedIoBufferStorage {
+    fn drop(&mut self) {
+        if let Self::Aligned { ptr, layout, .. } = self {
+            // SAFETY: ptr was allocated by alloc_zeroed with this layout and is
+            // solely owned by Self.
+            unsafe { dealloc(*ptr, *layout) };
+        }
+    }
+}
+
+impl fmt::Debug for OwnedIoBufferStorage {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Vec(buf) => f.debug_tuple("Vec").field(&buf.len()).finish(),
+            Self::Aligned { len, layout, .. } => f
+                .debug_struct("Aligned")
+                .field("len", len)
+                .field("layout", layout)
+                .finish(),
+        }
+    }
+}
+
+/// Owns host-memory buffer storage and the iovec array that points into it.
+///
+/// The retained iovec is valid for as long as this value is alive.
+/// When used for Async I/O this struct must remain valid for the duration of the op.
+#[derive(Debug)]
+pub struct OwnedIoBuffer {
+    storage: OwnedIoBufferStorage,
+    iovecs: Vec<libc::iovec>,
+}
+
+// SAFETY: OwnedIoBuffer owns the storage referenced by its single iovec, and moving the buffer
+// keeps the allocation address stable.
+unsafe impl Send for OwnedIoBuffer {}
+
+impl OwnedIoBuffer {
+    /// Creates a zeroed buffer with the requested logical length and alignment.
+    ///
+    /// An alignment of 0 or 1 uses ordinary `Vec` storage. Larger alignments use an explicitly
+    /// aligned allocation whose allocated size may be rounded up while the exposed slice length
+    /// remains `len`.
+    pub fn new(len: usize, alignment: usize) -> io::Result<Self> {
+        let mut storage = OwnedIoBufferStorage::new(len, alignment)?;
+        let iovec = libc::iovec {
+            iov_base: storage.as_mut_ptr().cast(),
+            iov_len: len,
+        };
+        Ok(Self {
+            storage,
+            iovecs: vec![iovec],
+        })
+    }
+
+    /// Creates an owned I/O buffer from an existing `Vec<u8>`.
+    ///
+    /// The generated iovec covers the full vector length and remains valid
+    /// until the OwnedIoBuffer is dropped.
+    pub fn from_vec(mut buf: Vec<u8>) -> Self {
+        let iovec = libc::iovec {
+            iov_base: buf.as_mut_ptr().cast(),
+            iov_len: buf.len(),
+        };
+        Self {
+            storage: OwnedIoBufferStorage::Vec(buf),
+            iovecs: vec![iovec],
+        }
+    }
+
+    /// Returns the logical buffer contents.
+    pub fn as_slice(&self) -> &[u8] {
+        self.storage.as_slice()
+    }
+
+    /// Returns the logical buffer contents mutably.
+    pub fn as_mut_slice(&mut self) -> &mut [u8] {
+        self.storage.as_mut_slice()
+    }
+
+    /// Returns the retained iovec array for kernel submission.
+    ///
+    /// The iovec pointers remain valid while this buffer is alive.
+    pub fn iovecs(&self) -> &[libc::iovec] {
+        &self.iovecs
+    }
+
+    /// Returns the total number of bytes described by the retained iovecs.
+    pub fn total_len(&self) -> usize {
+        self.iovecs.iter().map(|iov| iov.iov_len).sum()
+    }
+}

block/src/io/async_io/uring_data_io.rs

@@ -0,0 +1,357 @@
+// Copyright © 2021 Intel Corporation
+//
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use std::collections::{HashMap, VecDeque};
+use std::io;
+use std::os::fd::{AsRawFd, RawFd};
+
+use io_uring::{IoUring, opcode, squeue, types};
+use log::warn;
+use vmm_sys_util::eventfd::EventFd;
+
+use super::common::{duplicate_user_data_error, validate_batch};
+use super::{AsyncIoCompletion, AsyncIoOperation};
+
+/// `io_uring` wrapper for async I/O.
+///
+/// Holds the `IoUring` and its `EventFd`. Tracks ops that are pending.
+pub struct UringDataIo {
+    // Keep `io_uring` before `in_flight`. Rust drops fields in declaration
+    // order, so dropping the ring tears down kernel access before retained
+    // operations release the buffers referenced by their iovecs.
+    io_uring: IoUring,
+    // The `EventFd` for completion signals.
+    eventfd: EventFd,
+    // `in_flight` tracks every user_data value accepted by the kernel. Owned
+    // data operations store `Some(op)` so their iovecs and backing buffers
+    // remain valid until completion; metadata operations store `None`.
+    in_flight: HashMap<u64, Option<AsyncIoOperation>>,
+    // `injected` holds locally produced completions so synchronous failures
+    // and short-circuited requests use the same drain path as kernel CQEs.
+    injected: VecDeque<AsyncIoCompletion>,
+    // `needs_submit_retry` is set when SQEs have been published to the ring,
+    // but the submit syscall failed before confirming kernel ownership.
+    needs_submit_retry: bool,
+}
+
+impl UringDataIo {
+    /// Creates an io_uring queue and registers its completion eventfd.
+    pub fn new(ring_depth: u32) -> io::Result<Self> {
+        let io_uring = IoUring::new(ring_depth)?;
+        let eventfd = EventFd::new(libc::EFD_NONBLOCK)?;
+        io_uring.submitter().register_eventfd(eventfd.as_raw_fd())?;
+
+        Ok(Self {
+            io_uring,
+            eventfd,
+            in_flight: HashMap::new(),
+            injected: VecDeque::new(),
+            needs_submit_retry: false,
+        })
+    }
+
+    /// Returns the eventfd signaled when completions are available.
+    pub fn notifier(&self) -> &EventFd {
+        &self.eventfd
+    }
+
+    /// Submits one owned read or write operation to the queue.
+    pub fn submit_operation(&mut self, fd: RawFd, op: AsyncIoOperation) -> io::Result<()> {
+        self.submit_batch(fd, vec![op])
+    }
+
+    fn reserve_user_data(&mut self, user_data: u64) -> io::Result<()> {
+        if self.in_flight.contains_key(&user_data) {
+            return Err(duplicate_user_data_error(user_data));
+        }
+        self.in_flight.insert(user_data, None);
+
+        Ok(())
+    }
+
+    fn submit_kernel_entry(&mut self, user_data: u64, entry: &squeue::Entry) -> io::Result<()> {
+        self.reserve_user_data(user_data)?;
+
+        let (submitter, mut sq, _) = self.io_uring.split();
+        // SAFETY: the entry has no caller-owned buffer. `user_data` is retained
+        // in `in_flight` until the CQE is consumed.
+        if let Err(e) = unsafe { sq.push(entry) } {
+            self.in_flight.remove(&user_data);
+            return Err(io::Error::other(format!("Submission queue is full: {e:?}")));
+        }
+        sq.sync();
+
+        match submitter.submit() {
+            Ok(_) => self.needs_submit_retry = false,
+            Err(e) => {
+                self.needs_submit_retry = true;
+                warn!("io_uring submit failed after SQE was published: {e}");
+                self.eventfd.write(1).unwrap();
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Submits a batch of owned read and write operations.
+    ///
+    /// If the io_uring submission queue cannot accept the whole batch, each
+    /// operation is completed locally with `-EAGAIN` so callers can observe
+    /// every request through the normal completion path.
+    pub fn submit_batch(&mut self, fd: RawFd, batch: Vec<AsyncIoOperation>) -> io::Result<()> {
+        if batch.is_empty() {
+            return Ok(());
+        }
+
+        validate_batch(|user_data| self.in_flight.contains_key(&user_data), &batch)?;
+
+        let (submitter, mut sq, _) = self.io_uring.split();
+        let available = sq.capacity() - sq.len();
+        if batch.len() > available {
+            // Not enough space for the batch.
+            // Drop sq, which will re-publish an unmodified tail pointer
+            drop(sq);
+            for op in batch {
+                self.injected
+                    .push_back(AsyncIoCompletion::from_operation(op, -libc::EAGAIN));
+            }
+            self.eventfd.write(1).unwrap();
+            return Ok(());
+        }
+
+        let mut signal_completion = false;
+        let mut batch = batch.into_iter();
+        while let Some(op) = batch.next() {
+            let user_data = op.user_data();
+            let entry = Self::build_entry(fd, &op);
+            self.in_flight.insert(user_data, Some(op));
+
+            // SAFETY: the SQ capacity was just checked. Every iovec's pointer is retained in
+            // self.in_flight before the SQ tail is advanced by sync or drop. in_flight only
+            // drops the memory after a completion.
+            if let Err(e) = unsafe { sq.push(&entry) } {
+                Self::handle_push_failure(
+                    &mut self.in_flight,
+                    &mut self.injected,
+                    user_data,
+                    batch.by_ref(),
+                    &e,
+                );
+                signal_completion = true;
+                break;
+            }
+        }
+
+        sq.sync();
+        match submitter.submit() {
+            Ok(_) => self.needs_submit_retry = false,
+            Err(e) => {
+                self.needs_submit_retry = true;
+                warn!("io_uring submit failed after SQEs were published: {e}");
+                signal_completion = true;
+            }
+        }
+        if signal_completion {
+            self.eventfd.write(1).unwrap();
+        }
+
+        Ok(())
+    }
+
+    #[cold]
+    fn handle_push_failure(
+        in_flight: &mut HashMap<u64, Option<AsyncIoOperation>>,
+        injected: &mut VecDeque<AsyncIoCompletion>,
+        user_data: u64,
+        remaining: impl Iterator<Item = AsyncIoOperation>,
+        error: &squeue::PushError,
+    ) {
+        // Since capacity was just checked, this should only happen if the ring
+        // state changed unexpectedly. Keep all affected operations memory safe
+        // by returning local completions through the normal path.
+        let op = in_flight
+            .remove(&user_data)
+            .flatten()
+            .expect("pending operation missing after failed push");
+        injected.push_back(AsyncIoCompletion::from_operation(op, -libc::EAGAIN));
+        for op in remaining {
+            injected.push_back(AsyncIoCompletion::from_operation(op, -libc::EAGAIN));
+        }
+        warn!("io_uring submission queue became full after capacity check: {error:?}");
+    }
+
+    fn build_entry(fd: RawFd, op: &AsyncIoOperation) -> squeue::Entry {
+        let iovecs = op.iovecs();
+        let fd = types::Fd(fd);
+        if op.is_read() {
+            opcode::Readv::new(fd, iovecs.as_ptr(), iovecs.len() as u32)
+                .offset(op.offset() as u64)
+                .build()
+                .user_data(op.user_data())
+        } else {
+            opcode::Writev::new(fd, iovecs.as_ptr(), iovecs.len() as u32)
+                .offset(op.offset() as u64)
+                .build()
+                .user_data(op.user_data())
+        }
+    }
+
+    /// Submits an io_uring NOP carrying `user_data`.
+    pub fn submit_nop(&mut self, user_data: u64) -> io::Result<()> {
+        self.submit_kernel_entry(user_data, &opcode::Nop::new().build().user_data(user_data))
+    }
+
+    /// Submits an fsync operation carrying `user_data`.
+    pub fn submit_fsync(&mut self, fd: RawFd, user_data: u64) -> io::Result<()> {
+        self.submit_kernel_entry(
+            user_data,
+            &opcode::Fsync::new(types::Fd(fd))
+                .build()
+                .user_data(user_data),
+        )
+    }
+
+    /// Submits a fallocate operation carrying `user_data`.
+    pub fn submit_fallocate(
+        &mut self,
+        fd: RawFd,
+        offset: u64,
+        length: u64,
+        mode: i32,
+        user_data: u64,
+    ) -> io::Result<()> {
+        self.submit_kernel_entry(
+            user_data,
+            &opcode::Fallocate::new(types::Fd(fd), length)
+                .offset(offset)
+                .mode(mode)
+                .build()
+                .user_data(user_data),
+        )
+    }
+
+    /// Injects a completion that did not come from a kernel CQE.
+    ///
+    /// The notifier is signaled so callers can drain it with
+    /// [`Self::next_completion`].
+    pub fn inject_completion(&mut self, completion: AsyncIoCompletion) {
+        self.injected.push_back(completion);
+        self.eventfd.write(1).unwrap();
+    }
+
+    /// Returns the next kernel or injected completion if one is available.
+    ///
+    /// Consuming a kernel completion returns ownership of any buffer retained
+    /// by the corresponding operation.
+    pub fn next_completion(&mut self) -> Option<AsyncIoCompletion> {
+        if self.needs_submit_retry {
+            match self.io_uring.submitter().submit() {
+                Ok(_) => self.needs_submit_retry = false,
+                Err(e) => warn!("io_uring retry submit failed for retained SQEs: {e}"),
+            }
+        }
+
+        if let Some(entry) = self.io_uring.completion().next() {
+            let user_data = entry.user_data();
+            return Some(AsyncIoCompletion::new(
+                user_data,
+                entry.result(),
+                self.in_flight
+                    .remove(&user_data)
+                    .flatten()
+                    .and_then(AsyncIoOperation::into_completion_buffer),
+            ));
+        }
+
+        self.injected.pop_front()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::os::fd::AsRawFd;
+    use std::thread::sleep;
+    use std::time::Duration;
+
+    use vmm_sys_util::tempfile::TempFile;
+
+    use super::UringDataIo;
+    use crate::async_io::{AsyncIoCompletion, AsyncIoOperation, OwnedIoBuffer};
+
+    fn wait_for_completion(data_io: &mut UringDataIo) -> AsyncIoCompletion {
+        for _ in 0..1000 {
+            if let Some(completion) = data_io.next_completion() {
+                return completion;
+            }
+            sleep(Duration::from_millis(1));
+        }
+
+        panic!("timed out waiting for io_uring completion");
+    }
+
+    #[test]
+    fn uring_rejects_duplicate_user_data_for_metadata_ops() {
+        let file = TempFile::new().unwrap().into_file();
+        file.set_len(512).unwrap();
+        let fd = file.as_raw_fd();
+        let mut data_io = UringDataIo::new(8).unwrap();
+
+        data_io
+            .submit_operation(
+                fd,
+                AsyncIoOperation::read_to_vec(0, OwnedIoBuffer::from_vec(vec![0; 512]), 7),
+            )
+            .unwrap();
+
+        assert_eq!(
+            data_io.submit_fsync(fd, 7).unwrap_err().kind(),
+            std::io::ErrorKind::AlreadyExists
+        );
+        assert_eq!(
+            data_io.submit_nop(7).unwrap_err().kind(),
+            std::io::ErrorKind::AlreadyExists
+        );
+        assert_eq!(
+            data_io
+                .submit_fallocate(fd, 0, 512, 0, 7)
+                .unwrap_err()
+                .kind(),
+            std::io::ErrorKind::AlreadyExists
+        );
+
+        let completion = wait_for_completion(&mut data_io);
+        assert_eq!(completion.user_data, 7);
+        assert_eq!(completion.result, 512);
+    }
+
+    #[test]
+    fn uring_queue_full_batch_completes_each_operation() {
+        let file = TempFile::new().unwrap().into_file();
+        let fd = file.as_raw_fd();
+        let mut data_io = UringDataIo::new(1).unwrap();
+        let available = {
+            let (_, sq, _) = data_io.io_uring.split();
+            sq.capacity() - sq.len()
+        };
+        let batch_len = available + 1;
+        let batch: Vec<_> = (0..batch_len as u64)
+            .map(|user_data| {
+                AsyncIoOperation::read_to_vec(0, OwnedIoBuffer::from_vec(vec![0; 512]), user_data)
+            })
+            .collect();
+
+        data_io.submit_batch(fd, batch).unwrap();
+
+        let mut completed = Vec::new();
+        while let Some(completion) = data_io.next_completion() {
+            assert_eq!(completion.result, -libc::EAGAIN);
+            assert!(completion.buffer.is_some());
+            completed.push(completion.user_data);
+        }
+        completed.sort_unstable();
+        assert_eq!(completed, (0..batch_len as u64).collect::<Vec<_>>());
+    }
+}

block/src/io/fcntl.rs

@@ -0,0 +1,254 @@
+// Copyright © 2025 Cyberus Technology GmbH
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+//! Helpers for advisory file locking.
+//!
+//! Under the hood, the implementation uses OFD locks for the entire file,
+//! as described in [[0]]. The advantage over `F_SETLKW` (currently used by
+//! Rust std: `File::try_lock()`) is that only the very last `close()` on a
+//! file descriptor releases the lock. This prevents mistakes and unexpected
+//! behavior.
+//!
+//! [0]: <https://apenwarr.ca/log/20101213>.
+
+use std::fmt::Debug;
+use std::io;
+use std::os::fd::{AsRawFd, RawFd};
+use std::str::FromStr;
+
+use thiserror::Error;
+
+/// Errors that can happen when working with file locks.
+#[derive(Error, Debug)]
+pub enum LockError {
+    /// The file is already locked.
+    ///
+    /// A call to [`get_lock_state`] can help to identify the reason.
+    #[error("The file is already locked")]
+    AlreadyLocked,
+    /// IO error.
+    #[error("The lock state could not be checked or set")]
+    Io(#[source] io::Error),
+}
+
+/// Commands for use with [`fcntl`].
+#[allow(non_camel_case_types)]
+enum FcntlArg<'a> {
+    /// Set an OFD lock from the given lock description.
+    F_OFD_SETLK(&'a libc::flock),
+    /// Get the first OFD lock for the given lock description.
+    F_OFD_GETLK(&'a mut libc::flock),
+}
+
+/// Wrapper for [`libc::fcntl`] that properly sets the function arguments.
+fn fcntl(fd: RawFd, arg: FcntlArg) -> libc::c_int {
+    // SAFETY: We use a valid FD.
+    unsafe {
+        match arg {
+            FcntlArg::F_OFD_SETLK(flock) => libc::fcntl(fd, libc::F_OFD_SETLK, flock),
+            FcntlArg::F_OFD_GETLK(flock) => libc::fcntl(fd, libc::F_OFD_GETLK, flock),
+        }
+    }
+}
+
+/// Describes the type of lock you want to set.
+#[derive(Clone, Copy, Debug)]
+pub enum LockType {
+    /// Clear a lock.
+    Unlock,
+    /// Set a write lock (exclusive).
+    Write,
+    /// Set a read lock (shared).
+    Read,
+}
+
+impl LockType {
+    pub const fn to_libc_val(self) -> libc::c_int {
+        match self {
+            Self::Unlock => libc::F_UNLCK as libc::c_int,
+            Self::Write => libc::F_WRLCK as libc::c_int,
+            Self::Read => libc::F_RDLCK as libc::c_int,
+        }
+    }
+}
+
+/// Describes the current state of a lock.
+#[derive(Debug)]
+pub enum LockState {
+    /// No lock set.
+    Unlocked,
+    /// Locked for reading (non-exclusive).
+    SharedRead,
+    /// Locked for writing (exclusive mode).
+    ExclusiveWrite,
+}
+
+impl LockState {
+    fn new(value: libc::c_int) -> Self {
+        const F_UNLCK: libc::c_int = libc::F_UNLCK as libc::c_int;
+        const F_WRLCK: libc::c_int = libc::F_WRLCK as libc::c_int;
+        const F_RDLCK: libc::c_int = libc::F_RDLCK as libc::c_int;
+        match value {
+            F_UNLCK => Self::Unlocked,
+            F_WRLCK => Self::ExclusiveWrite,
+            F_RDLCK => Self::SharedRead,
+            // This is so unlikely that we want to avoid the complexity of
+            // coping with this error case. Can only fail if either Linux
+            // is broken or memory is messed up.
+            other => panic!("Unexpected lock state: {other}"),
+        }
+    }
+}
+
+/// The granularity of the advisory lock.
+///
+/// The granularity has significant implications in typical cloud deployments
+/// with network storage. The Linux kernel will sync advisory locks to network
+/// file systems, but these backends may have different policies and handle
+/// locks differently. For example, Netapp speaks a NFS API but will treat
+/// advisory OFD locks for the whole file as mandatory locks, whereas byte-range
+/// locks for the whole file will remain advisory [0].
+///
+/// As it is a valid use case to prevent multiple CHV instances from accessing
+/// the same disk but disk management software (e.g., Cinder in OpenStack)
+/// should be able to snapshot disks while VMs are running, we need special
+/// control over the lock granularity. Therefore, it is a valid use case to lock
+/// the whole byte range of a disk image without technically locking the whole
+/// file - to get the best of both worlds.
+///
+/// [0] https://kb.netapp.com/on-prem/ontap/da/NAS/NAS-KBs/How_is_Mandatory_Locking_supported_for_NFSv4_on_ONTAP_9
+#[derive(Clone, Copy, Debug)]
+pub enum LockGranularity {
+    WholeFile,
+    ByteRange(u64 /* from, inclusive */, u64 /* len */),
+}
+
+impl LockGranularity {
+    const fn l_start(self) -> u64 {
+        match self {
+            LockGranularity::WholeFile => 0,
+            LockGranularity::ByteRange(start, _) => start,
+        }
+    }
+
+    const fn l_len(self) -> u64 {
+        match self {
+            LockGranularity::WholeFile => 0, /* EOF */
+            LockGranularity::ByteRange(_, len) => len,
+        }
+    }
+}
+
+/// User-facing choice for the lock granularity.
+///
+/// This allows external management software to create snapshots of the disk
+/// image. Without a byte-range lock, some NFS implementations may treat the
+/// entire file as exclusively locked and prevent such operations (e.g. NetApp).
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+pub enum LockGranularityChoice {
+    /// Byte-range lock covering [0, size).
+    #[default]
+    ByteRange,
+    /// Whole-file lock (l_start=0, l_len=0) - original OFD whole-file lock behavior.
+    Full,
+}
+
+/// Error returned when parsing a [`LockGranularityChoice`] from a string.
+#[derive(Error, Debug)]
+#[error("Invalid lock granularity value: {0}, expected 'byte-range' or 'full'")]
+pub struct LockGranularityParseError(String);
+
+impl FromStr for LockGranularityChoice {
+    type Err = LockGranularityParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "byte-range" => Ok(LockGranularityChoice::ByteRange),
+            "full" => Ok(LockGranularityChoice::Full),
+            _ => Err(LockGranularityParseError(s.to_owned())),
+        }
+    }
+}
+
+/// Returns a [`struct@libc::flock`] structure for the whole file.
+const fn get_flock(lock_type: LockType, granularity: LockGranularity) -> libc::flock {
+    libc::flock {
+        l_type: lock_type.to_libc_val() as libc::c_short,
+        l_whence: libc::SEEK_SET as libc::c_short,
+        l_start: granularity.l_start() as libc::c_long,
+        l_len: granularity.l_len() as libc::c_long,
+        l_pid: 0, /* filled by callee */
+    }
+}
+
+/// Tries to acquire a lock using [`fcntl`] with respect to the given
+/// parameters.
+///
+/// Please note that `fcntl()` OFD locks are **advisory locks**, which do not
+/// prevent to `open()` a file if a lock is already placed.
+///
+/// # Parameters
+/// - `file`: The file to acquire a lock for [`LockType`]. The file's state will
+///   be logically mutated, but not technically.
+/// - `lock_type`: The [`LockType`]
+/// - `granularity`: The [`LockGranularity`].
+pub fn try_acquire_lock<Fd: AsRawFd>(
+    file: &Fd,
+    lock_type: LockType,
+    granularity: LockGranularity,
+) -> Result<(), LockError> {
+    let flock = get_flock(lock_type, granularity);
+
+    let res = fcntl(file.as_raw_fd(), FcntlArg::F_OFD_SETLK(&flock));
+    match res {
+        0 => Ok(()),
+        -1 => {
+            let io_error = io::Error::last_os_error();
+            let errno = io_error.raw_os_error().unwrap();
+            match errno {
+                // See man page for error code:
+                // <https://man7.org/linux/man-pages/man2/fcntl.2.html>
+                libc::EAGAIN | libc::EACCES => Err(LockError::AlreadyLocked),
+                _ => Err(LockError::Io(io_error)),
+            }
+        }
+        val => panic!("Unexpected return value from fcntl(): {val}"),
+    }
+}
+
+/// Clears a lock.
+///
+/// # Parameters
+/// - `file`: The file to clear all locks for [`LockType`].
+/// - `granularity`: The [`LockGranularity`].
+pub fn clear_lock<Fd: AsRawFd>(file: &Fd, granularity: LockGranularity) -> Result<(), LockError> {
+    try_acquire_lock(file, LockType::Unlock, granularity)
+}
+
+/// Returns the current lock state using [`fcntl`] with respect to the given
+/// parameters.
+///
+/// # Parameters
+/// - `file`: The file for which to get the lock state.
+/// - `granularity`: The [`LockGranularity`].
+pub fn get_lock_state<Fd: AsRawFd>(
+    file: &Fd,
+    granularity: LockGranularity,
+) -> Result<LockState, LockError> {
+    let mut flock = get_flock(LockType::Write, granularity);
+    let res = fcntl(file.as_raw_fd(), FcntlArg::F_OFD_GETLK(&mut flock));
+    match res {
+        0 => {
+            let state = flock.l_type as libc::c_int;
+            let state = LockState::new(state);
+            Ok(state)
+        }
+        -1 => {
+            let io_error = io::Error::last_os_error();
+            Err(LockError::Io(io_error))
+        }
+        val => panic!("Unexpected return value from fcntl(): {val}"),
+    }
+}

block/src/io/mod.rs

@@ -0,0 +1,12 @@
+// Copyright 2026 The Cloud Hypervisor Authors. All rights reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+//! Shared I/O infrastructure for all disk format backends.
+//!
+//! Contains the async I/O trait, request handling, and file locking
+//! helpers.
+
+pub mod async_io;
+pub mod fcntl;
+pub mod request;

block/src/io/request.rs

@@ -0,0 +1,658 @@
+// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Portions Copyright 2017 The Chromium OS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE-BSD-3-Clause file.
+//
+// Copyright © 2020 Intel Corporation
+//
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
+
+use std::io::{Read, Seek, SeekFrom, Write};
+use std::mem;
+use std::sync::Arc;
+use std::time::Instant;
+
+use log::{error, warn};
+use smallvec::SmallVec;
+use virtio_bindings::virtio_blk::{
+    VIRTIO_BLK_T_DISCARD, VIRTIO_BLK_T_WRITE_ZEROES, VIRTIO_BLK_WRITE_ZEROES_FLAG_UNMAP,
+    virtio_blk_discard_write_zeroes,
+};
+use virtio_queue::DescriptorChain;
+use vm_memory::bitmap::Bitmap;
+use vm_memory::{
+    Address as _, Bytes as _, GuestAddress, GuestMemory as _, GuestMemoryError,
+    GuestMemoryLoadGuard,
+};
+use vm_virtio::{AccessPlatform, Translatable as _};
+
+use crate::async_io::{
+    AsyncIo, AsyncIoCompletion, AsyncIoOperation, GuestMemoryTarget, OwnedIoBuffer,
+};
+use crate::{Error, ExecuteError, request_type, sector};
+
+const SECTOR_SHIFT: u8 = 9;
+pub const SECTOR_SIZE: u64 = 0x01 << SECTOR_SHIFT;
+
+/// Maximum number of segments per DISCARD or WRITE_ZEROES request.
+pub const MAX_DISCARD_WRITE_ZEROES_SEG: u32 = 1;
+/// Size and field offsets within `struct virtio_blk_discard_write_zeroes`.
+const DISCARD_WZ_SEG_SIZE: u32 = mem::size_of::<virtio_blk_discard_write_zeroes>() as u32;
+const DISCARD_WZ_MAX_PAYLOAD: u32 = DISCARD_WZ_SEG_SIZE * MAX_DISCARD_WRITE_ZEROES_SEG;
+const DISCARD_WZ_SECTOR_OFFSET: u64 =
+    mem::offset_of!(virtio_blk_discard_write_zeroes, sector) as u64;
+const DISCARD_WZ_NUM_SECTORS_OFFSET: u64 =
+    mem::offset_of!(virtio_blk_discard_write_zeroes, num_sectors) as u64;
+const DISCARD_WZ_FLAGS_OFFSET: u64 = mem::offset_of!(virtio_blk_discard_write_zeroes, flags) as u64;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RequestType {
+    In,
+    Out,
+    Flush,
+    GetDeviceId,
+    Discard,
+    WriteZeroes,
+    Unsupported(u32),
+}
+
+pub const DEFAULT_DESCRIPTOR_VEC_SIZE: usize = 32;
+
+pub struct ExecuteAsync {
+    // `true` if the execution will complete asynchronously
+    pub async_complete: bool,
+    // request need to be batched for submission if any
+    pub batch_request: Option<AsyncIoOperation>,
+}
+
+#[derive(Debug)]
+pub struct Request {
+    request_type: RequestType,
+    sector: u64,
+    data_descriptors: SmallVec<[(GuestAddress, u32); DEFAULT_DESCRIPTOR_VEC_SIZE]>,
+    status_addr: GuestAddress,
+    pub writeback: bool,
+    start: Instant,
+}
+
+impl Request {
+    pub fn parse<B: Bitmap + 'static>(
+        desc_chain: &mut DescriptorChain<GuestMemoryLoadGuard<vm_memory::GuestMemoryMmap<B>>>,
+        access_platform: Option<&dyn AccessPlatform>,
+    ) -> Result<Request, Error> {
+        let hdr_desc = desc_chain
+            .next()
+            .ok_or(Error::DescriptorChainTooShort)
+            .inspect_err(|_| {
+                error!("Missing head descriptor");
+            })?;
+
+        // The head contains the request type which MUST be readable.
+        if hdr_desc.is_write_only() {
+            return Err(Error::UnexpectedWriteOnlyDescriptor);
+        }
+
+        let hdr_desc_addr = hdr_desc
+            .addr()
+            .translate_gva(access_platform, hdr_desc.len() as usize)
+            .map_err(|e| Error::GuestMemory(GuestMemoryError::IOError(e)))?;
+
+        let mut req = Request {
+            request_type: request_type(desc_chain.memory(), hdr_desc_addr)?,
+            sector: sector(desc_chain.memory(), hdr_desc_addr)?,
+            data_descriptors: SmallVec::with_capacity(DEFAULT_DESCRIPTOR_VEC_SIZE),
+            status_addr: GuestAddress(0),
+            writeback: true,
+            start: Instant::now(),
+        };
+
+        let status_desc;
+        let mut desc = desc_chain
+            .next()
+            .ok_or(Error::DescriptorChainTooShort)
+            .inspect_err(|_| {
+                error!("Only head descriptor present: request = {req:?}");
+            })?;
+
+        if desc.has_next() {
+            req.data_descriptors.reserve_exact(1);
+            while desc.has_next() {
+                if desc.is_write_only() && req.request_type == RequestType::Out {
+                    return Err(Error::UnexpectedWriteOnlyDescriptor);
+                }
+                if desc.is_write_only() && req.request_type == RequestType::Discard {
+                    return Err(Error::UnexpectedWriteOnlyDescriptor);
+                }
+                if desc.is_write_only() && req.request_type == RequestType::WriteZeroes {
+                    return Err(Error::UnexpectedWriteOnlyDescriptor);
+                }
+                if !desc.is_write_only() && req.request_type == RequestType::In {
+                    return Err(Error::UnexpectedReadOnlyDescriptor);
+                }
+                if !desc.is_write_only() && req.request_type == RequestType::GetDeviceId {
+                    return Err(Error::UnexpectedReadOnlyDescriptor);
+                }
+
+                req.data_descriptors.push((
+                    desc.addr()
+                        .translate_gva(access_platform, desc.len() as usize)
+                        .map_err(|e| Error::GuestMemory(GuestMemoryError::IOError(e)))?,
+                    desc.len(),
+                ));
+                desc = desc_chain
+                    .next()
+                    .ok_or(Error::DescriptorChainTooShort)
+                    .inspect_err(|_| {
+                        error!("DescriptorChain corrupted: request = {req:?}");
+                    })?;
+            }
+            status_desc = desc;
+        } else {
+            status_desc = desc;
+            // Only flush requests are allowed to skip the data descriptor.
+            if req.request_type != RequestType::Flush {
+                error!("Need a data descriptor: request = {req:?}");
+                return Err(Error::DescriptorChainTooShort);
+            }
+        }
+
+        // The status MUST always be writable.
+        if !status_desc.is_write_only() {
+            return Err(Error::UnexpectedReadOnlyDescriptor);
+        }
+
+        if status_desc.len() < 1 {
+            return Err(Error::DescriptorLengthTooSmall);
+        }
+
+        req.status_addr = status_desc
+            .addr()
+            .translate_gva(access_platform, status_desc.len() as usize)
+            .map_err(|e| Error::GuestMemory(GuestMemoryError::IOError(e)))?;
+
+        Ok(req)
+    }
+
+    pub fn execute<T: Seek + Read + Write, B: Bitmap + 'static>(
+        &self,
+        disk: &mut T,
+        disk_nsectors: u64,
+        mem: &vm_memory::GuestMemoryMmap<B>,
+        serial: &[u8],
+    ) -> Result<u32, ExecuteError> {
+        self.check_data_bounds(disk_nsectors)?;
+
+        disk.seek(SeekFrom::Start(self.sector << SECTOR_SHIFT))
+            .map_err(ExecuteError::Seek)?;
+        let mut len = 0;
+        for (data_addr, data_len) in &self.data_descriptors {
+            match self.request_type {
+                RequestType::In => {
+                    let mut buf = vec![0u8; *data_len as usize];
+                    disk.read_exact(&mut buf).map_err(ExecuteError::ReadExact)?;
+                    mem.read_exact_volatile_from(
+                        *data_addr,
+                        &mut buf.as_slice(),
+                        *data_len as usize,
+                    )
+                    .map_err(ExecuteError::Read)?;
+                    len += data_len;
+                }
+                RequestType::Out => {
+                    let mut buf: Vec<u8> = Vec::new();
+                    mem.write_all_volatile_to(*data_addr, &mut buf, *data_len as usize)
+                        .map_err(ExecuteError::Write)?;
+                    disk.write_all(&buf).map_err(ExecuteError::WriteAll)?;
+                    if !self.writeback {
+                        disk.flush().map_err(ExecuteError::Flush)?;
+                    }
+                }
+                RequestType::Flush => disk.flush().map_err(ExecuteError::Flush)?,
+                RequestType::GetDeviceId => {
+                    if (*data_len as usize) < serial.len() {
+                        return Err(ExecuteError::BadRequest(Error::InvalidOffset));
+                    }
+                    mem.write_slice(serial, *data_addr)
+                        .map_err(ExecuteError::Write)?;
+                }
+                RequestType::Discard => {
+                    return Err(ExecuteError::Unsupported(VIRTIO_BLK_T_DISCARD));
+                }
+                RequestType::WriteZeroes => {
+                    return Err(ExecuteError::Unsupported(VIRTIO_BLK_T_WRITE_ZEROES));
+                }
+                RequestType::Unsupported(t) => return Err(ExecuteError::Unsupported(t)),
+            }
+        }
+        Ok(len)
+    }
+
+    pub fn execute_async<B: Bitmap + Send + Sync + 'static>(
+        &mut self,
+        mem: Arc<vm_memory::GuestMemoryMmap<B>>,
+        disk_nsectors: u64,
+        disk_image: &mut dyn AsyncIo,
+        serial: &[u8],
+        disable_sector0_writes: bool,
+        user_data: u64,
+    ) -> Result<ExecuteAsync, ExecuteError> {
+        let sector = self.sector;
+        let request_type = self.request_type;
+        let offset = (sector << SECTOR_SHIFT) as libc::off_t;
+        let alignment = disk_image.alignment();
+
+        self.check_data_bounds(disk_nsectors)?;
+
+        let mut ret = ExecuteAsync {
+            async_complete: true,
+            batch_request: None,
+        };
+        // Queue operations expected to be submitted.
+        match request_type {
+            RequestType::In => {
+                self.mark_read_dirty(&mem)?;
+                let op = self.build_data_operation(mem, offset, alignment, user_data)?;
+                if disk_image.batch_requests_enabled() {
+                    ret.batch_request = Some(op);
+                } else {
+                    match op {
+                        AsyncIoOperation::ReadToMemory {
+                            offset,
+                            target,
+                            user_data,
+                        } => disk_image
+                            .read_to_memory(offset, target, user_data)
+                            .map_err(ExecuteError::AsyncRead)?,
+                        AsyncIoOperation::ReadToVec {
+                            offset,
+                            buffer,
+                            user_data,
+                        } => disk_image
+                            .read_to_vec(offset, buffer, user_data)
+                            .map_err(ExecuteError::AsyncRead)?,
+                        _ => unreachable!("unexpected read operation"),
+                    }
+                }
+            }
+            RequestType::Out => {
+                let op = self.build_data_operation(mem, offset, alignment, user_data)?;
+                if disk_image.batch_requests_enabled() {
+                    ret.batch_request = Some(op);
+                } else {
+                    match op {
+                        AsyncIoOperation::WriteFromMemory {
+                            offset,
+                            target,
+                            user_data,
+                        } => disk_image
+                            .write_from_memory(offset, target, user_data)
+                            .map_err(ExecuteError::AsyncWrite)?,
+                        AsyncIoOperation::WriteFromVec {
+                            offset,
+                            buffer,
+                            user_data,
+                        } => disk_image
+                            .write_from_vec(offset, buffer, user_data)
+                            .map_err(ExecuteError::AsyncWrite)?,
+                        _ => unreachable!("unexpected write operation"),
+                    }
+                }
+            }
+            RequestType::Flush => {
+                disk_image
+                    .fsync(Some(user_data))
+                    .map_err(ExecuteError::AsyncFlush)?;
+            }
+            RequestType::GetDeviceId => {
+                let (data_addr, data_len) = if self.data_descriptors.len() == 1 {
+                    (self.data_descriptors[0].0, self.data_descriptors[0].1)
+                } else {
+                    return Err(ExecuteError::BadRequest(Error::TooManyDescriptors));
+                };
+                if (data_len as usize) < serial.len() {
+                    return Err(ExecuteError::BadRequest(Error::InvalidOffset));
+                }
+                mem.write_slice(serial, data_addr)
+                    .map_err(ExecuteError::Write)?;
+                ret.async_complete = false;
+                return Ok(ret);
+            }
+            RequestType::Discard => {
+                let (data_addr, data_len) = if self.data_descriptors.len() == 1 {
+                    (self.data_descriptors[0].0, self.data_descriptors[0].1)
+                } else {
+                    return Err(ExecuteError::BadRequest(Error::TooManyDescriptors));
+                };
+
+                if data_len < DISCARD_WZ_SEG_SIZE {
+                    return Err(ExecuteError::BadRequest(Error::DescriptorLengthTooSmall));
+                }
+                if data_len > DISCARD_WZ_MAX_PAYLOAD {
+                    return Err(ExecuteError::BadRequest(Error::TooManySegments(
+                        data_len.div_ceil(DISCARD_WZ_SEG_SIZE),
+                    )));
+                }
+
+                let mut discard_sector = [0u8; 8];
+                let mut discard_num_sectors = [0u8; 4];
+                let mut discard_flags = [0u8; 4];
+
+                let sector_addr = data_addr.checked_add(DISCARD_WZ_SECTOR_OFFSET).unwrap();
+                mem.read_slice(&mut discard_sector, sector_addr)
+                    .map_err(ExecuteError::Read)?;
+
+                let num_sectors_addr = data_addr
+                    .checked_add(DISCARD_WZ_NUM_SECTORS_OFFSET)
+                    .unwrap();
+                mem.read_slice(&mut discard_num_sectors, num_sectors_addr)
+                    .map_err(ExecuteError::Read)?;
+
+                let flags_addr = data_addr.checked_add(DISCARD_WZ_FLAGS_OFFSET).unwrap();
+                mem.read_slice(&mut discard_flags, flags_addr)
+                    .map_err(ExecuteError::Read)?;
+
+                let discard_flags = u32::from_le_bytes(discard_flags);
+                // Per virtio spec v1.2 reject discard if any flag is set, including unmap.
+                if discard_flags != 0 {
+                    warn!("Unsupported flags {discard_flags:#x} in discard request");
+                    return Err(ExecuteError::UnsupportedFlags {
+                        request_type: VIRTIO_BLK_T_DISCARD,
+                        flags: discard_flags,
+                    });
+                }
+
+                let discard_sector = u64::from_le_bytes(discard_sector);
+
+                if discard_sector == 0 && disable_sector0_writes {
+                    return Err(ExecuteError::BadRequest(Error::InvalidOffset));
+                }
+
+                let discard_num_sectors = u32::from_le_bytes(discard_num_sectors);
+
+                let top = discard_sector
+                    .checked_add(discard_num_sectors as u64)
+                    .ok_or(ExecuteError::BadRequest(Error::InvalidOffset))?;
+                if top > disk_nsectors {
+                    return Err(ExecuteError::BadRequest(Error::InvalidOffset));
+                }
+
+                let discard_offset = discard_sector * SECTOR_SIZE;
+                let discard_length = (discard_num_sectors as u64) * SECTOR_SIZE;
+
+                disk_image
+                    .punch_hole(discard_offset, discard_length, user_data)
+                    .map_err(ExecuteError::AsyncPunchHole)?;
+            }
+            RequestType::WriteZeroes => {
+                let (data_addr, data_len) = if self.data_descriptors.len() == 1 {
+                    (self.data_descriptors[0].0, self.data_descriptors[0].1)
+                } else {
+                    return Err(ExecuteError::BadRequest(Error::TooManyDescriptors));
+                };
+
+                if data_len < DISCARD_WZ_SEG_SIZE {
+                    return Err(ExecuteError::BadRequest(Error::DescriptorLengthTooSmall));
+                }
+                if data_len > DISCARD_WZ_MAX_PAYLOAD {
+                    return Err(ExecuteError::BadRequest(Error::TooManySegments(
+                        data_len.div_ceil(DISCARD_WZ_SEG_SIZE),
+                    )));
+                }
+
+                let mut wz_sector = [0u8; 8];
+                let mut wz_num_sectors = [0u8; 4];
+                let mut wz_flags = [0u8; 4];
+
+                let sector_addr = data_addr.checked_add(DISCARD_WZ_SECTOR_OFFSET).unwrap();
+                mem.read_slice(&mut wz_sector, sector_addr)
+                    .map_err(ExecuteError::Read)?;
+
+                let num_sectors_addr = data_addr
+                    .checked_add(DISCARD_WZ_NUM_SECTORS_OFFSET)
+                    .unwrap();
+                mem.read_slice(&mut wz_num_sectors, num_sectors_addr)
+                    .map_err(ExecuteError::Read)?;
+
+                let flags_addr = data_addr.checked_add(DISCARD_WZ_FLAGS_OFFSET).unwrap();
+                mem.read_slice(&mut wz_flags, flags_addr)
+                    .map_err(ExecuteError::Read)?;
+
+                let wz_sector = u64::from_le_bytes(wz_sector);
+                let wz_num_sectors = u32::from_le_bytes(wz_num_sectors);
+
+                let wz_flags = u32::from_le_bytes(wz_flags);
+                // Per virtio spec v1.2 reject write zeroes if any unknown flag is set.
+                if (wz_flags & !VIRTIO_BLK_WRITE_ZEROES_FLAG_UNMAP) != 0 {
+                    warn!("Unsupported flags {wz_flags:#x} in write zeroes request");
+                    return Err(ExecuteError::UnsupportedFlags {
+                        request_type: VIRTIO_BLK_T_WRITE_ZEROES,
+                        flags: wz_flags,
+                    });
+                }
+
+                let wz_offset = wz_sector * SECTOR_SIZE;
+                if wz_offset == 0 && disable_sector0_writes {
+                    return Err(ExecuteError::BadRequest(Error::InvalidOffset));
+                }
+
+                let top = wz_sector
+                    .checked_add(wz_num_sectors as u64)
+                    .ok_or(ExecuteError::BadRequest(Error::InvalidOffset))?;
+                if top > disk_nsectors {
+                    return Err(ExecuteError::BadRequest(Error::InvalidOffset));
+                }
+
+                let wz_length = (wz_num_sectors as u64) * SECTOR_SIZE;
+
+                if wz_flags & VIRTIO_BLK_WRITE_ZEROES_FLAG_UNMAP != 0 {
+                    disk_image
+                        .punch_hole(wz_offset, wz_length, user_data)
+                        .map_err(ExecuteError::AsyncPunchHole)?;
+                } else {
+                    disk_image
+                        .write_zeroes(wz_offset, wz_length, user_data)
+                        .map_err(ExecuteError::AsyncWriteZeroes)?;
+                }
+            }
+            RequestType::Unsupported(t) => return Err(ExecuteError::Unsupported(t)),
+        }
+
+        Ok(ret)
+    }
+
+    // Builds a read or write operation for IO to or from `mem`.
+    fn build_data_operation<B: Bitmap + Send + Sync + 'static>(
+        &self,
+        mem: Arc<vm_memory::GuestMemoryMmap<B>>,
+        offset: libc::off_t,
+        alignment: u64,
+        user_data: u64,
+    ) -> Result<AsyncIoOperation, ExecuteError> {
+        if self.guest_memory_is_aligned(&mem, alignment)? {
+            let target = GuestMemoryTarget::new(mem, &self.data_descriptors)
+                .map_err(ExecuteError::GetHostAddress)?;
+            return Ok(match self.request_type {
+                RequestType::In => AsyncIoOperation::read_to_memory(offset, target, user_data),
+                RequestType::Out => AsyncIoOperation::write_from_memory(offset, target, user_data),
+                _ => unreachable!("unexpected data operation type"),
+            });
+        }
+
+        // The guest-memory buffers are unaligned, so use an aligned bounce buffer.
+        let mut buffer = OwnedIoBuffer::new(self.data_len(), alignment as usize)
+            .map_err(ExecuteError::TemporaryBufferAllocation)?;
+
+        if self.request_type == RequestType::Out {
+            self.copy_guest_to_buffer(&mem, buffer.as_mut_slice())?;
+        }
+
+        Ok(match self.request_type {
+            RequestType::In => AsyncIoOperation::read_to_vec(offset, buffer, user_data),
+            RequestType::Out => AsyncIoOperation::write_from_vec(offset, buffer, user_data),
+            _ => unreachable!("unexpected data operation type"),
+        })
+    }
+
+    // Checks whether `self.data_descriptors` are aligned to `alignment`.
+    fn guest_memory_is_aligned<B: Bitmap + 'static>(
+        &self,
+        mem: &vm_memory::GuestMemoryMmap<B>,
+        alignment: u64,
+    ) -> Result<bool, ExecuteError> {
+        if alignment <= 1 {
+            return Ok(true);
+        }
+
+        for &(data_addr, data_len) in &self.data_descriptors {
+            let _: u32 = data_len;
+            const _: () = assert!(
+                core::mem::size_of::<u32>() <= core::mem::size_of::<usize>(),
+                "unsupported platform"
+            );
+            if data_len == 0 {
+                continue;
+            }
+            let data_len = data_len as usize;
+            let origin_ptr = mem
+                .get_slice(data_addr, data_len)
+                .map_err(ExecuteError::GetHostAddress)?;
+            let origin_ptr = origin_ptr.ptr_guard_mut();
+            if !(origin_ptr.as_ptr() as u64).is_multiple_of(alignment)
+                || !(origin_ptr.len() as u64).is_multiple_of(alignment)
+            {
+                return Ok(false);
+            }
+        }
+
+        Ok(true)
+    }
+
+    // Returns the sum of the lengths of `self.data_descriptors`.
+    fn data_len(&self) -> usize {
+        self.data_descriptors
+            .iter()
+            .map(|(_, len)| *len as usize)
+            .sum()
+    }
+
+    // Marks guest-memory read destinations dirty before submitting async IO.
+    fn mark_read_dirty<B: Bitmap + 'static>(
+        &self,
+        mem: &vm_memory::GuestMemoryMmap<B>,
+    ) -> Result<(), ExecuteError> {
+        for (data_addr, data_len) in &self.data_descriptors {
+            mem.get_slice(*data_addr, *data_len as usize)
+                .map_err(ExecuteError::GetHostAddress)?
+                .bitmap()
+                .mark_dirty(0, *data_len as usize);
+        }
+        Ok(())
+    }
+
+    // Copies guest descriptor contents into a contiguous host buffer.
+    fn copy_guest_to_buffer<B: Bitmap + 'static>(
+        &self,
+        mem: &vm_memory::GuestMemoryMmap<B>,
+        buffer: &mut [u8],
+    ) -> Result<(), ExecuteError> {
+        let mut offset = 0usize;
+        for (data_addr, data_len) in &self.data_descriptors {
+            let data_len = *data_len as usize;
+            mem.read_slice(&mut buffer[offset..offset + data_len], *data_addr)
+                .map_err(ExecuteError::Read)?;
+            offset += data_len;
+        }
+        Ok(())
+    }
+
+    // Copies a host completion buffer back into guest descriptors.
+    fn copy_buffer_to_guest<B: Bitmap + 'static>(
+        &self,
+        mem: &vm_memory::GuestMemoryMmap<B>,
+        buffer: &[u8],
+    ) -> Result<(), Error> {
+        let mut buffer_offset = 0usize;
+        for (data_addr, data_len) in &self.data_descriptors {
+            if buffer_offset >= buffer.len() {
+                break;
+            }
+            let data_len = (*data_len as usize).min(buffer.len() - buffer_offset);
+            mem.write_slice(&buffer[buffer_offset..buffer_offset + data_len], *data_addr)
+                .map_err(Error::GuestMemory)?;
+            buffer_offset += data_len;
+        }
+        Ok(())
+    }
+
+    pub fn complete_async<B: Bitmap + 'static>(
+        &mut self,
+        mem: &vm_memory::GuestMemoryMmap<B>,
+        completion: &mut AsyncIoCompletion,
+    ) -> Result<(), Error> {
+        if self.request_type == RequestType::In
+            && completion.result > 0
+            && let Some(buffer) = completion.buffer.take()
+        {
+            let len = (completion.result as usize).min(buffer.as_slice().len());
+            self.copy_buffer_to_guest(mem, &buffer.as_slice()[..len])?;
+        }
+
+        Ok(())
+    }
+
+    #[inline]
+    pub fn data_descriptors(
+        &self,
+    ) -> &SmallVec<[(GuestAddress, u32); DEFAULT_DESCRIPTOR_VEC_SIZE]> {
+        &self.data_descriptors
+    }
+
+    #[inline]
+    pub fn status_addr(&self) -> GuestAddress {
+        self.status_addr
+    }
+
+    #[inline]
+    pub fn start(&self) -> Instant {
+        self.start
+    }
+
+    #[inline]
+    pub fn sector(&self) -> u64 {
+        self.sector
+    }
+
+    #[inline]
+    pub fn request_type(&self) -> RequestType {
+        self.request_type
+    }
+
+    /// For In and Out requests, checks that the descriptors collectively fit in a backing disk of
+    /// the given size. Returns `Ok(())` if they fit, or `ExecuteError::BadRequest` otherwise.
+    fn check_data_bounds(&self, disk_nsectors: u64) -> Result<(), ExecuteError> {
+        if !matches!(self.request_type, RequestType::In | RequestType::Out) {
+            return Ok(());
+        }
+        let mut total_bytes: u64 = 0;
+        for (_, data_len) in &self.data_descriptors {
+            total_bytes = total_bytes
+                .checked_add(u64::from(*data_len))
+                .ok_or(ExecuteError::BadRequest(Error::InvalidOffset))?;
+        }
+        if total_bytes == 0 {
+            return Ok(());
+        }
+        let total_sectors = total_bytes.div_ceil(SECTOR_SIZE);
+        let end_sector = self
+            .sector
+            .checked_add(total_sectors)
+            .ok_or(ExecuteError::BadRequest(Error::InvalidOffset))?;
+        if end_sector > disk_nsectors {
+            return Err(ExecuteError::BadRequest(Error::InvalidOffset));
+        }
+        Ok(())
+    }
+}