From d9de2a3f4ea978c5b181b9e664ad2f0dab6db4d6 Mon Sep 17 00:00:00 2001 From: CBoYXD Date: Thu, 29 Oct 2026 21:09:57 +0200 Subject: [PATCH 1/2] update rustls-webpki, rustls and tokio to patched releases --- Cargo.lock | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ed4f1fa..7f45c36 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -989,9 +989,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.45" +version = "0.23.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" +checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" dependencies = [ "aws-lc-rs", "log", @@ -1034,9 +1034,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.15" +version = "0.103.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" dependencies = [ "aws-lc-rs", "ring", @@ -1274,9 +1274,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.53.1" +version = "1.53.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e" dependencies = [ "bytes", "pin-project-lite", From fa1c4de9c144bfdbfd000f3c85cd62db6983482a Mon Sep 17 00:00:00 2001 From: CBoYXD <135316445+CBoYXD@users.noreply.github.com> Date: Fri, 30 Oct 2026 13:03:08 +0200 Subject: [PATCH 2/2] fix: release add_reader waits before remove_reader returns On Unix every add_reader()/add_writer() readiness wait owns a duplicate of the descriptor. Those waits ran on the coordination thread, so remove_reader() only queued their cancellation, and the duplicate (and with it the kernel socket) could outlive the caller's close(). A closed connected datagram transport could therefore still hold its 4-tuple after connection_lost ran. macOS rejects reconnecting that 4-tuple with EADDRINUSE, which made test_connected_datagram_close_releases_socket_before_callback flaky. Waits armed while the loop runs on its own thread are now tasks on the loop's runtime, and remove_reader()/remove_writer() drop them before returning. Registrations made while the loop is stopped still wait on the coordination thread; a re-arm or replacement while running moves them to the loop thread, and the old wait is cancelled where it runs. The tests check through /proc/self/fd that no duplicate outlives remove_reader() or connection_lost, including registrations that move between the two threads. --- src/bindings/loop_api/watchers.rs | 28 ++++-- src/engine/dispatcher.rs | 2 +- src/engine/loop_core.rs | 161 ++++++++++++++++++++++++------ tests/test_compat.py | 135 ++++++++++++++++++++++++- 4 files changed, 282 insertions(+), 44 deletions(-) diff --git a/src/bindings/loop_api/watchers.rs b/src/bindings/loop_api/watchers.rs index e7ffb91..da04e0a 100644 --- a/src/bindings/loop_api/watchers.rs +++ b/src/bindings/loop_api/watchers.rs @@ -13,7 +13,7 @@ use pyo3::{prelude::*, types::PyTuple}; use super::PyLoop; use crate::{ context::capture_context, - engine::{CallbackKind, FdWatch, LoopCommand, LoopIoCommand, ReadyCallback}, + engine::{CallbackKind, FdWatch, ReadyCallback}, fd_ops, }; @@ -54,13 +54,16 @@ pub(super) fn add_reader( ); if let Some(previous) = previous { previous.ready.cancel(); + // The replaced wait may live on the other side of the loop/runtime + // thread split from the new one; stop it where it runs. + loop_ref + .core + .stop_fd_watch(raw_fd, false) + .map_err(PyLoop::map_loop_error)?; } loop_ref .core - .send_command(LoopCommand::Io(LoopIoCommand::StartReader { - fd: raw_fd, - callback: ready, - })) + .start_fd_watch(raw_fd, false, ready) .map_err(PyLoop::map_loop_error) } @@ -102,7 +105,7 @@ pub(super) fn remove_reader( }; loop_ref .core - .send_command(LoopCommand::Io(LoopIoCommand::StopReader(stop_fd))) + .stop_fd_watch(stop_fd, false) .map_err(PyLoop::map_loop_error)?; Ok(removed) } @@ -141,13 +144,16 @@ pub(super) fn add_writer( ); if let Some(previous) = previous { previous.ready.cancel(); + // The replaced wait may live on the other side of the loop/runtime + // thread split from the new one; stop it where it runs. + loop_ref + .core + .stop_fd_watch(raw_fd, true) + .map_err(PyLoop::map_loop_error)?; } loop_ref .core - .send_command(LoopCommand::Io(LoopIoCommand::StartWriter { - fd: raw_fd, - callback: ready, - })) + .start_fd_watch(raw_fd, true, ready) .map_err(PyLoop::map_loop_error) } @@ -186,7 +192,7 @@ pub(super) fn remove_writer( }; loop_ref .core - .send_command(LoopCommand::Io(LoopIoCommand::StopWriter(stop_fd))) + .stop_fd_watch(stop_fd, true) .map_err(PyLoop::map_loop_error)?; Ok(removed) } diff --git a/src/engine/dispatcher.rs b/src/engine/dispatcher.rs index ebebc49..fc32086 100644 --- a/src/engine/dispatcher.rs +++ b/src/engine/dispatcher.rs @@ -36,7 +36,7 @@ use crate::fd_ops; /// One readiness wait on the native reactor. A duplicate protects cancellation /// from descriptor reuse while user code closes or replaces its registration. #[cfg_attr(feature = "profile", hotpath::measure(future = true))] -async fn wait_fd_ready(fd: fd_ops::RawFd, writable: bool) -> std::io::Result<()> { +pub(super) async fn wait_fd_ready(fd: fd_ops::RawFd, writable: bool) -> std::io::Result<()> { let (read, write) = fd_ops::poll_fd(fd, !writable, writable, 0)?; if if writable { write } else { read } { return Ok(()); diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs index 4cf5c2f..e6f9b0a 100644 --- a/src/engine/loop_core.rs +++ b/src/engine/loop_core.rs @@ -44,6 +44,10 @@ use crate::{ fd_ops::RawFd, }; +/// Loop-thread readiness waits for one loop, keyed by `(fd, writable)`. +#[cfg(unix)] +type FdWatchTasks = HashMap<(RawFd, bool), crate::vibeio::JoinHandle<()>>; + thread_local! { /// Per-loop vibeio runtime hosted on the loop thread. Keyed by `LoopCore` /// pointer so a thread that runs several loops sequentially keeps each @@ -67,6 +71,14 @@ thread_local! { /// TLS on the loop thread. static IO_TASKS: RefCell>>> = RefCell::new(HashMap::new()); + + /// `add_reader()` / `add_writer()` readiness waits armed while the loop + /// runs, keyed by loop pointer then by `(fd, writable)`. Each wait owns a + /// duplicate of the descriptor; keeping it on the loop thread lets + /// removal drop that duplicate before `remove_reader()` returns, so a + /// caller that closes its socket next really releases it. + #[cfg(unix)] + static FD_WATCH_TASKS: RefCell> = RefCell::new(HashMap::new()); } /// Cross-thread wake state for the loop thread, kept separate from `LoopCore` @@ -1300,6 +1312,13 @@ impl LoopCore { } } }); + #[cfg(unix)] + if let Some(handles) = FD_WATCH_TASKS.with(|tasks| tasks.borrow_mut().remove(&runtime_key)) + { + for (_, handle) in handles { + handle.cancel(); + } + } LOOP_RUNTIMES.with(|runtimes| { if let Some(runtime) = runtimes.borrow_mut().remove(&runtime_key) { drop(std::mem::ManuallyDrop::into_inner(runtime)); @@ -1407,7 +1426,7 @@ impl LoopCore { } #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] - fn rearm_fd_watch_if_needed(&self, ready: &ReadyCallback) { + fn rearm_fd_watch_if_needed(self: &Arc, ready: &ReadyCallback) { // Readiness callbacks are one-shot at the runtime layer. Re-arm them // only after the current ready batch has drained so callbacks queued // by Future.set_result() can remove the registration first. This @@ -1417,39 +1436,125 @@ impl LoopCore { // Only re-arm when this callback is still the current registration // for the fd: a stale fire that outlived remove_reader()/add_reader() // must not restart a watcher for the superseded callback. - let command = match ready.kind() { - CallbackKind::Reader(fd) => self - .state - .lock() - .expect("poisoned loop state") - .reader_keepalive - .get(&fd) - .filter(|watch| std::ptr::eq(Arc::as_ptr(&watch.ready), ready)) - .map(|watch| { - LoopCommand::Io(LoopIoCommand::StartReader { - fd, - callback: Arc::clone(&watch.ready), - }) - }), - CallbackKind::Writer(fd) => self - .state - .lock() - .expect("poisoned loop state") - .writer_keepalive + let (fd, writable) = match ready.kind() { + CallbackKind::Reader(fd) => (fd, false), + CallbackKind::Writer(fd) => (fd, true), + _ => return, + }; + let callback = { + let state = self.state.lock().expect("poisoned loop state"); + let watches = if writable { + &state.writer_keepalive + } else { + &state.reader_keepalive + }; + watches .get(&fd) .filter(|watch| std::ptr::eq(Arc::as_ptr(&watch.ready), ready)) - .map(|watch| { - LoopCommand::Io(LoopIoCommand::StartWriter { - fd, - callback: Arc::clone(&watch.ready), + .map(|watch| Arc::clone(&watch.ready)) + }; + + if let Some(callback) = callback { + let _ = self.start_fd_watch(fd, writable, callback); + } + } + + /// Arms one readiness wait for an `add_reader()` / `add_writer()` + /// registration. While this loop runs on the calling thread the wait is a + /// task on the loop's own runtime, so `stop_fd_watch` can drop it (and the + /// descriptor duplicate it owns) synchronously. A registration made while + /// the loop is stopped, or from another thread, waits on the coordination + /// thread instead. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] + pub(crate) fn start_fd_watch( + self: &Arc, + fd: RawFd, + writable: bool, + callback: Arc, + ) -> Result<(), LoopCoreError> { + #[cfg(unix)] + let callback = { + let key = Arc::as_ptr(self) as usize; + let spawned = if self.on_runtime_thread() { + let core = Arc::clone(self); + let ready = Arc::clone(&callback); + LOOP_RUNTIMES.with(|runtimes| { + runtimes.borrow().get(&key).map(|runtime| { + runtime.spawn(async move { + if super::dispatcher::wait_fd_ready(fd, writable).await.is_ok() + && !ready.cancelled() + { + let _ = core.send_command(LoopCommand::ScheduleReady(ready)); + } + }) }) - }), - _ => None, + }) + } else { + None + }; + match spawned { + Some(handle) => { + // Cancel outside the borrow: dropping the replaced wait can + // release Python objects whose finalizers re-enter here. + let replaced = FD_WATCH_TASKS.with(|tasks| { + tasks + .borrow_mut() + .entry(key) + .or_default() + .insert((fd, writable), handle) + }); + if let Some(replaced) = replaced { + replaced.cancel(); + } + return Ok(()); + } + None => { + // Keep one owner per watch: a stale local wait must not + // make stop_fd_watch skip the coordination-thread wait. + let stale = FD_WATCH_TASKS.with(|tasks| { + tasks + .borrow_mut() + .get_mut(&key) + .and_then(|map| map.remove(&(fd, writable))) + }); + if let Some(stale) = stale { + stale.cancel(); + } + callback + } + } }; + self.send_command(LoopCommand::Io(if writable { + LoopIoCommand::StartWriter { fd, callback } + } else { + LoopIoCommand::StartReader { fd, callback } + })) + } - if let Some(command) = command { - let _ = self.send_command(command); + /// Stops the readiness wait armed for `fd`. A wait on this thread's loop + /// runtime is dropped before this returns, releasing its descriptor + /// duplicate; any other wait is cancelled through the coordination thread. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] + pub(crate) fn stop_fd_watch(&self, fd: RawFd, writable: bool) -> Result<(), LoopCoreError> { + #[cfg(unix)] + { + let key = self as *const LoopCore as usize; + let local = FD_WATCH_TASKS.with(|tasks| { + tasks + .borrow_mut() + .get_mut(&key) + .and_then(|map| map.remove(&(fd, writable))) + }); + if let Some(handle) = local { + handle.cancel(); + return Ok(()); + } } + self.send_command(LoopCommand::Io(if writable { + LoopIoCommand::StopWriter(fd) + } else { + LoopIoCommand::StopReader(fd) + })) } #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))] diff --git a/tests/test_compat.py b/tests/test_compat.py index 5e90c32..34ef51f 100644 --- a/tests/test_compat.py +++ b/tests/test_compat.py @@ -23,6 +23,31 @@ import rsloop._loop_compat as loop_compat EXCEPTION_GROUP = getattr(builtins, "ExceptionGroup", None) +HAS_PROC_FD = os.path.isdir("/proc/self/fd") + + +def _descriptor_duplicates(sock: socket.socket) -> list[int]: + """Other descriptors in this process that keep ``sock``'s socket open.""" + target = f"socket:[{os.fstat(sock.fileno()).st_ino}]" + duplicates = [] + for name in os.listdir("/proc/self/fd"): + if int(name) == sock.fileno(): + continue + try: + if os.readlink(f"/proc/self/fd/{name}") == target: + duplicates.append(int(name)) + except OSError: + pass + return duplicates + + +async def _wait_for_duplicate(sock: socket.socket) -> None: + # A parked readiness wait holds its own duplicate of the descriptor. + for _ in range(200): + if _descriptor_duplicates(sock): + return + await asyncio.sleep(0.005) + raise AssertionError("the readiness wait never armed") class TestCompatibility: @@ -1233,7 +1258,10 @@ async def main() -> None: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as peer: peer.bind(("127.0.0.1", 0)) port = None - for _ in range(2): + # Rebinding the same connected 4-tuple fails on BSD/macOS while + # any descriptor still holds the previous socket; /proc shows + # the same leak directly on Linux, where the rebind succeeds. + for _ in range(10): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) @@ -1244,20 +1272,119 @@ async def main() -> None: lost = loop.create_future() class Protocol(asyncio.DatagramProtocol): - def connection_lost(self, exc, result=lost): - result.set_result(exc) + def connection_lost(self, exc, result=lost, sock=sock): + # The transport closes sock after this callback, + # so the loop's own reader wait must be gone. + held = ( + _descriptor_duplicates(sock) if HAS_PROC_FD else [] + ) + result.set_result((exc, held)) transport, _ = await loop.create_datagram_endpoint( Protocol, sock=sock ) + # Let the reader park on the socket before closing. + await asyncio.sleep(0.01) transport.close() - assert await asyncio.wait_for(lost, 1) is None + assert await asyncio.wait_for(lost, 1) == (None, []) assert sock.fileno() == -1 finally: sock.close() rsloop.run(main()) + @pytest.mark.skipif(not HAS_PROC_FD, reason="requires /proc/self/fd") + def test_remove_reader_releases_wait_descriptor_before_returning(self) -> None: + # The readiness wait owns a duplicate of the descriptor. If it outlives + # remove_reader(), closing the socket next does not release it. + loop = rsloop.new_event_loop() + a, b = socket.socketpair() + try: + a.setblocking(False) + + async def remove_while_running() -> None: + loop.add_reader(a, lambda: None) + await _wait_for_duplicate(a) + assert loop.remove_reader(a) + assert _descriptor_duplicates(a) == [] + + async def arm() -> None: + loop.add_reader(a, lambda: None) + await _wait_for_duplicate(a) + + loop.run_until_complete(remove_while_running()) + # Armed during one run, removed while the loop is stopped. + loop.run_until_complete(arm()) + assert _descriptor_duplicates(a) != [] + assert loop.remove_reader(a) + assert _descriptor_duplicates(a) == [] + finally: + loop.close() + a.close() + b.close() + + @pytest.mark.skipif(not HAS_PROC_FD, reason="requires /proc/self/fd") + def test_reader_armed_while_stopped_hands_over_to_running_loop(self) -> None: + # A reader added while the loop is stopped waits on the runtime thread. + # Its re-arm, replacement and removal while running must neither lose + # events nor leave that wait holding the descriptor. + loop = rsloop.new_event_loop() + a, b = socket.socketpair() + fired: list[tuple[str, bytes]] = [] + + def reader(tag: str) -> None: + fired.append((tag, a.recv(1))) + + async def wait_fired(count: int) -> None: + for _ in range(200): + if len(fired) >= count: + return + await asyncio.sleep(0.005) + raise AssertionError(f"expected {count} reader calls, got {fired}") + + async def wait_released() -> None: + # Waits on the runtime thread are cancelled through its queue. + for _ in range(200): + if not _descriptor_duplicates(a): + return + await asyncio.sleep(0.005) + raise AssertionError("a removed reader still holds the descriptor") + + async def fire_then_remove() -> None: + b.send(b"x") + await wait_fired(1) + # The re-armed wait runs on the loop thread: removal is immediate. + await _wait_for_duplicate(a) + assert loop.remove_reader(a) + assert _descriptor_duplicates(a) == [] + + async def replace_then_remove() -> None: + await _wait_for_duplicate(a) + loop.add_reader(a, reader, "new") + b.send(b"y") + await wait_fired(2) + assert loop.remove_reader(a) + await wait_released() + + async def remove_unfired() -> None: + await _wait_for_duplicate(a) + assert loop.remove_reader(a) + await wait_released() + + try: + a.setblocking(False) + loop.add_reader(a, reader, "first") + loop.run_until_complete(fire_then_remove()) + loop.add_reader(a, reader, "old") + loop.run_until_complete(replace_then_remove()) + loop.add_reader(a, reader, "unfired") + loop.run_until_complete(remove_unfired()) + assert fired == [("first", b"x"), ("new", b"y")] + finally: + loop.close() + a.close() + b.close() + def test_sendfile_fallback_writes_file_contents(self) -> None: async def main(path: str, expected: bytes) -> tuple[int, bytes]: loop = asyncio.get_running_loop()