diff --git a/docs/features/l1-invalidation.md b/docs/features/l1-invalidation.md index 80b17415..77fcebec 100644 --- a/docs/features/l1-invalidation.md +++ b/docs/features/l1-invalidation.md @@ -291,9 +291,9 @@ A process created by `fork()`, such as a `multiprocessing` fork-context worker o The parent's entries stay in the child's memory, shared with the parent, until the child first stores a value in its L1. Then a background thread frees them. Freeing copies the memory they held into the child, about 2.3 times the size of the parent's L1 in cachekit's own measurement, because each free also writes to the memory allocator's bookkeeping beside the entry; the child's own L1 then reuses that memory. That copy avoids a larger one. The parent's background sweep rewrites those entries as they expire, within one L1 TTL, and a child that kept them would from then on hold a private copy that nothing can reach, beside its own L1. A master that fills no L1 before it forks avoids both. -uWSGI forks its workers without running Python's at-fork hooks. Run every app that uses cachekit with `--enable-threads --py-call-uwsgi-fork-hooks` (uWSGI 2.0.21 or later); that keeps the rule above. Without the fork-hook flag, and with the app loaded in the master (uWSGI's default), a worker keeps the L1 entries its master held at fork, runs no background sweep of expired L1 entries, and runs no invalidation listener, so its L1 heals by TTL. An expired entry is still never served: it is evicted when it is read. `--lazy-apps` is no substitute: without the flag, each worker first imports cachekit after the fork, so cachekit takes it for a fresh process and may start a thread there that hangs. Nor is `--py-call-osafterfork`: with `--enable-threads` it aborts every worker on Python 3.13 and later, and on earlier versions it does not hold the master's threads off at the fork, so a worker can hang. Under uWSGI without `--py-call-uwsgi-fork-hooks`, cachekit logs one WARNING `uWSGI is running without py-call-uwsgi-fork-hooks` when the master imports it. A process uWSGI forked logs nothing, because a log call there can hang on a handler lock that a master thread held at the fork. So under `--lazy-apps`, or when your app first imports cachekit in a worker, no WARNING appears. The [Deployment Reference](https://docs.cachekit.io/python/reference/deployment/) gives the full uWSGI setup. +uWSGI forks its workers without running Python's at-fork hooks. Run every app that uses cachekit with `--enable-threads --py-call-uwsgi-fork-hooks` (uWSGI 2.0.21 or later); that keeps the rule above. Without the fork-hook flag, and with the app loaded in the master (uWSGI's default), a worker keeps the L1 entries its master held at fork, runs no background sweep of expired L1 entries, and runs no invalidation listener, so its L1 heals by TTL. An expired entry is still never served: it is evicted when it is read. A sync function with `stale_ttl` serves a stale L2 entry there without revalidating it in the background, until the entry hard-expires; the first read after that recomputes it. `--lazy-apps` is no substitute: without the flag, each worker first imports cachekit after the fork, so cachekit takes it for a fresh process and may start a thread there that hangs. Nor is `--py-call-osafterfork`: with `--enable-threads` it aborts every worker on Python 3.13 and later, and on earlier versions it does not hold the master's threads off at the fork, so a worker can hang. Under uWSGI without `--py-call-uwsgi-fork-hooks`, cachekit logs one WARNING `uWSGI is running without py-call-uwsgi-fork-hooks` when the master imports it. A process uWSGI forked logs nothing, because a log call there can hang on a handler lock that a master thread held at the fork. So under `--lazy-apps`, or when your app first imports cachekit in a worker, no WARNING appears. The [Deployment Reference](https://docs.cachekit.io/python/reference/deployment/) gives the full uWSGI setup. -L1-only mode (`backend=None`, `@cache.local()`) has no L2 to refill from, so a forked child keeps the entries its parent held at fork, each expiring on its own TTL. The exception is a function whose cache a parent thread was using at the fork: that cache starts empty in the child. A background refresh that was running in the parent starts again on the child's next stale read. Under uWSGI this needs `py-call-uwsgi-fork-hooks`, as above: without it, a worker's first call to an L1-only cached function can hang if a master thread was using that function's cache at the fork. +L1-only mode (`backend=None`, `@cache.local()`) has no L2 to refill from, so a forked child keeps the entries its parent held at fork, each expiring on its own TTL. The exception is a function whose cache a parent thread was using at the fork: that cache starts empty in the child. A background refresh that was running in the parent starts again on the child's next stale read. Under uWSGI this needs `py-call-uwsgi-fork-hooks`, as above. Without it, a worker refreshes no entry of a sync function in the background: it serves the stale value until the entry expires, and the first read after that recomputes it. A worker's first call to an L1-only cached function can also hang if a master thread was using that function's cache at the fork. --- diff --git a/docs/free-threading.md b/docs/free-threading.md index 200aa62c..ccbcaa00 100644 --- a/docs/free-threading.md +++ b/docs/free-threading.md @@ -55,14 +55,14 @@ the ticket, plus what the free-threaded CI lane surfaced and sites added since: | `os.register_at_fork` handlers (session + stats registry) | Run in the child while single-threaded; replace locks wholesale | Safe — single-threaded by construction at execution time | | `backends/cachekitio/session.py` header cache | `threading.local` | Safe — per-thread state; its only cross-thread hazard was the session-identity publication above | | `decorators/stats_context.py` | `contextvars.ContextVar` | Safe by construction | -| `decorators/wrapper.py` L1/SWR + L2/SWR single-flight (`_RefreshPool`: `_l1_swr_pool`, `_l2_swr_pool`) | One `Lock` per pool over its holder map (an L2 holder carries its in-flight key) + PID-owner swap | Safe. The prune, the cap check and L2's per-key check run under the pool lock, so L2 has no check-then-add race; L1's per-key check is `ObjectCache`'s, under its own lock. A pruned refresh is cancelled with `call_soon_threadsafe` on its own loop. The fork-detection swap race is the documented-benign shape: a thread racing it can lose its new holder, whose key then waits for its hard TTL | +| `decorators/wrapper.py` L1/SWR + L2/SWR single-flight (`_RefreshPool`: `_l1_swr_pool`, `_l2_swr_pool`) | One `Lock` per pool over its holder map (an L2 holder carries its in-flight key) + PID-owner swap | Safe. The prune, the cap check and L2's per-key check run under the pool lock, so L2 has no check-then-add race; L1's per-key check is `ObjectCache`'s, under its own lock. A pruned refresh is cancelled with `call_soon_threadsafe` on its own loop. The fork-detection swap race is the documented-benign shape: a thread racing it can lose its new holder, whose key then waits for its hard TTL. In a child forked without at-fork hooks (the detection of the `l1_cache.py` row, so cachekit must be imported before the fork), neither sync path starts its refresh thread (`cachekit-swr-refresh`, `cachekit-swr-revalidate`), because a thread started there can hang in `Thread.start()`. The check is one `getpid()` on a stale hit that would refresh, made before a slot is taken, and it logs nothing. The stale value is served until the entry hard-expires, and the first read after that recomputes it. The async paths schedule a task, not a thread, and are unchanged. Tests in `tests/unit/test_swr_decorator.py::TestSWRForkIsolation` | | `decorators/single_flight.py` `AsyncFlights`, `ThreadFlights` (in-process single-flight of misses: one map per decorated function, used by `wrapper.py` and `local_wrapper.py`) | One `Lock` per map over its in-flight entries + PID-owner swap, checked by every method that takes the lock (`_FlightMap._owned`: `run`, `forget`, the settle paths). An async entry holds the call's task and its waiter count, which only that task's own loop touches. A sync waiter blocks on a per-call `Lock` its starter holds until the call settles | Safe. Lookup and insert run under the map lock, and an entry is removed only by the call that put it there (identity check), so a settling call never drops a newer one. A sync caller re-checks the cache under the map lock before it starts a call, so a call that settled between the caller's lookup and its map check is not run twice. An async caller joins only a task on its own running loop, never another thread's. An invalidation drops the key's entries under the map lock (`forget`); their callers keep waiting on the call they joined. An async starter inserts its call only if no `forget` ran since its lookup, so a call that an eager task factory ran up to its first await before an invalidation never reaches the map after it. A sync starter is settled in a `finally` that covers its insert, so an interrupt cannot leave its waiters blocked. The fork swap is the documented-benign shape: a forked child starts with an empty map, and a thread racing the swap starts its own call. Tests in `tests/unit/test_single_flight.py` | | `l1_cache.py` `L1CacheManager._take_over_if_forked` + `_empty_caches_after_fork` | Owner-PID fast path; on a PID change one thread takes over under a lock fetched by `dict.setdefault(pid, Lock())`. An `os.register_at_fork` hook gives every cache a fresh, empty state before the child's first `get()`, so a forked child starts with an empty L1; fork mechanics and the uWSGI limit are in the `_take_over_if_forked` docstring | Safe. `setdefault` is atomic under per-object locking, so every thread in a new child gets the same fresh lock, which no parent thread can have held at fork; the rest block on it and re-check. The fast path gates on `_owner_pid` alone, so a weak-memory reader may still see the inherited manager lock once: the same benign fork-swap race as the SWR row, since that lock is unheld unless a parent thread held it at fork. The hook runs while the child is single-threaded and replaces every cache's state, held lock or not, so it probes no lock, takes none and starts no thread. It does not free the inherited states, which would copy the parent's L1 pages into every child inside `fork()`: they wait in `_inherited_states` until the child's first put, whose take-over frees them in batches, giving up the GIL between batches, on the cleanup thread it restarts or, when the parent ran none, on a one-shot thread. Each state is taken off that list with one `list.pop()`, atomic under per-object locking, before it is freed, so two releasers running at once (the threads of two managers' take-overs) never take the same state and neither finds the list emptied under it. So a child bound for exec never pays. A child that uses its L1 pays once, up front: each free also writes the allocator's bookkeeping beside the entry, so its first put copies about 2.3 times the inherited L1 (36.5 MiB for a 16 MiB L1 in `tests/performance/test_l1_fork_memory.py`), and its own L1 then reuses that memory. Kept, those pages would cost nothing while the parent left them alone, as an idle preloaded master whose entries outlive its workers does. Freeing is the default anyway because a parent with a cleanup thread rewrites them within one L1 TTL, and from then on a child that kept them would hold a private copy nothing can reach, beside its own L1. A state whose lock is taken is only dropped: one a thread forked inside its critical section holds is freed with that thread's last reference, and one a parent thread held at fork can stay for the child's life, referenced by that thread's frame. Neither the hook nor the take-over clears shared state: each publishes a fresh `_L1State` (entries, byte count and lock) in one store, and a holder that is still running finishes on the state it bound (the forking thread, if it forked inside a critical section; without hooks, a live child thread holding a lock past the take-over's 1 s probe). Invalidations re-apply if a reset replaced the state meanwhile. Without hooks, the cost of a wrong guess is that namespace's L1 entries, once per child; `--py-call-uwsgi-fork-hooks` avoids it. Once it has run, the take-over skips its own timed cache-lock probe, which could otherwise drop a cache's entries under a live child thread. In a child the hook never reached (a fork made from C, as uWSGI does without `--py-call-uwsgi-fork-hooks`; any PID other than the importing process's and the one the hook last ran in), the take-over starts no thread and `start_background_cleanup` refuses to start one, including for a manager first built in that child: such a fork also skips CPython's own after-fork repair, so a thread started there can hang in `Thread.start()` or crash the interpreter. None of that logs, the take-over's lock-reset drop included, because the same fork skips `logging`'s at-fork lock reset, and a handler lock a parent thread held would hang the child. Expired entries in that child are evicted on read or under the memory bound instead. The detection needs cachekit imported before the fork: a process that first imports it after a fork made from C (uWSGI `--lazy-apps` without `--py-call-uwsgi-fork-hooks`) looks like a fresh process, and starts the cleanup thread. No check made at import tells the two apart | | `backends/cachekitio/client.py` client cache + `CachekitIOBackend._own_lease` | One urllib3 connection pool per config, shared by every thread of a process (urllib3's pool is thread-safe: a `LifoQueue` of connections, each request on its own connection). The process-wide lease cache is a `WeakValueDictionary` under a `Lock`, both held by one `_Leases` object that carries the PID that built it; each lease records its PID too, and the backend re-leases on a mismatch, or once `close_http_clients()` has closed its client, checked per request; the new lease is in the cache, so a later close reaches it. Async methods send on the same client through `asyncio.to_thread` | Safe across threads, with and without the GIL: `tests/unit/backends/test_cachekitio_thread_share.py` runs 8 threads on one backend, sync and through `to_thread`, and expects zero failures on 3.14 and on 3.14t with the GIL off. Safe across fork. A forked child never sends on or reads from its parent's pooled connections, which share the parent's TLS sessions: its first request finds the cache's PID is not its own, replaces the cache and its lock with new ones, and re-leases a new client, whether it was forked by `os.fork()` or from C with no at-fork hooks (uWSGI). An owner-PID check rather than an at-fork hook for that reason. The PID travels with the cache and its lock in one published reference, and with the client in the lease (single-assignment publication), so no thread pairs a new PID with an inherited lock or client, and no child thread takes the parent's lock, which a parent thread may have held at fork. Inherited clients are dropped unclosed: `close()` takes the pool's queue lock, for the same reason, and garbage collection closes the child's copies of the sockets without sending anything. Two child threads racing the first request may each build a cache and a client; the one replaced is simply not shared, and each request holds its lease until it returns, so that replacement never closes a client under an in-flight request. A child runs async calls on an event loop of its own (`asyncio.run`): a child that keeps running its parent's loop object waits forever on that loop's inherited default executor, whose worker threads did not survive the fork, as every async decorator's L2 operation already did. In a child forked from C, any logging call whose level check is not already cached, urllib3's per-request `DEBUG` check included, waits on `logging`'s module lock, and hangs if a parent thread held it at fork; a logging reconfiguration clears those caches. `--py-call-uwsgi-fork-hooks` avoids this. Tests in `tests/unit/backends/test_cachekitio_fork.py` | | `invalidation.py` dispatch map, start lock and listener thread | `_evictors` (registry id → `WeakSet` of evictors) under a per-PID `Lock` from `dict.setdefault` (`_dispatch_locks`): decoration registers under it, and the listener thread takes a snapshot under it and runs the evictors outside it. The listener is redis-py's own `PubSubWorkerThread`, started behind an owner-PID check under a second per-PID lock (`_start_locks`) taken non-blocking, so a cache operation never waits on another thread's start | Safe. `setdefault` is atomic under per-object locking, so threads racing to create a new process's lock share one, and a child forked while a parent thread holds either lock gets its own instead of waiting on a lock nobody will release. Walking the live `WeakSet` while a decoration adds to it would raise; the snapshot under the lock rules that out. The owner-PID check reads `_listener_pid`, which the starting thread publishes last, so a stale read only skips or retries a start. Evictors call `L1Cache.invalidate` and `invalidate_many`, which lock per state as in the row above. A child forked without at-fork hooks starts no listener and logs nothing (the same detection as the row above). Tests in `tests/unit/test_invalidation_listener.py` | | `backends/redis/provider.py` `with_timeout` windows (`_open_windows`: pool → configured `socket_timeout` and the open windows' timeouts) | Per-PID `Lock` from `_pid_lock` (`_windows_locks`) around each window's open and close and around `listener_pool()`'s read of the pool's kwargs | Safe. Opening and closing each read and write the record and the pool's `socket_timeout` together, and a window can open and close on any thread's event loop over one shared pool, so without the lock a window closing the last record could drop one another thread just joined. No `await` runs under the lock. A child gets its own lock, as in the row above. A child forked while a window is open inherits that window but not the task holding it, so it could never close it: an `os.register_at_fork` hook drops every inherited record and puts each pool's configured timeout back, and each record carries the PID that opened it, so a fork made from C, which runs no hook, drops it on the pool's next window or listener clone instead; until then such a child opens its connections at the inherited window's timeout (no per-operation PID check, as in the object-cache row; `--py-call-uwsgi-fork-hooks` avoids it). A close restores the pool's timeout before it drops the record, so a fork between the two steps still leaves the hook a record to reset from. Close-order and fork tests (single-threaded) in `tests/unit/test_invalidation_listener.py` | | `decorators/wrapper.py` `_cached_keys` | Builtin `set`, snapshot-copied before iteration in invalidation | Safe — single ops atomic; a copy racing an add can only miss a concurrently-written key, which invalidate-all semantics tolerate | -| `object_cache.py` `ObjectCache` + `_reset_caches_after_fork` | `RLock` on every public method. The entries, their byte total, their in-flight refresh markers and the lock form one `_ObjectCacheState`, which every critical section binds once. An `os.register_at_fork` hook walks a `WeakSet` of live caches in a forked child | Safe. The hook runs while the child is single-threaded, so it takes no lock that can block and starts no thread. A cache whose lock was free at fork keeps its entries and loses its in-flight refresh markers, since the threads running those refreshes did not survive the fork, so the child's next stale read refreshes again. A cache whose lock was held gets a fresh, empty state in one attribute store: a parent thread that held it is gone, and a forking thread that forked inside a critical section (from a signal handler or a finalizer) may never return there, as in a `multiprocessing` child. A forking thread that does return finishes on the state it bound. The generation counter stays on the cache, so no refresh token is reused across states. A child forked without at-fork hooks (uWSGI without `--py-call-uwsgi-fork-hooks`) gets no repair: its first call hangs if a parent thread held that cache's lock at fork, and a key whose refresh was in flight is served stale until it expires. The check that could catch that, an owner-PID check, would cost a `getpid()` on every cached call. Tests in `tests/unit/test_object_cache.py::TestObjectCacheFork` | +| `object_cache.py` `ObjectCache` + `_reset_caches_after_fork` | `RLock` on every public method. The entries, their byte total, their in-flight refresh markers and the lock form one `_ObjectCacheState`, which every critical section binds once. An `os.register_at_fork` hook walks a `WeakSet` of live caches in a forked child | Safe. The hook runs while the child is single-threaded, so it takes no lock that can block and starts no thread. A cache whose lock was free at fork keeps its entries and loses its in-flight refresh markers, since the threads running those refreshes did not survive the fork, so the child's next stale read refreshes again. A cache whose lock was held gets a fresh, empty state in one attribute store: a parent thread that held it is gone, and a forking thread that forked inside a critical section (from a signal handler or a finalizer) may never return there, as in a `multiprocessing` child. A forking thread that does return finishes on the state it bound. The generation counter stays on the cache, so no refresh token is reused across states. A child forked without at-fork hooks (uWSGI without `--py-call-uwsgi-fork-hooks`) gets no repair: its first call hangs if a parent thread held that cache's lock at fork, and a key whose refresh was in flight is served stale until it expires. The check that could catch that, an owner-PID check, would cost a `getpid()` on every cached call. A sync function refreshes no key there at all (the SWR row). Tests in `tests/unit/test_object_cache.py::TestObjectCacheFork` | | `reliability/metrics_collection.py` `get_async_metrics_collector` | Double-checked module-global singleton | Safe — single-assignment publication of a fully-constructed object; worst case a benign duplicate worker-restart check | | `reliability/async_metrics.py` `_metrics_cache` + `_metrics_cache_lock()` | Double-checked module dict of Prometheus metric objects; one `Lock` per PID, created with `dict.setdefault`; held across `os.fork` by at-fork hooks | Safe — the lock-free read sees `None` or a fully constructed metric (single-assignment publication). `setdefault` is atomic under per-object locking, so threads racing to create a new process's lock all get the same one; a child forked without the hooks never takes its parent's lock. Tests in `tests/unit/test_async_metrics_shared_series.py` | | `backends/memcached/backend.py` `MemcachedBackend` | pymemcache `HashClient(use_pooling=True)`: one `PooledClient` per server, whose `ObjectPool` `threading.Lock` guards only the free/used deque pop and append, never socket I/O | Safe across threads: each operation owns its checked-out connection, so no socket is shared. The pool raises at `max_pool_size` rather than waiting (`tests/unit/backends/test_memcached_pooling.py`). pymemcache's failover state (`HashClient._failed_clients`) is unlocked: while a failed server is retried, concurrent operations can raise a spurious `BackendError` wrapping `KeyError` (documented in `docs/backends/memcached.md#concurrency`). Not fork-safe, by design: a child that reused its parent's backend would share the parent's pooled sockets, and could deadlock on a pool lock a parent thread held at fork. A per-PID lock would cure only the deadlock, not the shared sockets, so the rule is that a forked child builds its own backend (documented in `docs/backends/memcached.md#concurrency`) | diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index e5bbe993..390e5f40 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/src/cachekit/decorators/wrapper.py @@ -16,7 +16,7 @@ from cachekit.hash_utils import _WarnThrottle, redact_error_for_log -from .. import invalidation +from .. import invalidation, l1_cache from ..backends.errors import BackendError, UnsupportedTenantError from ..cache_handler import ( CacheHit, @@ -1741,9 +1741,12 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 needs_refresh, version = False, 0 if found: _stats.record_l1_hit() - if needs_refresh: - # SWR: serve the stale value now, refresh on a daemon thread - # (sync functions have no event loop to schedule a task on) + # SWR: serve the stale value now, refresh on a daemon thread (sync functions have no + # event loop to schedule a task on). Not in a child forked without at-fork hooks: a + # thread started there can hang in Thread.start(). The refreshing marker get_with_swr + # set stays, so no later hit asks again; it goes with the entry, and the first read + # after expiry recomputes it. + if needs_refresh and not l1_cache._forked_without_hooks(): snapshot = _l1_swr_acquire(cache_key, version, args, kwargs) if snapshot is not None: slot, refresh_args, refresh_kwargs = snapshot @@ -1981,7 +1984,9 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 # SWR: stale hit — serve now, revalidate on a daemon thread. # Gated on _l2_swr_active: without a configured stale window this # decorator serves the mixed-reader hit but owns no revalidation. - if _sync_l2_stale and _l2_swr_active: + # Not in a child forked without at-fork hooks, where a thread can hang in + # Thread.start(): stale is served until it hard-expires, then a miss recomputes. + if _sync_l2_stale and _l2_swr_active and not l1_cache._forked_without_hooks(): _l2_swr_schedule(cache_key, args, kwargs, is_async=False) return result diff --git a/src/cachekit/object_cache.py b/src/cachekit/object_cache.py index f288a50c..63822479 100644 --- a/src/cachekit/object_cache.py +++ b/src/cachekit/object_cache.py @@ -141,6 +141,7 @@ class ObjectCache: ``--py-call-uwsgi-fork-hooks``) gets no repair: its first call to a cached function hangs if a parent thread held that cache's lock at fork, and a key whose refresh was in flight in the parent is served stale until it expires. + A sync function's decorator starts no refresh thread there at all. Examples: Basic usage with TTL: diff --git a/tests/unit/test_l1_memory_bounds.py b/tests/unit/test_l1_memory_bounds.py index c25eb7c2..37db34a9 100644 --- a/tests/unit/test_l1_memory_bounds.py +++ b/tests/unit/test_l1_memory_bounds.py @@ -13,17 +13,14 @@ import os import random import signal -import subprocess import sys import threading import time -from pathlib import Path -from typing import NoReturn import pytest from cachekit.l1_cache import CacheEntry, L1Cache, L1CacheManager -from tests.utils.fork_helpers import child_outcome, on_new_thread, report +from tests.utils.fork_helpers import child_outcome, on_new_thread, report, run_fresh MB = 1024 * 1024 @@ -391,19 +388,6 @@ def start(self) -> None: return child_outcome(child, r) -def _print_c_fork_child_outcome() -> NoReturn: - """Print _c_fork_child_outcome() and end without exit handlers: logging's would wait on a lock a gone thread holds.""" - import traceback - - try: - print(repr(_c_fork_child_outcome()), flush=True) - except BaseException: - traceback.print_exc() - sys.stderr.flush() - os._exit(1) - os._exit(0) - - @pytest.mark.unit class TestCleanupThreadAfterFork: """LAB-4772: a prefork child must run its own L1 cleanup thread.""" @@ -1070,11 +1054,7 @@ def test_child_of_a_c_fork_requests_no_thread_start(self): The fork runs in a fresh interpreter (_c_fork_child_outcome): this one has threads it cannot end (pytest-xdist's, other tests'), and one waiting for the GIL at fork hangs the child. """ - code = "from tests.unit.test_l1_memory_bounds import _print_c_fork_child_outcome as f; f()" - root = Path(__file__).resolve().parents[2] # where `tests` imports from - run = subprocess.run( # noqa: S603 - trusted: sys.executable + literal code - [sys.executable, "-c", code], cwd=root, capture_output=True, text=True, timeout=60 - ) + run = run_fresh(_c_fork_child_outcome) assert run.returncode == 0, run.stderr assert ast.literal_eval(run.stdout) == { diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py index 7b1031ce..b61d4c28 100644 --- a/tests/unit/test_swr_decorator.py +++ b/tests/unit/test_swr_decorator.py @@ -10,11 +10,13 @@ from __future__ import annotations +import ast import asyncio import faulthandler import logging import math import os +import signal import sys import threading import time @@ -26,6 +28,7 @@ from cachekit import cache, hash_utils from cachekit.config.decorator import DecoratorConfig from cachekit.config.validation import ConfigurationError +from tests.utils.fork_helpers import child_outcome, report, run_fresh _CAP = 2_592_000 # 30-day storage cap shared by ttl + stale_ttl @@ -328,6 +331,83 @@ def f() -> int: assert f() == 7 +def _forked_stale_hit_outcome(fork: str) -> object: + """What a forked child reports after one stale hit on an L1-only and an L2 SWR function. + + fork is "c" for libc's fork(), which runs no at-fork hook (uWSGI without --py-call-uwsgi-fork-hooks), + or "os" for os.fork(). Run only in a fresh interpreter (run_fresh), with no other thread alive at + fork: one waiting for the GIL at a fork from C hangs the child in no cachekit code. + """ + import ctypes + import types + + from cachekit import l1_cache, object_cache + from cachekit import logging as structured_logging + from cachekit.config import L1CacheConfig + + backend = FakeSWRBackend() + calls = {"l1": 0, "l2": 0} + + @cache(ttl=100, backend=None, l1=L1CacheConfig(swr_threshold_ratio=0.5)) + def l1_only() -> int: + calls["l1"] += 1 + return calls["l1"] + + @cache(backend=backend, ttl=60, stale_ttl=120, l1_enabled=False) + def l2_swr() -> int: + calls["l2"] += 1 + return calls["l2"] + + l1_only(), l2_swr() # misses: each stores 1 + manager = l1_cache.get_l1_cache_manager() + if manager._cleanup_thread is not None: # ends it, still referenced: the parent ran cleanup + manager._stop_cleanup.set() + manager._cleanup_thread.join() + for structured in structured_logging._logger_instances.values(): # their writers wake every flush interval + structured.writer.stop() + structured.writer.join() + if threading.active_count() != 1: # not an assert: python -O strips those + raise RuntimeError(f"a thread would be alive at fork: {threading.enumerate()}") + faulthandler.register(signal.SIGALRM, chain=True) # the child inherits it: a hang dumps its stack + r, w = os.pipe() + child = ctypes.PyDLL(None).fork() if fork == "c" else os.fork() # PyDLL keeps the GIL through the call + if child == 0: + try: + signal.alarm(5) + os.close(r) + requested: list[str] = [] + logged: list[str] = [] + + class _RecordingThread(threading.Thread): # keeps the child single-threaded + def start(self) -> None: + requested.append(self.name) + + class _Recorder(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + logged.append(record.getMessage()) + + threading.Thread = _RecordingThread # the child's copy only; it never returns + logging.getLogger("cachekit").addHandler(_Recorder(logging.WARNING)) # where the refresh paths report + now = time.monotonic() + object_cache.time = types.SimpleNamespace(monotonic=lambda: now + 60) # past ttl * ratio, short of ttl + backend.stale = True + outcome: object = { + "hookless": l1_cache._forked_without_hooks(), + "served": (l1_only(), l2_swr()), + "requested": requested, + "calls": calls, + "logged": logged, + } + except BaseException as e: + outcome = {"error": repr(e)} + try: + report(w, outcome) + finally: + os._exit(1) + os.close(w) + return child_outcome(child, r) + + class TestSWRForkIsolation: """LAB-506 follow-through: SWR scheduler state must not survive fork(). @@ -394,6 +474,28 @@ def child(q) -> None: finally: gate.set() # release the parent's parked revalidation thread + @pytest.mark.skipif(not sys.platform.startswith("linux"), reason="needs glibc fork() via ctypes") + @pytest.mark.parametrize( + ("fork", "hookless", "requested"), + [ + pytest.param("c", True, [], id="c-fork"), + pytest.param("os", False, ["cachekit-swr-refresh", "cachekit-swr-revalidate"], id="os-fork"), + ], + ) + def test_child_forked_without_hooks_serves_stale_and_starts_no_refresh(self, fork, hookless, requested) -> None: + """A thread started in a child forked from C can hang in Thread.start(), so neither sync SWR path + asks for one there. Stale is still served; an os.fork() child still refreshes both.""" + run = run_fresh(_forked_stale_hit_outcome, fork) + + assert run.returncode == 0, run.stderr + assert ast.literal_eval(run.stdout) == { + "hookless": hookless, + "served": (1, 1), # the parent's values, stale in the child + "requested": requested, + "calls": {"l1": 1, "l2": 1}, # and not recomputed in the call + "logged": [], # logging's own locks may be orphaned in this child + }, run.stderr # a hung child's stack, dumped on its alarm + """Branches the main flows don't reach: L1 refresh, no-lock backends, sync failure, slot exhaustion, operation-handler degradation.""" diff --git a/tests/utils/fork_helpers.py b/tests/utils/fork_helpers.py index 2509461c..10903cd5 100644 --- a/tests/utils/fork_helpers.py +++ b/tests/utils/fork_helpers.py @@ -6,8 +6,12 @@ import os import select import signal +import subprocess +import sys import threading +import traceback from collections.abc import Callable +from pathlib import Path from typing import NoReturn @@ -45,3 +49,31 @@ def on_new_thread(fn: Callable[[], object], timeout: float = 5.0) -> object: thread.start() thread.join(timeout) return out[0] if out else "hung" + + +def run_fresh(outcome: Callable[..., object], *args: object) -> subprocess.CompletedProcess[str]: + """Run outcome(*args) in a fresh interpreter; its stdout is the outcome's repr, for ast.literal_eval. + + For a test that forks from C: pytest's own interpreter has threads it cannot end (pytest-xdist's, + other tests'), and one waiting for the GIL at such a fork hangs the child in no cachekit code. + outcome must be a module-level function under `tests`, and args literals. + """ + code = ( + f"from {outcome.__module__} import {outcome.__name__} as outcome; " + f"from tests.utils.fork_helpers import print_outcome; print_outcome(outcome, *{args!r})" + ) + root = Path(__file__).resolve().parents[2] # where `tests` imports from + return subprocess.run( # noqa: S603 - trusted: sys.executable + code built from our own names and literals + [sys.executable, "-c", code], cwd=root, capture_output=True, text=True, timeout=60 + ) + + +def print_outcome(outcome: Callable[..., object], *args: object) -> NoReturn: + """Print outcome(*args) for run_fresh and end without exit handlers: logging's would wait on a lock a gone thread holds.""" + try: + print(repr(outcome(*args)), flush=True) + except BaseException: + traceback.print_exc() + sys.stderr.flush() + os._exit(1) + os._exit(0)