From ae5f4664acf97d94c422f3b697c8936c4ced9d24 Mon Sep 17 00:00:00 2001 From: Dylan Reid Date: Mon, 18 May 2026 16:38:45 -0700 Subject: [PATCH] block: Add AIO async struct Similar to uring io added in the parent commit. These async ops deal with buffer ownership across aio calls. This will be used in the (increasingly rare) case of io_uring not being available or desirable. Signed-off-by: Dylan Reid --- block/src/async_io.rs | 2 + block/src/async_io/aio_data_io.rs | 244 ++++++++++++++++++++++++++++++ block/src/async_io/common.rs | 1 - 3 files changed, 246 insertions(+), 1 deletion(-) create mode 100644 block/src/async_io/aio_data_io.rs diff --git a/block/src/async_io.rs b/block/src/async_io.rs index 7337ade54..f54ecb5a7 100644 --- a/block/src/async_io.rs +++ b/block/src/async_io.rs @@ -4,6 +4,7 @@ use std::marker::PhantomData; use std::os::fd::{AsRawFd, OwnedFd, RawFd}; +mod aio_data_io; mod common; mod completion; mod guest_memory_target; @@ -12,6 +13,7 @@ mod owned_io_buffer; #[cfg(feature = "io_uring")] mod uring_data_io; +pub use aio_data_io::AioDataIo; pub use completion::AsyncIoCompletion; pub use guest_memory_target::GuestMemoryTarget; pub use operation::AsyncIoOperation; diff --git a/block/src/async_io/aio_data_io.rs b/block/src/async_io/aio_data_io.rs new file mode 100644 index 000000000..410bce56c --- /dev/null +++ b/block/src/async_io/aio_data_io.rs @@ -0,0 +1,244 @@ +// 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, HashSet, 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. +/// +/// Submitted operations are kept in `pending` until their completion event is +/// consumed so the iovec pointers handed to the kernel remain valid for the +/// full lifetime of the operation. +pub struct AioDataIo { + // Keep this before `pending`: 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 user_data values accepted by the kernel submission + // path, including metadata operations that do not retain a buffer. + in_flight: HashSet, + // `pending` owns read/write operations until their events are consumed so + // their iovecs and backing buffers remain valid while the kernel may use + // them. + pending: HashMap, + // `completions` holds locally produced completions and kernel events that + // have been fetched but not yet returned to the caller. + completions: VecDeque, +} + +impl AioDataIo { + /// Creates a Linux AIO context and its completion eventfd. + pub fn new(queue_depth: u32) -> io::Result { + Ok(Self { + ctx: aio::IoContext::new(queue_depth)?, + eventfd: EventFd::new(libc::EFD_NONBLOCK)?, + in_flight: HashSet::new(), + pending: 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 { + // 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(&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); + self.pending.insert(user_data, 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 + .pending + .remove(&user_data) + .and_then(AsyncIoOperation::into_completion_buffer); + self.in_flight.remove(&user_data); + 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(&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); + 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 { + 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.in_flight.remove(&event.data); + self.completions.push_back(AsyncIoCompletion::new( + event.data, + event.res as i32, + self.pending + .remove(&event.data) + .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()); + } +} diff --git a/block/src/async_io/common.rs b/block/src/async_io/common.rs index 9f631a110..3270cffc7 100644 --- a/block/src/async_io/common.rs +++ b/block/src/async_io/common.rs @@ -10,7 +10,6 @@ use std::io; use super::AsyncIoOperation; /// Converts an I/O error into the negative errno form used in completions. -#[allow(dead_code)] pub(super) fn errno_result(error: &io::Error) -> i32 { -error.raw_os_error().unwrap_or(libc::EIO) }