From 34e6e87266b73146db451d7ee5bcccd1296ee407 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 10:30:04 -0500 Subject: [PATCH] fix(io): start no task when a stream with a file slot drops Dropping a stream that held a registered file slot spawned a task to give the slot back. A compio runtime being dropped clears its executor by dropping every unfinished task's future, so a stream owned by such a task spawned onto the queue being drained, tripping compio-executor's `inner.map.is_empty()` assertion. The task that takes the slot now keeps it: it waits for the stream to go and gives the slot back then, and dropping the stream only wakes it. A keeper dropped with its runtime leaves the slot to the ring's teardown. --- io/src/compio.rs | 94 +++++++++++++++++++++++++++++++++--------------- io/tests/io.rs | 30 ++++++++++++++-- 2 files changed, 93 insertions(+), 31 deletions(-) diff --git a/io/src/compio.rs b/io/src/compio.rs index 82f65ad..814f533 100644 --- a/io/src/compio.rs +++ b/io/src/compio.rs @@ -1033,7 +1033,7 @@ mod files { io, os::fd::{AsRawFd as _, RawFd}, rc::Rc, - task::Poll, + task::{Poll, Waker}, }; use ::compio::{ @@ -1171,14 +1171,23 @@ mod files { /// A stream's hold on its slot. pub(super) struct File { - state: Rc>, + slot: Rc, /// The socket's descriptor, which the slot replaces in an entry. fd: RawFd, } + /// What a stream and its slot's keeper share. + #[derive(Default)] + struct Slot { + state: Cell, + /// The keeper, waiting for the stream to go once it holds the slot. + keeper: Cell>, + } + #[derive(Default)] enum State { - /// The installer has not finished; operations name the descriptor. + /// The keeper has not taken the slot yet; operations name the + /// descriptor. Installing, /// The socket holds this slot. Fixed(u32), @@ -1188,41 +1197,46 @@ mod files { /// Taking a slot failed other than for want of room: the next /// operation reports it, and the rest name the descriptor. Failed(io::Error), - /// The stream is gone: the installer gives back the slot it takes. + /// The stream is gone: the keeper gives back the slot it holds or + /// takes. Dropped, } impl File { - /// Starts taking a slot for `socket`, on a task of the runtime, so - /// that no operation of the stream waits for it and none abandoned - /// midway loses the slot (and with it the socket, which the slot - /// keeps open). + /// Starts the keeper of a slot for `socket`, a task of the runtime + /// that takes the slot, so that no operation of the stream waits for + /// it and none abandoned midway loses the slot (and with it the + /// socket, which the slot keeps open), and gives it back once the + /// stream is gone, so that dropping a stream starts no task: a + /// runtime being dropped drops its tasks, and may be handed no new + /// one meanwhile. pub(super) fn new(socket: SharedFd, table: &super::FileTable) -> Self { let fd = socket.as_raw_fd(); - let state = Rc::new(Cell::new(State::Descriptor)); + let slot = Rc::new(Slot::default()); let slots = table.slots(); if slots > 0 { Runtime::with_current(|runtime| { if runtime.driver_type().is_iouring() { - state.set(State::Installing); + slot.state.set(State::Installing); runtime - .spawn(install(Rc::clone(&state), socket, slots)) + .spawn(keep(Rc::clone(&slot), socket, slots)) .detach(); } }); } - Self { state, fd } + Self { slot, fd } } fn slot(&self) -> io::Result> { - match self.state.take() { + let state = &self.slot.state; + match state.take() { State::Fixed(index) => { - self.state.set(State::Fixed(index)); + state.set(State::Fixed(index)); Ok(Some(index)) } State::Failed(error) => Err(error), - state => { - self.state.set(state); + other => { + state.set(other); Ok(None) } } @@ -1269,28 +1283,50 @@ mod files { impl Drop for File { fn drop(&mut self) { - if let State::Fixed(index) = self.state.replace(State::Dropped) { - // Without a runtime the ring is gone, and its table with it. - let _ = Runtime::try_with_current(|runtime| runtime.spawn(remove(index)).detach()); + self.slot.state.set(State::Dropped); + if let Some(keeper) = self.slot.keeper.take() { + keeper.wake(); } } } - /// Takes a slot for `socket` and leaves the outcome in `state`, or gives - /// the slot back at once when the stream is gone by then. - async fn install(state: Rc>, socket: SharedFd, slots: u32) { + /// Takes a slot for `socket` and leaves the outcome in `slot`; while + /// the stream holds the slot, waits for the stream to go, and gives the + /// slot back then, or at once when the stream is gone by the time the + /// slot is taken. + /// + /// A keeper dropped with its runtime leaves the slot taken, which is + /// right: the ring goes with the runtime, and its table with it. + async fn keep(slot: Rc, socket: SharedFd, slots: u32) { let taken = take_slot(socket, slots).await; - if let State::Dropped = state.take() { + if let State::Dropped = slot.state.take() { if let Ok(Some(index)) = taken { remove(index).await; } return; } - state.set(match taken { - Ok(Some(index)) => State::Fixed(index), - Ok(None) => State::Descriptor, - Err(error) => State::Failed(error), - }); + let index = match taken { + Ok(Some(index)) => index, + Ok(None) => { + slot.state.set(State::Descriptor); + return; + } + Err(error) => { + slot.state.set(State::Failed(error)); + return; + } + }; + slot.state.set(State::Fixed(index)); + std::future::poll_fn(|cx| match slot.state.take() { + State::Dropped => Poll::Ready(()), + held => { + slot.state.set(held); + slot.keeper.set(Some(cx.waker().clone())); + Poll::Pending + } + }) + .await; + remove(index).await; } /// The slot the kernel allocates for `socket`, registering the ring's @@ -1327,7 +1363,7 @@ mod files { fn register(slots: u32) -> io::Result<()> { let table = vec![-1; slots as usize]; match Runtime::with_current(|runtime| runtime.register_files(&table)) { - // Another stream's installer registered it meanwhile. + // Another stream's keeper registered it meanwhile. Err(error) if error.raw_os_error() == Some(libc::EBUSY) => Ok(()), res => res.map_err(|error| { io::Error::new( diff --git a/io/tests/io.rs b/io/tests/io.rs index f4227e4..e7bccc1 100644 --- a/io/tests/io.rs +++ b/io/tests/io.rs @@ -1621,6 +1621,32 @@ mod compio { super::assert_bytes(&peer.join().expect("peer thread"), &expected); } + /// A runtime dropped while a task still holds a stream with a slot tears + /// down cleanly: dropping the stream with the task hands the runtime no + /// new task while it is clearing its own. + #[cfg(target_os = "linux")] + #[test] + fn a_runtime_drops_while_a_task_holds_a_stream_with_a_slot() { + use std::io::Read as _; + + let peer = run(async { + let (stream, peer) = accept_one(|mut client| { + // The connection ends with the runtime, which drops the + // stream with the task holding it. + let _ = client.read_to_end(&mut Vec::new()); + }) + .await; + until_held(&socket_name(&stream), 1).await; + ::compio::runtime::spawn(async move { + let _stream = stream; + std::future::pending::<()>().await; + }) + .detach(); + peer + }); + peer.join().expect("peer thread"); + } + /// A socket the ring's table has no room for stays on its descriptor and /// works as any other. #[cfg(target_os = "linux")] @@ -1668,8 +1694,8 @@ mod compio { }, ) .await; - // Both installers have run by now; the one that came second found - // the table full. + // Both keepers have asked for a slot by now; the one that came + // second found the table full. ::compio::time::sleep(std::time::Duration::from_millis(20)).await; let held = names.each_ref().map(|name| slots_holding(name)); assert_eq!(held.iter().sum::(), 1, "slots held: {held:?}");