diff --git a/docs/source/reference-lowlevel.rst b/docs/source/reference-lowlevel.rst index 077f21cd27..4f173c799f 100644 --- a/docs/source/reference-lowlevel.rst +++ b/docs/source/reference-lowlevel.rst @@ -556,6 +556,8 @@ Low-level blocking .. autofunction:: wait_task_rescheduled .. autoclass:: Abort .. autofunction:: reschedule +.. autofunction:: hold_autojump + :with: Here's an example lock class implemented using :func:`wait_task_rescheduled` directly. This implementation has a diff --git a/newsfragments/3553.feature.rst b/newsfragments/3553.feature.rst new file mode 100644 index 0000000000..871897a3d7 --- /dev/null +++ b/newsfragments/3553.feature.rst @@ -0,0 +1,9 @@ +Added `trio.lowlevel.hold_autojump`, a context manager that keeps the +autojump clock (`trio.testing.MockClock` with a finite +``autojump_threshold``) from jumping while the block runs. It is meant for +code that parks a task with `trio.lowlevel.wait_task_rescheduled` until work +outside the run loop finishes -- in a thread, or on another event loop -- +and wakes it through `trio.lowlevel.TrioToken.run_sync_soon`. Without it +such a task looks idle, and virtual time can jump past timeouts while the +work is still in progress. `trio.to_thread.run_sync` already holds the +autojump for its own worker threads. diff --git a/newsfragments/456.bugfix.rst b/newsfragments/456.bugfix.rst new file mode 100644 index 0000000000..8abb256286 --- /dev/null +++ b/newsfragments/456.bugfix.rst @@ -0,0 +1,21 @@ +The autojump clock (`trio.testing.MockClock` with a finite +``autojump_threshold``) no longer jumps virtual time ahead while a task is +waiting for a thread started by `trio.to_thread.run_sync`. Such a task is now +treated as active, like a task blocked in +`trio.testing.wait_all_tasks_blocked`. Previously the run loop looked idle +while work was happening in the thread, so virtual time could race ahead of it +and fire timeouts in concurrent tasks spuriously. + +Not waited for: threads that have been abandoned (``abandon_on_cancel=True``, +after the waiting task was cancelled), threads that are currently inside +`trio.from_thread.run`, and the threads Trio itself uses only to wait for a +child process to exit or for a Windows handle, so virtual time can still pass +while a child process runs. + +This can change how existing tests behave. A virtual-time timeout no longer +fires while a `trio.to_thread.run_sync` call is still blocking; the call is +waited for in real time. A thread can also deadlock now: if it blocks, other +than through `trio.from_thread.run`, waiting for a task that first has to sleep +in virtual time (for example ``queue.get()`` on a queue that a task fills after +a `trio.sleep`), the clock waits for the thread and the thread for the clock. +Have the thread wait through `trio.from_thread.run` instead. diff --git a/src/trio/_core/__init__.py b/src/trio/_core/__init__.py index c53d9d59a0..81c2dca82c 100644 --- a/src/trio/_core/__init__.py +++ b/src/trio/_core/__init__.py @@ -46,6 +46,7 @@ current_task, current_time, current_trio_token, + hold_autojump, in_trio_run, in_trio_task, notify_closing, diff --git a/src/trio/_core/_mock_clock.py b/src/trio/_core/_mock_clock.py index e437464b65..dd495ac1f5 100644 --- a/src/trio/_core/_mock_clock.py +++ b/src/trio/_core/_mock_clock.py @@ -61,6 +61,21 @@ class MockClock(Clock): in `wait_all_tasks_blocked`, then the autojump feature treats that as active task and does *not* jump the clock. + Tasks waiting on a thread started by `trio.to_thread.run_sync` are + also treated as active: while the thread is running, real work is + still happening in real time, so the clock does *not* jump. + Virtual time advances again once the thread finishes. The threads + Trio itself uses only to wait for a child process to exit, or for + a Windows handle, do not count: virtual time can pass while a + child process runs. Beware of a thread that blocks, other than + through `trio.from_thread.run`, waiting for a task that first has + to sleep in virtual time (``queue.get()`` on a queue that a task + fills after a sleep): the clock waits for the thread and the + thread waits for the clock, a deadlock. + + Code that waits for other work outside the run loop can get the + same treatment with `trio.lowlevel.hold_autojump`. + """ def __init__(self, rate: float = 0.0, autojump_threshold: float = inf) -> None: diff --git a/src/trio/_core/_run.py b/src/trio/_core/_run.py index ddd4ea7ee2..195f8f9a7c 100644 --- a/src/trio/_core/_run.py +++ b/src/trio/_core/_run.py @@ -1810,6 +1810,10 @@ class Runner: # type: ignore[explicit-any] # If everything goes idle for this long, we call clock._autojump() clock_autojump_threshold: float = inf + # Nonzero while work that will wake a task is still running off the event + # loop; the autojump above is not armed until it returns to zero. See + # hold_autojump() below, and _AutojumpThreadHold in trio._threads. + autojump_holds: int = 0 # Guest mode stuff is_guest: bool = False @@ -2752,7 +2756,16 @@ def unrolled_run( # We use 'elif' here because if there are tasks in # wait_all_tasks_blocked, then those tasks will wake up without # jumping the clock, so we don't need to autojump. - elif runner.clock_autojump_threshold < timeout: + # + # Similarly, while anything holds the autojump (a + # to_thread.run_sync thread, or a hold_autojump() block), the + # system isn't really idle: real work is happening off the event + # loop, and it will eventually wake up a task. So we don't arm + # the autojump, and the IO wait simply sleeps until that work + # reports back. + elif ( + runner.clock_autojump_threshold < timeout and runner.autojump_holds == 0 + ): timeout = runner.clock_autojump_threshold idle_primed = IdlePrimedTypes.AUTOJUMP_CLOCK @@ -2987,6 +3000,47 @@ def current_task() -> Task: raise RuntimeError("must be called from async context") from None +@contextmanager +def hold_autojump() -> Generator[None, None, None]: + """Keep the autojump clock from jumping while the ``with`` block runs. + + Use this around a wait for work that happens outside the run loop, where + Trio cannot see it: for example, a task parked in + :func:`wait_task_rescheduled` until a thread or another event loop wakes + it through :meth:`TrioToken.run_sync_soon`. Such a task looks idle, so + without this a :class:`trio.testing.MockClock` with a finite + ``autojump_threshold`` may jump ahead, and fire timeouts, while the + work is still in progress. :func:`trio.to_thread.run_sync` already does + the equivalent for its worker threads. + + Only autojumping is affected. This does nothing under a real clock, and + tasks blocked in :func:`trio.testing.wait_all_tasks_blocked` still wake + up. Holds are counted, so they nest, and several tasks can hold at once. + While anything holds, the clock does not autojump: if the work never + finishes, a timeout that relies on autojumping never fires. + + The block must be entered and exited from inside a task of the same run. + :meth:`TrioToken.run_sync_soon` callbacks qualify, so one callback may + enter the block and a later one exit it; but such a hold only starts when + the callback runs, and the clock may jump first. A task that enters the + block itself and then waits has no such gap. + + Raises: + RuntimeError: if entered outside a task. + + """ + # Only tasks may take a hold. The run loop re-checks the count after every + # batch of task steps, and runs none between arming the autojump and + # firing it, so the count cannot rise in between. + current_task() + runner = GLOBAL_RUN_CONTEXT.runner + runner.autojump_holds += 1 + try: + yield + finally: + runner.autojump_holds -= 1 + + def current_effective_deadline() -> float: """Returns the current effective deadline for the current task. diff --git a/src/trio/_core/_tests/test_mock_clock.py b/src/trio/_core/_tests/test_mock_clock.py index 6131e7ba3f..7eecaf99d4 100644 --- a/src/trio/_core/_tests/test_mock_clock.py +++ b/src/trio/_core/_tests/test_mock_clock.py @@ -1,11 +1,13 @@ +import contextlib +import threading import time from math import inf import pytest -from trio import sleep +from trio import Event, move_on_after, sleep, sleep_forever -from ... import _core +from ... import _abc, _core from .. import wait_all_tasks_blocked from .._mock_clock import MockClock from .._run import GLOBAL_RUN_CONTEXT @@ -191,3 +193,171 @@ async def test_initialization_doesnt_mutate_runner() -> None: GLOBAL_RUN_CONTEXT.runner.clock_autojump_threshold, ) assert before == after + + +@pytest.mark.parametrize("hold", [False, True]) +async def test_hold_autojump_while_waiting_for_thread( + autojump_clock: MockClock, + hold: bool, +) -> None: + # A task parked in wait_task_rescheduled until a non-Trio thread wakes it + # through run_sync_soon looks idle to the run loop. Without a hold the + # clock jumps while the thread is still working; with one, virtual time + # waits for the thread. The thread works until ``proceed`` is set: by the + # sleeper (so only once the clock has jumped) or, when the clock is held + # and cannot jump, as soon as every task is blocked. + record = [] + proceed = threading.Event() + + async def wait_for_thread() -> None: + task = _core.current_task() + token = _core.current_trio_token() + + def thread_fn() -> None: + proceed.wait() + # Not needed by a correct clock; it gives a broken one time to jump. + time.sleep(0.05) + token.run_sync_soon(_core.reschedule, task) + + thread = threading.Thread(target=thread_fn) + thread.start() + try: + with _core.hold_autojump() if hold else contextlib.nullcontext(): + await _core.wait_task_rescheduled(lambda _: _core.Abort.FAILED) + finally: + proceed.set() + thread.join() + record.append(("thread", _core.current_time())) + + async def sleeper() -> None: + await sleep(10) + record.append(("sleeper", _core.current_time())) + proceed.set() + + async def release_thread_if_held() -> None: + await wait_all_tasks_blocked() + if hold: + proceed.set() + + async with _core.open_nursery() as nursery: + nursery.start_soon(wait_for_thread) + nursery.start_soon(sleeper) + nursery.start_soon(release_thread_if_held) + + if hold: + assert record == [("thread", 0), ("sleeper", 10)] + else: + assert record == [("sleeper", 10), ("thread", 10)] + + +async def test_hold_autojump_across_run_sync_soon_callbacks( + autojump_clock: MockClock, +) -> None: + # run_sync_soon callbacks run inside a task, so one callback can enter the + # block and a later one, here sent from another thread, can exit it. + token = _core.current_trio_token() + hold = _core.hold_autojump() + done = Event() + + def finished() -> None: + hold.__exit__(None, None, None) + done.set() + + def thread_fn() -> None: + time.sleep(0.2) + token.run_sync_soon(finished) + + # Queued before this task blocks, so it runs before the loop can go idle. + token.run_sync_soon(hold.__enter__) + thread = threading.Thread(target=thread_fn) + thread.start() + try: + with move_on_after(60) as scope: + await done.wait() + finally: + thread.join() + assert not scope.cancelled_caught + assert _core.current_time() == 0 + assert GLOBAL_RUN_CONTEXT.runner.autojump_holds == 0 + + +async def test_hold_autojump_nests_and_always_releases( + autojump_clock: MockClock, +) -> None: + runner = GLOBAL_RUN_CONTEXT.runner + with _core.hold_autojump(): + with _core.hold_autojump(): + assert runner.autojump_holds == 2 + assert runner.autojump_holds == 1 + assert runner.autojump_holds == 0 + + with pytest.raises(KeyError): + with _core.hold_autojump(): + raise KeyError + assert runner.autojump_holds == 0 + + with _core.CancelScope() as scope: + with _core.hold_autojump(): + scope.cancel() + await sleep_forever() + assert scope.cancelled_caught + assert runner.autojump_holds == 0 + + # ...and the clock autojumps again afterwards + start = _core.current_time() + await sleep(3) + assert _core.current_time() - start == 3 + + +def test_hold_autojump_with_real_clock() -> None: + # Under a real clock there is no autojump to hold; the block is harmless. + async def main() -> None: + with _core.hold_autojump(): + await sleep(0.01) + + _core.run(main) + + +def test_hold_autojump_outside_task() -> None: + with pytest.raises(RuntimeError, match=r"^must be called from async context$"): + with _core.hold_autojump(): + pass # pragma: no cover + + # Inside a run but between task steps is refused too: that is what keeps + # the count from changing between the run loop arming the autojump and + # firing it. (Exceptions in instrument hooks are swallowed, hence record.) + record = [] + + class TryToHold(_abc.Instrument): + def before_io_wait(self, timeout: float) -> None: + try: + with _core.hold_autojump(): + record.append("allowed") # pragma: no cover + except RuntimeError: + record.append("refused") + + _core.run(_core.checkpoint, instruments=[TryToHold()]) + assert record + assert set(record) == {"refused"} + + +async def test_hold_autojump_exit_inside_another_run( + autojump_clock: MockClock, +) -> None: + # A hold belongs to the run it was entered in. Exiting the block from + # inside a different run is a misuse, but it must release the original + # run's unit and leave the other run's count alone. + runner = GLOBAL_RUN_CONTEXT.runner + hold = _core.hold_autojump() + hold.__enter__() + other_run_count = [] + + async def exit_here() -> None: + hold.__exit__(None, None, None) + other_run_count.append(GLOBAL_RUN_CONTEXT.runner.autojump_holds) + + thread = threading.Thread(target=_core.run, args=(exit_here,)) + thread.start() + thread.join() + assert other_run_count == [0] + assert runner.autojump_holds == 0 diff --git a/src/trio/_subprocess_platform/waitid.py b/src/trio/_subprocess_platform/waitid.py index ebf83b4802..3d2b6820e9 100644 --- a/src/trio/_subprocess_platform/waitid.py +++ b/src/trio/_subprocess_platform/waitid.py @@ -6,7 +6,7 @@ from .. import _core, _subprocess from .._sync import CapacityLimiter, Event -from .._threads import to_thread_run_sync +from .._threads import _to_thread_run_sync assert (sys.platform != "win32" and sys.platform != "darwin") or not TYPE_CHECKING @@ -78,11 +78,13 @@ async def _waitid_system_task(pid: int, event: Event) -> None: # call to trio.run is shutting down. try: - await to_thread_run_sync( + await _to_thread_run_sync( sync_wait_reapable, - pid, + (pid,), + thread_name=None, abandon_on_cancel=True, limiter=waitid_limiter, + hold_autojump=False, ) except OSError: # If waitid fails, waitpid will fail too, so it still makes diff --git a/src/trio/_tests/test_subprocess.py b/src/trio/_tests/test_subprocess.py index dbd55f8282..e11c198c76 100644 --- a/src/trio/_tests/test_subprocess.py +++ b/src/trio/_tests/test_subprocess.py @@ -34,6 +34,7 @@ sleep_forever, ) from .._core._tests.tutil import skip_if_fbsd_pipes_broken, slow +from .._subprocess_platform import wait_child_exiting from ..lowlevel import open_process from ..testing import MockClock, assert_no_checkpoints, wait_all_tasks_blocked @@ -623,6 +624,44 @@ def on_alarm(sig: int, frame: FrameType | None) -> None: signal.signal(signal.SIGALRM, old_sigalrm) +@pytest.mark.skipif( + not wait_child_exiting.__module__.endswith("waitid"), + reason="needs the waitid backend", +) +async def test_autojump_clock_does_not_wait_for_waitid_thread( + autojump_clock: MockClock, +) -> None: + # The thread that the waitid backend uses only waits for the child to + # exit. The autojump clock must not wait for it: a virtual-time timeout + # has to fire while the child is still running, as it does on backends + # where waiting for a child is an I/O wait. + proc = await open_process(SLEEP(3600)) + try: + with move_on_after(5) as scope: + await wait_child_exiting(proc) + assert scope.cancelled_caught + assert proc.returncode is None + finally: + proc.kill() + await proc.wait() + + +async def test_autojump_clock_does_not_wait_for_child( + autojump_clock: MockClock, +) -> None: + # On every backend, a virtual-time timeout around a child process fires + # while the child is still running. + proc = await open_process(SLEEP(3600)) + try: + with move_on_after(5) as scope: + await proc.wait() + assert scope.cancelled_caught + assert proc.returncode is None + finally: + proc.kill() + await proc.wait() + + async def test_custom_deliver_cancel() -> None: custom_deliver_cancel_called = False diff --git a/src/trio/_tests/test_threads.py b/src/trio/_tests/test_threads.py index f349850ffe..4ebed51beb 100644 --- a/src/trio/_tests/test_threads.py +++ b/src/trio/_tests/test_threads.py @@ -27,9 +27,11 @@ sleep, sleep_forever, ) +from .._core._run import GLOBAL_RUN_CONTEXT from .._core._tests.test_ki import ki_self from .._core._tests.tutil import gc_collect_harder, slow from .._threads import ( + _AutojumpThreadHold, active_thread_count, current_default_thread_limiter, from_thread_check_cancelled, @@ -38,7 +40,7 @@ to_thread_run_sync, wait_all_threads_completed, ) -from ..testing import wait_all_tasks_blocked +from ..testing import MockClock, wait_all_tasks_blocked if TYPE_CHECKING: from collections.abc import AsyncGenerator, Awaitable, Callable @@ -1194,3 +1196,251 @@ async def wait_no_threads_left() -> None: async def test_wait_all_threads_completed_no_threads() -> None: await wait_all_threads_completed() assert active_thread_count() == 0 + + +async def test_autojump_clock_waits_for_threads(autojump_clock: MockClock) -> None: + # Virtual time must not jump ahead while a to_thread.run_sync thread is + # still running: the thread is real work, happening in real time. + # https://github.com/python-trio/trio/issues/456 + record = [] + + async def sleeper() -> None: + await sleep(10) + record.append(("sleeper", _core.current_time())) + + async def thread_task() -> None: + await to_thread_run_sync(time.sleep, 0.2) + record.append(("thread", _core.current_time())) + + async with _core.open_nursery() as nursery: + nursery.start_soon(sleeper) + nursery.start_soon(thread_task) + + # The thread finished first, with virtual time still untouched; only then + # did the clock jump ahead and wake the sleeper. + assert record == [("thread", 0), ("sleeper", 10)] + + +async def test_autojump_clock_and_from_thread_reentry( + autojump_clock: MockClock, +) -> None: + # A thread is not counted while it is inside from_thread.run, so a + # virtual-time sleep in the re-entrant call must still autojump. (If it + # did not, this test would deadlock.) + def thread_fn() -> None: + from_thread_run(sleep, 10) + + start = _core.current_time() + await to_thread_run_sync(thread_fn) + assert _core.current_time() - start == 10 + + +async def test_autojump_clock_and_from_thread_reentry_abandon_on_cancel( + autojump_clock: MockClock, +) -> None: + # With abandon_on_cancel=True, from_thread.run is dispatched to the + # system nursery instead of the waiting task; the thread must not be + # counted as active on that route either while the re-entrant call + # runs. (If it were, this test would deadlock.) + def thread_fn() -> None: + from_thread_run(sleep, 10) + + start = _core.current_time() + await to_thread_run_sync(thread_fn, abandon_on_cancel=True) + assert _core.current_time() - start == 10 + + +async def test_autojump_clock_and_from_thread_reentry_explicit_token( + autojump_clock: MockClock, +) -> None: + # Passing an explicit trio_token also sends from_thread.run to the system + # nursery, even from a to_thread.run_sync worker thread, and the thread + # must not be counted during the re-entrant call on that route either. + token = _core.current_trio_token() + + def thread_fn() -> None: + from_thread_run(sleep, 10, trio_token=token) + + start = _core.current_time() + await to_thread_run_sync(thread_fn) + assert _core.current_time() - start == 10 + + +@pytest.mark.parametrize("abandon_on_cancel", [False, True]) +async def test_autojump_clock_waits_again_after_from_thread_reentry( + autojump_clock: MockClock, + abandon_on_cancel: bool, +) -> None: + # When the re-entrant call returns, the thread is doing real work again + # and the clock must wait for it again, on both dispatch routes + # (abandon_on_cancel=True sends from_thread.run to the system nursery). + record = [] + + def thread_fn() -> None: + from_thread_run(sleep, 1) + time.sleep(0.2) + + async def thread_task() -> None: + await to_thread_run_sync(thread_fn, abandon_on_cancel=abandon_on_cancel) + record.append(("thread", _core.current_time())) + + async def sleeper() -> None: + await sleep(10) + record.append(("sleeper", _core.current_time())) + + async with _core.open_nursery() as nursery: + nursery.start_soon(thread_task) + nursery.start_soon(sleeper) + + assert record == [("thread", 1), ("sleeper", 10)] + + +async def test_autojump_clock_thread_abandoned_during_reentry( + autojump_clock: MockClock, +) -> None: + # If the waiting task is cancelled (abandoning the thread) while the + # thread's re-entrant from_thread.run call is still running in the + # system nursery, the thread must not be counted again when that call + # finishes: nobody is left to stop counting it. + finish_reentry = Event() + in_reentry = Event() + thread_finished = Event() + + async def reentrant_afn() -> None: + in_reentry.set() + await finish_reentry.wait() + + def thread_fn() -> None: + from_thread_run(reentrant_afn) + # The re-entrant call has been fully dealt with by the time it + # returns here, so this is a reliable signal for the checks below. + from_thread_run_sync(thread_finished.set) + + async with _core.open_nursery() as nursery: + nursery.start_soon( + partial(to_thread_run_sync, thread_fn, abandon_on_cancel=True), + ) + await in_reentry.wait() + nursery.cancel_scope.cancel() + + # The thread is abandoned and its re-entrant call is still pending. + assert GLOBAL_RUN_CONTEXT.runner.autojump_holds == 0 + finish_reentry.set() + await thread_finished.wait() + assert GLOBAL_RUN_CONTEXT.runner.autojump_holds == 0 + start = _core.current_time() + await sleep(7) + assert _core.current_time() - start == 7 + + +async def test_autojump_clock_cross_run_reentry(autojump_clock: MockClock) -> None: + # A worker thread can re-enter a *different* trio run by passing that + # run's token to from_thread.run. Its unit of autojump_holds belongs to + # its home run (A, this one): while the thread is inside run B, A must + # still count it, and B's count must not be touched. Run B relies on its + # own autojump *during* the re-entrant call, so this test hangs if B's + # count is driven negative; both runs' counts are checked afterwards. + a_runner = GLOBAL_RUN_CONTEXT.runner + a_count_during: list[int] = [] + b_token: list[_core.TrioToken] = [] + b_stop: list[Event] = [] + b_ready = threading.Event() + b_results: list[float] = [] + + def run_b() -> None: + async def b_main() -> None: + b_token.append(_core.current_trio_token()) + b_stop.append(Event()) + b_ready.set() + await b_stop[0].wait() + # After the cross-run re-entry: B's count must be untouched and + # B's autojump must still work. + b_results.append(GLOBAL_RUN_CONTEXT.runner.autojump_holds) + start = _core.current_time() + await sleep(5) + b_results.append(_core.current_time() - start) + + _core.run(b_main, clock=MockClock(autojump_threshold=0)) + + b_thread = threading.Thread(target=run_b) + b_thread.start() + + async def afn_in_b() -> int: + # Runs in B's system nursery. This virtual sleep needs B's autojump: + # if the cross-run dispatch decremented B's count, it hangs forever. + await sleep(2) + # Run A is quiescent, parked waiting for the worker thread. + a_count_during.append(a_runner.autojump_holds) + return 42 + + def worker() -> int: + b_ready.wait() + result: int = from_thread_run(afn_in_b, trio_token=b_token[0]) + return result + + try: + assert await to_thread_run_sync(worker) == 42 + finally: + # Stop run B even if the above failed, or its thread keeps pytest + # alive. B has started by now: the worker waits for it first. + b_token[0].run_sync_soon(b_stop[0].set) + await to_thread_run_sync(b_thread.join) + assert a_count_during == [1] + assert b_results == [0, 5] + # Run A's own count is back to zero, and A's autojump still works. + assert GLOBAL_RUN_CONTEXT.runner.autojump_holds == 0 + start = _core.current_time() + await sleep(3) + assert _core.current_time() - start == 3 + + +def test_autojump_clock_thread_hold_exit_outside_run() -> None: + # If a run dies while a task is parked in to_thread.run_sync, the task's + # coroutine is closed later, outside any run. Leaving the block must then + # be silent, as it is for the thread tracking that sits beside it. + holds: list[_AutojumpThreadHold] = [] + + async def main() -> None: + holds.append(_AutojumpThreadHold(GLOBAL_RUN_CONTEXT.runner)) + holds[0].__enter__() + + _core.run(main) + holds[0].__exit__(None, None, None) + assert holds[0].runner is None + + +async def test_autojump_clock_thread_and_hold_autojump_together( + autojump_clock: MockClock, +) -> None: + # to_thread.run_sync and trio.lowlevel.hold_autojump share one count, so + # the clock waits for whichever finishes last. wait_all_tasks_blocked is + # not held up by either. + record = [] + + async def sleeper() -> None: + await sleep(10) + record.append(("sleeper", _core.current_time())) + + proceed = threading.Event() + + def thread_fn() -> None: + proceed.wait() + # Not needed by a correct clock; it gives a broken one time to jump. + time.sleep(0.05) + + async def thread_task() -> None: + await to_thread_run_sync(thread_fn) + record.append(("thread", _core.current_time())) + + async with _core.open_nursery() as nursery: + nursery.start_soon(sleeper) + nursery.start_soon(thread_task) + try: + with _core.hold_autojump(): + await wait_all_tasks_blocked() + assert GLOBAL_RUN_CONTEXT.runner.autojump_holds == 2 + finally: + proceed.set() + + # The block ended before the thread did, and the clock still waited. + assert record == [("thread", 0), ("sleeper", 10)] diff --git a/src/trio/_threads.py b/src/trio/_threads.py index 46de93563d..a4af7ad0c5 100644 --- a/src/trio/_threads.py +++ b/src/trio/_threads.py @@ -23,14 +23,17 @@ enable_ki_protection, start_thread_soon, ) +from ._core._run import GLOBAL_RUN_CONTEXT from ._sync import CapacityLimiter, Event from ._util import coroutine_or_error if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Generator + from types import TracebackType from typing_extensions import TypeVarTuple, Unpack + from trio._core._run import Runner from trio._core._traps import RaiseCancelT Ts = TypeVarTuple("Ts") @@ -46,6 +49,7 @@ class _ParentTaskData(threading.local): abandon_on_cancel: bool cancel_register: list[RaiseCancelT | None] task_register: list[trio.lowlevel.Task | None] + autojump_hold: _AutojumpThreadHold PARENT_TASK_DATA = _ParentTaskData() @@ -84,6 +88,66 @@ def _track_active_thread() -> Generator[None, None, None]: active_threads_local.event = Event() +def _current_runner() -> Runner | None: + # A parked coroutine can be closed outside any run, after a crashed or + # abandoned run; there is no runner then, and the hold must do nothing. + return getattr(GLOBAL_RUN_CONTEXT, "runner", None) + + +@define +class _AutojumpThreadHold: + """One ``to_thread.run_sync`` call's unit of ``Runner.autojump_holds``. + + While the unit is taken (``held``), the autojump clock waits for this + call's thread. The unit is given up while the thread re-enters Trio + through ``from_thread.run``: the thread is then parked and the work is + back on the Trio side, where the run loop can see it, so a virtual-time + sleep there must still autojump. + + ``drop`` and ``restore`` do nothing unless called inside ``runner``, the + run the unit belongs to, and inside that run they are only called from + task steps. So the count only changes in task steps, and the run loop + runs none between arming the autojump and firing it. A thread that + re-enters a *different* run stays counted in its home run, which cannot + see the work done in the other one. + + ``runner`` is None if the thread is never counted, and becomes None when + the waiting task leaves, so a re-entrant call that finishes after the + thread was abandoned does not take the unit back. + """ + + runner: Runner | None + held: bool = False + + def __enter__(self) -> None: + self.restore() + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.drop() + self.runner = None + + def drop(self) -> None: + runner = self.runner + if self.held and runner is _current_runner() and runner is not None: + self.held = False + runner.autojump_holds -= 1 + + def restore(self) -> None: + runner = self.runner + if not self.held and runner is _current_runner() and runner is not None: + self.held = True + runner.autojump_holds += 1 + + +# For threads that to_thread.run_sync did not start: never counted. +_NOT_COUNTED = _AutojumpThreadHold(None) + + async def wait_all_threads_completed() -> None: """Wait until no threads are still running tasks. @@ -161,6 +225,14 @@ class Run(Generic[RetT]): # type: ignore[explicit-any] init=False, factory=stdlib_queue.SimpleQueue, ) + # The autojump unit of the to_thread.run_sync call this message comes + # from, or _NOT_COUNTED. Captured at construction, on the sending thread, + # because PARENT_TASK_DATA is thread-local. + hold: _AutojumpThreadHold = attrs.field( + init=False, + repr=False, + factory=lambda: getattr(PARENT_TASK_DATA, "autojump_hold", _NOT_COUNTED), + ) @disable_ki_protection async def unprotected_afn(self) -> RetT: @@ -179,7 +251,11 @@ async def run(self) -> None: self.queue.put_nowait(result) async def run_system(self) -> None: + # The calling thread is not counted during the re-entrant call; see + # _AutojumpThreadHold. + self.hold.drop() result = await outcome.acapture(self.unprotected_afn) + self.hold.restore() self.queue.put_nowait(result) def run_in_host_task(self, token: TrioToken) -> None: @@ -344,6 +420,30 @@ async def to_thread_run_sync( Exception: Whatever ``sync_fn(*args)`` raises. """ + return await _to_thread_run_sync( + sync_fn, + args, + thread_name=thread_name, + abandon_on_cancel=abandon_on_cancel, + limiter=limiter, + hold_autojump=True, + ) + + +@enable_ki_protection +async def _to_thread_run_sync( + sync_fn: Callable[[Unpack[Ts]], RetT], + args: tuple[Unpack[Ts]], + thread_name: str | None, + abandon_on_cancel: bool, + limiter: CapacityLimiter | None, + *, + hold_autojump: bool, +) -> RetT: + # hold_autojump=False is for Trio's own threads that only wait for + # something outside the process, like a child exiting. On other platforms + # that is an I/O wait, which leaves the run loop idle; the autojump clock + # must not treat it differently because this platform needs a thread. await trio.lowlevel.checkpoint_if_cancelled() # raise early if abandon_on_cancel.__bool__ raises # and give a new name to ensure mypy knows it's never None @@ -388,6 +488,7 @@ def worker_fn() -> RetT: PARENT_TASK_DATA.abandon_on_cancel = abandon_bool PARENT_TASK_DATA.cancel_register = cancel_register PARENT_TASK_DATA.task_register = task_register + PARENT_TASK_DATA.autojump_hold = hold try: ret = context.run(sync_fn, *args) @@ -405,6 +506,7 @@ def worker_fn() -> RetT: del PARENT_TASK_DATA.abandon_on_cancel del PARENT_TASK_DATA.cancel_register del PARENT_TASK_DATA.task_register + del PARENT_TASK_DATA.autojump_hold context = contextvars.copy_context() # Trio doesn't use current_async_library_cvar, but if someone @@ -421,7 +523,8 @@ def deliver_worker_fn_result(result: outcome.Outcome[RetT]) -> None: current_trio_token.run_sync_soon(report_back_in_trio_thread_fn, result) await limiter.acquire_on_behalf_of(placeholder) - with _track_active_thread(): + hold = _AutojumpThreadHold(GLOBAL_RUN_CONTEXT.runner if hold_autojump else None) + with _track_active_thread(), hold: try: start_thread_soon(worker_fn, deliver_worker_fn_result, thread_name) except: @@ -448,7 +551,14 @@ def abort(raise_cancel: RaiseCancelT) -> trio.lowlevel.Abort: if isinstance(msg_from_thread, outcome.Outcome): return msg_from_thread.unwrap() elif isinstance(msg_from_thread, Run): - await msg_from_thread.run() + # Not counted during the re-entrant call; see + # _AutojumpThreadHold. (RunSync, below, runs within this + # task step, so the run loop never observes it.) + hold.drop() + try: + await msg_from_thread.run() + finally: + hold.restore() elif isinstance(msg_from_thread, RunSync): msg_from_thread.run_sync() else: # pragma: no cover, internal debugging guard TODO: use assert_never diff --git a/src/trio/_wait_for_object.py b/src/trio/_wait_for_object.py index 53832513a3..cde507c7ff 100644 --- a/src/trio/_wait_for_object.py +++ b/src/trio/_wait_for_object.py @@ -13,6 +13,7 @@ kernel32, raise_winerror, ) +from ._threads import _to_thread_run_sync async def WaitForSingleObject(obj: int | CData) -> None: @@ -41,12 +42,13 @@ async def WaitForSingleObject(obj: int | CData) -> None: # that we can use to cancel the thread. cancel_handle = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL) try: - await trio.to_thread.run_sync( + await _to_thread_run_sync( WaitForMultipleObjects_sync, - handle, - cancel_handle, + (handle, cancel_handle), + thread_name=None, abandon_on_cancel=True, limiter=trio.CapacityLimiter(math.inf), + hold_autojump=False, ) finally: # Clean up our cancel handle. In case we get here because this task was diff --git a/src/trio/lowlevel.py b/src/trio/lowlevel.py index b6621d47ed..49904dea94 100644 --- a/src/trio/lowlevel.py +++ b/src/trio/lowlevel.py @@ -36,6 +36,7 @@ currently_ki_protected as currently_ki_protected, disable_ki_protection as disable_ki_protection, enable_ki_protection as enable_ki_protection, + hold_autojump as hold_autojump, in_trio_run as in_trio_run, in_trio_task as in_trio_task, notify_closing as notify_closing,