From c2d3069c67f1b176c090a63a3d01289fa80db691 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Tue, 4 Aug 2026 13:21:56 +0530 Subject: [PATCH 1/2] Run executor submissions inline at a seeded step run_in_executor no longer fences: the function runs synchronously at an ordinary ready-queue entry labelled executor:, so the seeded draw orders it against everything else and asyncio.to_thread works under simulation. The executor argument is never used, and a future cancelled before its step runs means the function never runs at all. call_soon_threadsafe from the loop's own thread is call_soon, which is all it ever was without a second thread; from any other thread it still fences, since a real thread's timing is outside the simulation. The probe-harness tests lean on add_reader as their canonical fence now. --- src/simloop/_loop.py | 120 +++++++++++++++++++++++++++++------- tests/test_loop.py | 143 ++++++++++++++++++++++++++++++++++++++++++- tests/test_probes.py | 6 +- 3 files changed, 240 insertions(+), 29 deletions(-) diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index 2c74049..fa27651 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -8,6 +8,7 @@ import random import socket import sys +import threading import weakref from array import array from asyncio import events @@ -45,8 +46,8 @@ class SimulationDeadlockError(RuntimeError): class SimulationFenceError(NotImplementedError): """The code under simulation touched an asyncio API simloop does not simulate. - Real I/O, executors, threads, signals and subprocesses reach outside the - simulation, so they fail loudly instead of silently breaking determinism. + Real I/O, threads, signals and subprocesses reach outside the simulation, + so they fail loudly instead of silently breaking determinism. """ @@ -74,6 +75,40 @@ def _label(callback: Callable[..., object]) -> str: return type(callback).__name__ +class _ExecutorJob: + """One ``run_in_executor`` submission, run inline at its scheduled step. + + The instance carries the submitted function's qualified name as its own + ``__qualname__``, so the trace labels the step with the work rather than + the wrapper. Outcomes land on the future the way an executor worker would + land them: any ``BaseException`` is stored rather than raised, and a + future already cancelled when the step runs means the function never runs + at all — the inline equivalent of cancelling a pending work item. + """ + + def __init__( + self, + func: Callable[..., object], + args: tuple[Any, ...], + future: asyncio.Future[Any], + ) -> None: + self._func = func + self._args = args + self._future = future + self.__qualname__ = f"executor:{_label(func)}" + + def __call__(self) -> None: + future = self._future + if future.cancelled(): + return + try: + result = self._func(*self._args) + except BaseException as exc: + future.set_exception(exc) + else: + future.set_result(result) + + def _host_of(handle: asyncio.Handle) -> str: """The simulated machine whose code this handle will run. @@ -168,7 +203,7 @@ class SimLoop(asyncio.AbstractEventLoop): Coroutine scheduling is inherited from the stdlib: ``asyncio.Task`` drives every step through ``call_soon``, so controlling ``call_soon`` dispatch is sufficient to control task interleaving. Anything this class does not - implement (networking, executors, signals, threads) raises + implement (threads, signals, subprocesses) raises ``NotImplementedError`` from the base class — unsupported code fails loudly instead of silently breaking determinism. """ @@ -223,6 +258,11 @@ def __init__(self, seed: int = 0) -> None: # aiohttp reaches the network; the address is only visible in the # first call, so it is parked here until the upgrade claims it. self._sock_targets: dict[Any, tuple[Any, int]] = {} + # The one thread the simulation lives in: the creating thread until a + # run starts, the running thread from then on. call_soon_threadsafe + # compares against it — from this thread the call is call_soon, from + # any other it is a real concurrent thread and fences. + self._thread_id = threading.get_ident() self._net = SimNetwork(self) @classmethod @@ -318,6 +358,55 @@ def call_soon( self._recorder.record("schedule", self._now, seq, label, _current_host.get()) return handle + def call_soon_threadsafe( + self, + callback: Callable[[Unpack[_Ts]], object], + *args: Unpack[_Ts], + context: Context | None = None, + ) -> asyncio.Handle: + # From the simulation's own thread this is call_soon by definition — + # libraries call it defensively without ever leaving the loop, and + # nothing about the schedule changes. From any other thread the caller + # is a real concurrent thread, whose timing no seed controls, so it + # fences rather than smuggle a race into a deterministic run. + if threading.get_ident() != self._thread_id: + raise SimulationFenceError( + "simloop does not simulate 'call_soon_threadsafe' from " + "another thread: a real thread's timing is outside the " + "simulation; see docs/supported-api.md for the supported " + "asyncio subset" + ) + return self.call_soon(callback, *args, context=context) + + def run_in_executor( + self, + executor: Any, + func: Callable[[Unpack[_Ts]], Any], + *args: Unpack[_Ts], + ) -> asyncio.Future[Any]: + """Run ``func`` inline at a scheduled step instead of on a thread. + + The submission becomes an ordinary ready-queue entry — labelled + ``executor:`` in the trace — so the seeded draw orders it + against everything else and a run stays reproducible. The function + executes synchronously when that step runs, costing no virtual time, + and its result or exception lands on the returned future exactly as + an executor worker would land it. ``asyncio.to_thread`` reaches the + loop through this call, so it works under simulation too. + + Two honest consequences of running inline: the ``executor`` argument + is never used — there is no pool, and nothing runs concurrently — and + a function that blocks waiting for loop progress (joining a thread + that needs a callback, waiting on a lock a coroutine holds) hangs the + process rather than deadlocking detectably. + """ + self._check_closed() + if not callable(func): + raise TypeError(f"a callable object is expected, got {func!r}") + future: asyncio.Future[Any] = self.create_future() + self.call_soon(_ExecutorJob(func, args, future)) + return future + def call_later( self, delay: float, @@ -426,6 +515,7 @@ def run_forever(self) -> None: if self._running: raise RuntimeError("this event loop is already running") self._running = True + self._thread_id = threading.get_ident() events._set_running_loop(self) try: while not self._stopping and (self._ready or self._timers): @@ -701,32 +791,16 @@ def _timer_handle_cancelled(self, handle: asyncio.TimerHandle) -> None: # Unsupported surface # ------------------------------------------------------------------ # - # Networking, executors, subprocesses, signals, file descriptors and - # thread-safe scheduling all reach outside the simulation, so they cannot - # participate in a deterministic virtual-time run. Each one fails loudly - # with NotImplementedError instead of quietly breaking reproducibility. + # Subprocesses, signals, file descriptors and real threads all reach + # outside the simulation, so they cannot participate in a deterministic + # virtual-time run. Each one fails loudly with NotImplementedError + # instead of quietly breaking reproducibility. # # These are declared explicitly rather than inherited because the base # class marks them abstract: the signatures mirror the stubs (reproducing # the callback/args type variable where one is present) so a subclass # remains a well-typed AbstractEventLoop. - def call_soon_threadsafe( - self, - callback: Callable[[Unpack[_Ts]], object], - *args: Unpack[_Ts], - context: Context | None = None, - ) -> asyncio.Handle: - _fence("call_soon_threadsafe") - - def run_in_executor( - self, - executor: Any, - func: Callable[[Unpack[_Ts]], Any], - *args: Unpack[_Ts], - ) -> Any: - _fence("run_in_executor") - def add_reader( self, fd: Any, diff --git a/tests/test_loop.py b/tests/test_loop.py index c412572..b7c1d76 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -1,6 +1,7 @@ import asyncio import contextvars import gc +import threading import time from typing import Any, cast @@ -168,10 +169,10 @@ async def main() -> None: def test_unsupported_apis_are_fenced() -> None: loop = SimLoop(seed=0) try: - with pytest.raises(SimulationFenceError, match="run_in_executor"): - loop.run_in_executor(None, print) + with pytest.raises(SimulationFenceError, match="add_reader"): + loop.add_reader(0, print) with pytest.raises(SimulationFenceError, match="supported-api"): - loop.call_soon_threadsafe(print) + loop.subprocess_shell(None, "true") # Callers written against the stdlib contract keep working. with pytest.raises(NotImplementedError): loop.add_signal_handler(2, print) @@ -179,6 +180,142 @@ def test_unsupported_apis_are_fenced() -> None: loop.close() +def test_run_in_executor_runs_the_function_inline() -> None: + # The executor object is never used: there is no pool and nothing runs + # concurrently, so passing one that would fail on first touch proves it. + class ExplodingExecutor: + def submit(self, *args: object) -> None: + raise AssertionError("the executor object must never be used") + + async def main(loop: SimLoop) -> tuple[str, float]: + upper = await loop.run_in_executor(ExplodingExecutor(), str.upper, "sim") + return upper, loop.time() + + loop = SimLoop(seed=0) + try: + # The function runs at a scheduled step and costs no virtual time. + assert loop.run_until_complete(main(loop)) == ("SIM", 0.0) + finally: + loop.close() + + +def test_asyncio_to_thread_reaches_the_inline_executor() -> None: + async def main() -> str: + return await asyncio.to_thread("-".join, ("a", "b")) + + loop = SimLoop(seed=0) + try: + assert loop.run_until_complete(main()) == "a-b" + finally: + loop.close() + + +def test_run_in_executor_delivers_the_exception() -> None: + def blow_up() -> None: + raise OSError("disk on fire") + + async def main(loop: SimLoop) -> None: + with pytest.raises(OSError, match="disk on fire"): + await loop.run_in_executor(None, blow_up) + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main(loop)) + finally: + loop.close() + + +def test_run_in_executor_rejects_a_non_callable() -> None: + loop = SimLoop(seed=0) + try: + with pytest.raises(TypeError, match="callable"): + loop.run_in_executor(None, cast(Any, "not callable")) + finally: + loop.close() + + +def test_a_cancelled_submission_never_runs() -> None: + ran: list[str] = [] + + async def main(loop: SimLoop) -> None: + future = loop.run_in_executor(None, ran.append, "ran") + future.cancel() + with pytest.raises(asyncio.CancelledError): + await future + # The clock only advances once the ready queue is empty, so sleeping + # guarantees the submission's step has come and gone without running. + await asyncio.sleep(1.0) + assert ran == [] + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main(loop)) + finally: + loop.close() + + +def test_executor_steps_are_traced_and_deterministic() -> None: + def work() -> int: + return 7 + + async def main(loop: SimLoop) -> int: + result: int = await loop.run_in_executor(None, work) + return result + + hashes: list[str] = [] + for _ in range(2): + loop = SimLoop(seed=3) + try: + assert loop.run_until_complete(main(loop)) == 7 + finally: + loop.close() + # The submission is an ordinary scheduling step, labelled with the + # submitted function rather than the wrapper that carried it. + labels = [event.label for event in loop.trace if event.kind == "run"] + assert any( + label.startswith("executor:") and label.endswith("work") + for label in labels + ) + hashes.append(loop.trace_hash()) + assert hashes[0] == hashes[1] + + +def test_call_soon_threadsafe_on_the_loop_thread_is_call_soon() -> None: + fired: list[str] = [] + + async def main(loop: SimLoop) -> None: + loop.call_soon_threadsafe(fired.append, "fired") + await asyncio.sleep(1.0) + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main(loop)) + finally: + loop.close() + assert fired == ["fired"] + + +def test_call_soon_threadsafe_from_another_thread_is_fenced() -> None: + caught: list[BaseException] = [] + loop = SimLoop(seed=0) + + def from_elsewhere() -> None: + try: + loop.call_soon_threadsafe(print) + except BaseException as exc: + caught.append(exc) + + thread = threading.Thread(target=from_elsewhere) + try: + thread.start() + thread.join() + finally: + loop.close() + (exc,) = caught + assert isinstance(exc, SimulationFenceError) + assert "another thread" in str(exc) + + def test_an_eager_task_start_is_fenced() -> None: # An eager first step would run at creation time, before the seeded draw # could order it, so asking for one must fail loudly. Declining it is the diff --git a/tests/test_probes.py b/tests/test_probes.py index a1f9b43..2fc4467 100644 --- a/tests/test_probes.py +++ b/tests/test_probes.py @@ -26,7 +26,7 @@ async def _returns(loop: SimLoop) -> str: async def _fences(loop: SimLoop) -> str: - loop.call_soon_threadsafe(print) + loop.add_reader(0, print) return "unreachable" @@ -38,7 +38,7 @@ async def _fences_inside_a_group(loop: SimLoop) -> str: async def _fences_behind_a_cause(loop: SimLoop) -> str: try: - loop.call_soon_threadsafe(print) + loop.add_reader(0, print) except SimulationFenceError as fence: raise RuntimeError("the library wrapped it") from fence return "unreachable" @@ -64,7 +64,7 @@ def test_a_probe_that_returns_reports_what_it_exercised() -> None: def test_a_fence_is_reported_verbatim() -> None: verdict = _runner.run(_fences) assert verdict == ( - "fenced: simloop does not simulate 'call_soon_threadsafe'; " + "fenced: simloop does not simulate 'add_reader'; " "see docs/supported-api.md for the supported asyncio subset" ) From 03f4d8c7e7695bbc2e8ae470b0b8c8c2d7c5f3bc Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Tue, 4 Aug 2026 13:22:09 +0530 Subject: [PATCH 2/2] Say what moved inside the fence and what stayed out The contract pages now carry the inline executor: supported-api gains rows for run_in_executor and same-thread call_soon_threadsafe, the fenced list keeps set_default_executor and the cross-thread call, and the README's honest-limits paragraph says executor submissions stay inside the line. design.md keeps its argument against passthrough to a real pool and marks inline execution as the one place the line moved without crossing it. anyio.to_thread is called out for what it is: real worker threads spawned through no loop API, where a fence, a hang or the caller's own timeout is a race between a real thread and a virtual clock. --- CHANGELOG.md | 14 ++++++++++++++ README.md | 9 +++++++-- docs/compatibility.md | 7 +++++-- docs/design.md | 15 ++++++++++++--- docs/supported-api.md | 16 +++++++++++----- 5 files changed, 49 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bf02aa..377f754 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -99,6 +99,20 @@ package, seeds are deliberately not a strategy (a seed has no size to shrink toward, and two shrinkers aimed at one failure fight), and Hypothesis is a dev dependency of this repository rather than something simloop imports. +- Executor submissions run inline instead of fencing: `loop.run_in_executor` + executes the function at an ordinary scheduled step — ordered by the + seeded draw, labelled `executor:` in the trace, costing no + virtual time — and its result or exception lands on the returned future + the way a worker would land it, so `asyncio.to_thread` works under + simulation. The executor argument is never used (there is no pool and + nothing runs concurrently), and `set_default_executor` still fences: a + pool that would never run anything is refused rather than accepted. + `call_soon_threadsafe` from the loop's own thread is now `call_soon`, + which is all it ever was without a second thread; from any other thread + it still fences. None of this reaches `anyio.to_thread`, whose worker + threads are real ones spawned through no loop API — a real thread racing + a virtual clock ends in the cross-thread fence, a hang, or the caller's + own timeout, whichever the race picks. - `server.sockets` on a simulated server answers with an empty tuple instead of not existing, which is all aiohttp's `web.TCPSite` and websockets' `serve()` need to start; both now run their documented diff --git a/README.md b/README.md index 07e223e..c61b1ee 100644 --- a/README.md +++ b/README.md @@ -282,9 +282,14 @@ and the campaign results: ## Honest limits Code that goes through the event-loop API is supported; code that -bypasses it is fenced: threads and executors, raw socket reads and -writes, subprocesses, signals, and loop-level TLS upgrades raise +bypasses it is fenced: threads, raw socket reads and writes, +subprocesses, signals, and loop-level TLS upgrades raise `SimulationFenceError` rather than silently breaking determinism. +Executor submissions stay inside the line: `run_in_executor` runs the +function inline at a seeded scheduling step — no pool, no thread — so +`asyncio.to_thread` works, and `call_soon_threadsafe` is `call_soon` +when the caller is the loop's own thread, while a real second thread +still fences. `sock_connect` on an `AF_INET` stream socket is the exception — it is simulated, so a client that connects a socket and hands it to `create_connection` runs, while the datagram and raw variants still fence. diff --git a/docs/compatibility.md b/docs/compatibility.md index 0ec7b74..3849514 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -140,6 +140,9 @@ rather than a row — no probe on this page requests `https://`. simloop fences `start_tls` and `create_connection(ssl=...)`, but a stack that runs its handshake in memory reaches neither — it reaches a simulated network with nothing on it that speaks TLS unless the test puts it there. -- Anything that reaches outside the loop by design — threads, executors, - subprocesses, signals, real DNS. Those are fences, listed in +- Anything that reaches outside the loop by design — threads, subprocesses, + signals, real DNS. Those are fences, listed in [docs/supported-api.md](supported-api.md), not compatibility questions. + Executors left this list: `run_in_executor` now runs the function inline, + which the same page describes; `anyio.to_thread` stays out because its + worker threads are real ones the loop never sees. diff --git a/docs/design.md b/docs/design.md index 737e8fe..8d2a473 100644 --- a/docs/design.md +++ b/docs/design.md @@ -134,14 +134,14 @@ production. ## Fail loudly: the fence policy -Anything that would reach outside the simulation — executors and threads, +Anything that would reach outside the simulation — real threads, signals, subprocesses, raw sockets, `add_reader`/`add_writer`, TLS, pipes, `sendfile` — raises `SimulationFenceError` naming the exact call, and optional stdlib kwargs that would smuggle those in (`ssl=`, `sock=`, …) are rejected the same way. -The tempting alternative was best-effort passthrough: let `run_in_executor` -actually run things, keep most libraries importable, appear more compatible. +The tempting alternative was best-effort passthrough: hand `run_in_executor` +a real thread pool, keep most libraries importable, appear more compatible. That is the worst possible failure mode for this tool — a harness that *claims* determinism while real threads race underneath produces unreproducible "reproducible" failures, and every hour a user spends on a @@ -149,6 +149,15 @@ replay that doesn't replay is trust that never comes back. A loud fence converts silent wrongness into a documented boundary ([supported-api.md](supported-api.md)) plus an honest error message. +`run_in_executor` is the one place the line moved without crossing it: the +submitted function runs inline at a seeded scheduling step — no pool, no +thread, no race — so `asyncio.to_thread` works and the schedule stays the +seed's. What makes that safe is what it refuses to do: a caller's executor +object is never used, and a genuine second thread calling in +(`call_soon_threadsafe` from anywhere but the loop's thread) still fences, +because that thread's timing is the nondeterminism the fence exists to +keep out. + The same posture applies to errors inside the simulation: a run must not look green while something failed. Unhandled exceptions from fire-and-forget tasks are collected by the loop's exception handler and re-raised from diff --git a/docs/supported-api.md b/docs/supported-api.md index a749a46..2d8f56a 100644 --- a/docs/supported-api.md +++ b/docs/supported-api.md @@ -17,6 +17,8 @@ determinism. Fenced APIs raise `SimulationFenceError` (a subclass of | `run_until_complete` / `run_forever` / `stop` / `close` | Deadlock detection: raises `SimulationDeadlockError` when nothing can run | | Handle / timer cancellation | Honored and recorded in the scheduling trace | | Exception handling | Unhandled failures fail the run at `run_until_complete`; `set_exception_handler` supported | +| `loop.run_in_executor` | The function runs inline at a scheduled step, not on a thread: the submission is an ordinary ready-queue entry — labelled `executor:` in the trace — so the seeded draw orders it against everything else, and it costs no virtual time. Its result or exception lands on the returned future the way an executor worker would land it, and cancelling the future before the step runs means the function never runs. The `executor` argument is never used: there is no pool and nothing runs concurrently. `asyncio.to_thread` reaches the loop through this call, so it works; `anyio.to_thread` does not — it spawns real worker threads through no loop API, so no fence can catch the escape at the source: the worker's report back is a cross-thread `call_soon_threadsafe` that does fence, but whether a run sees that fence, hangs, or hits its own timeout first is a race between a real thread and a virtual clock. A function that blocks waiting for loop progress hangs the run rather than deadlocking detectably | +| `loop.call_soon_threadsafe` | From the loop's own thread it is `call_soon` — which is all it ever was without a second thread involved. From any other thread it fences: a real thread's timing is outside the simulation | | `sim.random` / `sim.uuid4` / `sim.time` | Seed-derived streams inside a run; stdlib fallback outside | ## Works unchanged on top of the loop @@ -78,14 +80,18 @@ reliable by construction; and addressing is IPv4-only and entirely synthetic ## Fenced Anything that reaches outside the simulation raises `SimulationFenceError`: -executors and threads (`run_in_executor`, `call_soon_threadsafe`), signal -handlers, subprocesses, file-descriptor callbacks (`add_reader` / +real threads (`call_soon_threadsafe` from any thread but the loop's own), +signal handlers, subprocesses, file-descriptor callbacks (`add_reader` / `add_writer`), loop-level TLS upgrades (`start_tls`, `create_connection(ssl=...)`), `sendfile`, pipes, and an eager task start (`create_task(eager_start=True)`), which would run a task's first step at -creation time, before the seeded draw could order it against anything. TLS -a library performs in memory reaches no loop API and so reaches no fence; -what that means in practice is in +creation time, before the seeded draw could order it against anything. +Executor *submissions* are not in that list — `run_in_executor` runs the +function inline, as the table above says — but the pool machinery around +them still is: `set_default_executor` and `shutdown_default_executor` +fence, because an executor that would never be used is refused rather than +silently accepted. TLS a library performs in memory reaches no loop API +and so reaches no fence; what that means in practice is in [docs/compatibility.md](compatibility.md). The socket calls are fenced with one exception. `sock_connect` on an