From 331b2bba53bbcacfa8503c101899263870d4a065 Mon Sep 17 00:00:00 2001 From: Anatol Belski Date: Tue, 21 Jul 2026 07:58:02 +0200 Subject: [PATCH] block: Add a SyncCompletionQueue for the sync engines The raw, qcow, and vhdx sync engines each carried an EventFd and a VecDeque of completions and repeated the same push and signal idiom at every completion site. Bundle the two into a SyncCompletionQueue with complete, notifier, and next_completed, and route the three engines through it. The repeated eventfd signal now lives in one place. Signed-off-by: Anatol Belski --- block/src/formats/qcow/engine_sync.rs | 46 +++++++++++---------------- block/src/formats/raw/engine_sync.rs | 35 +++++++++----------- block/src/formats/vhdx/engine_sync.rs | 26 +++++++-------- block/src/io/async_io.rs | 1 + block/src/io/async_io/completion.rs | 36 +++++++++++++++++++++ 5 files changed, 82 insertions(+), 62 deletions(-) diff --git a/block/src/formats/qcow/engine_sync.rs b/block/src/formats/qcow/engine_sync.rs index e406f7bbe..c9e3f81fb 100644 --- a/block/src/formats/qcow/engine_sync.rs +++ b/block/src/formats/qcow/engine_sync.rs @@ -5,7 +5,6 @@ // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause use std::cmp::min; -use std::collections::VecDeque; use std::os::unix::fs::FileExt; use std::sync::Arc; @@ -18,7 +17,9 @@ use super::metadata::{ BackingRead, ClusterReadMapping, ClusterWriteMapping, DeallocAction, QcowMetadata, }; use super::qcow_raw_file::QcowRawFile; -use crate::async_io::{AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult}; +use crate::async_io::{ + AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult, SyncCompletionQueue, +}; pub(super) struct QcowSync { metadata: Arc, @@ -28,8 +29,7 @@ pub(super) struct QcowSync { sparse: bool, cluster_size: u64, decoder: Arc, - eventfd: EventFd, - completion_list: VecDeque, + completions: SyncCompletionQueue, } impl QcowSync { @@ -46,9 +46,7 @@ impl QcowSync { data_file, backing_file, sparse, - eventfd: EventFd::new(libc::EFD_NONBLOCK) - .expect("Failed creating EventFd for QcowSync"), - completion_list: VecDeque::new(), + completions: SyncCompletionQueue::new(), } } @@ -200,7 +198,7 @@ impl QcowSync { impl AsyncIo for QcowSync { fn notifier(&self) -> &EventFd { - &self.eventfd + self.completions.notifier() } fn submit_data_operation(&mut self, mut op: AsyncIoOperation) -> AsyncIoResult<()> { @@ -210,24 +208,22 @@ impl AsyncIo for QcowSync { } else { self.write_operation(&op)? }; - self.completion_list - .push_back(AsyncIoCompletion::from_operation(op, total_len as i32)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::from_operation(op, total_len as i32)); Ok(()) } fn fsync(&mut self, user_data: Option) -> AsyncIoResult<()> { self.metadata.flush().map_err(AsyncIoError::Fsync)?; if let Some(user_data) = user_data { - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, 0, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, 0, None)); } Ok(()) } fn next_completed_request(&mut self) -> Option { - self.completion_list.pop_front() + self.completions.next_completed() } fn punch_hole(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> { @@ -247,9 +243,8 @@ impl AsyncIo for QcowSync { for action in &actions { self.apply_dealloc_action(action); } - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, 0, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, 0, None)); Ok(()) } Err(e) => { @@ -258,9 +253,8 @@ impl AsyncIo for QcowSync { } else { -libc::EIO }; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, errno, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, errno, None)); Ok(()) } } @@ -283,9 +277,8 @@ impl AsyncIo for QcowSync { for action in &actions { self.apply_dealloc_action(action); } - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, 0, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, 0, None)); Ok(()) } Err(e) => { @@ -294,9 +287,8 @@ impl AsyncIo for QcowSync { } else { -libc::EIO }; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, errno, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, errno, None)); Ok(()) } } diff --git a/block/src/formats/raw/engine_sync.rs b/block/src/formats/raw/engine_sync.rs index ba0a72aef..1496738d6 100644 --- a/block/src/formats/raw/engine_sync.rs +++ b/block/src/formats/raw/engine_sync.rs @@ -4,20 +4,20 @@ // // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause -use std::collections::VecDeque; use std::io; use std::os::unix::io::AsRawFd; use vmm_sys_util::eventfd::EventFd; -use crate::async_io::{AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult}; +use crate::async_io::{ + AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult, SyncCompletionQueue, +}; use crate::sparse::{punch_hole, write_zeroes}; use crate::{AlignedFile, is_block_device}; pub(crate) struct RawSync { raw_file: AlignedFile, - eventfd: EventFd, - completion_list: VecDeque, + completions: SyncCompletionQueue, alignment: u64, is_block_device: bool, } @@ -28,8 +28,7 @@ impl RawSync { let alignment = raw_file.alignment() as u64; RawSync { raw_file, - eventfd: EventFd::new(libc::EFD_NONBLOCK).expect("Failed creating EventFd for RawFile"), - completion_list: VecDeque::new(), + completions: SyncCompletionQueue::new(), alignment, is_block_device, } @@ -38,7 +37,7 @@ impl RawSync { impl AsyncIo for RawSync { fn notifier(&self) -> &EventFd { - &self.eventfd + self.completions.notifier() } fn alignment(&self) -> u64 { @@ -61,9 +60,8 @@ impl AsyncIo for RawSync { unsafe { self.raw_file.write_vectored_at(iovecs, offset) } .map_err(AsyncIoError::WriteVectored)? } as i32; - self.completion_list - .push_back(AsyncIoCompletion::from_operation(op, result)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::from_operation(op, result)); Ok(()) } @@ -76,33 +74,30 @@ impl AsyncIo for RawSync { } if let Some(user_data) = user_data { - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, result, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, result, None)); } Ok(()) } fn next_completed_request(&mut self) -> Option { - self.completion_list.pop_front() + self.completions.next_completed() } fn punch_hole(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> { punch_hole(&mut self.raw_file, self.is_block_device, offset, length) .map_err(AsyncIoError::PunchHole)?; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, 0, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, 0, None)); Ok(()) } fn write_zeroes(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> { write_zeroes(&mut self.raw_file, self.is_block_device, offset, length) .map_err(AsyncIoError::WriteZeroes)?; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, 0, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, 0, None)); Ok(()) } } diff --git a/block/src/formats/vhdx/engine_sync.rs b/block/src/formats/vhdx/engine_sync.rs index 292a4d442..8e3c40971 100644 --- a/block/src/formats/vhdx/engine_sync.rs +++ b/block/src/formats/vhdx/engine_sync.rs @@ -4,19 +4,19 @@ // // SPDX-License-Identifier: Apache-2.0 -use std::collections::VecDeque; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::sync::{Arc, Mutex}; use vmm_sys_util::eventfd::EventFd; -use crate::async_io::{AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult}; +use crate::async_io::{ + AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult, SyncCompletionQueue, +}; use crate::formats::vhdx::Vhdx; pub(super) struct VhdxSync { vhdx_file: Arc>, - eventfd: EventFd, - completion_list: VecDeque, + completions: SyncCompletionQueue, size: u64, } @@ -24,9 +24,7 @@ impl VhdxSync { pub(super) fn new(vhdx_file: Arc>, size: u64) -> Self { VhdxSync { vhdx_file, - eventfd: EventFd::new(libc::EFD_NONBLOCK) - .expect("Failed creating EventFd for VhdxSync"), - completion_list: VecDeque::new(), + completions: SyncCompletionQueue::new(), size, } } @@ -61,7 +59,7 @@ impl VhdxSync { impl AsyncIo for VhdxSync { fn notifier(&self) -> &EventFd { - &self.eventfd + self.completions.notifier() } fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { @@ -74,9 +72,8 @@ impl AsyncIo for VhdxSync { self.write_operation(&op)? }; - self.completion_list - .push_back(AsyncIoCompletion::from_operation(op, result as i32)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::from_operation(op, result as i32)); Ok(()) } @@ -87,15 +84,14 @@ impl AsyncIo for VhdxSync { .flush() .map_err(AsyncIoError::Fsync)?; if let Some(user_data) = user_data { - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, 0, None)); - self.eventfd.write(1).unwrap(); + self.completions + .complete(AsyncIoCompletion::new(user_data, 0, None)); } Ok(()) } fn next_completed_request(&mut self) -> Option { - self.completion_list.pop_front() + self.completions.next_completed() } fn punch_hole(&mut self, _offset: u64, _length: u64, _user_data: u64) -> AsyncIoResult<()> { diff --git a/block/src/io/async_io.rs b/block/src/io/async_io.rs index ffe3a3ec8..a8b4c7703 100644 --- a/block/src/io/async_io.rs +++ b/block/src/io/async_io.rs @@ -17,6 +17,7 @@ use std::{io, result}; pub use aio_data_io::AioDataIo; pub use completion::AsyncIoCompletion; +pub(crate) use completion::SyncCompletionQueue; pub use guest_memory_target::GuestMemoryTarget; pub use operation::AsyncIoOperation; pub use owned_io_buffer::OwnedIoBuffer; diff --git a/block/src/io/async_io/completion.rs b/block/src/io/async_io/completion.rs index 3f07c9e29..01b664b9f 100644 --- a/block/src/io/async_io/completion.rs +++ b/block/src/io/async_io/completion.rs @@ -2,6 +2,10 @@ // // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause +use std::collections::VecDeque; + +use vmm_sys_util::eventfd::EventFd; + use super::{AsyncIoOperation, OwnedIoBuffer}; /// Completion returned by an owned async I/O backend. @@ -40,3 +44,35 @@ impl AsyncIoCompletion { Self::new(user_data, result, op.into_completion_buffer()) } } + +/// 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. +pub(crate) struct SyncCompletionQueue { + queue: VecDeque, + eventfd: EventFd, +} + +impl SyncCompletionQueue { + pub(crate) fn new() -> Self { + Self { + queue: VecDeque::new(), + eventfd: EventFd::new(libc::EFD_NONBLOCK) + .expect("Failed creating EventFd for sync completion queue"), + } + } + + pub(crate) fn notifier(&self) -> &EventFd { + &self.eventfd + } + + /// Enqueues a completion and signals the eventfd. + pub(crate) fn complete(&mut self, completion: AsyncIoCompletion) { + self.queue.push_back(completion); + self.eventfd.write(1).unwrap(); + } + + pub(crate) fn next_completed(&mut self) -> Option { + self.queue.pop_front() + } +}