Keep a worker's imported heap out of its cycle collections

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

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:

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)