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 new file mode 100644 index 0000000000..4796fbb624 --- /dev/null +++ b/newsfragments/3500.bugfix.rst @@ -0,0 +1 @@ +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/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/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..3e49e013e9 100644 --- a/src/trio/_core/_generated_io_kqueue.py +++ b/src/trio/_core/_generated_io_kqueue.py @@ -59,15 +59,43 @@ 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: + The `select.kevent` 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 80d0ce6296..576a9f50ca 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 @@ -38,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 @@ -67,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 @@ -90,6 +106,12 @@ def process_events(self, events: EventResult) -> None: if event.ident == self._force_wakeup_fd: self._force_wakeup.drain() continue + 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] @@ -97,17 +119,16 @@ def process_events(self, events: EventResult) -> None: _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 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: @@ -143,56 +164,104 @@ 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: + The `select.kevent` 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: # TODO: test this branch + self._deregister(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) + 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: + self._deregister(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 @@ -200,7 +269,7 @@ def abort(_: RaiseCancelT) -> Abort: raise return _core.Abort.SUCCEEDED - await self.wait_kevent(fd, filter, abort) + return await _core.wait_task_rescheduled(abort) # type: ignore[no-any-return] @_public async def wait_readable(self, fd: int | _HasFileNo) -> None: @@ -225,7 +294,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: @@ -240,7 +309,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: @@ -283,13 +352,16 @@ 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 - # 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] + self._deregister(key) else: # XX this is an interesting example of a case where being able # to close a queue would be useful... diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py index 7acbc0f082..32be96f0f3 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 @@ -755,22 +755,123 @@ 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", + [ + 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 - # 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: # 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) + + +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) + + +@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_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( diff --git a/src/trio/_core/_tests/test_kqueue.py b/src/trio/_core/_tests/test_kqueue.py new file mode 100644 index 0000000000..d315f1f6ce --- /dev/null +++ b/src/trio/_core/_tests/test_kqueue.py @@ -0,0 +1,131 @@ +"""Tests for the kqueue-specific lowlevel API (wait_kevent).""" + +from __future__ import annotations + +import select +import sys +import time +from typing import TYPE_CHECKING + +import pytest + +import trio +from trio.testing import wait_all_tasks_blocked + +pytestmark = pytest.mark.skipif( + not hasattr(select, "kqueue"), + 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. + 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: # pragma: no cover + pytest.fail("wait unexpectedly cancelled") + + 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)