From 45e4dc62566da72a7d7f5af27bff4b607430c44b Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Tue, 4 Aug 2026 20:55:21 +0530 Subject: [PATCH] Keep a worker's imported heap out of its cycle collections --- CHANGELOG.md | 9 ++++++++- src/simloop/_parallel.py | 31 +++++++++++++++++++++++++++++++ tests/test_parallel.py | 30 ++++++++++++++++++++++++++++++ 3 files changed, 69 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8789b48..496c4a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -67,7 +67,14 @@ recorded trace hash moves, once, for every workload. its report rebuilt by re-running that seed in the parent, which also proves the replay held across processes. Workloads must be picklable to cross that boundary, so lambdas, closures and fixture-taking tests stay - sequential and say so. + sequential and say so. Each worker also freezes its imported heap before + its first batch, which roughly doubles seeds per second: every run ends + with a cycle collection — that is what turns a dropped failing task into a + reported failure — and a full collection otherwise re-walks twenty + thousand modules and classes a run cannot make garbage. Nothing a run + builds escapes the collection, because freezing only applies to what + already exists. Sequential runs are unchanged: the freeze happens in + simloop's own worker processes, never in yours. - Every scheduling decision flows through a policy seam: seeded draws by default, making the same draws the loop made when it owned the PRNG itself, with recorded choice lists that can replay a schedule independently of its diff --git a/src/simloop/_parallel.py b/src/simloop/_parallel.py index 127dded..6d6429f 100644 --- a/src/simloop/_parallel.py +++ b/src/simloop/_parallel.py @@ -19,6 +19,7 @@ from __future__ import annotations import functools +import gc import importlib.util import inspect import pickle @@ -203,6 +204,35 @@ def record(self, failure: Failure | None) -> None: self.failure = failure +_heap_frozen = False + + +def _freeze_imports() -> None: + """Take the worker's imported heap out of every later cycle collection. + + Each run ends with a ``gc.collect()`` — that is what turns a failed + fire-and-forget task's reference cycle into a reported failure — and a + full collection walks every tracked object, which in a warmed worker is + some twenty thousand modules, classes and functions that a run cannot + make garbage. Freezing them moves them to the permanent generation, + which collections never traverse, so the same collection costs a + fraction of what it did and still sees everything a run built: freezing + applies to objects that already exist, and a run's cycles are all + allocated after this. + + Done on the first batch rather than at spawn, because the workload's own + module is imported when the batch that carries it is unpickled, and + freezing before that would leave it out. Once only: a later call would + make one batch's garbage permanent. + """ + global _heap_frozen + if _heap_frozen: + return + _heap_frozen = True + gc.collect() + gc.freeze() + + def _run_batch(fn: Workload, start: int, seeds: Sequence[int]) -> Failure | None: """Run ``seeds`` in order and report the first that failed. @@ -211,6 +241,7 @@ def _run_batch(fn: Workload, start: int, seeds: Sequence[int]) -> Failure | None away, because the parent re-runs the seed that matters anyway and shipping any of it back would mean pickling it. """ + _freeze_imports() for offset, seed in enumerate(seeds): loop = SimLoop(seed) try: diff --git a/tests/test_parallel.py b/tests/test_parallel.py index 5f6b4c4..f05d333 100644 --- a/tests/test_parallel.py +++ b/tests/test_parallel.py @@ -230,6 +230,36 @@ def test_one_seed_needs_no_worker_processes() -> None: assert report.seed == 4 +async def _boom() -> None: + raise RuntimeError("orphaned boom") + + +async def _orphans_a_failure_at(bad_seed: int) -> None: + """Drop a failing task on ``bad_seed`` and otherwise finish cleanly. + + Nobody awaits the task, so its exception reaches the loop only when the + cycle collector finalizes it — which is the one thing a worker's frozen + heap must not change. + """ + loop = asyncio.get_running_loop() + assert isinstance(loop, simloop.SimLoop) + if loop.seed == bad_seed: + asyncio.create_task(_boom()) + await asyncio.sleep(1.0) + + +def test_a_worker_still_surfaces_an_orphaned_failure() -> None: + # Workers take their imported heap out of every cycle collection, which + # is only sound because a run's own cycles are built after the freeze. + # A dropped failing task is exactly such a cycle, so finding this seed is + # what says the freeze cost the guarantee nothing. + report = explore(functools.partial(_orphans_a_failure_at, 5), range(32), jobs=2) + assert report is not None + assert report.seed == 5 + assert isinstance(report.exception, RuntimeError) + assert "orphaned boom" in str(report.exception) + + async def _fails_only_in_workers() -> None: loop = asyncio.get_running_loop() assert isinstance(loop, simloop.SimLoop)