From f85c9642b0c47385ee1371e7031ab2d6af340d83 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 17:29:35 +0000 Subject: [PATCH 1/2] Make the autojump clock wait for to_thread.run_sync threads A task blocked in to_thread.run_sync looks idle to the run loop: it is parked in wait_task_rescheduled, and the worker thread only touches the loop when it finishes. So with a finite autojump_threshold, MockClock could advance virtual time while a thread was still doing real work, making concurrent tasks observe spuriously-fired timeouts. Count the work that is in flight off the event loop in a plain counter on the Runner, autojump_holds (the same place clock_autojump_threshold lives, avoiding an import cycle with trio._threads), and don't arm the autojump while it is nonzero -- mirroring the existing special case for tasks blocked in wait_all_tasks_blocked. When the check suppresses arming, the IO wait just sleeps until a thread reports back, so there is no polling or busy-wait. Each to_thread.run_sync call owns one unit of the count through a small private object, _AutojumpThreadHold, which only acts inside the run it belongs to, so the count is only ever changed on that run's thread. Deliberate details: * A thread that has re-entered Trio via from_thread.run is NOT counted while the re-entrant call runs: the thread is parked and the real work is back on the Trio side, where the run loop can already see it. Without this, a virtual-time sleep inside the re-entrant call would deadlock. This applies on both dispatch routes -- the waiting task's message loop, and the system task that runs the call when the thread was started with abandon_on_cancel=True or passes an explicit trio_token. The unit is taken back when the re-entrant call returns. * When a task with abandon_on_cancel=True is cancelled, its unit is given up for good even though the abandoned OS thread is still running, and a re-entrant call still in flight at that moment will not take it back. The clock therefore tracks "threads some task is waiting on", which preserves today's cancel-then-jump behavior after abandonment. * A worker thread that re-enters a *different* run (explicit trio_token) stays counted in its home run, whose clock cannot see the work being done in the other run, and the other run's count is never touched. * The threads Trio itself uses only to wait for something outside the process -- a child exiting (the waitid backend) or a Windows handle -- are not counted. Where the platform allows, those waits are I/O waits that leave the run loop idle, and the clock must not behave differently because a platform needs a thread for them. They go through a private _to_thread_run_sync(..., hold_autojump=False); the public function's signature is unchanged. The cost: a thread can now deadlock under an autojump clock. If it blocks, other than through from_thread.run, waiting for a task that first has to sleep in virtual time (queue.get() on a queue a task fills after a sleep), the clock waits for the thread and the thread for the clock, where the clock used to jump and let it finish. Fixes #456. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01USWcRiSmFFfbTeL9SNRFE1 --- newsfragments/456.bugfix.rst | 21 +++ src/trio/_core/_mock_clock.py | 12 ++ src/trio/_core/_run.py | 14 +- src/trio/_subprocess_platform/waitid.py | 8 +- src/trio/_tests/test_subprocess.py | 39 +++++ src/trio/_tests/test_threads.py | 215 +++++++++++++++++++++++- src/trio/_threads.py | 114 ++++++++++++- src/trio/_wait_for_object.py | 8 +- 8 files changed, 421 insertions(+), 10 deletions(-) create mode 100644 newsfragments/456.bugfix.rst 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/_mock_clock.py b/src/trio/_core/_mock_clock.py index e437464b65..265c6b5814 100644 --- a/src/trio/_core/_mock_clock.py +++ b/src/trio/_core/_mock_clock.py @@ -61,6 +61,18 @@ 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. + """ 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..afbd903a25 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 + # _AutojumpThreadHold in trio._threads. + autojump_holds: int = 0 # Guest mode stuff is_guest: bool = False @@ -2752,7 +2756,15 @@ 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 a to_thread.run_sync thread is running, 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 the thread + # reports back. + elif ( + runner.clock_autojump_threshold < timeout and runner.autojump_holds == 0 + ): timeout = runner.clock_autojump_threshold idle_primed = IdlePrimedTypes.AUTOJUMP_CLOCK 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..e0678fdb8d 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,214 @@ 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 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 From 33ba976e5e6357abd118dd0f6b77863effd2d74a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 16 Sep 2026 17:22:21 +0000 Subject: [PATCH 2/2] Add trio.lowlevel.hold_autojump The autojump clock now waits for to_thread.run_sync threads, but a task can also be waiting for other work that happens outside the run loop: a task parked in wait_task_rescheduled until a foreign thread, or another event loop, wakes it through TrioToken.run_sync_soon. To the run loop such a task looks idle, so virtual time can jump past its timeouts while the work is still in progress, and code outside Trio had no way to say otherwise. hold_autojump() is a context manager over the same Runner.autojump_holds count that to_thread.run_sync uses: the autojump is not armed while any unit is held. It is synchronous, counted (so it nests), does nothing under a real clock, and does not hold up wait_all_tasks_blocked. It can only be entered from inside a task; run_sync_soon callbacks qualify because they run in a system task. 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, and no check is needed at the firing site. The runner is bound on entry and released on exit, so exiting the block from inside a different run cannot touch that run's count. to_thread.run_sync keeps its own private holder on the same count, because it has to let go of its unit and re-take it around from_thread re-entry, which a with block cannot express. --- docs/source/reference-lowlevel.rst | 2 + newsfragments/3553.feature.rst | 9 ++ src/trio/_core/__init__.py | 1 + src/trio/_core/_mock_clock.py | 3 + src/trio/_core/_run.py | 48 ++++++- src/trio/_core/_tests/test_mock_clock.py | 174 ++++++++++++++++++++++- src/trio/_tests/test_threads.py | 37 +++++ src/trio/lowlevel.py | 1 + 8 files changed, 270 insertions(+), 5 deletions(-) create mode 100644 newsfragments/3553.feature.rst 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/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 265c6b5814..dd495ac1f5 100644 --- a/src/trio/_core/_mock_clock.py +++ b/src/trio/_core/_mock_clock.py @@ -73,6 +73,9 @@ class MockClock(Clock): 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 afbd903a25..195f8f9a7c 100644 --- a/src/trio/_core/_run.py +++ b/src/trio/_core/_run.py @@ -1812,7 +1812,7 @@ class Runner: # type: ignore[explicit-any] 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 - # _AutojumpThreadHold in trio._threads. + # hold_autojump() below, and _AutojumpThreadHold in trio._threads. autojump_holds: int = 0 # Guest mode stuff @@ -2757,10 +2757,11 @@ def unrolled_run( # wait_all_tasks_blocked, then those tasks will wake up without # jumping the clock, so we don't need to autojump. # - # Similarly, while a to_thread.run_sync thread is running, the + # 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 the thread + # the autojump, and the IO wait simply sleeps until that work # reports back. elif ( runner.clock_autojump_threshold < timeout and runner.autojump_holds == 0 @@ -2999,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/_tests/test_threads.py b/src/trio/_tests/test_threads.py index e0678fdb8d..4ebed51beb 100644 --- a/src/trio/_tests/test_threads.py +++ b/src/trio/_tests/test_threads.py @@ -1407,3 +1407,40 @@ async def main() -> None: _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/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,