diff --git a/docs/free-threading.md b/docs/free-threading.md index ca73d642..c9c5aafe 100644 --- a/docs/free-threading.md +++ b/docs/free-threading.md @@ -63,7 +63,7 @@ covered, plus what the free-threaded CI lane surfaced and sites added since: | `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. In a child forked without at-fork hooks (the detection of the `l1_cache.py` row), 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 (L1-only: the object cache; backed: L1, through `L1Cache.peek`), so a call that settled between the caller's lookup and its map check is not run twice; a backed caller with L1 off has nothing to re-check and reads L2 on its own trip instead. The lock order is the map lock, then the cache's state lock, which an expiry sweep can hold for a scan of the whole cache; nothing that holds a cache lock takes a map lock. A joining caller's `on_join` runs outside the map lock, before it waits. 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` | +| `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 (L1-only: the object cache; backed: L1, through `L1Cache.peek`), so a call that settled between the caller's lookup and its map check is not run twice; a backed caller with L1 off has nothing to re-check and reads L2 on its own trip instead. The lock order is the map lock, then the cache's state lock; nothing that holds a cache lock takes a map lock. A joining caller's `on_join` runs outside the map lock, before it waits. 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` | | `decorators/write_behind.py` `WriteBehind` (opt-in `write_behind=True`: one writer per decorated function, shared by its sync and async callers) | A `_State` holding the PID, a `Condition`, the queued and in-flight jobs, the byte count, the thread count and the closed flag, published as one reference. Every method binds it once (`_current`), and each writer thread is started with the state it serves and touches no other | Safe. Admission, coalescing, the thread count and the queue are read and written under the state's condition. A thread that finds nothing to send gives up its place in the same hold as that check, so a submit never counts a thread that is about to quit. Two sends of one entry never overlap, and `cancel` waits on the jobs in flight at the call, by identity, so a later send of the same entry cannot hold it. Fork: a child's first caller takes over under a lock from `dict.setdefault(pid, Lock())`, re-checks, and publishes a fresh `_State` in one store, so no thread pairs the child's PID with the parent's lock or queue; the PID travels with the state, as in the `client.py` row. A child forked without at-fork hooks (the detection of the `l1_cache.py` row) starts no writer thread, and its misses write on the caller. Tests in `tests/unit/test_write_behind.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. A process that first imports cachekit after a fork made from C (uWSGI `--lazy-apps` without `--py-call-uwsgi-fork-hooks`) is detected at that import (`cachekit/_fork.py`) when its parent ran Python threads at the fork: `threading` still lists them, under native thread ids that are no task of this process (`/proc/self/task`), where an `os.fork()` child's after-fork hook would have dropped them. Such a process has no import PID, so every PID of it counts as hookless and every gate above applies; an `os.fork()` child of it runs the hooks and is safe again. When its parent ran no threads it looks like a fresh process and starts the cleanup thread, which is safe: no parent thread held anything at the fork. The check is Linux only and sees only threads `threading` registered. When it cannot tell (no `/proc/self/task`, or the current thread's own id absent from it: another PID namespace, or a patched `get_native_id`) it takes the process for a fresh one. gevent green threads are listed under the native id of the thread they ran on, so a parent's green threads alone do not count. The current and main threads never count, nor any thread under either's native id, because after a fork the forking thread's object may keep the parent's native id: always after a fork made from C, and after `os.fork()` before Python 3.13. So a fork made from C from a thread other than the main one loses the parent's main thread as evidence; uWSGI forks from its main thread. It lists threads under `threading`'s `_active_limbo_lock`, so a parent thread that held that lock at the fork, which a thread does only while it starts, ends or lists threads, hangs the import, as `logging`'s module lock already can. Tests in `tests/unit/test_fork_import_signal.py` | | `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` | @@ -71,7 +71,7 @@ covered, plus what the free-threaded CI lane surfaced and sites added since: | `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 | | `decorators/wrapper.py` `_key_memo` (the [key memo](performance.md#decorator--l1-hit-hot-path) of an L1-only function) | Builtin `dict`, no lock. The L1-only hit reads it (`get`); a memo miss writes one entry, after emptying it if it holds 1,024. After an invalidation's deletes, `_delete_then_forget` scans a `copy()` for the erased key's entries and pops each from the live memo, or empties it for invalidate-all | Safe. Each operation is one dict call, atomic under the GIL and under the free-threaded build's per-object locking, and `dict.copy()` copies under the dict's lock. An entry only ever holds the key the generator built for its arguments, so a racing write re-inserts that mapping or drops entries, and a dropped entry costs one key rebuild. Writers racing past the check can leave a few more than 1,024 entries until the next write empties it. A call made during an invalidation's deletes can memoise its arguments again, so the memo forgets after the deletes, as the calls in flight do (`test_key_memo.py`, gated-delete tests). A paired A/B of L1-only hits against `main` on 3.14t with the GIL off found no median below base at 1, 4, 8, 16 and 32 threads | -| `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` | +| `object_cache.py` `ObjectCache` + `_reset_caches_after_fork` | `RLock` on every public method. The entries, their byte total, their expiry heap, 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/metrics_collection.py` `MetricsCollector` + `get_all_metrics` + `_metrics_lock()` | Module dicts of counters and gauges (`cache_operations_total`, `circuit_breaker_state`, ...) behind one `RLock` per PID, created with `dict.setdefault` | **Fixed** — every read and write takes the lock. A child gets its own, so a lock a parent thread held at a fork from C, with no at-fork hook run, is never waited on; before it was one module `RLock`, so a hookless child's first health check or `StructuredLogger.cache_operation()` could hang. The child's counts start from the values it inherited. Tested in `tests/unit/test_structured_logging.py` (`TestEntriesAfterFork`) | | `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` | diff --git a/docs/performance.md b/docs/performance.md index 365ca294..a6df0624 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -93,6 +93,12 @@ The deterministic figure is the `l1_hit` row of the [instruction budgets](#instr The raw L1 byte-cache lookup, without the decorator, takes 354–362ns. An L1-only hit uses a separate object cache, whose lookup no guard times on its own. +### L1-Only Miss at a Full Budget + +An L1-only miss stores its result in the object cache. Once the function's byte budget (`l1_max_size_mb`, 100MB by default) is full, each store first removes the entries that have expired, then evicts least-recently-used ones until the new value fits. Expired entries go first wherever they sit in LRU order, so an entry read just before it expired never pushes out a live one. A min-heap of deadlines finds them: when nothing has expired, the store looks at the earliest deadline and evicts, without scanning the cache for expired entries. The one exception is the heap rebuild described below. Before the heap, every such store scanned every entry while holding the lock the function's hits wait on, so with `ttl=None` (stored for a year, so nothing expires) each one cost time in proportion to the number of entries cached. + +The `object_cache_put_full` row of the [instruction budgets](#instruction-budgets) measures a store into a full budget of 10,000 entries: about 160,000 instructions on CPython 3.14, most of it the size estimate of the stored value, against 5.64 million when every store scanned. The heap is memory the budget does not count: about 64 bytes per entry on CPython 3.12 and 72 on 3.14, plus one stale record for each entry removed or refreshed since the heap was last rebuilt. Once the stale records outnumber the entries, the store that finds it rebuilds the heap from the cache: one pass over the entries, about once per as many stores as there are entries. + ## L2 Backend (Redis) Performance An L2 hit with L1 disabled, the 100-user dict (23.5KB as MessagePack) and Redis in a container on the same runner takes 551 ± 18μs on the hosted EPYC 7763: the decorator, the round trip and deserialization. That is about 610 times an L1-only hit of the same dict on the same runner. A Redis on another machine adds its network latency on top; no guard measures that. @@ -290,23 +296,24 @@ The regression gate is the instruction budget: `make perf-ir` on your machine, a | Path | What one call does | CPython 3.12 | CPython 3.14 | |------|--------------------|-------------:|-------------:| -| `l1_hit` | `@cache(backend=None)` L1 hit | 11,619 | 10,911 | -| `minimal_l1_hit` | `@cache.minimal(backend=None)` L1 hit | 10,510 | 9,796 | -| `l2_hit` | `@cache`, L1 disabled, L2 hit, as one in-process single-flight trip | 189,164 | 181,850 | -| `miss` | `@cache`, L1 disabled, L2 miss, compute, L2 write, all as one in-process single-flight trip | 222,591 | 212,467 | -| `secure_l1_hit` | `@cache.secure` L1 hit (builds its key, decrypts the ciphertext L1 holds) | 158,029 | 156,881 | -| `l2_hit_async_metrics` | `l2_hit` with the metrics collector in batched mode | 178,747 | 172,693 | -| `serializer_default` | `StandardSerializer` round trip, small dict | 58,618 | 59,638 | -| `serializer_default_records` | `StandardSerializer` round trip, list of 100 six-field records (dict-heavy decode) | 1,423,191 | 1,424,545 | -| `serializer_auto` | `AutoSerializer` round trip | 65,821 | 66,683 | -| `serializer_orjson` | `OrjsonSerializer` round trip | 23,679 | 23,775 | -| `serializer_arrow` | `ArrowSerializer` round trip, 100-row DataFrame | 1,975,014 | 1,970,889 | -| `serializer_arrow_usgs` | `ArrowSerializer` round trip, real 12,535-row DataFrame (see Real data below) | 61,849,178 | 61,414,623 | -| `serializer_auto_usgs` | `AutoSerializer` round trip, same frame (it hands the frame to Arrow) | 61,813,019 | 61,434,085 | -| `serializer_encrypted` | `EncryptionWrapper` encrypt + decrypt round trip | 99,816 | 101,866 | -| `byte_storage_store_40b` | `ByteStorage.store()` alone, 40-byte compressible input (under cachekit-core's 64-byte literals-only LZ4 threshold) | 2,656 | 2,527 | -| `byte_storage_store_64k` | `ByteStorage.store()` alone, 64 KiB of address-like text | 720,633 | 720,645 | -| `file_set` | `FileBackend.set()` overwriting one key in a 1,000-entry cache (no eviction) | 75,556 | 73,669 | +| `l1_hit` | `@cache(backend=None)` L1 hit | 11,632 | 10,896 | +| `minimal_l1_hit` | `@cache.minimal(backend=None)` L1 hit | 10,505 | 9,790 | +| `l2_hit` | `@cache`, L1 disabled, L2 hit, as one in-process single-flight trip | 188,634 | 182,083 | +| `miss` | `@cache`, L1 disabled, L2 miss, compute, L2 write, all as one in-process single-flight trip | 222,372 | 212,493 | +| `secure_l1_hit` | `@cache.secure` L1 hit (builds its key, decrypts the ciphertext L1 holds) | 157,389 | 155,969 | +| `l2_hit_async_metrics` | `l2_hit` with the metrics collector in batched mode | 178,846 | 172,851 | +| `serializer_default` | `StandardSerializer` round trip, small dict | 58,743 | 59,601 | +| `serializer_default_records` | `StandardSerializer` round trip, list of 100 six-field records (dict-heavy decode) | 1,421,931 | 1,422,896 | +| `serializer_auto` | `AutoSerializer` round trip | 65,943 | 66,651 | +| `serializer_orjson` | `OrjsonSerializer` round trip | 22,068 | 22,688 | +| `serializer_arrow` | `ArrowSerializer` round trip, 100-row DataFrame | 1,980,225 | 1,982,998 | +| `serializer_arrow_usgs` | `ArrowSerializer` round trip, real 12,535-row DataFrame (see Real data below) | 61,768,020 | 61,261,299 | +| `serializer_auto_usgs` | `AutoSerializer` round trip, same frame (it hands the frame to Arrow) | 61,770,277 | 61,264,653 | +| `serializer_encrypted` | `EncryptionWrapper` encrypt + decrypt round trip | 100,048 | 101,948 | +| `byte_storage_store_40b` | `ByteStorage.store()` alone, 40-byte compressible input (under cachekit-core's 64-byte literals-only LZ4 threshold) | 2,656 | 2,540 | +| `byte_storage_store_64k` | `ByteStorage.store()` alone, 64 KiB of address-like text | 720,634 | 720,648 | +| `file_set` | `FileBackend.set()` overwriting one key in a 1,000-entry cache (no eviction) | 75,881 | 73,987 | +| `object_cache_put_full` | `ObjectCache.put()` of a new key into a full 10,000-entry byte budget (evicts the LRU entry; nothing expired) | 164,507 | 160,678 | Budgets are per interpreter (minor version, build flavour, machine); an interpreter without budgets fails with `no budget`. They were recorded on CPython 3.12.12 and 3.14.3, x86_64, glibc 2.39, with the release extension that `uv sync` builds. Counts depend on that whole build, so on a different interpreter, extension or C library, record a baseline on `main` first (`--update --allow-increase`) and compare your branch against it. Batched mode costs the caller about 9,000 to 10,000 fewer instructions per L2 hit than synchronous recording, because the Prometheus update moves to the worker thread. The gap was about 50,000 until each label tuple's Prometheus series was bound once instead of looked up per call. diff --git a/src/cachekit/object_cache.py b/src/cachekit/object_cache.py index 65b94640..2b25db56 100644 --- a/src/cachekit/object_cache.py +++ b/src/cachekit/object_cache.py @@ -7,6 +7,7 @@ from __future__ import annotations +import heapq import math import os import random @@ -67,18 +68,24 @@ class _Entry: class _ObjectCacheState: - """The entries, their byte total, their in-flight refresh markers, and the lock guarding all three. + """The entries, their byte total, their expiry heap, their in-flight refresh markers, and the lock guarding them. One object so the at-fork hook can replace all of it with a single attribute store: a thread still inside a critical section finishes on the state it bound. The stats and the generation counter stay on the cache, so a swap keeps them. """ - __slots__ = ("lock", "refreshing", "size_bytes", "store") + __slots__ = ("expiry", "lock", "refreshing", "size_bytes", "store") def __init__(self) -> None: self.store: OrderedDict[str, _Entry] = OrderedDict() self.size_bytes = 0 + # Min-heap of (expires_at, key), one per stored or refreshed deadline, so the over-bound + # eviction finds expired entries without scanning the store. Deletion is lazy: a removal or a + # refresh leaves the old tuple behind. A tuple is live while its key holds an entry with that + # deadline, and every entry's current deadline has one, so a dead tuple is safe to drop. + # push() rebuilds the heap from the store once dead tuples outnumber live entries. + self.expiry: list[tuple[float, str]] = [] # In-flight refresh ownership: key -> owning generation. A refresh captures the entry's # generation at read time and only lands — or clears/cancels its marker — for that exact # generation, so a stale refresh from a replaced entry can neither resurrect data nor @@ -102,13 +109,46 @@ def remove(self, key: str) -> None: self.size_bytes -= entry.size_bytes self.refreshing.pop(key, None) + def push(self, key: str, entry: _Entry) -> None: + """Record a stored entry's deadline in the expiry heap; call with lock held, after the store write. + + A rebuild from the store drops the dead tuples once they outnumber the live entries. It costs + O(n) in the entries left, and it at least halves the heap, so it amortises to O(1) per push. + """ + if len(self.expiry) >= 2 * len(self.store): + self.expiry = [(e.expires_at, k) for k, e in self.store.items()] + heapq.heapify(self.expiry) + else: + heapq.heappush(self.expiry, (entry.expires_at, key)) + + def remove_expired(self, now: float) -> None: + """Remove every entry whose deadline has passed; call with lock held. + + Pops heap tuples only while the earliest deadline is due, so when nothing has expired it + returns after one comparison. Each due tuple costs O(log n). A due tuple is dead when its key + holds no entry with that deadline, as a replaced or refreshed entry leaves behind; it is dropped. + """ + heap = self.expiry + while heap and heap[0][0] <= now: + expires_at, key = heapq.heappop(heap) + entry = self.store.get(key) + if entry is not None and entry.expires_at == expires_at: + self.remove(key) + class ObjectCache: """Thread-safe in-memory cache storing Python object references directly. Uses an OrderedDict for O(1) LRU ordering. On a put() when a bound is - exceeded, expired entries are swept first; if still over, the oldest fresh - entries are evicted (LRU). + exceeded, expired entries are removed first; if still over, the oldest fresh + entries are evicted (LRU). A min-heap of deadlines finds the expired entries, + so a put at capacity costs O(log n) per expired entry it removes and does not + scan for expired entries. The heap is memory the byte bound does not + count: about 64 to 72 bytes per entry (one 2-tuple and its list slot on 64-bit + CPython), plus the tuples of removed or refreshed entries. Once those + outnumber the live ones, the put that finds it rebuilds the heap: an O(n) + pause once per about n puts at capacity, against a full scan on every one + before the heap. Bounds (at least one required): - max_entries: entry-count bound (default 256, pass None to disable) @@ -373,6 +413,7 @@ def complete_refresh(self, key: str, version: int, value: Any, ttl: float) -> bo entry.size_bytes = size entry.refresh_failed_at = None # success ends any retry back-off s.store.move_to_end(key) + s.push(key, entry) # the old deadline's tuple is now dead # New value may be larger — restore the byte bound by evicting LRU others self._evict(s, extra_bytes=0, need_slot=False) return True @@ -418,7 +459,9 @@ def put(self, key: str, value: Any, ttl: int) -> None: """Store a value in the cache. When a bound is exceeded: - 1. Expired entries are removed first. + 1. Expired entries are removed first, wherever they sit in LRU order. + The expiry heap finds them, so no entry is scanned to look for them; the one + put in about n that rebuilds the heap makes one pass over the entries. 2. If still over, the oldest (LRU) fresh entries are evicted. A value whose estimated size exceeds the entire byte budget is never @@ -454,9 +497,10 @@ def put(self, key: str, value: Any, ttl: int) -> None: now = time.monotonic() self._generation += 1 - s.store[key] = _Entry(value=value, expires_at=now + ttl, cached_at=now, size_bytes=size, generation=self._generation) + entry = _Entry(value=value, expires_at=now + ttl, cached_at=now, size_bytes=size, generation=self._generation) + s.store[key] = entry # a new key appends at the MRU end, so no move_to_end s.size_bytes += size - # No move_to_end needed — OrderedDict.__setitem__ appends new keys to end + s.push(key, entry) def delete(self, key: str) -> bool: """Remove a single entry from the cache. @@ -488,6 +532,7 @@ def clear(self) -> None: with s.lock: s.store.clear() s.size_bytes = 0 + s.expiry.clear() s.refreshing.clear() # ------------------------------------------------------------------ @@ -540,7 +585,8 @@ def _evict(self, s: _ObjectCacheState, extra_bytes: int, need_slot: bool) -> Non Must be called with s.lock held. Strategy: - 1. If any bound is exceeded, remove all expired entries first. + 1. If any bound is exceeded, remove all expired entries first, popping them off the + expiry heap: one comparison when none has expired, never a scan of the store. 2. While still over a bound, evict the oldest (LRU) fresh entry. Args: @@ -558,11 +604,7 @@ def over_bounds() -> bool: if not over_bounds(): return - # Sweep expired entries first - now = time.monotonic() - expired_keys = [k for k, e in s.store.items() if now >= e.expires_at] - for k in expired_keys: - s.remove(k) + s.remove_expired(time.monotonic()) # Still over a bound — evict the least-recently-used fresh entries while s.store and over_bounds(): diff --git a/tests/performance/ir_baselines.json b/tests/performance/ir_baselines.json index e60cb7cf..68f94cda 100644 --- a/tests/performance/ir_baselines.json +++ b/tests/performance/ir_baselines.json @@ -3,44 +3,46 @@ "cpython-3.12-x86_64": { "budgets": { "byte_storage_store_40b": 2656, - "byte_storage_store_64k": 720633, - "file_set": 75556, - "l1_hit": 11619, - "l2_hit": 189164, - "l2_hit_async_metrics": 178747, - "minimal_l1_hit": 10510, - "miss": 222591, - "secure_l1_hit": 158029, - "serializer_arrow": 1975014, - "serializer_arrow_usgs": 61849178, - "serializer_auto": 65821, - "serializer_auto_usgs": 61813019, - "serializer_default": 58618, - "serializer_default_records": 1423191, - "serializer_encrypted": 99816, - "serializer_orjson": 23679 + "byte_storage_store_64k": 720634, + "file_set": 75881, + "l1_hit": 11632, + "l2_hit": 188634, + "l2_hit_async_metrics": 178846, + "minimal_l1_hit": 10505, + "miss": 222372, + "object_cache_put_full": 164507, + "secure_l1_hit": 157389, + "serializer_arrow": 1980225, + "serializer_arrow_usgs": 61768020, + "serializer_auto": 65943, + "serializer_auto_usgs": 61770277, + "serializer_default": 58743, + "serializer_default_records": 1421931, + "serializer_encrypted": 100048, + "serializer_orjson": 22068 }, "python": "3.12.12" }, "cpython-3.14-x86_64": { "budgets": { - "byte_storage_store_40b": 2527, - "byte_storage_store_64k": 720645, - "file_set": 73669, - "l1_hit": 10911, - "l2_hit": 181850, - "l2_hit_async_metrics": 172693, - "minimal_l1_hit": 9796, - "miss": 212467, - "secure_l1_hit": 156881, - "serializer_arrow": 1970889, - "serializer_arrow_usgs": 61414623, - "serializer_auto": 66683, - "serializer_auto_usgs": 61434085, - "serializer_default": 59638, - "serializer_default_records": 1424545, - "serializer_encrypted": 101866, - "serializer_orjson": 23775 + "byte_storage_store_40b": 2540, + "byte_storage_store_64k": 720648, + "file_set": 73987, + "l1_hit": 10896, + "l2_hit": 182083, + "l2_hit_async_metrics": 172851, + "minimal_l1_hit": 9790, + "miss": 212493, + "object_cache_put_full": 160678, + "secure_l1_hit": 155969, + "serializer_arrow": 1982998, + "serializer_arrow_usgs": 61261299, + "serializer_auto": 66651, + "serializer_auto_usgs": 61264653, + "serializer_default": 59601, + "serializer_default_records": 1422896, + "serializer_encrypted": 101948, + "serializer_orjson": 22688 }, "python": "3.14.3" } diff --git a/tests/performance/ir_budget.py b/tests/performance/ir_budget.py index dd70aa13..95857967 100644 --- a/tests/performance/ir_budget.py +++ b/tests/performance/ir_budget.py @@ -112,6 +112,7 @@ "byte_storage_store_40b", "byte_storage_store_64k", "file_set", + "object_cache_put_full", ) # ── gate ──────────────────────────────────────────────────────────────────────────────────── diff --git a/tests/performance/ir_workload.py b/tests/performance/ir_workload.py index 45cac2f7..4c63cafc 100644 --- a/tests/performance/ir_workload.py +++ b/tests/performance/ir_workload.py @@ -18,6 +18,7 @@ from typing import Any FILE_SET_ENTRIES = 1000 +OBJECT_CACHE_ENTRIES = 10_000 # ByteStorage.store() inputs. 40 B is under cachekit-core's 64 B literals-only LZ4 threshold, which # every serializer path's input clears (the smallest, serializer_default's msgpack, is 79 B). The # 64 KiB input is address-like text, so the encoder finds many short matches rather than one run. @@ -150,6 +151,19 @@ def call(*args: Any) -> dict: backend.set(f"entry:{i}", payload) return lambda: backend.set("entry:0", payload) + if path == "object_cache_put_full": + # ObjectCache.put() of a new key into a full byte budget, as an L1-only miss stores its result + # once the budget holds OBJECT_CACHE_ENTRIES fresh entries: each put evicts the LRU head. + import itertools + + from cachekit.object_cache import ObjectCache, _estimate_object_size + + oc = ObjectCache(max_entries=None, max_size_bytes=OBJECT_CACHE_ENTRIES * _estimate_object_size(value)) + keys = itertools.count() + for _ in range(OBJECT_CACHE_ENTRIES): + oc.put(f"entry:{next(keys)}", value, ttl=3600) # pinned clocks never reach the deadline + return lambda: oc.put(f"entry:{next(keys)}", value, ttl=3600) + if path in BYTE_STORAGE_PAYLOADS: # ByteStorage.store() alone: LZ4, xxHash3-64 and the envelope, with no serializer around it. from cachekit._rust_serializer import ByteStorage diff --git a/tests/unit/test_object_cache.py b/tests/unit/test_object_cache.py index 206fac97..f08ece9e 100644 --- a/tests/unit/test_object_cache.py +++ b/tests/unit/test_object_cache.py @@ -12,6 +12,7 @@ import time import types import weakref +from collections import OrderedDict from collections.abc import Callable from typing import Any @@ -128,6 +129,124 @@ def test_put_evicts_expired_before_lru(self, monkeypatch: pytest.MonkeyPatch) -> assert found_a is False assert found_b is False + def test_put_evicts_expired_entry_deep_in_lru_order_before_live_head(self, monkeypatch: pytest.MonkeyPatch) -> None: + """An expired entry read just before it expired sits at the MRU end; it still goes before the live LRU head.""" + fake = _fake_clock(monkeypatch) + oc = ObjectCache(max_entries=3) + oc.put("head", "H", ttl=100) + oc.put("short", "S", ttl=10) # expires at 1010 + oc.put("mid", "M", ttl=100) + assert oc.get("short")[0] is True # LRU order now: head, mid, short + + fake.monotonic = lambda: 1015.0 + oc.put("new", "N", ttl=100) + + assert oc.peek("head")[0] is True + assert oc.peek("short")[0] is False + assert oc.size == 3 + + def test_put_at_capacity_scans_no_entry_when_none_has_expired(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A put that evicts while every entry is fresh looks only at the earliest deadline, not at each entry.""" + _fake_clock(monkeypatch) + oc = ObjectCache(max_entries=100) + scans = [] + + class Store(OrderedDict): + def items(self): + scans.append("items") + return super().items() + + def values(self): + scans.append("values") + return super().values() + + oc._state.store = Store() + for i in range(100): + oc.put(f"k{i}", i, ttl=600) + scans.clear() + + for i in range(100, 150): + oc.put(f"k{i}", i, ttl=600) # each evicts the LRU head + + assert scans == [] + assert oc.size == 100 + assert oc.peek("k49")[0] is False and oc.peek("k50")[0] is True + + def test_refreshed_entry_is_not_evicted_at_its_old_deadline(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A refresh moves the deadline, so the heap tuple of the old one must not remove the live entry.""" + fake = _fake_clock(monkeypatch) + oc = ObjectCache(max_entries=2, swr_threshold_ratio=0.5) + oc.put("k", "v1", ttl=10) # expires at 1010 + oc.put("other", "o", ttl=100) + + fake.monotonic = lambda: 1006.0 + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=10) + assert needs_refresh + assert oc.complete_refresh("k", version, "v2", ttl=10) is True # expires at 1016; LRU: other, k + + fake.monotonic = lambda: 1012.0 # past the old deadline only + oc.put("new", "n", ttl=100) + + assert oc.peek("k") == (True, "v2") + assert oc.peek("other")[0] is False # the LRU head went, as for any live entries + assert _consistent(oc) + + def test_replaced_entry_is_not_evicted_at_the_old_entrys_deadline(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A put over a key leaves the old entry's heap tuple behind; when it comes due it must not remove the new entry.""" + fake = _fake_clock(monkeypatch) + oc = ObjectCache(max_entries=2) + oc.put("k", "v1", ttl=10) # expires at 1010 + oc.put("other", "o", ttl=100) + oc.put("k", "v2", ttl=10) # same deadline, second tuple; LRU: other, k + + fake.monotonic = lambda: 1005.0 + oc.put("k", "v3", ttl=100) # expires at 1105 + fake.monotonic = lambda: 1012.0 + oc.put("new", "n", ttl=100) + + assert oc.peek("k") == (True, "v3") + assert oc.peek("other")[0] is False + + def test_complete_refresh_over_the_byte_bound_evicts_expired_before_lru(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A refresh that grows its value restores the byte bound through the same expired-first eviction.""" + fake = _fake_clock(monkeypatch) + small, large = "x" * 100, "x" * 1000 + budget = _estimate_object_size(large) + 2 * _estimate_object_size(small) + oc = ObjectCache(max_entries=None, max_size_bytes=budget, swr_threshold_ratio=0.5) + oc.put("live", small, ttl=100) # the LRU head + oc.put("short", small, ttl=10) # expires at 1010 + oc.put("k", small, ttl=20) + assert oc.get("short")[0] is True # LRU order now: live, k, short + + fake.monotonic = lambda: 1011.0 + _, _, needs_refresh, version = oc.get_with_swr("k", ttl=20) + assert needs_refresh + assert oc.complete_refresh("k", version, large, ttl=20) is True + + assert oc.peek("live")[0] is True + assert oc.peek("short")[0] is False + assert oc.size_bytes <= budget + + def test_expiry_heap_stays_bounded_at_capacity(self) -> None: + """Every put at capacity leaves a dead heap tuple; the heap is rebuilt before it outgrows the entries twice over. + + The hot entry is read after every put, so it stays out of LRU reach while its tuple, the + earliest deadline, holds the heap top: dead tuples pile up behind it until a rebuild. + """ + oc = ObjectCache(max_entries=50) + oc.put("hot", "h", ttl=600) + longest = 0 + for i in range(1000): + oc.put(f"k{i}", i, ttl=600) + assert oc.get("hot")[0] is True + longest = max(longest, len(oc._state.expiry)) + assert len(oc._state.expiry) <= 2 * oc.size + assert longest == 2 * oc.size # the heap reached its bound and was rebuilt there, never passing it + assert _consistent(oc) + + oc.clear() + assert oc._state.expiry == [] + @pytest.mark.unit class TestObjectCacheLRU: @@ -707,10 +826,11 @@ def test_rejects_negative_or_nan_interval(self, bad: float) -> None: def _consistent(oc: ObjectCache) -> bool: - """Whether the byte total and the in-flight refresh markers agree with the entries.""" + """Whether the byte total, the expiry heap and the in-flight refresh markers agree with the entries.""" s = oc._state sized = s.size_bytes == sum(entry.size_bytes for entry in s.store.values()) - return sized and all(key in s.store and s.store[key].generation == g for key, g in s.refreshing.items()) + tracked = {(e.expires_at, k) for k, e in s.store.items()} <= set(s.expiry) + return sized and tracked and all(key in s.store and s.store[key].generation == g for key, g in s.refreshing.items()) @pytest.mark.unit