From 1b598c99eb8fcafe9ae27d7fb5e78fc099f8ecc8 Mon Sep 17 00:00:00 2001 From: Caitlyn Lye <57263725+LyundipLye@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:35:14 +0100 Subject: [PATCH 1/5] Handle kqueue cancellation after event fetch Fixes #3500. --- newsfragments/3500.bugfix.rst | 1 + src/trio/_core/_io_kqueue.py | 4 +++- src/trio/_core/_tests/test_guest_mode.py | 16 ++++++++++++++++ 3 files changed, 20 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..436f0ec8b4 --- /dev/null +++ b/newsfragments/3500.bugfix.rst @@ -0,0 +1 @@ +Fixed a race on kqueue backends where cancelling an I/O wait in guest mode after its event had been fetched could raise :exc:`TrioInternalError`. diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index 80d0ce6296..70695e5e2d 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -90,7 +90,9 @@ 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: + 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 7acbc0f082..f8e07fc25b 100644 --- a/src/trio/_core/_tests/test_guest_mode.py +++ b/src/trio/_core/_tests/test_guest_mode.py @@ -774,3 +774,19 @@ def uh_oh() -> None: trivial_guest_run(trio_main) for sock in pair: sock.close() + + +@pytest.mark.skipif( + sys.platform in {"win32", "linux"}, + reason="requires the kqueue backend", +) +def test_cancel_io_after_events_fetched() -> None: + async def trio_main(in_host: InHost) -> None: + left, right = socket.socketpair() + with left, right, trio.CancelScope() as scope: + in_host(scope.cancel) + await trio.lowlevel.wait_writable(left) + + assert scope.cancelled_caught + + trivial_guest_run(trio_main) From 6e9ce0507ba0d366c9e6d0459adc59d4a3f6ee7b Mon Sep 17 00:00:00 2001 From: Caitlyn Lye <57263725+LyundipLye@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:55:14 +0100 Subject: [PATCH 2/5] Limit stale kqueue handling to cancelled waits --- src/trio/_core/_io_kqueue.py | 14 +++++++++++--- src/trio/_core/_tests/test_io.py | 17 +++++++++++++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py index 70695e5e2d..6ce6658ca3 100644 --- a/src/trio/_core/_io_kqueue.py +++ b/src/trio/_core/_io_kqueue.py @@ -38,6 +38,7 @@ class KqueueIOManager: _registered: dict[tuple[int, int], Task | UnboundedQueue[select.kevent]] = ( attrs.Factory(dict) ) + _cancelled_keys: set[tuple[int, int]] = attrs.Factory(set) _force_wakeup: WakeupSocketpair = attrs.Factory(WakeupSocketpair) _force_wakeup_fd: int | None = None @@ -85,14 +86,20 @@ def get_events(self, timeout: float) -> EventResult: return events def process_events(self, events: EventResult) -> None: + cancelled_keys = self._cancelled_keys + self._cancelled_keys = set() for event in events: key = (event.ident, event.filter) if event.ident == self._force_wakeup_fd: self._force_wakeup.drain() continue - receiver = self._registered.get(key) - if receiver is None: - continue + try: + receiver = self._registered[key] + except KeyError: + if key in cancelled_keys: + # See test_cancel_io_after_events_fetched. + continue + raise if event.flags & select.KQ_EV_ONESHOT: # TODO: test this branch del self._registered[key] if isinstance(receiver, _core.Task): @@ -163,6 +170,7 @@ async def wait_kevent( def abort(raise_cancel: RaiseCancelT) -> Abort: r = abort_func(raise_cancel) if r is _core.Abort.SUCCEEDED: # TODO: test this branch + self._cancelled_keys.add(key) del self._registered[key] return r diff --git a/src/trio/_core/_tests/test_io.py b/src/trio/_core/_tests/test_io.py index 379daa025e..ea7d3837e1 100644 --- a/src/trio/_core/_tests/test_io.py +++ b/src/trio/_core/_tests/test_io.py @@ -422,6 +422,23 @@ def check( check(expected_monitors=0, expected_readers=1, expected_writers=0) +@pytest.mark.skipif( + sys.platform in {"win32", "linux"}, + reason="requires the kqueue backend", +) +def test_kqueue_process_events_rejects_unknown_key() -> None: + if sys.platform != "win32" and sys.platform != "linux": + from .._io_kqueue import KqueueIOManager + + manager = KqueueIOManager() + try: + event = select.kevent(1_000_000, select.KQ_FILTER_READ) + with pytest.raises(KeyError): + manager.process_events([event]) + finally: + manager.close() + + async def test_can_survive_unnotified_close() -> None: # An "unnotified" close is when the user closes an fd/socket/handle # directly, without calling notify_closing first. This should never happen From a844504bfcc0da1e0ef1a380f5980cc228136a1d Mon Sep 17 00:00:00 2001 From: Local Bridge via ChatGPT <57263725+LyundipLye@users.noreply.github.com> Date: Sat, 19 Sep 2026 18:19:49 +0100 Subject: [PATCH 3/5] Remove redundant platform guard --- src/trio/_core/_tests/test_io.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/src/trio/_core/_tests/test_io.py b/src/trio/_core/_tests/test_io.py index ea7d3837e1..b1438e409f 100644 --- a/src/trio/_core/_tests/test_io.py +++ b/src/trio/_core/_tests/test_io.py @@ -427,16 +427,15 @@ def check( reason="requires the kqueue backend", ) def test_kqueue_process_events_rejects_unknown_key() -> None: - if sys.platform != "win32" and sys.platform != "linux": - from .._io_kqueue import KqueueIOManager + from .._io_kqueue import KqueueIOManager - manager = KqueueIOManager() - try: - event = select.kevent(1_000_000, select.KQ_FILTER_READ) - with pytest.raises(KeyError): - manager.process_events([event]) - finally: - manager.close() + manager = KqueueIOManager() + try: + event = select.kevent(1_000_000, select.KQ_FILTER_READ) + with pytest.raises(KeyError): + manager.process_events([event]) + finally: + manager.close() async def test_can_survive_unnotified_close() -> None: From 258a21281b4a4231ff5f20ab75fbc19759b49a3a Mon Sep 17 00:00:00 2001 From: Local Bridge via ChatGPT <57263725+LyundipLye@users.noreply.github.com> Date: Sat, 19 Sep 2026 18:31:24 +0100 Subject: [PATCH 4/5] Avoid type-checking kqueue-only regression test --- src/trio/_core/_tests/test_io.py | 24 +++++++++++------------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/src/trio/_core/_tests/test_io.py b/src/trio/_core/_tests/test_io.py index b1438e409f..d8d05055fa 100644 --- a/src/trio/_core/_tests/test_io.py +++ b/src/trio/_core/_tests/test_io.py @@ -422,20 +422,18 @@ def check( check(expected_monitors=0, expected_readers=1, expected_writers=0) -@pytest.mark.skipif( - sys.platform in {"win32", "linux"}, - reason="requires the kqueue backend", -) -def test_kqueue_process_events_rejects_unknown_key() -> None: - from .._io_kqueue import KqueueIOManager +if sys.platform not in {"win32", "linux"} and not TYPE_CHECKING: - manager = KqueueIOManager() - try: - event = select.kevent(1_000_000, select.KQ_FILTER_READ) - with pytest.raises(KeyError): - manager.process_events([event]) - finally: - manager.close() + def test_kqueue_process_events_rejects_unknown_key() -> None: + from .._io_kqueue import KqueueIOManager + + manager = KqueueIOManager() + try: + event = select.kevent(1_000_000, select.KQ_FILTER_READ) + with pytest.raises(KeyError): + manager.process_events([event]) + finally: + manager.close() async def test_can_survive_unnotified_close() -> None: From 55753d813a235e898f4c7f4dc8c44211b16bcc64 Mon Sep 17 00:00:00 2001 From: Local Bridge via ChatGPT <57263725+LyundipLye@users.noreply.github.com> Date: Sun, 20 Sep 2026 10:01:20 +0100 Subject: [PATCH 5/5] Use skip marker for kqueue-only regression test --- src/trio/_core/_tests/test_io.py | 29 ++++++++++++++++++----------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/src/trio/_core/_tests/test_io.py b/src/trio/_core/_tests/test_io.py index d8d05055fa..860dd063a7 100644 --- a/src/trio/_core/_tests/test_io.py +++ b/src/trio/_core/_tests/test_io.py @@ -422,18 +422,25 @@ def check( check(expected_monitors=0, expected_readers=1, expected_writers=0) -if sys.platform not in {"win32", "linux"} and not TYPE_CHECKING: - - def test_kqueue_process_events_rejects_unknown_key() -> None: - from .._io_kqueue import KqueueIOManager +@pytest.mark.skipif( + sys.platform in {"win32", "linux"}, + reason="requires the kqueue backend", +) +def test_kqueue_process_events_rejects_unknown_key() -> None: + from .._io_kqueue import ( # type: ignore[attr-defined, unused-ignore] + KqueueIOManager, + ) - manager = KqueueIOManager() - try: - event = select.kevent(1_000_000, select.KQ_FILTER_READ) - with pytest.raises(KeyError): - manager.process_events([event]) - finally: - manager.close() + manager = KqueueIOManager() + try: + event = select.kevent( # type: ignore[attr-defined, unused-ignore] + 1_000_000, + select.KQ_FILTER_READ, # type: ignore[attr-defined, unused-ignore] + ) + with pytest.raises(KeyError): + manager.process_events([event]) + finally: + manager.close() async def test_can_survive_unnotified_close() -> None: