From 7e2e7a164baa9881ba66f70f54756877e83f2a7d Mon Sep 17 00:00:00 2001 From: Rob Bradford Date: Wed, 17 Jun 2026 13:02:59 +0100 Subject: [PATCH] block: Drain io_uring in-flight operations on teardown Closing an io_uring fd does not synchronously finish requests that already reached the kernel. During block worker teardown this can let an io-wq worker keep using retained guest-memory iovecs after reset. Drain UringDataIo in Drop: retry any published SQEs and wait for CQEs until no retained operation remains. If draining fails, leak retained buffers. Drop QcowAsync's ring before its data fd so retrying published SQEs still uses a valid descriptor. To avoid a potential infinite loop when completions fail to be delivered cap the number of iterations of the loop (2x the number of inflight requests). Assisted-by: Codex:GPT-5 Signed-off-by: Rob Bradford --- block/src/formats/qcow/worker/async_uring.rs | 5 +- block/src/io/async_io/uring_data_io.rs | 78 ++++++++++++++++++-- 2 files changed, 76 insertions(+), 7 deletions(-) diff --git a/block/src/formats/qcow/worker/async_uring.rs b/block/src/formats/qcow/worker/async_uring.rs index 6e93ac609..c76cea324 100644 --- a/block/src/formats/qcow/worker/async_uring.rs +++ b/block/src/formats/qcow/worker/async_uring.rs @@ -40,6 +40,8 @@ use crate::async_io::{ /// before the host offset is known. pub struct QcowAsync { metadata: Arc, + // Drop before data_file so pending SQEs can be submitted while fd is valid. + data_io: UringDataIo, data_file: QcowRawFile, backing_file: Option>, sparse: bool, @@ -49,7 +51,6 @@ pub struct QcowAsync { io_alignment: u64, cluster_size: u64, decoder: Arc, - data_io: UringDataIo, } impl QcowAsync { @@ -67,12 +68,12 @@ impl QcowAsync { cluster_size: metadata.cluster_size(), decoder: metadata.decoder(), metadata, + data_io: UringDataIo::new(ring_depth)?, data_file, backing_file, sparse, alignment, io_alignment, - data_io: UringDataIo::new(ring_depth)?, }) } diff --git a/block/src/io/async_io/uring_data_io.rs b/block/src/io/async_io/uring_data_io.rs index b9445ef3a..12167983e 100644 --- a/block/src/io/async_io/uring_data_io.rs +++ b/block/src/io/async_io/uring_data_io.rs @@ -5,11 +5,11 @@ // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause use std::collections::{HashMap, VecDeque}; -use std::io; use std::os::fd::{AsRawFd, RawFd}; +use std::{io, mem}; use io_uring::{IoUring, opcode, squeue, types}; -use log::warn; +use log::{error, warn}; use vmm_sys_util::eventfd::EventFd; use super::common::{duplicate_user_data_error, validate_batch}; @@ -19,9 +19,6 @@ use super::{AsyncIoCompletion, AsyncIoOperation}; /// /// Holds the `IoUring` and its `EventFd`. Tracks ops that are pending. pub struct UringDataIo { - // Keep `io_uring` before `in_flight`. Rust drops fields in declaration - // order, so dropping the ring tears down kernel access before retained - // operations release the buffers referenced by their iovecs. io_uring: IoUring, // The `EventFd` for completion signals. eventfd: EventFd, @@ -270,6 +267,54 @@ impl UringDataIo { } } +impl Drop for UringDataIo { + fn drop(&mut self) { + // Closing the ring fd does not cancel io_uring ops that have started. + // Wait for CQEs before releasing retained iovecs. + if self.needs_submit_retry { + if let Err(e) = self.io_uring.submitter().submit() { + warn!("io_uring drain submit failed for retained SQEs: {e}"); + } + self.needs_submit_retry = false; + } + + let max_drain_iterations = self.in_flight.len().saturating_mul(2); + let mut drain_iterations = 0; + while !self.in_flight.is_empty() { + if drain_iterations == max_drain_iterations { + error!( + "io_uring drain abandoned with {} operations still in flight after {} drain iterations", + self.in_flight.len(), + drain_iterations + ); + // Keep retained buffers mapped if the ring cannot be drained. + mem::forget(mem::take(&mut self.in_flight)); + break; + } + drain_iterations += 1; + + if let Some(entry) = self.io_uring.completion().next() { + self.in_flight.remove(&entry.user_data()); + continue; + } + + // No completion ready: block in the kernel until at least one is. + if let Err(e) = self.io_uring.submitter().submit_and_wait(1) { + if e.kind() == io::ErrorKind::Interrupted { + continue; + } + error!( + "io_uring drain abandoned with {} operations still in flight: {e}", + self.in_flight.len() + ); + // Keep retained buffers mapped if the ring cannot be drained. + mem::forget(mem::take(&mut self.in_flight)); + break; + } + } + } +} + #[cfg(test)] mod tests { use std::io; @@ -328,6 +373,29 @@ mod tests { assert_eq!(completion.result, 512); } + #[test] + fn uring_drop_drains_in_flight_operations() { + let file = TempFile::new().unwrap().into_file(); + file.set_len(8192).unwrap(); + let fd = file.as_raw_fd(); + let mut data_io = UringDataIo::new(8).unwrap(); + + for user_data in 0..4 { + data_io + .submit_operation( + fd, + AsyncIoOperation::read_to_vec( + 0, + OwnedIoBuffer::from_vec(vec![0; 512]), + user_data, + ), + ) + .unwrap(); + } + + drop(data_io); + } + #[test] fn uring_queue_full_batch_completes_each_operation() { let file = TempFile::new().unwrap().into_file();