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..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,12 +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[key] + 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): @@ -161,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_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) diff --git a/src/trio/_core/_tests/test_io.py b/src/trio/_core/_tests/test_io.py index 379daa025e..860dd063a7 100644 --- a/src/trio/_core/_tests/test_io.py +++ b/src/trio/_core/_tests/test_io.py @@ -422,6 +422,27 @@ 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 ( # type: ignore[attr-defined, unused-ignore] + KqueueIOManager, + ) + + 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: # An "unnotified" close is when the user closes an fd/socket/handle # directly, without calling notify_closing first. This should never happen