From 09d5b46eaa48eee74dbd31c15423efb83e1a74cc Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Fri, 18 Sep 2026 14:30:35 +0100 Subject: [PATCH 1/9] Slightly more robust fix for notify_closing on dead events for kqueue --- src/trio/_core/_io_kqueue.py | 2 +- src/trio/_core/_tests/test_guest_mode.py | 15 ++++++++++----- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index 80d0ce6296..fcb386be68 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -283,7 +283,7 @@ def notify_closing(self, fd: int | _HasFileNo) -> None: try: self._kqueue.control([event], 0) except OSError as e: - if e.errno == errno.ENOENT: # pragma: no branch + if e.errno in (errno.EBADF, errno.ENOENT): # pragma: no branch # the event isn't in kqueue continue raise # pragma: no cover diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py index 7acbc0f082..cf6aa587a1 100644 --- a/src/trio/_core/_tests/test_guest_mode.py +++ b/src/trio/_core/_tests/test_guest_mode.py @@ -755,7 +755,10 @@ async def trio_main() -> None: assert record == {("asyncio", "asyncio", True), ("trio", "trio", True)} -def test_notify_closing_after_events() -> None: +@pytest.mark.parametrize( + "close_first", [False, True], ids=["notify-then-close", "close-then-notify"], +) +def test_notify_closing_after_events(close_first: bool) -> None: # inspired by wrong repro in https://github.com/python-trio/trio/pull/3502 # either the program should silently pass or wait_writable should fail. pair = socket.socketpair() @@ -769,8 +772,10 @@ async def trio_main(in_host: InHost) -> None: def uh_oh() -> None: # this will run after trio gets events but before they are processed - trio.lowlevel.notify_closing(pair[0]) + fd = pair[0].fileno() + if close_first: + pair[0].close() + trio.lowlevel.notify_closing(fd) - trivial_guest_run(trio_main) - for sock in pair: - sock.close() + with pair[0], pair[1]: + trivial_guest_run(trio_main) From 57a4020966e7f346e7b03c34c8d4178ab9119426 Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Sat, 19 Sep 2026 13:34:34 +0100 Subject: [PATCH 2/9] Fix #3500: TrioInternalError on macOS/BSD in guest mode The kqueue IO manager tries to keep two things matching: the events registered with the OS kqueue, and its `_registered` dict that tracks them. In guest mode, host callbacks can run between get_events() (which removes events from the kqueue, because wait_readable/wait_writable register them as one-shot) and process_events() (which removes items from _registered). If a host callback cancels the waiter in that window, its registration is removed while its event is already in the fetched batch, and process_events() then raised KeyError, which surfaced as TrioInternalError. This change makes process_events() ignore events whose registration has already been removed. --- newsfragments/3500.bugfix.rst | 1 + src/trio/_core/_io_kqueue.py | 8 +++++++- src/trio/_core/_tests/test_guest_mode.py | 24 ++++++++++++++++++++++++ 3 files changed, 32 insertions(+), 1 deletion(-) create mode 100644 newsfragments/3500.bugfix.rst diff --git a/newsfragments/3500.bugfix.rst b/newsfragments/3500.bugfix.rst new file mode 100644 index 0000000000..9ad10ce834 --- /dev/null +++ b/newsfragments/3500.bugfix.rst @@ -0,0 +1 @@ +Avoid raising `trio.TrioInternalError` when, in guest mode on BSD/MacOS, a callback in the host loop cancels a task waiting for a readable/writable wait that has just completed. diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index fcb386be68..ea9af71ed7 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -90,7 +90,13 @@ def process_events(self, events: EventResult) -> None: if event.ident == self._force_wakeup_fd: self._force_wakeup.drain() continue - receiver = self._registered[key] + receiver = self._registered.get(key) + if receiver is None: + # In guest mode, host callbacks can run between get_events() + # and process_events(). If one cancels a wait that has already + # completed then it will remove the receiver from _registered, + # so it will be missing when we get here; we can just drop it. + continue if event.flags & select.KQ_EV_ONESHOT: # TODO: test this branch del self._registered[key] if isinstance(receiver, _core.Task): diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py index cf6aa587a1..00da7b12b0 100644 --- a/src/trio/_core/_tests/test_guest_mode.py +++ b/src/trio/_core/_tests/test_guest_mode.py @@ -779,3 +779,27 @@ def uh_oh() -> None: with pair[0], pair[1]: trivial_guest_run(trio_main) + + +def test_cancel_from_host_after_events() -> None: + # Regression test for https://github.com/python-trio/trio/issues/3500 + # + # In guest mode, host-loop callbacks can run between the IO manager + # fetching a batch of events and Trio processing it. If such a callback + # cancels a task that is blocked in wait_readable/wait_writable, the + # kqueue backend used to delete the task's registration while its event + # was already in that batch, and then crash with a KeyError (wrapped in + # TrioInternalError) when the stale event was processed. + a, b = socket.socketpair() + with a, b: + + async def trio_main(in_host: InHost) -> None: + with trio.CancelScope() as cancel_scope: + # This runs in the host loop after the current tick has + # already fetched the "writable" event for `a`, but before + # the tick that would deliver it. + in_host(cancel_scope.cancel) + await trio.lowlevel.wait_writable(a) # `a` is writable already + assert cancel_scope.cancelled_caught + + trivial_guest_run(trio_main) From 94558520a27838e3f108a9a28ca2770bbf20ed44 Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Sun, 20 Sep 2026 11:50:24 +0100 Subject: [PATCH 3/9] wait_kevent: do the kevent registration itself (part of #578) trio.lowlevel.wait_kevent now actually registers the kevent (as a one shot event) rather than leaving that to the caller, and unregisters on cancellation rather than leaving it to the user supplied (and now deprecated) abort_func parameter. This is a more convenient API, and more reliably handles conflicts if multiple callers attempt to wait for the same event. As a side effect, this fixes an issue in Process.wait (in #3500) where it could raise an exception in guest mode if the process finished just before the wait was cancelled. --- docs/source/reference-lowlevel.rst | 2 +- newsfragments/3500.bugfix.rst | 2 +- newsfragments/578.deprecated.rst | 5 + newsfragments/578.feature.rst | 5 + src/trio/_core/_generated_io_kqueue.py | 39 +++++-- src/trio/_core/_io_kqueue.py | 126 ++++++++++++++++------- src/trio/_core/_tests/test_guest_mode.py | 63 +++++++++++- src/trio/_core/_tests/test_kqueue.py | 126 +++++++++++++++++++++++ src/trio/_subprocess_platform/kqueue.py | 17 +-- 9 files changed, 323 insertions(+), 62 deletions(-) create mode 100644 newsfragments/578.deprecated.rst create mode 100644 newsfragments/578.feature.rst create mode 100644 src/trio/_core/_tests/test_kqueue.py diff --git a/docs/source/reference-lowlevel.rst b/docs/source/reference-lowlevel.rst index 077f21cd27..6d9eea5f21 100644 --- a/docs/source/reference-lowlevel.rst +++ b/docs/source/reference-lowlevel.rst @@ -276,7 +276,7 @@ anything real. See `#26 .. autofunction:: current_kqueue() -.. autofunction:: wait_kevent(ident, filter, abort_func) +.. autofunction:: wait_kevent(ident, filter, *, fflags=0, data=0) :async: .. autofunction:: monitor_kevent(ident, filter) diff --git a/newsfragments/3500.bugfix.rst b/newsfragments/3500.bugfix.rst index 9ad10ce834..4796fbb624 100644 --- a/newsfragments/3500.bugfix.rst +++ b/newsfragments/3500.bugfix.rst @@ -1 +1 @@ -Avoid raising `trio.TrioInternalError` when, in guest mode on BSD/MacOS, a callback in the host loop cancels a task waiting for a readable/writable wait that has just completed. +Avoid raising `trio.TrioInternalError` when, in guest mode on macOS/BSD, a callback in the host loop cancels a task waiting for a readable/writable wait that has just completed. Also avoid raising an exception in a similar situation for `trio.Process.wait`. diff --git a/newsfragments/578.deprecated.rst b/newsfragments/578.deprecated.rst new file mode 100644 index 0000000000..48ef89235e --- /dev/null +++ b/newsfragments/578.deprecated.rst @@ -0,0 +1,5 @@ +`trio.lowlevel.wait_kevent` previously required the caller to actually +register with the kqueue (and unregister in the ``abort_func`` callback); +this usage is deprecated. Instead, do not register the event with the kqueue, +and do not pass ``abort_func``; then `trio.lowlevel.wait_kevent` will do the +registration (and unregistration on cancellation). diff --git a/newsfragments/578.feature.rst b/newsfragments/578.feature.rst new file mode 100644 index 0000000000..0a85a4233c --- /dev/null +++ b/newsfragments/578.feature.rst @@ -0,0 +1,5 @@ +`trio.lowlevel.wait_kevent` now actually registers the kevent (as a one shot +event) rather than leaving that to the caller, and unregisters on cancellation +rather than leaving it to the user supplied (and now deprecated) +``abort_func`` parameter. This is a more convenient API, and more reliably +handles conflicts if multiple callers attempt to wait for the same event. diff --git a/src/trio/_core/_generated_io_kqueue.py b/src/trio/_core/_generated_io_kqueue.py index 556d29e1f2..25dccf9e24 100644 --- a/src/trio/_core/_generated_io_kqueue.py +++ b/src/trio/_core/_generated_io_kqueue.py @@ -59,15 +59,42 @@ def monitor_kevent( @enable_ki_protection async def wait_kevent( - ident: int, filter: int, abort_func: Callable[[RaiseCancelT], Abort] -) -> Abort: - """TODO: these are implemented, but are currently more of a sketch than - anything real. See `#26 - `__. + ident: int | _HasFileNo, + filter: int, + abort_func: Callable[[RaiseCancelT], Abort] | None = None, + *, + fflags: int = 0, + data: int = 0, +) -> select.kevent: + """Waits for a one-shot kevent to happen. + + This is a low-level function that lets you wait on a specific kevent, in case + you have a use case not covered by the IO primitives in Trio. + + This registers ``kevent(ident, filter, flags, fflags, data)``, where + ``flags`` is set to ``select.KQ_EV_ADD | select.KQ_EV_ONESHOT``, and waits + for it to complete. If cancelled then it is removed with ``flags`` set to + ``select.KQ_EV_DELETE``. + + Args: + ident: Value used to identify the event. The interpretation depends on the + filter but it's usually the file descriptor. + filter: Name of the kernel filter e.g. ``select.KQ_FILTER_READ``. + fflags: Filter-specific flags. + data: Filter-specific data. + + Returns: + select.kevent: The event returned from kqueue. + + Raises: + BusyResourceError: if another task is waiting for this (ident, filter) pair. + OSError: if the kqueue rejects the registration (for example, + `ProcessLookupError` for an ``EVFILT_PROC`` ident that has + already exited). """ try: return await GLOBAL_RUN_CONTEXT.runner.io_manager.wait_kevent( - ident, filter, abort_func + ident, filter, abort_func, fflags=fflags, data=data ) except AttributeError: raise RuntimeError("must be called from async context") from None diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index ea9af71ed7..86ca8202d1 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -10,6 +10,7 @@ import outcome from .. import _core +from .._deprecate import warn_deprecated from ._run import _public from ._wakeup_socketpair import WakeupSocketpair @@ -105,15 +106,12 @@ def process_events(self, events: EventResult) -> None: receiver.put_nowait(event) # TODO: test this line # kevent registration is complicated -- e.g. aio submission can - # implicitly perform a EV_ADD, and EVFILT_PROC with NOTE_TRACK will - # automatically register filters for child processes. So our lowlevel - # API is *very* low-level: we expose the kqueue itself for adding - # events or sticking into AIO submission structs, and split waiting - # off into separate methods. It's your responsibility to make sure - # that handle_io never receives an event without a corresponding - # registration! This may be challenging if you want to be careful - # about e.g. KeyboardInterrupt. Possibly this API could be improved to - # be more ergonomic... + # implicitly perform an EV_ADD, and EVFILT_PROC with NOTE_TRACK will + # automatically register filters for child processes. Earlier revisions + # had a *very* low-level API, where the caller had responsibility for + # registering events with the OS kqueue. wait_kevent now takes care of + # that for one-shot waits. monitor_kevent has not yet been updated + # similarly, so its callers still need to register events themselves. @_public def current_kqueue(self) -> select.kqueue: @@ -149,56 +147,105 @@ def monitor_kevent( @_public async def wait_kevent( self, - ident: int, + ident: int | _HasFileNo, filter: int, - abort_func: Callable[[RaiseCancelT], Abort], - ) -> Abort: - """TODO: these are implemented, but are currently more of a sketch than - anything real. See `#26 - `__. + abort_func: Callable[[RaiseCancelT], Abort] | None = None, + *, + fflags: int = 0, + data: int = 0, + ) -> select.kevent: + """Waits for a one-shot kevent to happen. + + This is a low-level function that lets you wait on a specific kevent, + in case you have a use case not covered by the IO primitives in Trio. + + This registers ``kevent(ident, filter, flags, fflags, data)``, where + ``flags`` is set to ``select.KQ_EV_ADD | select.KQ_EV_ONESHOT``, and + waits for it to complete. If cancelled then it is removed with ``flags`` + set to ``select.KQ_EV_DELETE``. + + Args: + ident: Value used to identify the event. The interpretation depends + on the filter but it's usually the file descriptor. + filter: Name of the kernel filter e.g. ``select.KQ_FILTER_READ``. + fflags: Filter-specific flags. + data: Filter-specific data. + + Returns: + select.kevent: The event returned from kqueue. + + Raises: + BusyResourceError: if another task is waiting for this + ``(ident, filter)`` pair. + OSError: if the kqueue rejects the registration (for example, + `ProcessLookupError` for an ``EVFILT_PROC`` ident that has + already exited). """ + if not isinstance(ident, int): + ident = ident.fileno() + + if abort_func is not None: + warn_deprecated( + "wait_kevent(..., abort_func=...)", + "0.35.0", + issue=578, + instead="wait_kevent(ident, filter, fflags=..., data=...)," + " which registers the kevent itself", + ) + key = (ident, filter) if key in self._registered: raise _core.BusyResourceError( "attempt to register multiple listeners for same ident/filter pair", ) - self._registered[key] = _core.current_task() - def abort(raise_cancel: RaiseCancelT) -> Abort: - r = abort_func(raise_cancel) - if r is _core.Abort.SUCCEEDED: # TODO: test this branch - del self._registered[key] - return r + if abort_func is not None: + abort_fn = abort_func + self._registered[key] = _core.current_task() - # wait_task_rescheduled does not have its return type typed - return await _core.wait_task_rescheduled(abort) # type: ignore[no-any-return] + def abort_deprecated(raise_cancel: RaiseCancelT) -> Abort: + r = abort_fn(raise_cancel) + if r is _core.Abort.SUCCEEDED: + del self._registered[key] + return r - async def _wait_common( - self, - fd: int | _HasFileNo, - filter: int, - ) -> None: - if not isinstance(fd, int): - fd = fd.fileno() - flags = select.KQ_EV_ADD | select.KQ_EV_ONESHOT - event = select.kevent(fd, filter, flags) + # wait_task_rescheduled does not have its return type typed + return await _core.wait_task_rescheduled( # type: ignore[no-any-return] + abort_deprecated, + ) + + # Register the event before updating _registered in case this throws + event = select.kevent( + ident, + filter, + select.KQ_EV_ADD | select.KQ_EV_ONESHOT, + fflags, + data, + ) self._kqueue.control([event], 0) + self._registered[key] = _core.current_task() - def abort(_: RaiseCancelT) -> Abort: - event = select.kevent(fd, filter, select.KQ_EV_DELETE) + def abort(raise_cancel: RaiseCancelT) -> Abort: + del self._registered[key] try: + event = select.kevent(ident, filter, select.KQ_EV_DELETE) self._kqueue.control([event], 0) except OSError as exc: # kqueue tracks individual fds (*not* the underlying file # object, see _io_epoll.py for a long discussion of why this # distinction matters), and automatically deregisters an event # if the fd is closed. So if kqueue.control says that it - # doesn't know about this event, then probably it's because + # doesn't know about this event, it could be because # the fd was closed behind our backs. (Too bad we can't ask it # to wake us up when this happens, versus discovering it after # the fact... oh well, you can't have everything.) - # # FreeBSD reports this using EBADF. macOS uses ENOENT. + # + # This can also happen if, in guest mode, the cancellation + # happens just after the event was fetched by get_events() + # (which removes it, since it's oneshot) but before it has been + # processed it in process_events(). This is reported as + # errno.ENOENT on both platforms. if exc.errno in (errno.EBADF, errno.ENOENT): # pragma: no branch pass else: # pragma: no cover @@ -206,7 +253,8 @@ def abort(_: RaiseCancelT) -> Abort: raise return _core.Abort.SUCCEEDED - await self.wait_kevent(fd, filter, abort) + # wait_task_rescheduled does not have its return type typed + return await _core.wait_task_rescheduled(abort) # type: ignore[no-any-return] @_public async def wait_readable(self, fd: int | _HasFileNo) -> None: @@ -231,7 +279,7 @@ async def wait_readable(self, fd: int | _HasFileNo) -> None: if another task calls :func:`notify_closing` while this function is still working. """ - await self._wait_common(fd, select.KQ_FILTER_READ) + await self.wait_kevent(fd, select.KQ_FILTER_READ) @_public async def wait_writable(self, fd: int | _HasFileNo) -> None: @@ -246,7 +294,7 @@ async def wait_writable(self, fd: int | _HasFileNo) -> None: if another task calls :func:`notify_closing` while this function is still working. """ - await self._wait_common(fd, select.KQ_FILTER_WRITE) + await self.wait_kevent(fd, select.KQ_FILTER_WRITE) @_public def notify_closing(self, fd: int | _HasFileNo) -> None: diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py index 00da7b12b0..3da93191aa 100644 --- a/src/trio/_core/_tests/test_guest_mode.py +++ b/src/trio/_core/_tests/test_guest_mode.py @@ -30,7 +30,7 @@ import trio.testing from trio.abc import Clock, Instrument -from .tutil import gc_collect_harder, restore_unraisablehook +from .tutil import gc_collect_harder, restore_unraisablehook, slow if TYPE_CHECKING: from trio._channel import MemorySendChannel @@ -803,3 +803,64 @@ async def trio_main(in_host: InHost) -> None: assert cancel_scope.cancelled_caught trivial_guest_run(trio_main) + + +@slow +def test_cancel_from_host_after_process_exit_event() -> None: + # Regression test for the Process.wait() variant of + # https://github.com/python-trio/trio/issues/3500: on the kqueue backend, + # cancelling from the host after the child's exit event had been fetched + # but before it was processed raised an OSError out of cancel(). + # + # This doesn't use trivial_guest_run because the host callback needs to + # know when Trio's worker thread has fetched a batch of events and posted + # its delivery callback, i.e. when run_sync_soon_threadsafe is called. + todo: queue.Queue[Callable[[], object] | Outcome[None]] = queue.Queue() + events_fetched = threading.Event() + + def run_sync_soon_threadsafe(fn: Callable[[], object]) -> None: + events_fetched.set() + todo.put(fn) + + def run_sync_soon_not_threadsafe(fn: Callable[[], object]) -> None: + todo.put(fn) + + def done_callback(outcome: Outcome[None]) -> None: + todo.put(outcome) + + async def trio_main() -> None: + # The child must still be alive when Process.wait() registers its + # interest, and then exit while Trio is idle in the worker thread. + proc = await trio.lowlevel.open_process( + [sys.executable, "-c", "import time; time.sleep(1)"], + ) + with trio.CancelScope() as cancel_scope: + async with trio.open_nursery() as nursery: + nursery.start_soon(proc.wait) + await trio.testing.wait_all_tasks_blocked() + # No fetch can be in progress while a task is running, so from + # here on the event can only be set by the fetch that returns + # the child's exit event. + events_fetched.clear() + + def cancel_from_host() -> None: + assert events_fetched.wait(30) + cancel_scope.cancel() + + run_sync_soon_not_threadsafe(cancel_from_host) + assert cancel_scope.cancelled_caught + # Reap the child now that we're no longer cancelled. + assert await proc.wait() == 0 + + trio.lowlevel.start_guest_run( + trio_main, + run_sync_soon_threadsafe=run_sync_soon_threadsafe, + run_sync_soon_not_threadsafe=run_sync_soon_not_threadsafe, + done_callback=done_callback, + ) + while True: + item = todo.get() + if isinstance(item, Outcome): + item.unwrap() + return + item() diff --git a/src/trio/_core/_tests/test_kqueue.py b/src/trio/_core/_tests/test_kqueue.py new file mode 100644 index 0000000000..b13039df51 --- /dev/null +++ b/src/trio/_core/_tests/test_kqueue.py @@ -0,0 +1,126 @@ +"""Tests for the kqueue-specific lowlevel API (wait_kevent).""" + +from __future__ import annotations + +import select +import time + +import pytest + +import trio +from trio.testing import wait_all_tasks_blocked + +pytestmark = pytest.mark.skipif( + not hasattr(select, "kqueue"), + reason="kqueue platforms only", +) + + +async def test_wait_kevent_timer() -> None: + # Basic test of trio.lowlevel.wait_kevent() - a 20ms timer. + event = await trio.lowlevel.wait_kevent(1, select.KQ_FILTER_TIMER, data=20) + assert event.ident == 1 + assert event.filter == select.KQ_FILTER_TIMER + assert event.data >= 1 + + +async def test_wait_kevent_cancellation_deregisters() -> None: + # Tests that cancelling trio.lowlevel.wait_kevent() actually does unregister + # the event (by registering a new one and checking that it doesn't raise + # BusyResourceError). + with trio.move_on_after(0.05) as scope: + await trio.lowlevel.wait_kevent(2, select.KQ_FILTER_TIMER, data=10_000) + assert scope.cancelled_caught + event = await trio.lowlevel.wait_kevent(2, select.KQ_FILTER_TIMER, data=20) + assert event.data >= 1 + + +async def test_wait_kevent_busy_loser_arms_nothing() -> None: + # Tests that calling trio.lowlevel.wait_kevent() when an existing wait for + # that (ident, filter) pair is in progress does not affect the existing + # waiter (unlike the old wait_kevent API, where the caller would do the + # OS registration so could modify parameters of an existing wait). + # The test works by waiting for a 200ms timer then, during that, waiting + # for a 10ms timer. + result: select.kevent | None = None + + async def waiter() -> None: + nonlocal result + result = await trio.lowlevel.wait_kevent( + 3, + select.KQ_FILTER_TIMER, + data=200, + ) + + start = time.monotonic() + async with trio.open_nursery() as nursery: + nursery.start_soon(waiter) + await wait_all_tasks_blocked() + with pytest.raises(trio.BusyResourceError): + await trio.lowlevel.wait_kevent(3, select.KQ_FILTER_TIMER, data=10) + assert result is not None + # Kernel timers never fire early, so this only fails if the loser's + # 10ms registration replaced the winner's 200ms one. + assert time.monotonic() - start >= 0.15 + + +async def test_wait_kevent_deprecated_abort_func() -> None: + # Tests that the old deprecated API for trio.lowlevel.wait_kevent() still + # works. This is the case where the wait is not cancelled. + kq = trio.lowlevel.current_kqueue() + kq.control( + [ + select.kevent( + 4, + select.KQ_FILTER_TIMER, + select.KQ_EV_ADD | select.KQ_EV_ONESHOT, + 0, + 20, + ), + ], + 0, + ) + + def abort(_: object) -> trio.lowlevel.Abort: + kq.control([select.kevent(4, select.KQ_FILTER_TIMER, select.KQ_EV_DELETE)], 0) + return trio.lowlevel.Abort.SUCCEEDED + + with pytest.deprecated_call(): + event = await trio.lowlevel.wait_kevent(4, select.KQ_FILTER_TIMER, abort) + assert event.ident == 4 + assert event.data >= 1 + + +async def test_wait_kevent_deprecated_abort_func_cancel() -> None: + # Tests that the old deprecated API for trio.lowlevel.wait_kevent() still + # works. This is the case where the wait is cancelled (in particular, it + # checks that a new wait on the same (ident, filter) pair can be started + # immediately afterwards). + kq = trio.lowlevel.current_kqueue() + kq.control( + [ + select.kevent( + 5, + select.KQ_FILTER_TIMER, + select.KQ_EV_ADD | select.KQ_EV_ONESHOT, + 0, + 10_000, + ), + ], + 0, + ) + aborted = False + + def abort(_: object) -> trio.lowlevel.Abort: + nonlocal aborted + aborted = True + kq.control([select.kevent(5, select.KQ_FILTER_TIMER, select.KQ_EV_DELETE)], 0) + return trio.lowlevel.Abort.SUCCEEDED + + with trio.move_on_after(0.05) as scope: + with pytest.deprecated_call(): + await trio.lowlevel.wait_kevent(5, select.KQ_FILTER_TIMER, abort) + assert scope.cancelled_caught + assert aborted + event = await trio.lowlevel.wait_kevent(5, select.KQ_FILTER_TIMER, data=20) + assert event.data >= 1 diff --git a/src/trio/_subprocess_platform/kqueue.py b/src/trio/_subprocess_platform/kqueue.py index 0bff66ddb0..36025359a7 100644 --- a/src/trio/_subprocess_platform/kqueue.py +++ b/src/trio/_subprocess_platform/kqueue.py @@ -10,7 +10,6 @@ async def wait_child_exiting(process: _subprocess.Process) -> None: - kqueue = _core.current_kqueue() try: from select import KQ_NOTE_EXIT except ImportError: # pragma: no cover @@ -19,16 +18,12 @@ async def wait_child_exiting(process: _subprocess.Process) -> None: # I verified this value against both Darwin and FreeBSD KQ_NOTE_EXIT = 0x80000000 # type: ignore[misc] - def make_event(flags: int) -> select.kevent: - return select.kevent( + try: + await _core.wait_kevent( process.pid, - filter=select.KQ_FILTER_PROC, - flags=flags, + select.KQ_FILTER_PROC, fflags=KQ_NOTE_EXIT, ) - - try: - kqueue.control([make_event(select.KQ_EV_ADD | select.KQ_EV_ONESHOT)], 0) except ProcessLookupError: # pragma: no cover # This can supposedly happen if the process is in the process # of exiting, and it can even be the case that kqueue says the @@ -40,9 +35,3 @@ def make_event(flags: int) -> select.kevent: # for the same process simultaneously, but given the explanation # in Chromium it seems we should still keep the check. return - - def abort(_: _core.RaiseCancelT) -> _core.Abort: - kqueue.control([make_event(select.KQ_EV_DELETE)], 0) - return _core.Abort.SUCCEEDED - - await _core.wait_kevent(process.pid, select.KQ_FILTER_PROC, abort) From 3c8ee3ab85e00ec024b69f9dbb0c9fef1d23d4c9 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Sun, 20 Sep 2026 11:02:45 +0000 Subject: [PATCH 4/9] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/trio/_core/_tests/test_guest_mode.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py index 3da93191aa..2fb97d2e0b 100644 --- a/src/trio/_core/_tests/test_guest_mode.py +++ b/src/trio/_core/_tests/test_guest_mode.py @@ -756,7 +756,9 @@ async def trio_main() -> None: @pytest.mark.parametrize( - "close_first", [False, True], ids=["notify-then-close", "close-then-notify"], + "close_first", + [False, True], + ids=["notify-then-close", "close-then-notify"], ) def test_notify_closing_after_events(close_first: bool) -> None: # inspired by wrong repro in https://github.com/python-trio/trio/pull/3502 From 18129b0e7ad865116dfc4e1af5aa0fa4443626a8 Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:38:19 +0100 Subject: [PATCH 5/9] Fix CI failures --- src/trio/_core/_generated_io_kqueue.py | 17 +++++++++-------- src/trio/_core/_io_kqueue.py | 2 +- src/trio/_core/_tests/test_guest_mode.py | 13 +++++++++++-- src/trio/_core/_tests/test_kqueue.py | 11 ++++++++--- 4 files changed, 29 insertions(+), 14 deletions(-) diff --git a/src/trio/_core/_generated_io_kqueue.py b/src/trio/_core/_generated_io_kqueue.py index 25dccf9e24..ac53bb73ac 100644 --- a/src/trio/_core/_generated_io_kqueue.py +++ b/src/trio/_core/_generated_io_kqueue.py @@ -68,17 +68,17 @@ async def wait_kevent( ) -> select.kevent: """Waits for a one-shot kevent to happen. - This is a low-level function that lets you wait on a specific kevent, in case - you have a use case not covered by the IO primitives in Trio. + This is a low-level function that lets you wait on a specific kevent, + in case you have a use case not covered by the IO primitives in Trio. This registers ``kevent(ident, filter, flags, fflags, data)``, where - ``flags`` is set to ``select.KQ_EV_ADD | select.KQ_EV_ONESHOT``, and waits - for it to complete. If cancelled then it is removed with ``flags`` set to - ``select.KQ_EV_DELETE``. + ``flags`` is set to ``select.KQ_EV_ADD | select.KQ_EV_ONESHOT``, and + waits for it to complete. If cancelled then it is removed with ``flags`` + set to ``select.KQ_EV_DELETE``. Args: - ident: Value used to identify the event. The interpretation depends on the - filter but it's usually the file descriptor. + ident: Value used to identify the event. The interpretation depends + on the filter but it's usually the file descriptor. filter: Name of the kernel filter e.g. ``select.KQ_FILTER_READ``. fflags: Filter-specific flags. data: Filter-specific data. @@ -87,7 +87,8 @@ async def wait_kevent( select.kevent: The event returned from kqueue. Raises: - BusyResourceError: if another task is waiting for this (ident, filter) pair. + BusyResourceError: if another task is waiting for this + ``(ident, filter)`` pair. OSError: if the kqueue rejects the registration (for example, `ProcessLookupError` for an ``EVFILT_PROC`` ident that has already exited). diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index 86ca8202d1..f8dff9d750 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -205,7 +205,7 @@ async def wait_kevent( def abort_deprecated(raise_cancel: RaiseCancelT) -> Abort: r = abort_fn(raise_cancel) - if r is _core.Abort.SUCCEEDED: + if r is _core.Abort.SUCCEEDED: # TODO: test this branch del self._registered[key] return r diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py index 2fb97d2e0b..88fafd708b 100644 --- a/src/trio/_core/_tests/test_guest_mode.py +++ b/src/trio/_core/_tests/test_guest_mode.py @@ -757,8 +757,17 @@ async def trio_main() -> None: @pytest.mark.parametrize( "close_first", - [False, True], - ids=["notify-then-close", "close-then-notify"], + [ + pytest.param(False, id="notify-then-close"), + pytest.param( + True, + id="close-then-notify", + marks=pytest.mark.skipif( + sys.platform == "win32", + reason="Windows backend can't notify_closing an already-closed socket", + ), + ), + ], ) def test_notify_closing_after_events(close_first: bool) -> None: # inspired by wrong repro in https://github.com/python-trio/trio/pull/3502 diff --git a/src/trio/_core/_tests/test_kqueue.py b/src/trio/_core/_tests/test_kqueue.py index b13039df51..d315f1f6ce 100644 --- a/src/trio/_core/_tests/test_kqueue.py +++ b/src/trio/_core/_tests/test_kqueue.py @@ -3,7 +3,9 @@ from __future__ import annotations import select +import sys import time +from typing import TYPE_CHECKING import pytest @@ -15,6 +17,10 @@ reason="kqueue platforms only", ) +assert ( + sys.platform != "linux" and sys.platform != "win32" +) or not TYPE_CHECKING # Skip type checking when not on a kqueue platform + async def test_wait_kevent_timer() -> None: # Basic test of trio.lowlevel.wait_kevent() - a 20ms timer. @@ -81,9 +87,8 @@ async def test_wait_kevent_deprecated_abort_func() -> None: 0, ) - def abort(_: object) -> trio.lowlevel.Abort: - kq.control([select.kevent(4, select.KQ_FILTER_TIMER, select.KQ_EV_DELETE)], 0) - return trio.lowlevel.Abort.SUCCEEDED + def abort(_: object) -> trio.lowlevel.Abort: # pragma: no cover + pytest.fail("wait unexpectedly cancelled") with pytest.deprecated_call(): event = await trio.lowlevel.wait_kevent(4, select.KQ_FILTER_TIMER, abort) From 0bf983268d19c67a740b04cd512f961f7325e3e6 Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Tue, 22 Sep 2026 07:14:55 +0100 Subject: [PATCH 6/9] Fix notify_closing after close on kqueue failing to interrupt waiters With the kqueue IO manager (on BSD/macOS), if a socket is already closed then notify_closing would (until recently) raise an exception or (more recently) silently pass. Either way, any current waiters would remain hung forever. This commit keeps the recent change that ignores the exception, but instead of immediately passing it falls through to the code that wakes any current waiters with ClosedResourceError. --- newsfragments/3520.bugfix.rst | 1 + src/trio/_core/_io_kqueue.py | 16 ++++++++----- src/trio/_core/_tests/test_guest_mode.py | 4 ++-- src/trio/_core/_tests/test_io.py | 30 ++++++++++++++++++++++++ 4 files changed, 43 insertions(+), 8 deletions(-) create mode 100644 newsfragments/3520.bugfix.rst diff --git a/newsfragments/3520.bugfix.rst b/newsfragments/3520.bugfix.rst new file mode 100644 index 0000000000..d207f17a90 --- /dev/null +++ b/newsfragments/3520.bugfix.rst @@ -0,0 +1 @@ +Fix `trio.lowlevel.notify_closing` after ``close`` on kqueue failing to interrupt current waiters. diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index f8dff9d750..f72f4ffb8e 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -95,8 +95,9 @@ def process_events(self, events: EventResult) -> None: if receiver is None: # In guest mode, host callbacks can run between get_events() # and process_events(). If one cancels a wait that has already - # completed then it will remove the receiver from _registered, - # so it will be missing when we get here; we can just drop it. + # completed, or calls notify_closing() on its fd, then it will + # remove the receiver from _registered, so it will be missing + # when we get here; we can just drop it. continue if event.flags & select.KQ_EV_ONESHOT: # TODO: test this branch del self._registered[key] @@ -337,10 +338,13 @@ def notify_closing(self, fd: int | _HasFileNo) -> None: try: self._kqueue.control([event], 0) except OSError as e: - if e.errno in (errno.EBADF, errno.ENOENT): # pragma: no branch - # the event isn't in kqueue - continue - raise # pragma: no cover + # kqueue no longer knows about this event. Either the fd + # was already closed, which deregisters it automatically + # (FreeBSD reports EBADF, macOS ENOENT), or, in guest + # mode, the one-shot event was fetched by get_events() + # but not yet processed (ENOENT on both platforms). + if e.errno not in (errno.EBADF, errno.ENOENT): + raise # pragma: no cover exc = _core.ClosedResourceError("another task closed this fd") _core.reschedule(receiver, outcome.Error(exc)) del self._registered[key] diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py index 88fafd708b..32be96f0f3 100644 --- a/src/trio/_core/_tests/test_guest_mode.py +++ b/src/trio/_core/_tests/test_guest_mode.py @@ -771,14 +771,14 @@ async def trio_main() -> None: ) def test_notify_closing_after_events(close_first: bool) -> None: # inspired by wrong repro in https://github.com/python-trio/trio/pull/3502 - # either the program should silently pass or wait_writable should fail. + # wait_writable must raise ClosedResourceError pair = socket.socketpair() for sock in pair: sock.setblocking(False) async def trio_main(in_host: InHost) -> None: in_host(uh_oh) - with contextlib.suppress(trio.ClosedResourceError): + with pytest.raises(trio.ClosedResourceError): await trio.lowlevel.wait_writable(pair[0]) # blocks def uh_oh() -> None: diff --git a/src/trio/_core/_tests/test_io.py b/src/trio/_core/_tests/test_io.py index 379daa025e..4ea2021592 100644 --- a/src/trio/_core/_tests/test_io.py +++ b/src/trio/_core/_tests/test_io.py @@ -217,6 +217,36 @@ async def writer() -> None: notify_closing(a) +@pytest.mark.skipif( + sys.platform == "win32", + reason="Windows backend can't notify_closing an already-closed socket", +) +async def test_interrupted_by_close_before_notify(socketpair: SocketPair) -> None: + # Like test_interrupted_by_close, but in the alternate order the + # notify_closing docs allow: close first, then notify_closing, with no + # checkpoint in between. This ordering is only possible with a saved fd, + # since a closed socket object's fileno() is -1. + a, _b = socketpair + + async def reader() -> None: + with pytest.raises(_core.ClosedResourceError): + await trio.lowlevel.wait_readable(a) + + async def writer() -> None: + with pytest.raises(_core.ClosedResourceError): + await trio.lowlevel.wait_writable(a) + + fill_socket(a) + + async with _core.open_nursery() as nursery: + nursery.start_soon(reader) + nursery.start_soon(writer) + await wait_all_tasks_blocked() + fd = a.fileno() + a.close() + trio.lowlevel.notify_closing(fd) + + @read_socket_test @write_socket_test async def test_socket_simultaneous_read_write( From 3a72f8144b78088d74a80481c08a1d9fd511f380 Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Tue, 22 Sep 2026 09:59:34 +0100 Subject: [PATCH 7/9] kqueue: Only tolerate spurious events if we know their reason --- src/trio/_core/_io_kqueue.py | 36 ++++++++++++++++++++++++++---------- 1 file changed, 26 insertions(+), 10 deletions(-) diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index f72f4ffb8e..5fafdcae41 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -39,6 +39,13 @@ class KqueueIOManager: _registered: dict[tuple[int, int], Task | UnboundedQueue[select.kevent]] = ( attrs.Factory(dict) ) + # Keeps track of whether we are running host callbacks in guest mode + # and, if so, which events have been discarded by those callbacks + # cancelling waiters or calling notify_closing. Needed in case they + # have just completed, in which case they will appear to be spurious + # in process_events. + _events_pending: bool = False + _deregistered: set[tuple[int, int]] = attrs.Factory(set) _force_wakeup: WakeupSocketpair = attrs.Factory(WakeupSocketpair) _force_wakeup_fd: int | None = None @@ -68,7 +75,15 @@ def close(self) -> None: def force_wakeup(self) -> None: self._force_wakeup.wakeup_thread_and_signal_safe() + def _deregister(self, key: tuple[int, int]) -> None: + del self._registered[key] + if self._events_pending: + self._deregistered.add(key) + def get_events(self, timeout: float) -> EventResult: + # In guest mode, note that we are now processing host callbacks, so + # need to record any discarded events. + self._events_pending = True # max_events must be > 0 or kqueue gets cranky # and we generally want this to be strictly larger than the actual # number of events we get, so that we can tell that we've gotten @@ -91,20 +106,21 @@ def process_events(self, events: EventResult) -> None: if event.ident == self._force_wakeup_fd: self._force_wakeup.drain() continue - receiver = self._registered.get(key) - if receiver is None: - # In guest mode, host callbacks can run between get_events() - # and process_events(). If one cancels a wait that has already - # completed, or calls notify_closing() on its fd, then it will - # remove the receiver from _registered, so it will be missing - # when we get here; we can just drop it. + if key in self._deregistered: + # A host callback deregistered this key after the batch was + # fetched (see _deregistered above); its receiver is gone and + # this stale event should just be dropped. Any other missing + # key would be a bug, so let the lookup below fail loudly. continue + receiver = self._registered[key] if event.flags & select.KQ_EV_ONESHOT: # TODO: test this branch del self._registered[key] if isinstance(receiver, _core.Task): _core.reschedule(receiver, outcome.Value(event)) else: receiver.put_nowait(event) # TODO: test this line + self._events_pending = False + self._deregistered.clear() # kevent registration is complicated -- e.g. aio submission can # implicitly perform an EV_ADD, and EVFILT_PROC with NOTE_TRACK will @@ -207,7 +223,7 @@ async def wait_kevent( def abort_deprecated(raise_cancel: RaiseCancelT) -> Abort: r = abort_fn(raise_cancel) if r is _core.Abort.SUCCEEDED: # TODO: test this branch - del self._registered[key] + self._deregister(key) return r # wait_task_rescheduled does not have its return type typed @@ -227,7 +243,7 @@ def abort_deprecated(raise_cancel: RaiseCancelT) -> Abort: self._registered[key] = _core.current_task() def abort(raise_cancel: RaiseCancelT) -> Abort: - del self._registered[key] + self._deregister(key) try: event = select.kevent(ident, filter, select.KQ_EV_DELETE) self._kqueue.control([event], 0) @@ -347,7 +363,7 @@ def notify_closing(self, fd: int | _HasFileNo) -> None: raise # pragma: no cover exc = _core.ClosedResourceError("another task closed this fd") _core.reschedule(receiver, outcome.Error(exc)) - del self._registered[key] + self._deregister(key) else: # XX this is an interesting example of a case where being able # to close a queue would be useful... From a9ba4a59ca0b0ddb6dd5e770e9f678c8bb98773d Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Thu, 1 Oct 2026 16:45:37 +0100 Subject: [PATCH 8/9] Remove redundant comments about wait_task_rescheduled return type --- src/trio/_core/_io_kqueue.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index 5fafdcae41..1354b54a4f 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -226,7 +226,6 @@ def abort_deprecated(raise_cancel: RaiseCancelT) -> Abort: self._deregister(key) return r - # wait_task_rescheduled does not have its return type typed return await _core.wait_task_rescheduled( # type: ignore[no-any-return] abort_deprecated, ) @@ -270,7 +269,6 @@ def abort(raise_cancel: RaiseCancelT) -> Abort: raise return _core.Abort.SUCCEEDED - # wait_task_rescheduled does not have its return type typed return await _core.wait_task_rescheduled(abort) # type: ignore[no-any-return] @_public From fafd60fa56ffb7fd70c1a8ab543ef080e377066c Mon Sep 17 00:00:00 2001 From: Arthur Tacca <20324099+arthur-tacca@users.noreply.github.com> Date: Thu, 1 Oct 2026 20:37:15 +0100 Subject: [PATCH 9/9] Fix docs target not found (select.kevent as ref.class) --- src/trio/_core/_generated_io_kqueue.py | 2 +- src/trio/_core/_io_kqueue.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/trio/_core/_generated_io_kqueue.py b/src/trio/_core/_generated_io_kqueue.py index ac53bb73ac..3e49e013e9 100644 --- a/src/trio/_core/_generated_io_kqueue.py +++ b/src/trio/_core/_generated_io_kqueue.py @@ -84,7 +84,7 @@ async def wait_kevent( data: Filter-specific data. Returns: - select.kevent: The event returned from kqueue. + The `select.kevent` returned from kqueue. Raises: BusyResourceError: if another task is waiting for this diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index 1354b54a4f..576a9f50ca 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -189,7 +189,7 @@ async def wait_kevent( data: Filter-specific data. Returns: - select.kevent: The event returned from kqueue. + The `select.kevent` returned from kqueue. Raises: BusyResourceError: if another task is waiting for this