From 440129ced713badf345281a0bf684f801cd33450 Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sun, 1 Nov 2026 19:00:25 +1100 Subject: [PATCH 1/2] test(write-behind): pin the L1-off fork window, tighten the parking tests, use wait_for The fork test that ran with write_behind=False pinned L1's sweeper, not the writer: it moves to the L1 sweep tests, and a new test pins the writer's own window with L1 off, where a waiting thread alone makes os.fork() warn until it ends. The forked-child test caps parking at the parent's one thread, so a child that inherited the parent's count would fail it. _work waits with Condition.wait_for instead of a hand-written deadline loop, and the writers' lock comment states the lock order. No behaviour change. --- src/cachekit/decorators/write_behind.py | 9 ++--- tests/unit/test_l1_expiry_sweep.py | 30 ++++++++++++++- tests/unit/test_write_behind.py | 49 ++++++++++++++----------- tests/utils/fork_helpers.py | 12 ++++++ 4 files changed, 72 insertions(+), 28 deletions(-) diff --git a/src/cachekit/decorators/write_behind.py b/src/cachekit/decorators/write_behind.py index 6b637a09..9c0591f5 100644 --- a/src/cachekit/decorators/write_behind.py +++ b/src/cachekit/decorators/write_behind.py @@ -185,10 +185,8 @@ def _work(self, s: _State) -> None: entry = _runnable(s) if entry is None and _park(s): try: - deadline = time.monotonic() + IDLE_S - while entry is None and not s.closed and (left := deadline - time.monotonic()) > 0: - s.wake.wait(left) - entry = _runnable(s) + s.wake.wait_for(lambda: s.closed or _runnable(s) is not None, timeout=IDLE_S) + entry = _runnable(s) finally: _unpark(s) if entry is None: @@ -265,7 +263,8 @@ def _unpark(s: _State) -> None: _writers: weakref.WeakSet[WriteBehind] = weakref.WeakSet() # Guards _writers and _parked: a decoration can add while an invalidation walks it. Per PID, so a child never waits on -# a lock a parent thread held at fork. +# a lock a parent thread held at fork. Lock order: a state's cond first, then this lock (as _park and _unpark take +# them), never the reverse. _writers_locks: dict[int, threading.Lock] = {} diff --git a/tests/unit/test_l1_expiry_sweep.py b/tests/unit/test_l1_expiry_sweep.py index 8c00832b..71b61123 100644 --- a/tests/unit/test_l1_expiry_sweep.py +++ b/tests/unit/test_l1_expiry_sweep.py @@ -7,10 +7,15 @@ from __future__ import annotations +import ast +import sys +import threading + import pytest -from cachekit import l1_cache +from cachekit import cache, l1_cache from cachekit.l1_cache import _INVALIDATE_BATCH, _SWEEP_BATCH, L1Cache +from tests.utils.fork_helpers import fork_warns, run_fresh class _Clock: @@ -223,3 +228,26 @@ def test_put_rebuilds_the_heap_where_no_sweep_runs(clock: _Clock) -> None: clock.now += 100 + 999 assert l1.cleanup_expired() == 10 # every entry kept its latest deadline through the rebuilds assert not l1._state.cache + + +def _fork_warns_after_a_minimal_miss() -> object: + """Run fresh: does os.fork() warn after one default @cache.minimal miss?""" + + @cache.minimal(backend=None) + def fn(x: int) -> int: + return x + + fn(1) + return {"threads": sorted(t.name for t in threading.enumerate()), "warned": fork_warns()} + + +@pytest.mark.unit +@pytest.mark.skipif(sys.version_info < (3, 12), reason="os.fork() warns about threads from 3.12") +def test_the_sweeper_makes_fork_warn_for_good() -> None: + """A process with an L1 cache runs the sweeper thread for good, so os.fork() warns there whatever else runs. + That is why write-behind's waiting threads need no before-fork hook (test_write_behind pins the L1-off case).""" + run = run_fresh(_fork_warns_after_a_minimal_miss) + assert run.returncode == 0, run.stderr + outcome = ast.literal_eval(run.stdout) + assert outcome["warned"] is True, outcome + assert any("cleanup_worker" in name for name in outcome["threads"]), outcome # the sweeper's target diff --git a/tests/unit/test_write_behind.py b/tests/unit/test_write_behind.py index a9c04b2a..9f3a8919 100644 --- a/tests/unit/test_write_behind.py +++ b/tests/unit/test_write_behind.py @@ -37,7 +37,7 @@ from cachekit.decorators.orchestrator import FeatureOrchestrator from cachekit.interop import InteropError from cachekit.serializers.encryption_wrapper import KeyringConfigurationError -from tests.utils.fork_helpers import child_outcome, report, run_fresh +from tests.utils.fork_helpers import child_outcome, fork_warns, report, run_fresh pytestmark = pytest.mark.unit @@ -969,7 +969,8 @@ def _until(holds: Callable[[], bool]) -> None: @pytest.fixture def no_park_cap(monkeypatch: pytest.MonkeyPatch) -> None: """Other tests' threads may still be waiting out IDLE_S; the cap is TestParking's own subject only.""" - monkeypatch.setattr(wb, "MAX_PARKED", 10_000) + # raising=False: a mutation run against a writer without the cap fails these tests, rather than their setup. + monkeypatch.setattr(wb, "MAX_PARKED", 10_000, raising=False) def _three_writers_under_a_cap_of_two() -> object: @@ -989,8 +990,9 @@ def _three_writers_under_a_cap_of_two() -> object: def _forked_child_parks_from_zero() -> object: - """Run fresh: a parent with a waiting thread forks; what the child's writer and count start from.""" - wb.IDLE_S = 3600.0 + """Run fresh: a parent with a waiting thread forks; what the child's writer and count start from. The cap is + the parent's one thread, so a child that inherited the parent's count could not park its own.""" + wb.MAX_PARKED, wb.IDLE_S = 1, 3600.0 writer = wb.WriteBehind(lambda *_: None) writer.submit(("", "k"), 1, lambda: None) _until(lambda: writer._state.idle == 1) @@ -1014,25 +1016,28 @@ def _forked_child_parks_from_zero() -> object: return {"child": child_outcome(child, r), "parent_parked": wb._parked[os.getpid()]} -def _fork_warns_without_write_behind() -> object: - """Run fresh: does os.fork() already warn after a default @cache.minimal miss, write-behind off?""" - import warnings - +def _l1_off_fork_warns_only_while_a_thread_waits() -> object: + """Run fresh: with L1 off no sweeper runs, so the writer's waiting thread is the process's only other thread. + Does os.fork() warn while it waits, and not once it has waited out IDLE_S and ended?""" + wb.IDLE_S = 3600.0 backend = _GatedBackend() + backend.gate.set() - @cache.minimal(backend=backend, write_behind=False) + @cache(backend=backend, ttl=60, write_behind=True, l1_enabled=False) def fn(x: int) -> int: return x fn(1) + (writer,) = wb._all_writers() + s = writer._state + _until(lambda: s.idle == 1) # sent, and waiting for the next write threads = sorted(t.name for t in threading.enumerate()) - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - child = os.fork() - if child == 0: - os._exit(0) - os.waitpid(child, 0) - return {"threads": threads, "warned": any("multi-threaded" in str(w.message) for w in caught)} + while_waiting = fork_warns() + wb.IDLE_S = 0.05 + fn(2) # the waiting thread sends it, then waits the new IDLE_S out and ends + # The OS thread, not the Thread object: fork's warning counts /proc/self/task, which outlives join() briefly. + _until(lambda: s.workers == 0 and len(os.listdir("/proc/self/task")) == 1) + return {"threads": threads, "while_waiting": while_waiting, "after": fork_warns(), "sent": len(backend.store)} @pytest.mark.usefixtures("no_park_cap") @@ -1104,11 +1109,11 @@ def test_a_forked_child_starts_with_no_waiting_threads(self) -> None: assert outcome["parent_parked"] == 1 @pytest.mark.skipif(sys.version_info < (3, 12), reason="os.fork() warns about threads from 3.12") - def test_minimal_already_forks_multi_threaded_without_write_behind(self) -> None: - """Why a waiting thread needs no before-fork hook: a @cache.minimal process already runs the L1 - sweeper for good, so os.fork() warns with write-behind off too, and a waiting thread changes nothing.""" - run = run_fresh(_fork_warns_without_write_behind) + @pytest.mark.skipif(sys.platform != "linux", reason="counts the process's threads in /proc/self/task") + def test_with_l1_off_fork_warns_only_while_a_thread_waits(self) -> None: + """L1's sweeper makes any process with an L1 cache multi-threaded for good (test_l1_expiry_sweep pins it). + With L1 off, a waiting writer thread opens a window of up to IDLE_S after a write in which os.fork() warns.""" + run = run_fresh(_l1_off_fork_warns_only_while_a_thread_waits) assert run.returncode == 0, run.stderr outcome = ast.literal_eval(run.stdout) - assert outcome["warned"] is True, outcome - assert _WRITER not in outcome["threads"], outcome + assert outcome == {"threads": ["MainThread", _WRITER], "while_waiting": True, "after": False, "sent": 2} diff --git a/tests/utils/fork_helpers.py b/tests/utils/fork_helpers.py index 10903cd5..cba87ce6 100644 --- a/tests/utils/fork_helpers.py +++ b/tests/utils/fork_helpers.py @@ -10,6 +10,7 @@ import sys import threading import traceback +import warnings from collections.abc import Callable from pathlib import Path from typing import NoReturn @@ -42,6 +43,17 @@ def child_outcome(pid: int, r: int, timeout: float = 20.0) -> object: return f"no outcome: exited {os.WEXITSTATUS(status)}" +def fork_warns() -> bool: + """Whether os.fork() here warns that the process is multi-threaded (it does from Python 3.12). The child exits at once.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + child = os.fork() + if child == 0: + os._exit(0) + os.waitpid(child, 0) + return any("multi-threaded" in str(w.message) for w in caught) + + def on_new_thread(fn: Callable[[], object], timeout: float = 5.0) -> object: """fn's result from a new thread, or "hung" if it holds the thread past timeout.""" out: list[object] = [] From c560c2e06c88be7d49c566c9c90f6ad2c781e73b Mon Sep 17 00:00:00 2001 From: Ray Walker Date: Sun, 1 Nov 2026 19:01:48 +1100 Subject: [PATCH 2/2] test(write-behind): name the thread count fork's warning reads --- tests/unit/test_write_behind.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_write_behind.py b/tests/unit/test_write_behind.py index 9f3a8919..35a770b2 100644 --- a/tests/unit/test_write_behind.py +++ b/tests/unit/test_write_behind.py @@ -1035,7 +1035,8 @@ def fn(x: int) -> int: while_waiting = fork_warns() wb.IDLE_S = 0.05 fn(2) # the waiting thread sends it, then waits the new IDLE_S out and ends - # The OS thread, not the Thread object: fork's warning counts /proc/self/task, which outlives join() briefly. + # OS threads, not Thread objects: fork's warning reads the count in /proc/self/stat, where a thread can stay + # briefly after join() returns. /proc/self/task lists the same threads. _until(lambda: s.workers == 0 and len(os.listdir("/proc/self/task")) == 1) return {"threads": threads, "while_waiting": while_waiting, "after": fork_warns(), "sent": len(backend.store)}