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..2954f0273d --- /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 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`. Such a task +otherwise looks idle, so virtual time could jump past its timeouts while the +work is still in progress. `trio.to_thread.run_sync` gained a matching +``hold_autojump`` argument. diff --git a/newsfragments/456.bugfix.rst b/newsfragments/456.bugfix.rst new file mode 100644 index 0000000000..8f75a12725 --- /dev/null +++ b/newsfragments/456.bugfix.rst @@ -0,0 +1,7 @@ +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`, which made timeouts +in concurrent tasks fire spuriously. A thread that waits until a task has +slept in virtual time (for example on a `queue.Queue` that a task fills after +a `trio.sleep`) must now be started with ``hold_autojump=False``, or it +deadlocks. 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..0e177a6ea0 100644 --- a/src/trio/_core/_mock_clock.py +++ b/src/trio/_core/_mock_clock.py @@ -61,6 +61,14 @@ class MockClock(Clock): in `wait_all_tasks_blocked`, then the autojump feature treats that as active task and does *not* jump the clock. + Likewise, the clock does *not* jump while a task is waiting for a + thread started by `trio.to_thread.run_sync`, unless that thread + was started with ``hold_autojump=False``; a thread that waits + until a task has slept in virtual time needs that, or the two wait + for each other forever. 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..650b80b1b5 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,52 @@ 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` does the + equivalent for its worker threads, unless given ``hold_autojump=False``. + + 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 + # Checked only on normal exit, so that an exception already on its + # way out is never replaced; the finally releases the hold either way. + assert ( + getattr(GLOBAL_RUN_CONTEXT, "runner", None) is runner + ), "hold_autojump() block exited outside the run it was entered in" + 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..ec359ddcf7 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,173 @@ 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, which fails an assertion; but it + # still releases the original run's unit and leaves 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: + with pytest.raises(AssertionError, match="outside the run"): + 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..3fdb467b0c 100644 --- a/src/trio/_subprocess_platform/waitid.py +++ b/src/trio/_subprocess_platform/waitid.py @@ -83,6 +83,10 @@ async def _waitid_system_task(pid: int, event: Event) -> None: pid, abandon_on_cancel=True, limiter=waitid_limiter, + # This thread only waits for the child to exit. Where the platform + # allows, that is an I/O wait, which does not hold the autojump + # clock either. + 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..88d916bb06 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, Runner 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,279 @@ 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, which exits the block with + # GeneratorExit. That must be silent, as it is for the thread tracking + # that sits beside it, and must still release the hold. + holds: list[_AutojumpThreadHold] = [] + runners: list[Runner] = [] + + async def main() -> None: + runners.append(GLOBAL_RUN_CONTEXT.runner) + holds.append(_AutojumpThreadHold(_core.current_trio_token())) + holds[0].__enter__() + assert runners[0].autojump_holds == 1 + + _core.run(main) + holds[0].__exit__(GeneratorExit, GeneratorExit(), None) + assert holds[0].token is None + assert runners[0].autojump_holds == 0 + + +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)] + + +async def test_autojump_clock_hold_autojump_false(autojump_clock: MockClock) -> None: + # A thread that waits until a task has slept in virtual time must be + # started with hold_autojump=False: with the default, the clock would wait + # for the thread and the thread for the clock. + record = [] + proceed = threading.Event() + + async def sleeper() -> None: + await sleep(10) + record.append(("sleeper", _core.current_time())) + proceed.set() + + async def thread_task() -> None: + await to_thread_run_sync(proceed.wait, hold_autojump=False) + record.append(("thread", _core.current_time())) + + async with _core.open_nursery() as nursery: + nursery.start_soon(thread_task) + nursery.start_soon(sleeper) + + assert record == [("sleeper", 10), ("thread", 10)] diff --git a/src/trio/_threads.py b/src/trio/_threads.py index 46de93563d..c2ac2dd668 100644 --- a/src/trio/_threads.py +++ b/src/trio/_threads.py @@ -28,6 +28,8 @@ if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Generator + from contextlib import AbstractContextManager + from types import TracebackType from typing_extensions import TypeVarTuple, Unpack @@ -46,6 +48,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 +87,77 @@ def _track_active_thread() -> Generator[None, None, None]: active_threads_local.event = Event() +@define +class _AutojumpThreadHold: + """Holds the autojump clock while one ``to_thread.run_sync`` thread works. + + While the thread runs, it holds its home run's clock through + `trio.lowlevel.hold_autojump`. Three things need more than a ``with`` + block, which is why this class exists: + + * While the thread is inside ``from_thread.run``, the work is back in + Trio, where the run loop can see it, so the hold is dropped (a + virtual-time sleep in that call must still autojump) and taken again + when the call returns. That can happen in the waiting task or, for + ``abandon_on_cancel=True`` or an explicit ``trio_token``, in a system + task. + * Once the waiting task has left (for example because the thread was + abandoned on cancel), nobody would release a new hold, so a re-entrant + call that returns later must not take one. ``token`` is cleared then. + * A thread may re-enter a *different* run by passing that run's token. + That run must not be held, and the home run keeps holding, since it + cannot see the work done there. + + ``drop`` and ``restore`` only act inside a task of the home run, as + `trio.lowlevel.hold_autojump` requires. + """ + + # The home run's token; None if the thread is never held, or once the + # waiting task has left. + token: TrioToken | None + _hold: AbstractContextManager[None] | None = attrs.field( + default=None, + init=False, + ) + + def _at_home(self) -> bool: + return ( + self.token is not None and self.token is trio.lowlevel.current_trio_token() + ) + + def restore(self) -> None: + if self._hold is None and self._at_home(): + self._hold = trio.lowlevel.hold_autojump() + self._hold.__enter__() + + def drop(self) -> None: + if self._hold is not None and self._at_home(): + hold, self._hold = self._hold, None + hold.__exit__(None, None, None) + + def __enter__(self) -> None: + self.restore() + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + # The waiting task is leaving. This can be outside any run, if the run + # died and the parked coroutine is closed later; the hold is bound to + # its own run, so releasing it is still safe. Passing the exception on + # lets the hold treat that exit as the abnormal exit it is. + self.token = None + if self._hold is not None: + hold, self._hold = self._hold, None + hold.__exit__(exc_type, exc_value, traceback) + + +# 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 +235,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 +261,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: @@ -260,6 +346,7 @@ async def to_thread_run_sync( thread_name: str | None = None, abandon_on_cancel: bool = False, limiter: CapacityLimiter | None = None, + hold_autojump: bool = True, ) -> RetT: """Convert a blocking operation into an async operation using a thread. @@ -299,6 +386,17 @@ async def to_thread_run_sync( If None (the default), uses the default `~trio.CapacityLimiter`, as returned by :func:`current_default_thread_limiter`. + hold_autojump (bool): Whether the autojump clock + (`trio.testing.MockClock` with a finite ``autojump_threshold``) + waits for this thread. If True (the default), virtual time does + not jump ahead while the thread runs, except while it is inside + `trio.from_thread.run` or after it has been abandoned. Pass False + for a thread that only waits for something, the way an I/O + operation would; in particular, a thread that waits until a task + has slept in virtual time (for example on a `queue.Queue` that a + task fills after a `trio.sleep`) needs False, or the clock waits + for the thread and the thread for the clock. See also + `trio.lowlevel.hold_autojump`. **Cancellation handling**: Cancellation is a tricky issue here, because neither Python nor the operating systems it runs on provide any general @@ -388,6 +486,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 +504,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 +521,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(current_trio_token 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 +549,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..231064c6c7 100644 --- a/src/trio/_wait_for_object.py +++ b/src/trio/_wait_for_object.py @@ -47,6 +47,8 @@ async def WaitForSingleObject(obj: int | CData) -> None: cancel_handle, abandon_on_cancel=True, limiter=trio.CapacityLimiter(math.inf), + # Only waits for a handle, like an I/O wait; see waitid.py. + 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,