From 1b2326fde4526f7dfd171d85aa7efe37212807f8 Mon Sep 17 00:00:00 2001 From: Dylan Reid Date: Tue, 19 May 2026 14:25:12 -0700 Subject: [PATCH] block: Remove legacy async I/O API Drop the borrowed iovec AsyncIo entry points now that all callers use owned operations. Rename the transitional owned batch and completion methods to the final trait names and remove the borrowed submission helpers from the queue wrappers. This removes a bunch of known safety foot-guns so future-us don't accidentally use them. Assisted-by: Codex:GPT-5.5 Signed-off-by: Dylan Reid --- block/src/aligned_operation.rs | 90 ----------- block/src/async_io.rs | 43 +----- block/src/async_io/aio_data_io.rs | 51 +------ block/src/async_io/uring_data_io.rs | 98 +----------- block/src/fixed_vhd_async.rs | 55 +------ block/src/fixed_vhd_disk.rs | 2 +- block/src/fixed_vhd_sync.rs | 42 +----- block/src/lib.rs | 117 +-------------- block/src/qcow_async.rs | 149 +------------------ block/src/qcow_sync.rs | 48 +----- block/src/raw_async.rs | 56 +------ block/src/raw_async_aio.rs | 24 +-- block/src/raw_async_io_tests.rs | 2 +- block/src/raw_sync.rs | 54 +------ block/src/request.rs | 6 - block/src/vhdx_sync.rs | 32 +--- performance-metrics/src/micro_bench_block.rs | 8 +- performance-metrics/src/util.rs | 4 +- virtio-devices/src/block.rs | 4 +- 19 files changed, 54 insertions(+), 831 deletions(-) delete mode 100644 block/src/aligned_operation.rs diff --git a/block/src/aligned_operation.rs b/block/src/aligned_operation.rs deleted file mode 100644 index 6096a4f93..000000000 --- a/block/src/aligned_operation.rs +++ /dev/null @@ -1,90 +0,0 @@ -// Copyright (c) 2026 Meta Platforms, Inc. and affiliates. -// -// SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause - -use std::alloc::{Layout, alloc_zeroed, dealloc}; -use std::io; - -use vm_memory::GuestAddress; - -/// Owns an aligned bounce buffer used when a guest descriptor's host VA -/// does not meet the disk backend's alignment requirement. -#[derive(Debug)] -pub struct AlignedOperation { - data_addr: GuestAddress, - aligned_ptr: *mut u8, - size: usize, - layout: Layout, -} - -impl AlignedOperation { - /// Allocate a zero-initialized buffer of `size` bytes aligned to - /// `alignment`. Returns `InvalidInput` if `size` is zero; - /// `alignment` must be a power of two and not exceed `isize::MAX` - /// after rounding up. - pub fn new(data_addr: GuestAddress, size: usize, alignment: usize) -> io::Result { - if size == 0 { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "AlignedOperation requires a non-zero size", - )); - } - let layout = Layout::from_size_align(size, alignment) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - // SAFETY: size is non-zero (checked above) and Layout::from_size_align - // rejects alignments that are not a power of two or that overflow. - let aligned_ptr = unsafe { alloc_zeroed(layout) }; - if aligned_ptr.is_null() { - return Err(io::Error::last_os_error()); - } - Ok(Self { - data_addr, - aligned_ptr, - size, - layout, - }) - } - - /// Gets the raw pointer to the aligned buffer. - pub fn as_mut_ptr(&mut self) -> *mut u8 { - self.aligned_ptr - } - - /// Returns the aligned buffer as a slice. - pub fn as_bytes(&self) -> &[u8] { - // SAFETY: `new` allocates `size` bytes via alloc_zeroed (so they - // are initialized) and AlignedOperation owns the buffer - // exclusively. - unsafe { std::slice::from_raw_parts(self.aligned_ptr, self.size) } - } - - /// Returns the aligned buffer as a mutable slice. - pub fn as_bytes_mut(&mut self) -> &mut [u8] { - // SAFETY: same invariant as as_bytes; &mut self rules out other - // simultaneous borrows. - unsafe { std::slice::from_raw_parts_mut(self.aligned_ptr, self.size) } - } - - /// Returns the guest address for this op. - pub fn data_addr(&self) -> GuestAddress { - self.data_addr - } -} - -impl Drop for AlignedOperation { - fn drop(&mut self) { - // SAFETY: `new` is the only constructor, and it stores a pointer - // returned by `alloc_zeroed` paired with the exact `layout` used - // for that allocation. Ownership has not escaped (the type is - // neither `Clone` nor `Copy`). - unsafe { - dealloc(self.aligned_ptr, self.layout); - } - } -} - -// SAFETY: AlignedOperation owns its heap allocation exclusively (no Clone/ -// Copy, no shared aliases) and the allocation's lifetime is tied to the -// value's. Moving an AlignedOperation between threads transfers that -// ownership — the same rationale Box uses for its Send impl. -unsafe impl Send for AlignedOperation {} diff --git a/block/src/async_io.rs b/block/src/async_io.rs index eb97e8b05..64e7a742d 100644 --- a/block/src/async_io.rs +++ b/block/src/async_io.rs @@ -25,7 +25,7 @@ use thiserror::Error; pub use uring_data_io::UringDataIo; use vmm_sys_util::eventfd::EventFd; -use crate::{BatchRequest, SECTOR_SIZE}; +use crate::SECTOR_SIZE; #[derive(Error, Debug)] pub enum DiskFileError { @@ -101,33 +101,13 @@ pub type AsyncIoResult = std::result::Result; pub trait AsyncIo: Send { fn notifier(&self) -> &EventFd; - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()>; - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()>; /// 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<()> { - let error = - io::Error::other("owned async I/O operations are not supported by this backend"); - if op.is_read() { - Err(AsyncIoError::ReadVectored(error)) - } else { - Err(AsyncIoError::WriteVectored(error)) - } - } + fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()>; /// Submits a read from `offset` into guest memory. fn read_to_memory( @@ -178,32 +158,17 @@ pub trait AsyncIo: Send { /// Returns the next owned completion, if one is available. /// /// Read completions from owned host-memory buffers return that buffer here. - fn next_completion(&mut self) -> Option { - self.next_completed_request() - .map(|(user_data, result)| AsyncIoCompletion::new(user_data, result, None)) - } - - fn next_completed_request(&mut self) -> Option<(u64, i32)> { - self.next_completion() - .map(|completion| (completion.user_data, completion.result)) - } + fn next_completed_request(&mut self) -> Option; fn batch_requests_enabled(&self) -> bool { false } - fn submit_batch_requests(&mut self, _batch_request: &[BatchRequest]) -> AsyncIoResult<()> { - Ok(()) - } - /// 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_operations( - &mut self, - batch_request: Vec, - ) -> AsyncIoResult<()> { + fn submit_batch_requests(&mut self, batch_request: Vec) -> AsyncIoResult<()> { if batch_request.is_empty() { Ok(()) } else { diff --git a/block/src/async_io/aio_data_io.rs b/block/src/async_io/aio_data_io.rs index d0a97f099..3072a4e40 100644 --- a/block/src/async_io/aio_data_io.rs +++ b/block/src/async_io/aio_data_io.rs @@ -27,8 +27,7 @@ pub struct AioDataIo { 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 and legacy borrowed operations - // store `None`. + // 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. @@ -105,54 +104,6 @@ impl AioDataIo { Ok(()) } - /// Submits one borrowed read or write operation to the queue. - /// - /// This is only for the legacy `AsyncIo` interface. The caller must keep - /// the iovec array and the buffers it references alive until the matching - /// completion is consumed. - pub fn submit_borrowed_operation( - &mut self, - fd: RawFd, - offset: libc::off_t, - is_read: bool, - iovecs: &[libc::iovec], - 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); - - let opcode = if 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: offset, - aio_data: user_data, - aio_flags: aio::IOCB_FLAG_RESFD, - aio_resfd: self.eventfd.as_raw_fd() as u32, - ..Default::default() - }; - - match Self::submit_iocbs(&self.ctx, &[&mut iocb]) { - Ok(1) => Ok(()), - Ok(_) => { - self.in_flight.remove(&user_data); - Err(io::Error::from_raw_os_error(libc::EAGAIN)) - } - Err(e) => { - self.in_flight.remove(&user_data); - Err(e) - } - } - } - /// 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) { diff --git a/block/src/async_io/uring_data_io.rs b/block/src/async_io/uring_data_io.rs index 2857e429f..97e4199e7 100644 --- a/block/src/async_io/uring_data_io.rs +++ b/block/src/async_io/uring_data_io.rs @@ -4,7 +4,7 @@ // // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, VecDeque}; use std::io; use std::os::fd::{AsRawFd, RawFd}; @@ -27,8 +27,7 @@ pub struct UringDataIo { 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 and legacy borrowed operations - // store `None`. + // 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. @@ -64,29 +63,6 @@ impl UringDataIo { self.submit_batch(fd, vec![op]) } - /// Submits one borrowed read or write operation to the queue. - /// - /// This is only for the legacy `AsyncIo` interface. The caller must keep - /// the iovec array and the buffers it references alive until the matching - /// completion is consumed. - /// - /// # Safety - /// - /// The caller must guarantee that `iovecs` and all buffers described by - /// them remain valid for kernel access until the completion carrying - /// `user_data` is consumed. - pub unsafe fn submit_borrowed_operation( - &mut self, - fd: RawFd, - offset: libc::off_t, - is_read: bool, - iovecs: &[libc::iovec], - user_data: u64, - ) -> io::Result<()> { - let entry = Self::build_borrowed_entry(fd, offset, is_read, iovecs, user_data); - self.submit_kernel_entry(user_data, &entry) - } - 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)); @@ -96,56 +72,6 @@ impl UringDataIo { Ok(()) } - /// Submits a batch of borrowed read and write operations to the queue. - /// - /// This is only for the legacy `AsyncIo` interface. The caller must keep - /// every iovec array and referenced buffer alive until its matching - /// completion is consumed. - /// - /// # Safety - /// - /// The caller must guarantee that every borrowed iovec array and every - /// buffer described by those arrays remains valid for kernel access until - /// the matching completion is consumed. - pub unsafe fn submit_borrowed_batch( - &mut self, - fd: RawFd, - batch: &[(libc::off_t, bool, &[libc::iovec], u64)], - ) -> io::Result<()> { - if batch.is_empty() { - return Ok(()); - } - - let mut seen = HashSet::with_capacity(batch.len()); - for &(_, _, _, user_data) in batch { - if self.in_flight.contains_key(&user_data) || !seen.insert(user_data) { - return Err(duplicate_user_data_error(user_data)); - } - } - - let (submitter, mut sq, _) = self.io_uring.split(); - let available = sq.capacity() - sq.len(); - if batch.len() > available { - return Err(io::Error::other("io_uring submission queue is full")); - } - - for &(offset, is_read, iovecs, user_data) in batch { - let entry = Self::build_borrowed_entry(fd, offset, is_read, iovecs, user_data); - self.in_flight.insert(user_data, None); - - // SAFETY: capacity was checked above, and the legacy caller keeps - // the borrowed iovec storage alive until completion. - 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(); - submitter.submit()?; - Ok(()) - } - fn submit_kernel_entry(&mut self, user_data: u64, entry: &squeue::Entry) -> io::Result<()> { self.reserve_user_data(user_data)?; @@ -259,27 +185,17 @@ impl UringDataIo { fn build_entry(fd: RawFd, op: &AsyncIoOperation) -> squeue::Entry { let iovecs = op.iovecs(); - Self::build_borrowed_entry(fd, op.offset(), op.is_read(), iovecs, op.user_data()) - } - - fn build_borrowed_entry( - fd: RawFd, - offset: libc::off_t, - is_read: bool, - iovecs: &[libc::iovec], - user_data: u64, - ) -> squeue::Entry { let fd = types::Fd(fd); - if is_read { + if op.is_read() { opcode::Readv::new(fd, iovecs.as_ptr(), iovecs.len() as u32) - .offset(offset as u64) + .offset(op.offset() as u64) .build() - .user_data(user_data) + .user_data(op.user_data()) } else { opcode::Writev::new(fd, iovecs.as_ptr(), iovecs.len() as u32) - .offset(offset as u64) + .offset(op.offset() as u64) .build() - .user_data(user_data) + .user_data(op.user_data()) } } diff --git a/block/src/fixed_vhd_async.rs b/block/src/fixed_vhd_async.rs index 975ac6be4..38a3d2c79 100644 --- a/block/src/fixed_vhd_async.rs +++ b/block/src/fixed_vhd_async.rs @@ -8,7 +8,6 @@ use std::os::unix::io::RawFd; use vmm_sys_util::eventfd::EventFd; -use crate::BatchRequest; use crate::async_io::{AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult}; use crate::error::BlockResult; use crate::raw_async::RawFileAsync; @@ -65,45 +64,6 @@ impl AsyncIo for FixedVhdAsync { self.raw_file_async.notifier() } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - if offset as u64 >= self.size { - return Err(AsyncIoError::ReadVectored(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!( - "Invalid offset {}, can't be larger than file size {}", - offset, self.size - ), - ))); - } - - self.raw_file_async.read_vectored(offset, iovecs, user_data) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - if offset as u64 >= self.size { - return Err(AsyncIoError::WriteVectored(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!( - "Invalid offset {}, can't be larger than file size {}", - offset, self.size - ), - ))); - } - - self.raw_file_async - .write_vectored(offset, iovecs, user_data) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { self.validate_operation_bounds(&op)?; self.raw_file_async.submit_data_operation(op) @@ -113,8 +73,8 @@ impl AsyncIo for FixedVhdAsync { self.raw_file_async.fsync(user_data) } - fn next_completion(&mut self) -> Option { - self.raw_file_async.next_completion() + fn next_completed_request(&mut self) -> Option { + self.raw_file_async.next_completed_request() } fn punch_hole(&mut self, _offset: u64, _length: u64, _user_data: u64) -> AsyncIoResult<()> { @@ -133,18 +93,11 @@ impl AsyncIo for FixedVhdAsync { true } - fn submit_batch_requests(&mut self, batch_request: &[BatchRequest]) -> AsyncIoResult<()> { - self.raw_file_async.submit_batch_requests(batch_request) - } - - fn submit_batch_operations( - &mut self, - batch_request: Vec, - ) -> AsyncIoResult<()> { + fn submit_batch_requests(&mut self, batch_request: Vec) -> AsyncIoResult<()> { for op in &batch_request { self.validate_operation_bounds(op)?; } - self.raw_file_async.submit_batch_operations(batch_request) + self.raw_file_async.submit_batch_requests(batch_request) } } diff --git a/block/src/fixed_vhd_disk.rs b/block/src/fixed_vhd_disk.rs index d8c0332be..854ccb6ab 100644 --- a/block/src/fixed_vhd_disk.rs +++ b/block/src/fixed_vhd_disk.rs @@ -203,7 +203,7 @@ mod unit_tests { let op = AsyncIoOperation::read_to_vec(0x800, OwnedIoBuffer::from_vec(vec![0; 0x900]), 1); assert!(matches!( - async_io.submit_batch_operations(vec![op]), + async_io.submit_batch_requests(vec![op]), Err(crate::async_io::AsyncIoError::ReadVectored(_)) )); } diff --git a/block/src/fixed_vhd_sync.rs b/block/src/fixed_vhd_sync.rs index 57bc15ac5..e422fc812 100644 --- a/block/src/fixed_vhd_sync.rs +++ b/block/src/fixed_vhd_sync.rs @@ -30,44 +30,6 @@ impl AsyncIo for FixedVhdSync { self.raw_file_sync.notifier() } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - if offset as u64 >= self.size { - return Err(AsyncIoError::ReadVectored(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!( - "Invalid offset {}, can't be larger than file size {}", - offset, self.size - ), - ))); - } - - self.raw_file_sync.read_vectored(offset, iovecs, user_data) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - if offset as u64 >= self.size { - return Err(AsyncIoError::WriteVectored(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!( - "Invalid offset {}, can't be larger than file size {}", - offset, self.size - ), - ))); - } - - self.raw_file_sync.write_vectored(offset, iovecs, user_data) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { let offset = op.offset(); if offset as u64 >= self.size { @@ -92,8 +54,8 @@ impl AsyncIo for FixedVhdSync { self.raw_file_sync.fsync(user_data) } - fn next_completion(&mut self) -> Option { - self.raw_file_sync.next_completion() + fn next_completed_request(&mut self) -> Option { + self.raw_file_sync.next_completed_request() } fn punch_hole(&mut self, _offset: u64, _length: u64, _user_data: u64) -> AsyncIoResult<()> { diff --git a/block/src/lib.rs b/block/src/lib.rs index dfeec593c..972e3e559 100644 --- a/block/src/lib.rs +++ b/block/src/lib.rs @@ -8,7 +8,6 @@ // // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause -mod aligned_operation; pub mod async_io; pub mod disk_file; pub mod error; @@ -41,10 +40,9 @@ pub mod vhdx; pub mod vhdx_sync; use std::alloc::{Layout, alloc_zeroed}; -use std::collections::VecDeque; use std::fmt::{self, Debug}; use std::fs::{File, OpenOptions}; -use std::io::{self, IoSlice, IoSliceMut, Read, Seek, SeekFrom, Write}; +use std::io::{self, Read, Seek, Write}; use std::os::linux::fs::MetadataExt; use std::os::unix::fs::FileTypeExt; use std::os::unix::io::{AsRawFd, RawFd}; @@ -52,26 +50,23 @@ use std::path::Path; use std::str::FromStr; use std::{cmp, mem, result}; -pub use aligned_operation::AlignedOperation; #[cfg(feature = "io_uring")] use io_uring::{IoUring, Probe, opcode}; use libc::{ FALLOC_FL_KEEP_SIZE, FALLOC_FL_PUNCH_HOLE, FALLOC_FL_ZERO_RANGE, S_IFBLK, S_IFMT, ioctl, }; use log::{debug, info, warn}; -pub use request::{BatchRequest, ExecuteAsync, MAX_DISCARD_WRITE_ZEROES_SEG, Request, RequestType}; +pub use request::{ExecuteAsync, MAX_DISCARD_WRITE_ZEROES_SEG, Request, RequestType}; use serde::{Deserialize, Serialize}; -use smallvec::SmallVec; use thiserror::Error; use virtio_bindings::virtio_blk::*; use vm_memory::bitmap::Bitmap; use vm_memory::{ByteValued, Bytes, GuestAddress, GuestMemory, GuestMemoryError}; -use vmm_sys_util::eventfd::EventFd; use vmm_sys_util::{aio, ioctl_io_nr, ioctl_ior_nr}; -use crate::async_io::{AsyncIoError, AsyncIoResult}; +use crate::async_io::AsyncIoError; use crate::error::{BlockError, BlockErrorKind, BlockResult, ErrorOp}; -use crate::request::{DEFAULT_DESCRIPTOR_VEC_SIZE, SECTOR_SIZE}; +use crate::request::SECTOR_SIZE; use crate::vhdx::VhdxError; #[derive(Error, Debug)] @@ -453,110 +448,6 @@ pub fn preallocate_disk>(file: &File, path: P) { } } -pub trait AsyncAdaptor { - fn read_vectored_sync( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - eventfd: &EventFd, - completion_list: &mut VecDeque<(u64, i32)>, - ) -> AsyncIoResult<()> - where - Self: Read + Seek, - { - // Convert libc::iovec into IoSliceMut - let mut slices: SmallVec<[IoSliceMut; DEFAULT_DESCRIPTOR_VEC_SIZE]> = - SmallVec::with_capacity(iovecs.len()); - for iovec in iovecs.iter() { - // SAFETY: on Linux IoSliceMut wraps around libc::iovec - slices.push(IoSliceMut::new(unsafe { - std::mem::transmute::(*iovec) - })); - } - - let result = { - // Move the cursor to the right offset - self.seek(SeekFrom::Start(offset as u64)) - .map_err(AsyncIoError::ReadVectored)?; - - let mut r = 0; - for b in slices.iter_mut() { - r += self.read(b).map_err(AsyncIoError::ReadVectored)?; - } - r - }; - - completion_list.push_back((user_data, result as i32)); - eventfd.write(1).unwrap(); - - Ok(()) - } - - fn write_vectored_sync( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - eventfd: &EventFd, - completion_list: &mut VecDeque<(u64, i32)>, - ) -> AsyncIoResult<()> - where - Self: Write + Seek, - { - // Convert libc::iovec into IoSlice - let mut slices: SmallVec<[IoSlice; DEFAULT_DESCRIPTOR_VEC_SIZE]> = - SmallVec::with_capacity(iovecs.len()); - for iovec in iovecs.iter() { - // SAFETY: on Linux IoSlice wraps around libc::iovec - slices.push(IoSlice::new(unsafe { - std::mem::transmute::(*iovec) - })); - } - - let result = { - // Move the cursor to the right offset - self.seek(SeekFrom::Start(offset as u64)) - .map_err(AsyncIoError::WriteVectored)?; - - let mut r = 0; - for b in slices.iter() { - r += self.write(b).map_err(AsyncIoError::WriteVectored)?; - } - r - }; - - completion_list.push_back((user_data, result as i32)); - eventfd.write(1).unwrap(); - - Ok(()) - } - - fn fsync_sync( - &mut self, - user_data: Option, - eventfd: &EventFd, - completion_list: &mut VecDeque<(u64, i32)>, - ) -> AsyncIoResult<()> - where - Self: Write, - { - let result: i32 = { - // Flush - self.flush().map_err(AsyncIoError::Fsync)?; - - 0 - }; - - if let Some(user_data) = user_data { - completion_list.push_back((user_data, result)); - eventfd.write(1).unwrap(); - } - - Ok(()) - } -} - #[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default)] pub enum ImageType { FixedVhd, diff --git a/block/src/qcow_async.rs b/block/src/qcow_async.rs index 680aa8f77..c4db510ab 100644 --- a/block/src/qcow_async.rs +++ b/block/src/qcow_async.rs @@ -16,6 +16,7 @@ use std::sync::Arc; use vmm_sys_util::eventfd::EventFd; use vmm_sys_util::write_zeroes::{PunchHole, WriteZeroesAt}; +use crate::SECTOR_SIZE; use crate::async_io::{ AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult, UringDataIo, }; @@ -28,7 +29,6 @@ use crate::qcow_common::{ AlignedBuf, aligned_pread, aligned_pwrite, decompress_cluster, gather_from_iovecs_into, pread_alloc, pread_exact, pwrite_all, scatter_to_iovecs, zero_fill_iovecs, }; -use crate::{BatchRequest, RequestType, SECTOR_SIZE}; /// Per queue QCOW2 I/O worker using io_uring. /// @@ -175,69 +175,6 @@ impl AsyncIo for QcowAsync { self.data_io.notifier() } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - let total_len: usize = iovecs.iter().map(|v| v.iov_len).sum(); - - if let Some(host_offset) = Self::resolve_read( - &self.metadata, - &self.data_file, - &self.backing_file, - offset as u64, - iovecs, - total_len, - self.alignment, - self.cluster_size, - &*self.decoder, - )? { - // SAFETY: this legacy trait method's caller must keep the - // borrowed iovecs and writable buffers valid until completion. - unsafe { - self.data_io.submit_borrowed_operation( - self.data_file.as_raw_fd(), - host_offset as libc::off_t, - true, - iovecs, - user_data, - ) - } - .map_err(AsyncIoError::ReadVectored)?; - } else { - self.data_io.inject_completion(AsyncIoCompletion::new( - user_data, - total_len as i32, - None, - )); - } - Ok(()) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - Self::cow_write_sync( - offset as u64, - iovecs, - &self.metadata, - &self.data_file, - &self.backing_file, - self.alignment, - self.cluster_size, - )?; - - let total_len: usize = iovecs.iter().map(|v| v.iov_len).sum(); - self.data_io - .inject_completion(AsyncIoCompletion::new(user_data, total_len as i32, None)); - Ok(()) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { if op.is_read() { match self.prepare_read_operation(op) { @@ -264,7 +201,7 @@ impl AsyncIo for QcowAsync { Ok(()) } - fn next_completion(&mut self) -> Option { + fn next_completed_request(&mut self) -> Option { self.data_io.next_completion() } @@ -344,77 +281,7 @@ impl AsyncIo for QcowAsync { self.io_alignment } - fn submit_batch_requests(&mut self, batch_request: &[BatchRequest]) -> AsyncIoResult<()> { - let mut async_reads = Vec::new(); - - for req in batch_request { - match req.request_type { - RequestType::In => { - let total_len: usize = req.iovecs.iter().map(|v| v.iov_len).sum(); - - if let Some(host_offset) = Self::resolve_read( - &self.metadata, - &self.data_file, - &self.backing_file, - req.offset as u64, - &req.iovecs, - total_len, - self.alignment, - self.cluster_size, - &*self.decoder, - )? { - async_reads.push(( - host_offset as libc::off_t, - true, - req.iovecs.as_slice(), - req.user_data, - )); - } else { - self.data_io.inject_completion(AsyncIoCompletion::new( - req.user_data, - total_len as i32, - None, - )); - } - } - RequestType::Out => { - let total_len: usize = req.iovecs.iter().map(|v| v.iov_len).sum(); - Self::cow_write_sync( - req.offset as u64, - &req.iovecs, - &self.metadata, - &self.data_file, - &self.backing_file, - self.alignment, - self.cluster_size, - )?; - self.data_io.inject_completion(AsyncIoCompletion::new( - req.user_data, - total_len as i32, - None, - )); - } - _ => unreachable!("Unexpected batch request type: {:?}", req.request_type), - } - } - - if !async_reads.is_empty() { - // SAFETY: this legacy trait method's caller must keep every - // borrowed iovec array and buffer valid until its completion. - unsafe { - self.data_io - .submit_borrowed_batch(self.data_file.as_raw_fd(), &async_reads) - } - .map_err(AsyncIoError::SubmitBatchRequests)?; - } - - Ok(()) - } - - fn submit_batch_operations( - &mut self, - batch_request: Vec, - ) -> AsyncIoResult<()> { + fn submit_batch_requests(&mut self, batch_request: Vec) -> AsyncIoResult<()> { let mut async_reads = Vec::new(); for op in batch_request { @@ -738,7 +605,7 @@ mod unit_tests { fn wait_for_completion(async_io: &mut dyn AsyncIo) -> AsyncIoCompletion { loop { - if let Some(c) = async_io.next_completion() { + if let Some(c) = async_io.next_completed_request() { return c; } // Block until the eventfd is signaled (io_uring or synthetic). @@ -801,7 +668,7 @@ mod unit_tests { let mut async_io = disk.create_async_io(1).unwrap(); async_io.punch_hole(offset, data.len() as u64, 100).unwrap(); - let completion = async_io.next_completion().unwrap(); + let completion = async_io.next_completed_request().unwrap(); let (user_data, result) = completion_tuple(&completion); assert_eq!(user_data, 100); assert_eq!(result, 0, "punch_hole should succeed"); @@ -824,7 +691,7 @@ mod unit_tests { async_io .write_zeroes(offset, data.len() as u64, 200) .unwrap(); - let completion = async_io.next_completion().unwrap(); + let completion = async_io.next_completed_request().unwrap(); let (user_data, result) = completion_tuple(&completion); assert_eq!(user_data, 200); assert_eq!(result, 0, "write_zeroes should succeed"); @@ -949,7 +816,7 @@ mod unit_tests { ), ]; - async_io.submit_batch_operations(batch).unwrap(); + async_io.submit_batch_requests(batch).unwrap(); let mut completions = [ completion_tuple(&wait_for_completion(async_io.as_mut())), @@ -975,7 +842,7 @@ mod unit_tests { ), ]; - async_io.submit_batch_operations(read_batch).unwrap(); + async_io.submit_batch_requests(read_batch).unwrap(); let mut completion_a = wait_for_completion(async_io.as_mut()); let mut completion_b = wait_for_completion(async_io.as_mut()); diff --git a/block/src/qcow_sync.rs b/block/src/qcow_sync.rs index 2afc8d6fc..b794a0e2c 100644 --- a/block/src/qcow_sync.rs +++ b/block/src/qcow_sync.rs @@ -253,38 +253,6 @@ impl AsyncIo for QcowSync { &self.eventfd } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: the legacy caller is responsible for keeping the borrowed - // iovecs valid for this synchronous call and for ensuring that turning - // their pointers into Rust slices does not violate aliasing rules. - let total_len = unsafe { self.read_iovecs(offset, iovecs)? }; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, total_len as i32, None)); - self.eventfd.write(1).unwrap(); - Ok(()) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: the legacy caller is responsible for keeping the borrowed - // iovecs valid for this synchronous call and for ensuring that turning - // their pointers into Rust slices does not violate aliasing rules. - let total_len = unsafe { self.write_iovecs(offset, iovecs)? }; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, total_len as i32, None)); - self.eventfd.write(1).unwrap(); - Ok(()) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { let offset = op.offset(); let is_read = op.is_read(); @@ -318,7 +286,7 @@ impl AsyncIo for QcowSync { Ok(()) } - fn next_completion(&mut self) -> Option { + fn next_completed_request(&mut self) -> Option { self.completion_list.pop_front() } @@ -518,7 +486,7 @@ mod unit_tests { } fn next_completion(async_io: &mut dyn AsyncIo) -> (u64, i32) { - completion_tuple(&async_io.next_completion().unwrap()) + completion_tuple(&async_io.next_completed_request().unwrap()) } fn async_read(disk: &QcowDisk, offset: u64, len: usize) -> Vec { @@ -530,7 +498,7 @@ mod unit_tests { 1, ) .unwrap(); - let mut completion = async_io.next_completion().unwrap(); + let mut completion = async_io.next_completed_request().unwrap(); let (user_data, result) = completion_tuple(&completion); assert_eq!(user_data, 1); assert_eq!(result as usize, len, "read should return requested length"); @@ -616,12 +584,12 @@ mod unit_tests { .unwrap(); let mut async_io = disk.create_async_io(1).unwrap(); async_io.write_zeroes(0, cluster_size, 200).unwrap(); - let (user_data, result) = async_io.next_completed_request().unwrap(); + let (user_data, result) = next_completion(async_io.as_mut()); assert_eq!(user_data, 200); assert_eq!(result, 0); async_io.fsync(Some(201)).unwrap(); - let (user_data, result) = async_io.next_completed_request().unwrap(); + let (user_data, result) = next_completion(async_io.as_mut()); assert_eq!(user_data, 201); assert_eq!(result, 0); drop(async_io); @@ -1042,7 +1010,7 @@ mod unit_tests { let mut async_io = disk.create_async_io(1).unwrap(); async_io.write_zeroes(offset, cluster_size, 42).unwrap(); - let (user_data, result) = async_io.next_completed_request().unwrap(); + let (user_data, result) = next_completion(async_io.as_mut()); assert_eq!(user_data, 42); assert_eq!(result, 0); drop(async_io); @@ -1077,7 +1045,7 @@ mod unit_tests { let mut async_io = disk.create_async_io(1).unwrap(); async_io.write_zeroes(offset, cluster_size, 42).unwrap(); - let (_user_data, result) = async_io.next_completed_request().unwrap(); + let (_user_data, result) = next_completion(async_io.as_mut()); assert_eq!(result, 0); drop(async_io); @@ -2012,7 +1980,7 @@ mod unit_tests { let mut aio = disk.create_async_io(1).unwrap(); aio.read_to_vec(0, OwnedIoBuffer::from_vec(vec![0; total]), 10) .unwrap(); - let mut completion = aio.next_completion().unwrap(); + let mut completion = aio.next_completed_request().unwrap(); let (ud, res) = completion_tuple(&completion); assert_eq!(ud, 10); assert_eq!(res as usize, total); diff --git a/block/src/raw_async.rs b/block/src/raw_async.rs index 813271d23..b19f576fb 100644 --- a/block/src/raw_async.rs +++ b/block/src/raw_async.rs @@ -14,7 +14,7 @@ use crate::async_io::{ }; use crate::error::{BlockError, BlockErrorKind, BlockResult}; use crate::sparse::{blkdiscard, blkzeroout}; -use crate::{BatchRequest, RequestType, SECTOR_SIZE, is_block_device}; +use crate::{SECTOR_SIZE, is_block_device}; pub struct RawFileAsync { fd: RawFd, @@ -47,36 +47,6 @@ impl AsyncIo for RawFileAsync { self.alignment } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: this legacy trait method's caller must keep the borrowed - // iovecs and writable buffers valid until completion. - unsafe { - self.data_io - .submit_borrowed_operation(self.fd, offset, true, iovecs, user_data) - } - .map_err(AsyncIoError::ReadVectored) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: this legacy trait method's caller must keep the borrowed - // iovecs and readable buffers valid until completion. - unsafe { - self.data_io - .submit_borrowed_operation(self.fd, offset, false, iovecs, user_data) - } - .map_err(AsyncIoError::WriteVectored) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { let is_read = op.is_read(); self.data_io.submit_operation(self.fd, op).map_err(|e| { @@ -101,7 +71,7 @@ impl AsyncIo for RawFileAsync { Ok(()) } - fn next_completion(&mut self) -> Option { + fn next_completed_request(&mut self) -> Option { self.data_io.next_completion() } @@ -109,27 +79,7 @@ impl AsyncIo for RawFileAsync { true } - fn submit_batch_requests(&mut self, batch_request: &[BatchRequest]) -> AsyncIoResult<()> { - let mut batch = Vec::with_capacity(batch_request.len()); - for req in batch_request { - let is_read = match req.request_type { - RequestType::In => true, - RequestType::Out => false, - _ => unreachable!("Unexpected batch request type: {:?}", req.request_type), - }; - batch.push((req.offset, is_read, req.iovecs.as_slice(), req.user_data)); - } - - // SAFETY: this legacy trait method's caller must keep every borrowed - // iovec array and buffer valid until its completion. - unsafe { self.data_io.submit_borrowed_batch(self.fd, &batch) } - .map_err(AsyncIoError::SubmitBatchRequests) - } - - fn submit_batch_operations( - &mut self, - batch_request: Vec, - ) -> AsyncIoResult<()> { + fn submit_batch_requests(&mut self, batch_request: Vec) -> AsyncIoResult<()> { self.data_io .submit_batch(self.fd, batch_request) .map_err(AsyncIoError::SubmitBatchRequests) diff --git a/block/src/raw_async_aio.rs b/block/src/raw_async_aio.rs index e910bf675..e9c35e4dc 100644 --- a/block/src/raw_async_aio.rs +++ b/block/src/raw_async_aio.rs @@ -49,28 +49,6 @@ impl AsyncIo for RawFileAsyncAio { self.alignment } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - self.data_io - .submit_borrowed_operation(self.fd, offset, true, iovecs, user_data) - .map_err(AsyncIoError::ReadVectored) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - self.data_io - .submit_borrowed_operation(self.fd, offset, false, iovecs, user_data) - .map_err(AsyncIoError::WriteVectored) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { let is_read = op.is_read(); self.data_io.submit_operation(self.fd, op).map_err(|e| { @@ -95,7 +73,7 @@ impl AsyncIo for RawFileAsyncAio { Ok(()) } - fn next_completion(&mut self) -> Option { + fn next_completed_request(&mut self) -> Option { self.data_io.next_completion() } diff --git a/block/src/raw_async_io_tests.rs b/block/src/raw_async_io_tests.rs index 393adf4b4..eb0c14023 100644 --- a/block/src/raw_async_io_tests.rs +++ b/block/src/raw_async_io_tests.rs @@ -16,7 +16,7 @@ use std::io::{Read, Seek, SeekFrom, Write}; use crate::async_io::{AsyncIo, AsyncIoError}; fn next_completion(async_io: &mut dyn AsyncIo) -> (u64, i32) { - let completion = async_io.next_completion().expect("No completion"); + let completion = async_io.next_completed_request().expect("No completion"); (completion.user_data, completion.result) } diff --git a/block/src/raw_sync.rs b/block/src/raw_sync.rs index 2c4a79c01..4daa16ceb 100644 --- a/block/src/raw_sync.rs +++ b/block/src/raw_sync.rs @@ -43,58 +43,6 @@ impl AsyncIo for RawFileSync { self.alignment } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: FFI call with valid arguments - let result = unsafe { - libc::preadv( - self.fd as libc::c_int, - iovecs.as_ptr(), - iovecs.len() as libc::c_int, - offset, - ) - }; - if result < 0 { - return Err(AsyncIoError::ReadVectored(std::io::Error::last_os_error())); - } - - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, result as i32, None)); - self.eventfd.write(1).unwrap(); - - Ok(()) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: FFI call with valid arguments - let result = unsafe { - libc::pwritev( - self.fd as libc::c_int, - iovecs.as_ptr(), - iovecs.len() as libc::c_int, - offset, - ) - }; - if result < 0 { - return Err(AsyncIoError::WriteVectored(std::io::Error::last_os_error())); - } - - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, result as i32, None)); - self.eventfd.write(1).unwrap(); - - Ok(()) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { let offset = op.offset(); let is_read = op.is_read(); @@ -157,7 +105,7 @@ impl AsyncIo for RawFileSync { Ok(()) } - fn next_completion(&mut self) -> Option { + fn next_completed_request(&mut self) -> Option { self.completion_list.pop_front() } diff --git a/block/src/request.rs b/block/src/request.rs index af7d08c83..b7529e922 100644 --- a/block/src/request.rs +++ b/block/src/request.rs @@ -60,12 +60,6 @@ pub enum RequestType { } pub const DEFAULT_DESCRIPTOR_VEC_SIZE: usize = 32; -pub struct BatchRequest { - pub offset: libc::off_t, - pub iovecs: SmallVec<[libc::iovec; DEFAULT_DESCRIPTOR_VEC_SIZE]>, - pub user_data: u64, - pub request_type: RequestType, -} pub struct ExecuteAsync { // `true` if the execution will complete asynchronously diff --git a/block/src/vhdx_sync.rs b/block/src/vhdx_sync.rs index 856c1fe0f..76a4a8bd7 100644 --- a/block/src/vhdx_sync.rs +++ b/block/src/vhdx_sync.rs @@ -191,36 +191,6 @@ impl AsyncIo for VhdxSync { &self.eventfd } - fn read_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: the legacy caller is responsible for keeping the borrowed - // iovecs and writable buffers valid until completion. This is unsound, but only temporary. - let result = unsafe { self.read_iovecs(offset, iovecs)? }; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, result as i32, None)); - self.eventfd.write(1).unwrap(); - Ok(()) - } - - fn write_vectored( - &mut self, - offset: libc::off_t, - iovecs: &[libc::iovec], - user_data: u64, - ) -> AsyncIoResult<()> { - // SAFETY: the legacy caller is responsible for keeping the borrowed - // iovecs and readable buffers valid until completion. This is unsound, but only temporary. - let result = unsafe { self.write_iovecs(offset, iovecs)? }; - self.completion_list - .push_back(AsyncIoCompletion::new(user_data, result as i32, None)); - self.eventfd.write(1).unwrap(); - Ok(()) - } - fn submit_data_operation(&mut self, op: AsyncIoOperation) -> AsyncIoResult<()> { let offset = op.offset(); let is_read = op.is_read(); @@ -259,7 +229,7 @@ impl AsyncIo for VhdxSync { Ok(()) } - fn next_completion(&mut self) -> Option { + fn next_completed_request(&mut self) -> Option { self.completion_list.pop_front() } diff --git a/performance-metrics/src/micro_bench_block.rs b/performance-metrics/src/micro_bench_block.rs index 22d210df6..c217f3c08 100644 --- a/performance-metrics/src/micro_bench_block.rs +++ b/performance-metrics/src/micro_bench_block.rs @@ -22,7 +22,7 @@ use crate::util::{ }; /// Submit num_ops AIO writes, wait for them all to land, then time -/// how long it takes to drain every completion via next_completion(). +/// how long it takes to drain every completion via next_completed_request(). /// /// Returns the drain wall clock time in seconds. pub fn micro_bench_aio_drain(control: &PerformanceTestControl) -> f64 { @@ -50,7 +50,7 @@ pub fn micro_bench_aio_drain(control: &PerformanceTestControl) -> f64 { let start = Instant::now(); let mut drained = 0usize; while drained < num_ops { - if aio.next_completion().is_some() { + if aio.next_completed_request().is_some() { drained += 1; } } @@ -421,7 +421,7 @@ pub fn micro_bench_qcow_batch_read(control: &PerformanceTestControl) -> f64 { let start = Instant::now(); async_io - .submit_batch_operations(batch) + .submit_batch_requests(batch) .expect("submit_batch_requests failed"); // Drain all io_uring completions before stopping the clock. @@ -631,7 +631,7 @@ pub fn micro_bench_qcow_batch_write(control: &PerformanceTestControl) -> f64 { let start = Instant::now(); async_io - .submit_batch_operations(batch) + .submit_batch_requests(batch) .expect("submit_batch_requests failed"); drain_async_completions(async_io.as_mut(), num_ops); diff --git a/performance-metrics/src/util.rs b/performance-metrics/src/util.rs index 6fb82e3dd..eed4d5e14 100644 --- a/performance-metrics/src/util.rs +++ b/performance-metrics/src/util.rs @@ -81,7 +81,7 @@ pub fn qcow_async_tempfile(num_clusters: usize) -> (TempFile, QcowDisk) { /// Drain `count` completions from a synchronous async_io backend. pub fn drain_completions(async_io: &mut dyn AsyncIo, count: usize) { for _ in 0..count { - async_io.next_completion(); + async_io.next_completed_request(); } } @@ -187,7 +187,7 @@ pub fn drain_async_completions(async_io: &mut dyn AsyncIo, count: usize) { let mut drained = 0usize; while drained < count { wait_for_eventfd(async_io.notifier()); - while async_io.next_completion().is_some() { + while async_io.next_completed_request().is_some() { drained += 1; } } diff --git a/virtio-devices/src/block.rs b/virtio-devices/src/block.rs index 6cd939ab8..04a58db9d 100644 --- a/virtio-devices/src/block.rs +++ b/virtio-devices/src/block.rs @@ -358,7 +358,7 @@ impl BlockEpollHandler { } if !batch_requests.is_empty() { - match self.disk_image.submit_batch_operations(batch_requests) { + match self.disk_image.submit_batch_requests(batch_requests) { Ok(()) => { self.inflight_requests.extend(batch_inflight_requests); } @@ -447,7 +447,7 @@ impl BlockEpollHandler { let mut read_ops = Wrapping(0); let mut write_ops = Wrapping(0); - while let Some(mut completion) = self.disk_image.next_completion() { + while let Some(mut completion) = self.disk_image.next_completed_request() { let result = completion.result; let desc_index = completion.user_data as u16;