diff --git a/block/src/io/async_io/aio_data_io.rs b/block/src/io/async_io/aio_data_io.rs index f62bd61cb..6914ca139 100644 --- a/block/src/io/async_io/aio_data_io.rs +++ b/block/src/io/async_io/aio_data_io.rs @@ -6,7 +6,7 @@ // // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause -use std::collections::{HashMap, VecDeque}; +use std::collections::HashMap; use std::os::fd::{AsRawFd, RawFd}; use std::{io, slice}; @@ -15,7 +15,7 @@ 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}; +use super::{AsyncIoCompletion, AsyncIoOperation, CompletionCommon}; /// Retained Linux AIO queue for owned async data I/O operations. pub struct AioDataIo { @@ -23,15 +23,11 @@ pub struct AioDataIo { // 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>, - // `completions` holds locally produced completions and kernel events that - // have been fetched but not yet returned to the caller. - completions: VecDeque, + completions: CompletionCommon, } impl AioDataIo { @@ -39,15 +35,14 @@ impl AioDataIo { pub fn new(queue_depth: u32) -> io::Result { Ok(Self { ctx: aio::IoContext::new(queue_depth)?, - eventfd: EventFd::new(libc::EFD_NONBLOCK)?, in_flight: HashMap::new(), - completions: VecDeque::new(), + completions: CompletionCommon::new(), }) } /// Returns the eventfd signaled when completions are available. pub fn notifier(&self) -> &EventFd { - &self.eventfd + self.completions.notifier() } #[allow(unused_unsafe)] @@ -84,7 +79,7 @@ impl AioDataIo { aio_offset: op.offset(), aio_data: user_data, aio_flags: aio::IOCB_FLAG_RESFD, - aio_resfd: self.eventfd.as_raw_fd() as u32, + aio_resfd: self.completions.notifier().as_raw_fd() as u32, ..Default::default() }; self.in_flight.insert(user_data, Some(op)); @@ -115,7 +110,7 @@ impl AioDataIo { 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, + aio_resfd: self.completions.notifier().as_raw_fd() as u32, ..Default::default() }; self.in_flight.insert(user_data, None); @@ -135,8 +130,7 @@ impl AioDataIo { /// 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(); + self.completions.complete(completion); } /// Returns the next kernel or injected completion if one is available. @@ -144,28 +138,30 @@ impl AioDataIo { /// 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.completions.push_back(AsyncIoCompletion::new( - event.data, - event.res as i32, - self.in_flight - .remove(&event.data) - .flatten() - .and_then(AsyncIoOperation::into_completion_buffer), - )); - } + if let Some(completion) = self.completions.next_completed() { + return Some(completion); } - self.completions.pop_front() + 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.complete(AsyncIoCompletion::new( + event.data, + event.res as i32, + self.in_flight + .remove(&event.data) + .flatten() + .and_then(AsyncIoOperation::into_completion_buffer), + )); + } + + self.completions.next_completed() } } diff --git a/block/src/io/async_io/completion.rs b/block/src/io/async_io/completion.rs index 20d9eba55..6a4191111 100644 --- a/block/src/io/async_io/completion.rs +++ b/block/src/io/async_io/completion.rs @@ -46,8 +46,8 @@ impl AsyncIoCompletion { } /// Pending completions plus the eventfd that signals the device to -/// drain them. Sync engines run each operation inline, enqueue its -/// completion, and signal the eventfd. +/// drain them. The sync engines and the async backends enqueue their +/// completions here and wake the device through the eventfd. pub(crate) struct CompletionCommon { queue: VecDeque, eventfd: EventFd, diff --git a/block/src/io/async_io/uring_data_io.rs b/block/src/io/async_io/uring_data_io.rs index c0920af54..dca0ab829 100644 --- a/block/src/io/async_io/uring_data_io.rs +++ b/block/src/io/async_io/uring_data_io.rs @@ -4,7 +4,7 @@ // // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause -use std::collections::{HashMap, VecDeque}; +use std::collections::HashMap; use std::os::fd::{AsRawFd, RawFd}; use std::{io, mem}; @@ -13,22 +13,18 @@ use log::{error, warn}; use vmm_sys_util::eventfd::EventFd; use super::common::{duplicate_user_data_error, validate_batch}; -use super::{AsyncIoCompletion, AsyncIoOperation}; +use super::{AsyncIoCompletion, AsyncIoOperation, CompletionCommon}; /// `io_uring` wrapper for async I/O. /// -/// Holds the `IoUring` and its `EventFd`. Tracks ops that are pending. +/// Holds the `IoUring` and tracks the ops that are pending. pub struct UringDataIo { 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>, - // `injected` holds locally produced completions so synchronous failures - // and short-circuited requests use the same drain path as kernel CQEs. - injected: VecDeque, + completions: CompletionCommon, // `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, @@ -38,21 +34,22 @@ impl UringDataIo { /// Creates an io_uring queue and registers its completion eventfd. pub fn new(ring_depth: u32) -> io::Result { let io_uring = IoUring::new(ring_depth)?; - let eventfd = EventFd::new(libc::EFD_NONBLOCK)?; - io_uring.submitter().register_eventfd(eventfd.as_raw_fd())?; + let completions = CompletionCommon::new(); + io_uring + .submitter() + .register_eventfd(completions.notifier().as_raw_fd())?; Ok(Self { io_uring, - eventfd, in_flight: HashMap::new(), - injected: VecDeque::new(), + completions, needs_submit_retry: false, }) } /// Returns the eventfd signaled when completions are available. pub fn notifier(&self) -> &EventFd { - &self.eventfd + self.completions.notifier() } /// Submits one owned read or write operation to the queue. @@ -86,7 +83,7 @@ impl UringDataIo { Err(e) => { self.needs_submit_retry = true; warn!("io_uring submit failed after SQE was published: {e}"); - self.eventfd.write(1).unwrap(); + self.completions.notifier().write(1).unwrap(); } } @@ -112,10 +109,9 @@ impl UringDataIo { // 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.completions + .complete(AsyncIoCompletion::from_operation(op, -libc::EAGAIN)); } - self.eventfd.write(1).unwrap(); return Ok(()); } @@ -132,7 +128,7 @@ impl UringDataIo { if let Err(e) = unsafe { sq.push(&entry) } { Self::handle_push_failure( &mut self.in_flight, - &mut self.injected, + &mut self.completions, user_data, batch.by_ref(), &e, @@ -152,7 +148,7 @@ impl UringDataIo { } } if signal_completion { - self.eventfd.write(1).unwrap(); + self.completions.notifier().write(1).unwrap(); } Ok(()) @@ -161,7 +157,7 @@ impl UringDataIo { #[cold] fn handle_push_failure( in_flight: &mut HashMap>, - injected: &mut VecDeque, + completions: &mut CompletionCommon, user_data: u64, remaining: impl Iterator, error: &squeue::PushError, @@ -173,9 +169,9 @@ impl UringDataIo { .remove(&user_data) .flatten() .expect("pending operation missing after failed push"); - injected.push_back(AsyncIoCompletion::from_operation(op, -libc::EAGAIN)); + completions.complete(AsyncIoCompletion::from_operation(op, -libc::EAGAIN)); for op in remaining { - injected.push_back(AsyncIoCompletion::from_operation(op, -libc::EAGAIN)); + completions.complete(AsyncIoCompletion::from_operation(op, -libc::EAGAIN)); } warn!("io_uring submission queue became full after capacity check: {error:?}"); } @@ -216,8 +212,7 @@ impl UringDataIo { /// 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(); + self.completions.complete(completion); } /// Returns the next kernel or injected completion if one is available. @@ -244,7 +239,7 @@ impl UringDataIo { )); } - self.injected.pop_front() + self.completions.next_completed() } }