diff --git a/block/src/async_io/uring_data_io.rs b/block/src/async_io/uring_data_io.rs index 97e4199e7..2857e429f 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, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::io; use std::os::fd::{AsRawFd, RawFd}; @@ -27,7 +27,8 @@ 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 operations store `None`. + // remain valid until completion; metadata and legacy borrowed 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. @@ -63,6 +64,29 @@ 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)); @@ -72,6 +96,56 @@ 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)?; @@ -185,17 +259,27 @@ 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 op.is_read() { + if is_read { opcode::Readv::new(fd, iovecs.as_ptr(), iovecs.len() as u32) - .offset(op.offset() as u64) + .offset(offset as u64) .build() - .user_data(op.user_data()) + .user_data(user_data) } else { opcode::Writev::new(fd, iovecs.as_ptr(), iovecs.len() as u32) - .offset(op.offset() as u64) + .offset(offset as u64) .build() - .user_data(op.user_data()) + .user_data(user_data) } } diff --git a/block/src/raw_async.rs b/block/src/raw_async.rs index daf6775a9..813271d23 100644 --- a/block/src/raw_async.rs +++ b/block/src/raw_async.rs @@ -1,72 +1,46 @@ // Copyright © 2021 Intel Corporation // +// Copyright (c) Meta Platforms, Inc. and affiliates. +// // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause -use std::io::Error; -use std::os::unix::io::{AsRawFd, RawFd}; +use std::os::unix::io::RawFd; -use io_uring::{IoUring, opcode, types}; use libc::{FALLOC_FL_KEEP_SIZE, FALLOC_FL_PUNCH_HOLE, FALLOC_FL_ZERO_RANGE}; use vmm_sys_util::eventfd::EventFd; -use crate::async_io::{AsyncIo, AsyncIoError, AsyncIoResult}; +use crate::async_io::{ + AsyncIo, AsyncIoCompletion, AsyncIoError, AsyncIoOperation, AsyncIoResult, UringDataIo, +}; use crate::error::{BlockError, BlockErrorKind, BlockResult}; use crate::sparse::{blkdiscard, blkzeroout}; use crate::{BatchRequest, RequestType, SECTOR_SIZE, is_block_device}; pub struct RawFileAsync { fd: RawFd, - io_uring: IoUring, - eventfd: EventFd, + data_io: UringDataIo, alignment: u64, is_block_device: bool, } impl RawFileAsync { pub fn new(fd: RawFd, ring_depth: u32) -> BlockResult { - let io_uring = - IoUring::new(ring_depth).map_err(|e| BlockError::new(BlockErrorKind::Io, e))?; - let eventfd = - EventFd::new(libc::EFD_NONBLOCK).map_err(|e| BlockError::new(BlockErrorKind::Io, e))?; - - // Register the io_uring eventfd that will notify when something in - // the completion queue is ready. - io_uring - .submitter() - .register_eventfd(eventfd.as_raw_fd()) - .map_err(|e| BlockError::new(BlockErrorKind::Io, e))?; - + let data_io = + UringDataIo::new(ring_depth).map_err(|e| BlockError::new(BlockErrorKind::Io, e))?; let is_block_device = is_block_device(fd); Ok(RawFileAsync { fd, - io_uring, - eventfd, + data_io, alignment: SECTOR_SIZE, is_block_device, }) } - - /// Queue an `IORING_OP_NOP` carrying `user_data` so a synchronously - /// completed operation (e.g. a BLK* ioctl) is reaped through the normal - /// io_uring completion path. - fn submit_nop(&mut self, user_data: u64) -> Result<(), Error> { - let (submitter, mut sq, _) = self.io_uring.split(); - // SAFETY: Nop carries no buffer; only `user_data` is consumed by the - // kernel. - unsafe { - sq.push(&opcode::Nop::new().build().user_data(user_data)) - .map_err(|e| Error::other(format!("Submission queue is full: {e:?}")))?; - }; - sq.sync(); - submitter.submit()?; - Ok(()) - } } impl AsyncIo for RawFileAsync { fn notifier(&self) -> &EventFd { - &self.eventfd + self.data_io.notifier() } fn alignment(&self) -> u64 { @@ -79,28 +53,13 @@ impl AsyncIo for RawFileAsync { iovecs: &[libc::iovec], user_data: u64, ) -> AsyncIoResult<()> { - let (submitter, mut sq, _) = self.io_uring.split(); - - // SAFETY: we know the file descriptor is valid and we - // relied on vm-memory to provide the buffer address. + // SAFETY: this legacy trait method's caller must keep the borrowed + // iovecs and writable buffers valid until completion. unsafe { - sq.push( - &opcode::Readv::new(types::Fd(self.fd), iovecs.as_ptr(), iovecs.len() as u32) - .offset(offset.try_into().unwrap()) - .build() - .user_data(user_data), - ) - .map_err(|e| { - AsyncIoError::ReadVectored(Error::other(format!("Submission queue is full: {e:?}"))) - })?; - }; - - // Update the submission queue and submit new operations to the - // io_uring instance. - sq.sync(); - submitter.submit().map_err(AsyncIoError::ReadVectored)?; - - Ok(()) + self.data_io + .submit_borrowed_operation(self.fd, offset, true, iovecs, user_data) + } + .map_err(AsyncIoError::ReadVectored) } fn write_vectored( @@ -109,52 +68,31 @@ impl AsyncIo for RawFileAsync { iovecs: &[libc::iovec], user_data: u64, ) -> AsyncIoResult<()> { - let (submitter, mut sq, _) = self.io_uring.split(); - - // SAFETY: we know the file descriptor is valid and we - // relied on vm-memory to provide the buffer address. + // SAFETY: this legacy trait method's caller must keep the borrowed + // iovecs and readable buffers valid until completion. unsafe { - sq.push( - &opcode::Writev::new(types::Fd(self.fd), iovecs.as_ptr(), iovecs.len() as u32) - .offset(offset.try_into().unwrap()) - .build() - .user_data(user_data), - ) - .map_err(|e| { - AsyncIoError::WriteVectored(Error::other(format!( - "Submission queue is full: {e:?}" - ))) - })?; - }; + self.data_io + .submit_borrowed_operation(self.fd, offset, false, iovecs, user_data) + } + .map_err(AsyncIoError::WriteVectored) + } - // Update the submission queue and submit new operations to the - // io_uring instance. - sq.sync(); - submitter.submit().map_err(AsyncIoError::WriteVectored)?; - - Ok(()) + 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| { + if is_read { + AsyncIoError::ReadVectored(e) + } else { + AsyncIoError::WriteVectored(e) + } + }) } fn fsync(&mut self, user_data: Option) -> AsyncIoResult<()> { if let Some(user_data) = user_data { - let (submitter, mut sq, _) = self.io_uring.split(); - - // SAFETY: we know the file descriptor is valid. - unsafe { - sq.push( - &opcode::Fsync::new(types::Fd(self.fd)) - .build() - .user_data(user_data), - ) - .map_err(|e| { - AsyncIoError::Fsync(Error::other(format!("Submission queue is full: {e:?}"))) - })?; - }; - - // Update the submission queue and submit new operations to the - // io_uring instance. - sq.sync(); - submitter.submit().map_err(AsyncIoError::Fsync)?; + self.data_io + .submit_fsync(self.fd, user_data) + .map_err(AsyncIoError::Fsync)?; } else { // SAFETY: FFI call with a valid fd unsafe { libc::fsync(self.fd) }; @@ -163,11 +101,8 @@ impl AsyncIo for RawFileAsync { Ok(()) } - fn next_completed_request(&mut self) -> Option<(u64, i32)> { - self.io_uring - .completion() - .next() - .map(|entry| (entry.user_data(), entry.result())) + fn next_completion(&mut self) -> Option { + self.data_io.next_completion() } fn batch_requests_enabled(&self) -> bool { @@ -175,84 +110,29 @@ impl AsyncIo for RawFileAsync { } fn submit_batch_requests(&mut self, batch_request: &[BatchRequest]) -> AsyncIoResult<()> { - if !self.batch_requests_enabled() { - return Ok(()); - } - - let (submitter, mut sq, _) = self.io_uring.split(); - let mut submitted = false; - - // Refuse the whole batch if it can't fit in the SQ to avoid having to unroll a partially - // successful push. - if batch_request.len() > sq.capacity() - sq.len() { - return Err(AsyncIoError::SubmitBatchRequests(Error::other( - "io_uring submission queue is full", - ))); - } - + let mut batch = Vec::with_capacity(batch_request.len()); for req in batch_request { - match req.request_type { - RequestType::In => { - // SAFETY: we know the file descriptor is valid and we - // relied on vm-memory to provide the buffer address. - unsafe { - sq.push( - &opcode::Readv::new( - types::Fd(self.fd), - req.iovecs.as_ptr(), - req.iovecs.len() as u32, - ) - .offset(req.offset as u64) - .build() - .user_data(req.user_data), - ) - .map_err(|e| { - AsyncIoError::ReadVectored(Error::other(format!( - "Submission queue is full: {e:?}" - ))) - })?; - }; - submitted = true; - } - RequestType::Out => { - // SAFETY: we know the file descriptor is valid and we - // relied on vm-memory to provide the buffer address. - unsafe { - sq.push( - &opcode::Writev::new( - types::Fd(self.fd), - req.iovecs.as_ptr(), - req.iovecs.len() as u32, - ) - .offset(req.offset as u64) - .build() - .user_data(req.user_data), - ) - .map_err(|e| { - AsyncIoError::WriteVectored(Error::other(format!( - "Submission queue is full: {e:?}" - ))) - })?; - }; - submitted = true; - } - _ => { - unreachable!("Unexpected batch request type: {:?}", req.request_type) - } - } + 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)); } - // Only submit if we actually queued something - if submitted { - // Update the submission queue and submit new operations to the - // io_uring instance. - sq.sync(); - submitter - .submit() - .map_err(AsyncIoError::SubmitBatchRequests)?; - } + // 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) + } - Ok(()) + fn submit_batch_operations( + &mut self, + batch_request: Vec, + ) -> AsyncIoResult<()> { + self.data_io + .submit_batch(self.fd, batch_request) + .map_err(AsyncIoError::SubmitBatchRequests) } fn punch_hole(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> { @@ -265,31 +145,17 @@ impl AsyncIo for RawFileAsync { // Deliver the completion through the normal io_uring path by // queuing a NOP carrying `user_data`. The registered eventfd will // fire when it completes, just like any other request. - return self.submit_nop(user_data).map_err(AsyncIoError::PunchHole); + return self + .data_io + .submit_nop(user_data) + .map_err(AsyncIoError::PunchHole); } - let (submitter, mut sq, _) = self.io_uring.split(); - let mode = FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE; - // SAFETY: The file descriptor is known to be valid. - unsafe { - sq.push( - &opcode::Fallocate::new(types::Fd(self.fd), length) - .offset(offset) - .mode(mode) - .build() - .user_data(user_data), - ) - .map_err(|e| { - AsyncIoError::PunchHole(Error::other(format!("Submission queue is full: {e:?}"))) - })?; - }; - - sq.sync(); - submitter.submit().map_err(AsyncIoError::PunchHole)?; - - Ok(()) + self.data_io + .submit_fallocate(self.fd, offset, length, mode, user_data) + .map_err(AsyncIoError::PunchHole) } fn write_zeroes(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> { @@ -297,31 +163,15 @@ impl AsyncIo for RawFileAsync { if self.is_block_device { blkzeroout(self.fd, offset, length).map_err(AsyncIoError::WriteZeroes)?; return self + .data_io .submit_nop(user_data) .map_err(AsyncIoError::WriteZeroes); } - let (submitter, mut sq, _) = self.io_uring.split(); - let mode = FALLOC_FL_ZERO_RANGE | FALLOC_FL_KEEP_SIZE; - // SAFETY: The file descriptor is known to be valid. - unsafe { - sq.push( - &opcode::Fallocate::new(types::Fd(self.fd), length) - .offset(offset) - .mode(mode) - .build() - .user_data(user_data), - ) - .map_err(|e| { - AsyncIoError::WriteZeroes(Error::other(format!("Submission queue is full: {e:?}"))) - })?; - }; - - sq.sync(); - submitter.submit().map_err(AsyncIoError::WriteZeroes)?; - - Ok(()) + self.data_io + .submit_fallocate(self.fd, offset, length, mode, user_data) + .map_err(AsyncIoError::WriteZeroes) } }