From 683e5b7ea23693538d11adec0a3b76fdb0db364a Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 09:46:17 +0530 Subject: [PATCH 1/8] Fence unsupported asyncio APIs with a named error --- src/simloop/__init__.py | 10 ++++- src/simloop/_loop.py | 87 ++++++++++++++++++++++++----------------- tests/test_loop.py | 9 ++++- 3 files changed, 66 insertions(+), 40 deletions(-) diff --git a/src/simloop/__init__.py b/src/simloop/__init__.py index abaf976..aee4371 100644 --- a/src/simloop/__init__.py +++ b/src/simloop/__init__.py @@ -1,8 +1,14 @@ """simloop — deterministic simulation testing for Python asyncio.""" -from simloop._loop import SimLoop, SimulationDeadlockError +from simloop._loop import SimLoop, SimulationDeadlockError, SimulationFenceError from simloop._trace import TraceEvent __version__ = "0.0.1.dev0" -__all__ = ["SimLoop", "SimulationDeadlockError", "TraceEvent", "__version__"] +__all__ = [ + "SimLoop", + "SimulationDeadlockError", + "SimulationFenceError", + "TraceEvent", + "__version__", +] diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index 75dcf96..b32f042 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -10,7 +10,7 @@ from asyncio import events from collections.abc import Callable from contextvars import Context -from typing import Any, TypeVarTuple, Unpack +from typing import Any, NoReturn, TypeVarTuple, Unpack from simloop._trace import TraceEvent, TraceRecorder @@ -25,6 +25,21 @@ 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. + """ + + +def _fence(api: str) -> NoReturn: + raise SimulationFenceError( + f"simloop does not simulate {api!r}; " + "see docs/supported-api.md for the supported asyncio subset" + ) + + def _label(callback: Callable[..., object]) -> str: # Labels feed the trace hash, so they must be stable across processes: # qualified names only, never repr() (which can embed memory addresses). @@ -303,7 +318,7 @@ def call_soon_threadsafe( *args: Unpack[_Ts], context: Context | None = None, ) -> asyncio.Handle: - raise NotImplementedError("call_soon_threadsafe is not supported") + _fence("call_soon_threadsafe") def run_in_executor( self, @@ -311,7 +326,7 @@ def run_in_executor( func: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts], ) -> Any: - raise NotImplementedError("run_in_executor is not supported") + _fence("run_in_executor") def add_reader( self, @@ -319,7 +334,7 @@ def add_reader( callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts], ) -> None: - raise NotImplementedError("add_reader is not supported") + _fence("add_reader") def add_writer( self, @@ -327,7 +342,7 @@ def add_writer( callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts], ) -> None: - raise NotImplementedError("add_writer is not supported") + _fence("add_writer") def add_signal_handler( self, @@ -335,94 +350,94 @@ def add_signal_handler( callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts], ) -> None: - raise NotImplementedError("add_signal_handler is not supported") + _fence("add_signal_handler") def set_default_executor(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("set_default_executor is not supported") + _fence("set_default_executor") def set_task_factory(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("set_task_factory is not supported") + _fence("set_task_factory") def get_task_factory(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("get_task_factory is not supported") + _fence("get_task_factory") def set_exception_handler(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("set_exception_handler is not supported") + _fence("set_exception_handler") def get_exception_handler(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("get_exception_handler is not supported") + _fence("get_exception_handler") def shutdown_asyncgens(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("shutdown_asyncgens is not supported") + _fence("shutdown_asyncgens") def shutdown_default_executor(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("shutdown_default_executor is not supported") + _fence("shutdown_default_executor") def getaddrinfo(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("getaddrinfo is not supported") + _fence("getaddrinfo") def getnameinfo(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("getnameinfo is not supported") + _fence("getnameinfo") def create_connection(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("create_connection is not supported") + _fence("create_connection") def create_server(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("create_server is not supported") + _fence("create_server") def start_tls(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("start_tls is not supported") + _fence("start_tls") def sendfile(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sendfile is not supported") + _fence("sendfile") def sock_sendfile(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_sendfile is not supported") + _fence("sock_sendfile") def create_datagram_endpoint(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("create_datagram_endpoint is not supported") + _fence("create_datagram_endpoint") def connect_read_pipe(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("connect_read_pipe is not supported") + _fence("connect_read_pipe") def connect_write_pipe(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("connect_write_pipe is not supported") + _fence("connect_write_pipe") def subprocess_shell(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("subprocess_shell is not supported") + _fence("subprocess_shell") def subprocess_exec(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("subprocess_exec is not supported") + _fence("subprocess_exec") def remove_reader(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("remove_reader is not supported") + _fence("remove_reader") def remove_writer(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("remove_writer is not supported") + _fence("remove_writer") def remove_signal_handler(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("remove_signal_handler is not supported") + _fence("remove_signal_handler") def sock_recv(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_recv is not supported") + _fence("sock_recv") def sock_recv_into(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_recv_into is not supported") + _fence("sock_recv_into") def sock_sendall(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_sendall is not supported") + _fence("sock_sendall") def sock_connect(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_connect is not supported") + _fence("sock_connect") def sock_accept(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_accept is not supported") + _fence("sock_accept") def sock_sendto(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_sendto is not supported") + _fence("sock_sendto") def sock_recvfrom(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_recvfrom is not supported") + _fence("sock_recvfrom") def sock_recvfrom_into(self, *args: Any, **kwargs: Any) -> Any: - raise NotImplementedError("sock_recvfrom_into is not supported") + _fence("sock_recvfrom_into") diff --git a/tests/test_loop.py b/tests/test_loop.py index bed7bae..63c42fe 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -4,7 +4,7 @@ import pytest -from simloop import SimLoop, SimulationDeadlockError +from simloop import SimLoop, SimulationDeadlockError, SimulationFenceError def test_runs_a_coroutine_to_completion() -> None: @@ -166,8 +166,13 @@ async def main() -> None: def test_unsupported_apis_are_fenced() -> None: loop = SimLoop(seed=0) try: - with pytest.raises(NotImplementedError): + with pytest.raises(SimulationFenceError, match="run_in_executor"): loop.run_in_executor(None, print) + with pytest.raises(SimulationFenceError, match="supported-api"): + loop.call_soon_threadsafe(print) + # Callers written against the stdlib contract keep working. + with pytest.raises(NotImplementedError): + loop.add_signal_handler(2, print) finally: loop.close() From c753bacbd04b5fc43b8a96ebb46f14e5fb2b941f Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 09:50:49 +0530 Subject: [PATCH 2/8] Trace cancelled callbacks and timers --- src/simloop/_loop.py | 10 +++++++-- src/simloop/_trace.py | 2 +- tests/test_loop.py | 52 +++++++++++++++++++++++++++++++++++++++++++ tests/test_trace.py | 8 +++++++ 4 files changed, 69 insertions(+), 3 deletions(-) diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index b32f042..de58cd2 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -154,13 +154,17 @@ def _step(self) -> None: index = self._rng.randrange(len(self._ready)) seq, label, handle = self._ready.pop(index) if handle.cancelled(): + # The draw itself is a scheduling decision, so a skipped handle + # must appear in the trace for the replay proof to stay complete. + self._recorder.record("cancel", self._now, seq, label) return self._recorder.record("run", self._now, seq, label) handle._run() def _advance_clock(self) -> None: while self._timers and self._timers[0][3].cancelled(): - heapq.heappop(self._timers) + _, seq, label, _timer = heapq.heappop(self._timers) + self._recorder.record("cancel", self._now, seq, label) if not self._timers: raise SimulationDeadlockError( "nothing left to run: no ready callbacks and no pending timers" @@ -169,7 +173,9 @@ def _advance_clock(self) -> None: self._recorder.record("advance", self._now, -1, "") while self._timers and self._timers[0][0] <= self._now: _, seq, label, timer = heapq.heappop(self._timers) - if not timer.cancelled(): + if timer.cancelled(): + self._recorder.record("cancel", self._now, seq, label) + else: self._ready.append((seq, label, timer)) # ------------------------------------------------------------------ diff --git a/src/simloop/_trace.py b/src/simloop/_trace.py index 12411f7..c62af99 100644 --- a/src/simloop/_trace.py +++ b/src/simloop/_trace.py @@ -10,7 +10,7 @@ from dataclasses import dataclass from typing import Literal -EventKind = Literal["schedule", "run", "advance"] +EventKind = Literal["schedule", "run", "advance", "cancel"] @dataclass(frozen=True, slots=True) diff --git a/tests/test_loop.py b/tests/test_loop.py index 63c42fe..db5bf75 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -189,3 +189,55 @@ async def main() -> None: kinds = {event.kind for event in loop.trace} assert kinds == {"schedule", "run", "advance"} assert len(loop.trace_hash()) == 64 + + +def test_cancelled_callback_is_traced_not_run() -> None: + fired: list[str] = [] + + async def main() -> None: + loop = asyncio.get_running_loop() + handle = loop.call_soon(fired.append, "x") + handle.cancel() + await asyncio.sleep(1.0) + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main()) + finally: + loop.close() + assert fired == [] + assert "cancel" in {event.kind for event in loop.trace} + + +def test_cancelled_timer_is_traced() -> None: + async def main() -> None: + loop = asyncio.get_running_loop() + timer = loop.call_later(1.0, print, "never") + timer.cancel() + timer.cancel() + await asyncio.sleep(2.0) + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main()) + finally: + loop.close() + cancels = [event for event in loop.trace if event.kind == "cancel"] + assert len(cancels) == 1 + + +def test_call_at_in_the_past_runs_at_current_time() -> None: + fired: list[float] = [] + + async def main() -> None: + loop = asyncio.get_running_loop() + await asyncio.sleep(5.0) + loop.call_at(1.0, lambda: fired.append(loop.time())) + await asyncio.sleep(0.1) + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main()) + finally: + loop.close() + assert fired == [5.0] diff --git a/tests/test_trace.py b/tests/test_trace.py index dfa6959..134fff8 100644 --- a/tests/test_trace.py +++ b/tests/test_trace.py @@ -27,3 +27,11 @@ def test_different_order_hashes_differently() -> None: second.record("run", 0.0, 1, "g") second.record("run", 0.0, 0, "f") assert first.hash() != second.hash() + + +def test_cancel_events_change_the_hash() -> None: + first, second = TraceRecorder(), TraceRecorder() + first.record("run", 0.0, 0, "f") + second.record("run", 0.0, 0, "f") + second.record("cancel", 0.0, 1, "g") + assert first.hash() != second.hash() From 032a447c3699af211b795465fe50989aa166ad69 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 09:56:16 +0530 Subject: [PATCH 3/8] Allow installing a custom exception handler --- src/simloop/_loop.py | 37 +++++++++++++++++------ tests/test_loop.py | 72 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 100 insertions(+), 9 deletions(-) diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index de58cd2..300afd0 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -16,6 +16,8 @@ _Ts = TypeVarTuple("_Ts") +_ExceptionHandler = Callable[[asyncio.AbstractEventLoop, dict[str, Any]], object] + class SimulationDeadlockError(RuntimeError): """No runnable callbacks or timers remain, but the awaited future is not done. @@ -83,6 +85,7 @@ def __init__(self, seed: int = 0) -> None: # Exceptions from callbacks and fire-and-forget tasks accumulate here # and are re-raised from run_until_complete once the loop stops. self._unhandled: list[BaseException] = [] + self._exception_handler: _ExceptionHandler | None = None # ------------------------------------------------------------------ # Introspection @@ -276,7 +279,32 @@ def create_task( # Error handling # ------------------------------------------------------------------ + def set_exception_handler(self, handler: _ExceptionHandler | None) -> None: + if handler is not None and not callable(handler): + raise TypeError( + f"a callable object or None is expected, got {handler!r}" + ) + self._exception_handler = handler + + def get_exception_handler(self) -> _ExceptionHandler | None: + return self._exception_handler + def call_exception_handler(self, context: dict[str, Any]) -> None: + if self._exception_handler is None: + self.default_exception_handler(context) + return + try: + self._exception_handler(self, context) + except (SystemExit, KeyboardInterrupt): + raise + except BaseException as handler_error: + # A broken exception handler is itself a failure that must not + # vanish: surface it, and fall back to the default policy for + # the original context. + self._unhandled.append(handler_error) + self.default_exception_handler(context) + + def default_exception_handler(self, context: dict[str, Any]) -> None: # A simulation must not swallow errors. Collect real failures so that # run_until_complete re-raises them once the loop stops. This covers # fire-and-forget tasks, whose exceptions otherwise reach here only @@ -292,9 +320,6 @@ def call_exception_handler(self, context: dict[str, Any]) -> None: "simloop:", context.get("message", "unhandled error"), file=sys.stderr ) - def default_exception_handler(self, context: dict[str, Any]) -> None: - self.call_exception_handler(context) - def get_debug(self) -> bool: return False @@ -367,12 +392,6 @@ def set_task_factory(self, *args: Any, **kwargs: Any) -> Any: def get_task_factory(self, *args: Any, **kwargs: Any) -> Any: _fence("get_task_factory") - def set_exception_handler(self, *args: Any, **kwargs: Any) -> Any: - _fence("set_exception_handler") - - def get_exception_handler(self, *args: Any, **kwargs: Any) -> Any: - _fence("get_exception_handler") - def shutdown_asyncgens(self, *args: Any, **kwargs: Any) -> Any: _fence("shutdown_asyncgens") diff --git a/tests/test_loop.py b/tests/test_loop.py index db5bf75..e3562a1 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -1,6 +1,7 @@ import asyncio import gc import time +from typing import Any, cast import pytest @@ -241,3 +242,74 @@ async def main() -> None: finally: loop.close() assert fired == [5.0] + + +def test_custom_exception_handler_takes_responsibility() -> None: + seen: list[dict[str, Any]] = [] + + async def fail() -> None: + raise ValueError("handled elsewhere") + + async def main() -> str: + asyncio.create_task(fail()) + await asyncio.sleep(1.0) + return "finished" + + loop = SimLoop(seed=0) + loop.set_exception_handler(lambda _loop, context: seen.append(context)) + try: + assert loop.run_until_complete(main()) == "finished" + finally: + loop.close() + assert len(seen) == 1 + assert isinstance(seen[0].get("exception"), ValueError) + + +def test_broken_exception_handler_fails_the_run() -> None: + def broken(loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None: + raise KeyError("handler bug") + + async def fail() -> None: + raise ValueError("original") + + async def main() -> None: + asyncio.create_task(fail()) + await asyncio.sleep(1.0) + + loop = SimLoop(seed=0) + loop.set_exception_handler(broken) + try: + with pytest.raises(KeyError, match="handler bug"): + loop.run_until_complete(main()) + finally: + loop.close() + + +def test_clearing_the_handler_restores_failing_by_default() -> None: + async def fail() -> None: + raise ValueError("must surface") + + async def main() -> None: + asyncio.create_task(fail()) + await asyncio.sleep(1.0) + + loop = SimLoop(seed=0) + handler = lambda _loop, _context: None # noqa: E731 + loop.set_exception_handler(handler) + assert loop.get_exception_handler() is handler + loop.set_exception_handler(None) + assert loop.get_exception_handler() is None + try: + with pytest.raises(ValueError, match="must surface"): + loop.run_until_complete(main()) + finally: + loop.close() + + +def test_non_callable_exception_handler_is_rejected() -> None: + loop = SimLoop(seed=0) + try: + with pytest.raises(TypeError): + loop.set_exception_handler(cast(Any, 42)) + finally: + loop.close() From a20888722e19b3990d0e7fcc540748746641f5cb Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 10:10:10 +0530 Subject: [PATCH 4/8] Support task factories in create_task --- src/simloop/_loop.py | 34 ++++++++++++++----- tests/test_loop.py | 77 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 103 insertions(+), 8 deletions(-) diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index 300afd0..2a14a5f 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -10,7 +10,10 @@ from asyncio import events from collections.abc import Callable from contextvars import Context -from typing import Any, NoReturn, TypeVarTuple, Unpack +from typing import TYPE_CHECKING, Any, NoReturn, TypeVarTuple, Unpack + +if TYPE_CHECKING: + from asyncio.events import _TaskFactory from simloop._trace import TraceEvent, TraceRecorder @@ -86,6 +89,7 @@ def __init__(self, seed: int = 0) -> None: # and are re-raised from run_until_complete once the loop stops. self._unhandled: list[BaseException] = [] self._exception_handler: _ExceptionHandler | None = None + self._task_factory: _TaskFactory | None = None # ------------------------------------------------------------------ # Introspection @@ -273,7 +277,27 @@ def create_task( context: Context | None = None, ) -> asyncio.Task[Any]: self._check_closed() - return asyncio.Task(coro, loop=self, name=name, context=context) + if self._task_factory is None: + return asyncio.Task(coro, loop=self, name=name, context=context) + factory: Any = self._task_factory + task: asyncio.Task[Any] = ( + factory(self, coro) + if context is None + else factory(self, coro, context=context) + ) + if name is not None: + set_name = getattr(task, "set_name", None) + if set_name is not None: + set_name(name) + return task + + def set_task_factory(self, factory: _TaskFactory | None) -> None: + if factory is not None and not callable(factory): + raise TypeError("task factory must be a callable or None") + self._task_factory = factory + + def get_task_factory(self) -> _TaskFactory | None: + return self._task_factory # ------------------------------------------------------------------ # Error handling @@ -386,12 +410,6 @@ def add_signal_handler( def set_default_executor(self, *args: Any, **kwargs: Any) -> Any: _fence("set_default_executor") - def set_task_factory(self, *args: Any, **kwargs: Any) -> Any: - _fence("set_task_factory") - - def get_task_factory(self, *args: Any, **kwargs: Any) -> Any: - _fence("get_task_factory") - def shutdown_asyncgens(self, *args: Any, **kwargs: Any) -> Any: _fence("shutdown_asyncgens") diff --git a/tests/test_loop.py b/tests/test_loop.py index e3562a1..89b8e3c 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -1,4 +1,5 @@ import asyncio +import contextvars import gc import time from typing import Any, cast @@ -313,3 +314,79 @@ def test_non_callable_exception_handler_is_rejected() -> None: loop.set_exception_handler(cast(Any, 42)) finally: loop.close() + + +def test_task_factory_is_used_by_create_task() -> None: + created: list[asyncio.Task[Any]] = [] + + def factory(loop: asyncio.AbstractEventLoop, coro: Any) -> asyncio.Task[Any]: + task = asyncio.Task(coro, loop=loop) + created.append(task) + return task + + async def add(a: int, b: int) -> int: + return a + b + + async def main() -> int: + task = asyncio.get_running_loop().create_task(add(2, 3), name="adder") + return await task + + loop = SimLoop(seed=0) + loop.set_task_factory(factory) + try: + assert loop.get_task_factory() is factory + assert loop.run_until_complete(main()) == 5 + finally: + loop.close() + assert any(task.get_name() == "adder" for task in created) + + +def test_task_factory_receives_context_kwarg() -> None: + received: list[dict[str, Any]] = [] + + def factory( + loop: asyncio.AbstractEventLoop, coro: Any, **kwargs: Any + ) -> asyncio.Task[Any]: + received.append(kwargs) + return asyncio.Task(coro, loop=loop, **kwargs) + + async def noop() -> None: + return None + + async def main() -> None: + loop = asyncio.get_running_loop() + await loop.create_task(noop()) + await loop.create_task(noop(), context=contextvars.copy_context()) + + loop = SimLoop(seed=0) + loop.set_task_factory(factory) + try: + loop.run_until_complete(main()) + finally: + loop.close() + kw_sets = [set(kwargs) for kwargs in received] + assert set() in kw_sets # a call with no context kwarg + assert {"context"} in kw_sets # the context-carrying call + + +def test_clearing_the_task_factory_restores_default() -> None: + async def add(a: int, b: int) -> int: + return a + b + + loop = SimLoop(seed=0) + loop.set_task_factory(lambda loop, coro: asyncio.Task(coro, loop=loop)) + loop.set_task_factory(None) + try: + assert loop.get_task_factory() is None + assert loop.run_until_complete(add(2, 3)) == 5 + finally: + loop.close() + + +def test_non_callable_task_factory_is_rejected() -> None: + loop = SimLoop(seed=0) + try: + with pytest.raises(TypeError): + loop.set_task_factory(cast(Any, 42)) + finally: + loop.close() From 8df04f88dd355c5869792ffa5f8040e50de8b619 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 10:15:26 +0530 Subject: [PATCH 5/8] Add seeded shims for random, uuid and time --- src/simloop/__init__.py | 3 ++ src/simloop/_loop.py | 7 ++++ src/simloop/_sim.py | 56 +++++++++++++++++++++++++++++++ tests/test_sim.py | 73 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 139 insertions(+) create mode 100644 src/simloop/_sim.py create mode 100644 tests/test_sim.py diff --git a/src/simloop/__init__.py b/src/simloop/__init__.py index aee4371..6a3fd3c 100644 --- a/src/simloop/__init__.py +++ b/src/simloop/__init__.py @@ -1,14 +1,17 @@ """simloop — deterministic simulation testing for Python asyncio.""" from simloop._loop import SimLoop, SimulationDeadlockError, SimulationFenceError +from simloop._sim import Sim, sim from simloop._trace import TraceEvent __version__ = "0.0.1.dev0" __all__ = [ + "Sim", "SimLoop", "SimulationDeadlockError", "SimulationFenceError", "TraceEvent", "__version__", + "sim", ] diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index 2a14a5f..34dfe9a 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -73,6 +73,13 @@ class SimLoop(asyncio.AbstractEventLoop): def __init__(self, seed: int = 0) -> None: self._seed = seed self._rng = random.Random(seed) + # Streams for user-facing entropy, derived from the seed but kept + # separate from the scheduler's RNG: user draws must never perturb + # scheduling order, and scheduling must never perturb user values. + # String seeding hashes via SHA-512, so the streams are stable + # across processes and interpreter versions. + self._user_random = random.Random(f"{seed}:random") + self._uuid_random = random.Random(f"{seed}:uuid") self._now = 0.0 # Ready entries are (seq, label, handle); seq is a global creation # counter that gives every scheduled callback a stable identity. diff --git a/src/simloop/_sim.py b/src/simloop/_sim.py new file mode 100644 index 0000000..42d717d --- /dev/null +++ b/src/simloop/_sim.py @@ -0,0 +1,56 @@ +"""Seeded stand-ins for common sources of nondeterminism in user code. + +Inside a running SimLoop, ``sim.random``, ``sim.uuid4`` and ``sim.time`` draw +from streams derived from the loop's seed, so their values replay exactly. +Outside a simulation they fall back to the stdlib, so code written against +them behaves normally in production. +""" + +from __future__ import annotations + +import asyncio +import random +import time +import uuid + +from simloop._loop import SimLoop + +_fallback_random = random.Random() + + +def _running_sim_loop() -> SimLoop | None: + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return None + return loop if isinstance(loop, SimLoop) else None + + +class Sim: + """Facade over the running SimLoop's user-facing entropy and clock.""" + + @property + def random(self) -> random.Random: + """A ``random.Random``: seeded per loop inside a run, real outside.""" + loop = _running_sim_loop() + if loop is None: + return _fallback_random + return loop._user_random + + def uuid4(self) -> uuid.UUID: + """A version-4 UUID: seed-derived inside a run, real outside.""" + loop = _running_sim_loop() + if loop is None: + return uuid.uuid4() + return uuid.UUID(int=loop._uuid_random.getrandbits(128), version=4) + + def time(self) -> float: + """Seconds: virtual loop time inside a run (starting at 0.0, not the + epoch), wall-clock ``time.time()`` outside.""" + loop = _running_sim_loop() + if loop is None: + return time.time() + return loop.time() + + +sim = Sim() diff --git a/tests/test_sim.py b/tests/test_sim.py new file mode 100644 index 0000000..4a668c5 --- /dev/null +++ b/tests/test_sim.py @@ -0,0 +1,73 @@ +import asyncio +import time +import uuid as uuid_module + +from simloop import SimLoop, sim + + +def _collect(seed: int) -> tuple[list[float], list[str], float]: + async def main() -> tuple[list[float], list[str], float]: + draws = [sim.random.random() for _ in range(5)] + uuids = [str(sim.uuid4()) for _ in range(3)] + await asyncio.sleep(1.5) + return draws, uuids, sim.time() + + loop = SimLoop(seed) + try: + result: tuple[list[float], list[str], float] = loop.run_until_complete(main()) + return result + finally: + loop.close() + + +def test_shims_replay_exactly_per_seed() -> None: + assert _collect(1) == _collect(1) + + +def test_shims_diverge_across_seeds() -> None: + assert _collect(1) != _collect(2) + + +def test_sim_time_is_virtual_inside_a_run() -> None: + assert _collect(0)[2] == 1.5 + + +def test_sim_uuid4_is_a_valid_version_4_uuid() -> None: + for text in _collect(3)[1]: + assert uuid_module.UUID(text).version == 4 + + +def test_user_draws_do_not_perturb_scheduling() -> None: + async def with_draws() -> None: + for _ in range(10): + sim.random.random() + sim.uuid4() + await asyncio.sleep(1.0) + + async def without_draws() -> None: + await asyncio.sleep(1.0) + + hashes = [] + for main in (with_draws, without_draws): + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main()) + finally: + loop.close() + hashes.append(loop.trace_hash()) + assert hashes[0] == hashes[1] + + +def test_shims_fall_back_to_stdlib_outside_a_simulation() -> None: + assert len({sim.random.random() for _ in range(3)}) == 3 + assert sim.uuid4() != sim.uuid4() + assert abs(sim.time() - time.time()) < 5.0 + + +def test_shims_fall_back_on_the_stock_loop() -> None: + async def main() -> float: + sim.random.random() + assert sim.uuid4() != sim.uuid4() + return sim.time() + + assert abs(asyncio.run(main()) - time.time()) < 5.0 From 34b6f476ef0722cc90380bda0ee6a32a239bb5c2 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 10:19:17 +0530 Subject: [PATCH 6/8] Add cross-process replay stability tests --- .github/workflows/ci.yml | 2 +- pyproject.toml | 2 ++ tests/replay_workload.py | 62 ++++++++++++++++++++++++++++++++++++++++ tests/test_hardening.py | 56 ++++++++++++++++++++++++++++++++++++ 4 files changed, 121 insertions(+), 1 deletion(-) create mode 100644 tests/replay_workload.py create mode 100644 tests/test_hardening.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 87c2b00..60f7c76 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,5 +19,5 @@ jobs: with: python-version: ${{ matrix.python }} - run: uv sync - - run: uv run pytest -q + - run: uv run pytest -q -m "slow or not slow" - run: uv run mypy diff --git a/pyproject.toml b/pyproject.toml index 9d05408..df0e692 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,8 @@ build-backend = "hatchling.build" [tool.pytest.ini_options] testpaths = ["tests"] +addopts = "-m 'not slow'" +markers = ["slow: long-running replay-stability checks"] [tool.mypy] strict = true diff --git a/tests/replay_workload.py b/tests/replay_workload.py new file mode 100644 index 0000000..1734587 --- /dev/null +++ b/tests/replay_workload.py @@ -0,0 +1,62 @@ +"""Reference client/server workload for replay-stability checks. + +Importable for in-process runs; also runnable as a script — +``python tests/replay_workload.py `` prints one line: +`` ``. Jittered sleeps and UUID-tagged replies +pull the seeded shim streams into the replay proof alongside scheduling. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import sys + +from simloop import SimLoop, sim + +_CLIENTS = ("alice", "bob", "carol") +_REQUESTS_EACH = 3 + +_Request = tuple[str, int, "asyncio.Queue[str]"] + + +async def _serve(requests: "asyncio.Queue[_Request]", total: int) -> None: + for _ in range(total): + name, number, inbox = await requests.get() + await asyncio.sleep(sim.random.uniform(0.001, 0.01)) + await inbox.put(f"{name}:{number}:{sim.uuid4()}") + + +async def _request_all(name: str, requests: "asyncio.Queue[_Request]") -> list[str]: + inbox: asyncio.Queue[str] = asyncio.Queue() + replies: list[str] = [] + for number in range(_REQUESTS_EACH): + await requests.put((name, number, inbox)) + await asyncio.sleep(0.01 * (number + 1)) + replies.append(await inbox.get()) + return replies + + +async def _exchange() -> dict[str, list[str]]: + requests: asyncio.Queue[_Request] = asyncio.Queue() + server = asyncio.create_task(_serve(requests, len(_CLIENTS) * _REQUESTS_EACH)) + clients = { + name: asyncio.create_task(_request_all(name, requests)) for name in _CLIENTS + } + replies = {name: await task for name, task in clients.items()} + await server + return replies + + +def run(seed: int) -> str: + loop = SimLoop(seed) + try: + replies = loop.run_until_complete(_exchange()) + finally: + loop.close() + digest = hashlib.sha256(repr(sorted(replies.items())).encode()).hexdigest() + return f"{loop.trace_hash()} {digest}" + + +if __name__ == "__main__": + print(run(int(sys.argv[1]))) diff --git a/tests/test_hardening.py b/tests/test_hardening.py new file mode 100644 index 0000000..f90b93f --- /dev/null +++ b/tests/test_hardening.py @@ -0,0 +1,56 @@ +"""Long-running replay-stability checks (marked slow; CI runs them).""" + +from __future__ import annotations + +import importlib.util +import os +import subprocess +import sys +from pathlib import Path +from typing import Any + +import pytest + +_SCRIPT = Path(__file__).with_name("replay_workload.py") + + +def _load_workload() -> Any: + spec = importlib.util.spec_from_file_location("replay_workload", _SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _run_child(seed: int, hashseed: str | None) -> str: + env = os.environ.copy() + env.pop("PYTHONHASHSEED", None) + if hashseed is not None: + env["PYTHONHASHSEED"] = hashseed + result = subprocess.run( + [sys.executable, str(_SCRIPT), str(seed)], + capture_output=True, + text=True, + env=env, + check=True, + timeout=60, + ) + assert result.stderr == "" + return result.stdout.strip() + + +@pytest.mark.slow +def test_hundred_reruns_per_seed_are_stable() -> None: + workload = _load_workload() + for seed in range(5): + results = {workload.run(seed) for _ in range(100)} + assert len(results) == 1, f"seed {seed} produced diverging runs" + + +@pytest.mark.slow +def test_replay_is_stable_across_processes_and_hash_seeds() -> None: + workload = _load_workload() + for seed in (0, 7): + results = {_run_child(seed, hs) for hs in (None, "0", "1", "random")} + results.add(workload.run(seed)) + assert len(results) == 1, f"seed {seed} diverged across processes" From b6ca937fea1ffa37caf39d6fe467d5a7b2561246 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 10:24:03 +0530 Subject: [PATCH 7/8] Add scheduling overhead benchmark --- .github/workflows/ci.yml | 1 + benchmarks/overhead.py | 60 ++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- 3 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 benchmarks/overhead.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 60f7c76..f9a1ce2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,3 +21,4 @@ jobs: - run: uv sync - run: uv run pytest -q -m "slow or not slow" - run: uv run mypy + - run: uv run python benchmarks/overhead.py --tasks 10 --rounds 10 diff --git a/benchmarks/overhead.py b/benchmarks/overhead.py new file mode 100644 index 0000000..e93919a --- /dev/null +++ b/benchmarks/overhead.py @@ -0,0 +1,60 @@ +"""Compare scheduling overhead: SimLoop against the stock asyncio loop. + +The workload passes a token around a ring of queue-connected tasks, so the +measured cost is almost purely task switching and queue hand-off. Run with +``python benchmarks/overhead.py [--tasks N] [--rounds M]``. SimLoop's number +includes trace recording — that is the honest price of replayability. +""" + +from __future__ import annotations + +import argparse +import asyncio +import time + +from simloop import SimLoop + + +async def _token_ring(n_tasks: int, rounds: int) -> None: + queues: list[asyncio.Queue[int]] = [asyncio.Queue() for _ in range(n_tasks)] + + async def worker(index: int) -> None: + for _ in range(rounds): + token = await queues[index].get() + await queues[(index + 1) % n_tasks].put(token + 1) + + workers = [asyncio.create_task(worker(i)) for i in range(n_tasks)] + await queues[0].put(0) + await asyncio.gather(*workers) + + +def _measure(loop: asyncio.AbstractEventLoop, n_tasks: int, rounds: int) -> float: + start = time.perf_counter() + try: + loop.run_until_complete(_token_ring(n_tasks, rounds)) + finally: + loop.close() + return time.perf_counter() - start + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Measure SimLoop scheduling overhead vs stock asyncio." + ) + parser.add_argument("--tasks", type=int, default=100) + parser.add_argument("--rounds", type=int, default=200) + args = parser.parse_args() + hops = args.tasks * args.rounds + + stock = _measure(asyncio.new_event_loop(), args.tasks, args.rounds) + simulated = _measure(SimLoop(seed=0), args.tasks, args.rounds) + + print(f"{args.tasks} tasks x {args.rounds} rounds = {hops} hops") + print(f"{'loop':<10}{'total s':>10}{'us/hop':>10}") + for label, seconds in (("asyncio", stock), ("SimLoop", simulated)): + print(f"{label:<10}{seconds:>10.4f}{seconds / hops * 1e6:>10.2f}") + print(f"overhead: {simulated / stock:.2f}x") + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index df0e692..294ef17 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,4 +33,4 @@ markers = ["slow: long-running replay-stability checks"] [tool.mypy] strict = true -files = ["src", "tests"] +files = ["src", "tests", "benchmarks"] From 102fccd150dd4c463c8990e954f8ec483ced8260 Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Sun, 12 Jul 2026 10:29:08 +0530 Subject: [PATCH 8/8] Document the supported asyncio subset --- README.md | 10 ++-- docs/supported-api.md | 41 ++++++++++++++ tests/test_stdlib_surface.py | 105 +++++++++++++++++++++++++++++++++++ 3 files changed, 152 insertions(+), 4 deletions(-) create mode 100644 docs/supported-api.md create mode 100644 tests/test_stdlib_surface.py diff --git a/README.md b/README.md index 6f5e9b9..3cccf7d 100644 --- a/README.md +++ b/README.md @@ -13,10 +13,12 @@ a test that deadlocks *every* time at seed 4217, in milliseconds, under a debugg ## Status -**Early development — nothing usable yet.** First goal: a from-scratch -deterministic event loop that runs a toy client/server pair with identical -trace hashes across repeated runs at the same seed, and different orderings -across seeds. +**Early development.** The deterministic loop core works: seeded scheduling, +virtual time, replay-proving trace hashes, deadlock detection, and seeded +`sim.random` / `sim.uuid4` / `sim.time` shims, with the stdlib coordination +primitives (`Queue`, `Event`, `Lock`, `gather`, `TaskGroup`, `timeout`) +running unchanged on top. No simulated network yet. See +[the supported subset](docs/supported-api.md) for the exact contract. ## Planned diff --git a/docs/supported-api.md b/docs/supported-api.md new file mode 100644 index 0000000..073b5d4 --- /dev/null +++ b/docs/supported-api.md @@ -0,0 +1,41 @@ +# Supported asyncio subset + +simloop runs real, unmodified asyncio code on a simulated event loop. This +page is the honest contract: what is simulated, what works unchanged on top, +and what is fenced off because it would escape the simulation and break +determinism. Fenced APIs raise `SimulationFenceError` (a subclass of +`NotImplementedError`) naming the offending call. + +## Simulated by SimLoop + +| API | Behavior under simulation | +|---|---| +| `loop.call_soon` / `call_later` / `call_at` | Seeded ready-queue ordering; `(deadline, seq)` timer tie-break | +| `loop.time()` / `asyncio.sleep` | Virtual clock starting at 0.0; never waits on wall time | +| `loop.create_task` / `asyncio.create_task` | Real stdlib `Task`s, including custom task factories | +| `loop.create_future` | Real stdlib `Future`s | +| `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 | +| `sim.random` / `sim.uuid4` / `sim.time` | Seed-derived streams inside a run; stdlib fallback outside | + +## Works unchanged on top of the loop + +`asyncio.Queue`, `asyncio.Event`, `asyncio.Lock`, `asyncio.Semaphore`, +`asyncio.gather`, `asyncio.TaskGroup`, `asyncio.timeout`, `asyncio.wait_for` +— everything built purely on futures, tasks and timers. Timeouts fire in +virtual time. Each of these claims is exercised by the test suite. + +## Planned: simulated network + +`create_connection`, `create_server`, `create_datagram_endpoint` and the +streams API, backed by in-memory transports with seeded fault injection +(latency, drops, duplication, reordering, partitions). + +## Fenced + +Anything that reaches outside the simulation raises `SimulationFenceError`: +executors and threads (`run_in_executor`, `call_soon_threadsafe`), signal +handlers, subprocesses, raw sockets (`sock_*`), file-descriptor callbacks +(`add_reader` / `add_writer`), name resolution (`getaddrinfo` / +`getnameinfo`), TLS upgrades, `sendfile`, and pipes. diff --git a/tests/test_stdlib_surface.py b/tests/test_stdlib_surface.py new file mode 100644 index 0000000..dfce0ec --- /dev/null +++ b/tests/test_stdlib_surface.py @@ -0,0 +1,105 @@ +"""The stdlib coordination surface the docs promise works on SimLoop.""" + +import asyncio + +import pytest + +from simloop import SimLoop + + +def test_gather_and_taskgroup_run_deterministically() -> None: + async def double(x: int) -> int: + await asyncio.sleep(0.01) + return 2 * x + + async def main() -> list[int]: + gathered = await asyncio.gather(double(1), double(2), double(3)) + async with asyncio.TaskGroup() as group: + tasks = [group.create_task(double(x)) for x in (4, 5)] + return list(gathered) + [task.result() for task in tasks] + + loop = SimLoop(seed=0) + try: + assert loop.run_until_complete(main()) == [2, 4, 6, 8, 10] + finally: + loop.close() + + +def test_timeout_uses_virtual_time() -> None: + async def main() -> None: + async with asyncio.timeout(10.0): + await asyncio.sleep(1.0) + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main()) + finally: + loop.close() + assert loop.time() == 1.0 + + +def test_wait_for_times_out_in_virtual_time() -> None: + async def main() -> None: + await asyncio.wait_for(asyncio.sleep(60.0), timeout=1.0) + + loop = SimLoop(seed=0) + try: + with pytest.raises(TimeoutError): + loop.run_until_complete(main()) + finally: + loop.close() + assert loop.time() == 1.0 + + +def test_event_lock_and_semaphore_coordinate_tasks() -> None: + order: list[str] = [] + + async def main() -> None: + event = asyncio.Event() + lock = asyncio.Lock() + semaphore = asyncio.Semaphore(1) + + async def waiter() -> None: + await event.wait() + async with lock, semaphore: + order.append("waiter") + + task = asyncio.create_task(waiter()) + async with lock: + order.append("main") + event.set() + await asyncio.sleep(0.01) + await task + + loop = SimLoop(seed=0) + try: + loop.run_until_complete(main()) + finally: + loop.close() + assert order == ["main", "waiter"] + + +def test_queue_coordinates_producer_and_consumer() -> None: + async def main() -> list[int]: + queue: asyncio.Queue[int] = asyncio.Queue() + received: list[int] = [] + + async def producer() -> None: + for value in range(3): + await queue.put(value) + await asyncio.sleep(0.01) + + async def consumer() -> None: + for _ in range(3): + received.append(await queue.get()) + + async with asyncio.TaskGroup() as group: + group.create_task(producer()) + group.create_task(consumer()) + return received + + loop = SimLoop(seed=0) + try: + assert loop.run_until_complete(main()) == [0, 1, 2] + finally: + loop.close()