diff --git a/docs/api-reference.md b/docs/api-reference.md index 93659618..a3ce6d57 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -360,11 +360,12 @@ orchestrator = FeatureOrchestrator( These components work behind the scenes to provide reliability features: #### `AsyncMetricsCollector` -Non-blocking metrics collection system that prevents performance degradation: -- Queue-based collection with overflow protection -- Background thread processing +Metrics collection kept off the shared hot path: +- Cache operations record into per-thread cells, which the Prometheus series sum when scraped: a record takes no + shared lock, and a scrape counts every operation recorded before it +- Generic counters and histograms record synchronously, or queue for a background worker once a collector is busy, + with overflow protection - Self-healing worker thread management -- Zero impact on critical path latency #### `get_cached_redis_client()` Thread-local Redis client caching (`cachekit.backends.redis.client`): diff --git a/docs/features/prometheus-metrics.md b/docs/features/prometheus-metrics.md index e1e867b0..bbbe1143 100644 --- a/docs/features/prometheus-metrics.md +++ b/docs/features/prometheus-metrics.md @@ -331,7 +331,9 @@ field list and an example stats endpoint. ## Performance Impact -No wall-clock guard times metrics recording, so the Performance Guide publishes no timing for it. Its [Instruction Budgets](../performance.md#instruction-budgets) give the instruction cost of an L2 hit with synchronous and with batched metrics recording. +No wall-clock guard times metrics recording, so the Performance Guide publishes no timing for it. Its [Instruction Budgets](../performance.md#instruction-budgets) give the instruction cost of an L2 hit with a synchronous and with a batched collector. + +Each thread records a cache operation into cells of its own, and the series add those cells up when Prometheus scrapes them. So a record takes no lock another thread waits on, which matters on free-threaded Python, and a scrape counts every operation recorded before it, in either collector mode. With `PROMETHEUS_MULTIPROC_DIR` set, a scrape reads each process's files instead of these values, so cache operations then record into the series directly. --- diff --git a/docs/free-threading.md b/docs/free-threading.md index c9c5aafe..1a583b0f 100644 --- a/docs/free-threading.md +++ b/docs/free-threading.md @@ -75,8 +75,9 @@ covered, plus what the free-threaded CI lane surfaced and sites added since: | `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` | +| `reliability/async_metrics.py` `AsyncMetricsCollector.record_cache_operation` + `_ThreadState` + `_Shards` + `_ShardView` | Every backed cache operation counts once per collector and updates three Prometheus series. It used to write a shared `_operation_count` (unlocked, so lossy with the GIL off) and either the series' value locks or the batching queue | **Fixed** — each thread writes only its own `_ThreadState`, held in a `threading.local`: one cell per series it records into, and one operation count per collector. The series' values are `_ShardView`s that sum the live cells plus an exact retired total, read under one per-PID lock that no record takes, so a scrape and `get_stats()` count every operation recorded before them. Freed at thread exit, a `_ThreadState` moves its counts into the retired totals and drops its cells in one locked step, so reads never decrease and cells stay bounded to the live threads; sums are retired as exact partials (`math.fsum`), so a retirement never re-rounds a total one ulp lower, which `rate()` would read as a counter reset (an infinite term collapses the sum to `inf`). The lock is reentrant, because a garbage collection under it can run a finalizer on the same thread. A count move such a finalizer starts there is deferred to the next section that is not nested. One example is a cyclic collection freeing a thread's state that a traceback kept alive past the exit, so no read sees a count half moved. The at-fork hooks hold the lock across `os.fork`; a forked child starts from the counts it inherited. In prometheus_client's multiprocess mode cache operations record into the series directly. Tests in `tests/unit/test_async_metrics_per_thread.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`) | -| `reliability/async_metrics.py` `AsyncMetricsCollector._maybe_switch_mode`, `shutdown`, `_take_over_if_forked` | Auto sync/batched switch; `_should_check_mode` is an unlocked check-then-set, so two producers can both run a switch. A forked child inherits the batching state (queue, stop event, worker, pool lock) and prometheus_client's metric locks | **Fixed** — one `Lock` held across the switch, so concurrent switches start one worker. The lock is per PID (a per-instance dict filled with `setdefault`, like `_metrics_locks`), so a child forked while a thread is mid-switch never waits on the copy that thread still holds. `shutdown()` takes the same lock, bars any later switch from starting a worker, and sets sync mode before stopping the worker, so every later record lands synchronously; a record another thread had already chosen to queue can miss the worker's final drain (at most one per such thread). Returning to batched mode starts a worker on the existing queue only once the previous worker has exited (its exit drain assumes a single consumer); while it is alive the collector stays synchronous and retries at the next mode check, never joining on the caller's thread. Within a process the queue is never replaced, so producers always find one. **Fork:** an owner-PID take-over gives an inherited collector fresh batching state and sync mode in a forked child. The at-fork hook (after a fork made from C, the first take-over or collector built in the child instead) gives cachekit's own metrics and the default registry fresh prometheus_client locks; an application's own metrics are not covered, and neither are values in prometheus_client's multiprocess mode (`PROMETHEUS_MULTIPROC_DIR`), which share one lock the reset cannot reach, so a child forked while a parent thread holds it can still hang. No collector, inherited or built there, starts a thread in the child of a fork made from C, where at-fork hooks are skipped and `is_alive()` stays stale, nor in a child that first imports cachekit after such a fork of a parent that ran threads (uWSGI `--lazy-apps`; the detection of the `l1_cache.py` row). On free-threaded builds such a fork without `--py-call-uwsgi-fork-hooks` is unsupported: any Python code in that child can hang on a per-object lock a parent thread held, before cachekit can act. The `_take_over_if_forked`, `_reset_metric_locks` and `_reset_metric_locks_once` docstrings give the mechanics and the open cases. There is no per-record `getpid()` on the sync path. Tests in `tests/unit/test_async_metrics_mode_switch.py` | +| `reliability/async_metrics.py` `AsyncMetricsCollector._maybe_switch_mode`, `shutdown`, `_take_over_if_forked` | Auto sync/batched switch; `_should_check_mode` is an unlocked check-then-set, so two producers can both run a switch. A forked child inherits the batching state (queue, stop event, worker, pool lock) and prometheus_client's metric locks | **Fixed** — one `Lock` held across the switch, so concurrent switches start one worker. The lock is per PID (a per-instance dict filled with `setdefault`, like `_metrics_locks`), so a child forked while a thread is mid-switch never waits on the copy that thread still holds. `shutdown()` takes the same lock, bars any later switch from starting a worker, and sets sync mode before stopping the worker, so every later record lands synchronously; a record another thread had already chosen to queue can miss the worker's final drain (at most one per such thread). Returning to batched mode starts a worker on the existing queue only once the previous worker has exited (its exit drain assumes a single consumer); while it is alive the collector stays synchronous and retries at the next mode check, never joining on the caller's thread. Within a process the queue is never replaced, so producers always find one. **Fork:** an owner-PID take-over gives an inherited collector fresh batching state and sync mode in a forked child. The at-fork hook (after a fork made from C, the first take-over or collector built in the child instead) gives cachekit's own metrics and the default registry fresh prometheus_client locks; an application's own metrics are not covered, and neither are values in prometheus_client's multiprocess mode (`PROMETHEUS_MULTIPROC_DIR`), which share one lock the reset cannot reach, so a child forked while a parent thread holds it can still hang. No collector, inherited or built there, starts a thread in the child of a fork made from C, where at-fork hooks are skipped and `is_alive()` stays stale, nor in a child that first imports cachekit after such a fork of a parent that ran threads (uWSGI `--lazy-apps`; the detection of the `l1_cache.py` row). On free-threaded builds such a fork without `--py-call-uwsgi-fork-hooks` is unsupported: any Python code in that child can hang on a per-object lock a parent thread held, before cachekit can act. The `_take_over_if_forked`, `_reset_metric_locks` and `_reset_metric_locks_once` docstrings give the mechanics and the open cases. Cache operations never queue, so only generic metrics and mode checks touch the batching state. There is no per-record `getpid()` on the sync path. Tests in `tests/unit/test_async_metrics_mode_switch.py` | | Rust extension (`rust/src/`, cachekit-core 0.7.2) | `#[pymodule(gil_used = false)]`; every `#[pyclass]` exposes only `&self` methods; the nonce counters (the process-wide instance counter and each encryptor's own), core's only shared state in the features cachekit-py enables, are `AtomicU64`; PyO3 enforces `Send + Sync` on pyclasses at compile time | Safe — declared free-threading-ready. (The wasm32 `Cell` nonce variant is single-threaded by target.) | ## The CI safety net diff --git a/docs/performance.md b/docs/performance.md index a6df0624..a82bb091 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -290,36 +290,36 @@ The regression gate is the instruction budget: `make perf-ir` on your machine, a - Each path runs at five heap layouts (0 to 880 extra objects held before the workload) and its figure is the median. Any code change moves the layout, and the allocators take shorter or longer paths in some heap states: single layouts of the orjson round trip and the minimal L1 hit sat up to 2.4% from the rest. The median ignores one or two such layouts. The cheapest layout is not used, because a lucky one can sit 1% below the rest, and a budget recorded there makes every later run look like a regression. - `ir_workload.py`'s text is compiled into every measured process, so editing it moves the layout the budgets were recorded at: re-record after any change to it. The gate itself (`ir_budget.py`) is kept out of the measured process, so editing it does not move a budget. - The L2 paths use an in-process dict backend, so they cost instructions only: no sockets, retries or timeouts. -- The metrics collector starts synchronous and switches to batched mode (each call queues its record for a worker thread) when its 5 s check sees more than 100 records/s. Pinned clocks never reach that check, so the plain paths measure synchronous recording, and `l2_hit_async_metrics` measures the batched mode a busy long-lived process runs. Every metrics path records once per call through the same method, so one batched path covers that mode. +- The metrics collector starts synchronous and switches to batched mode (generic metrics queue for a worker thread) when its 5 s check sees more than 100 records/s. Cache operations record into per-thread cells in either mode. Pinned clocks never reach that check, so the plain paths measure a synchronous collector, and `l2_hit_async_metrics` the batched collector a busy long-lived process runs; it checks that no cache operation queues. Every metrics path records once per call through the same method, so one batched path covers that mode. **Budgets** (instructions per call, `tests/performance/ir_baselines.json`): | Path | What one call does | CPython 3.12 | CPython 3.14 | |------|--------------------|-------------:|-------------:| -| `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. +| `l1_hit` | `@cache(backend=None)` L1 hit | 11,614 | 10,904 | +| `minimal_l1_hit` | `@cache.minimal(backend=None)` L1 hit | 10,496 | 9,788 | +| `l2_hit` | `@cache`, L1 disabled, L2 hit, as one in-process single-flight trip | 171,988 | 167,408 | +| `miss` | `@cache`, L1 disabled, L2 miss, compute, L2 write, all as one in-process single-flight trip | 211,434 | 203,521 | +| `secure_l1_hit` | `@cache.secure` L1 hit (builds its key, decrypts the ciphertext L1 holds) | 139,943 | 140,926 | +| `l2_hit_async_metrics` | `l2_hit` with the metrics collector in batched mode | 171,370 | 167,447 | +| `serializer_default` | `StandardSerializer` round trip, small dict | 58,661 | 59,595 | +| `serializer_default_records` | `StandardSerializer` round trip, list of 100 six-field records (dict-heavy decode) | 1,422,079 | 1,422,120 | +| `serializer_auto` | `AutoSerializer` round trip | 65,971 | 66,624 | +| `serializer_orjson` | `OrjsonSerializer` round trip | 22,085 | 22,677 | +| `serializer_arrow` | `ArrowSerializer` round trip, 100-row DataFrame | 1,972,713 | 1,982,226 | +| `serializer_arrow_usgs` | `ArrowSerializer` round trip, real 12,535-row DataFrame (see Real data below) | 61,750,733 | 61,321,806 | +| `serializer_auto_usgs` | `AutoSerializer` round trip, same frame (it hands the frame to Arrow) | 61,806,031 | 61,327,994 | +| `serializer_encrypted` | `EncryptionWrapper` encrypt + decrypt round trip | 100,164 | 101,927 | +| `byte_storage_store_40b` | `ByteStorage.store()` alone, 40-byte compressible input (under cachekit-core's 64-byte literals-only LZ4 threshold) | 2,658 | 2,530 | +| `byte_storage_store_64k` | `ByteStorage.store()` alone, 64 KiB of address-like text | 720,667 | 720,663 | +| `file_set` | `FileBackend.set()` overwriting one key in a 1,000-entry cache (no eviction) | 75,887 | 74,132 | +| `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,514 | 160,511 | + +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. `l2_hit` and `l2_hit_async_metrics` now cost the same within 0.4%: a cache operation records into per-thread cells in either mode. Against the previous budgets that is 14,700 to 16,600 fewer instructions per L2 hit than the synchronous Prometheus update, and 5,400 to 7,500 fewer than queueing the record for the batching worker. **Sensitivity:** before the key memo, one extra BLAKE2b hash of the cache key per call raised every key-generating path by 4,700 to 4,900 instructions (`l1_hit` +6.1%, `l2_hit_async_metrics` +1.5%, `miss` +1.4%) and failed the gate, while the serializer paths, which generate no key, stayed within 0.11%. An interleaved wall-clock run agreed in sign (+286ns per L1 hit, median of 12 paired processes). -**Limits:** instruction counts do not weight cache misses or branch mispredictions. A claimed speed-up still needs an interleaved wall-clock comparison; the instruction count only guarantees the work did not grow. Paths that wait on a network backend are not covered. Cyclic-GC cost is outside the budgets; allocation and reference counting are inside. In batched mode the worker's Prometheus update runs on its own thread and is not budgeted, and so does part of pyarrow's work on the real-data paths: about half of their instructions run on pyarrow's threads. The orjson round trip's 1 KB output buffer comes from glibc malloc, whose path length depends on heap state that no layout sample pins, so its figure moved 1.3% between unrelated changes; it is gated at 2%. Unrelated changes can still move another path's figure by up to 0.6% (a `WARN`), and a `LOWER` verdict on a path the change did not touch is a layout shift, not a saving: ratchet only the paths the change touched (`--update --path `). A change to the Rust extension moves the paths that run through it more, with no change in their work: `main`'s own source, built with and without symbol stripping, moved the `ByteStorage` round trip of `serializer_default` by 1.9%, while every function in it cost the same over 5,000 round trips, so the move is a one-time allocator event that the binary's layout puts inside one loop size. After an extension change, re-record the serializer paths that fail (`--update --allow-increase --path `) and say why in the PR. +**Limits:** instruction counts do not weight cache misses or branch mispredictions. A claimed speed-up still needs an interleaved wall-clock comparison; the instruction count only guarantees the work did not grow. Paths that wait on a network backend are not covered. Cyclic-GC cost is outside the budgets; allocation and reference counting are inside. A Prometheus scrape, which sums each thread's cache-operation cells, runs outside the measured calls and is not budgeted. Part of pyarrow's work on the real-data paths is not budgeted either: about half of their instructions run on pyarrow's threads. The orjson round trip's 1 KB output buffer comes from glibc malloc, whose path length depends on heap state that no layout sample pins, so its figure moved 1.3% between unrelated changes; it is gated at 2%. Unrelated changes can still move another path's figure by up to 0.6% (a `WARN`), and a `LOWER` verdict on a path the change did not touch is a layout shift, not a saving: ratchet only the paths the change touched (`--update --path `). A change to the Rust extension moves the paths that run through it more, with no change in their work: `main`'s own source, built with and without symbol stripping, moved the `ByteStorage` round trip of `serializer_default` by 1.9%, while every function in it cost the same over 5,000 round trips, so the move is a one-time allocator event that the binary's layout puts inside one loop size. After an extension change, re-record the serializer paths that fail (`--update --allow-increase --path `) and say why in the PR. ```bash make perf-ir # gate: fail on a >=1% per-call regression (orjson 2%; needs valgrind; 170 runs, several minutes) diff --git a/src/cachekit/reliability/async_metrics.py b/src/cachekit/reliability/async_metrics.py index 10a4801d..1f66054c 100644 --- a/src/cachekit/reliability/async_metrics.py +++ b/src/cachekit/reliability/async_metrics.py @@ -1,18 +1,25 @@ """Async metrics collection for high-performance reliability features. -This module provides asynchronous, batched metrics collection to eliminate -synchronous Prometheus updates from the hot path. +Cache operations record into per-thread cells that their Prometheus series sum when read, so the hot path takes no +shared lock and a scrape is exact at once. Generic counters and histograms record synchronously or through a +batching worker. """ import logging +import math import numbers import os import queue import threading import time import warnings +import weakref +from bisect import bisect_left from collections import defaultdict -from typing import Any, Optional, Union +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from functools import partial +from typing import Any, Optional, TypeVar, Union from cachekit import _fork from cachekit.hash_utils import redact_error_for_log @@ -39,18 +46,20 @@ # Bytes, powers of 4: 16 B through 256 MiB, above the default max_value_size (100 MiB). SIZE_BUCKETS_BYTES = tuple(float(4**i) for i in range(2, 15)) -# A batched flush's cache_operations_total series: (operation, namespace, success, serializer). +# A cache operation's label tuple: (operation, namespace, success, serializer). _CacheOpKey = tuple[str, str, bool, str] -# One batched record's histogram values: (operation, namespace, success, serializer, duration_ms, size_bytes). None -# marks a value the sync path would not observe. -_CacheObservation = tuple[str, str, bool, str, Optional[float], Optional[int]] try: from prometheus_client import REGISTRY, Counter, Gauge, Histogram # type: ignore[assignment] + from prometheus_client import values as _prometheus_values PROMETHEUS_AVAILABLE = True + # Per-thread cells need in-process values. In multiprocess mode (PROMETHEUS_MULTIPROC_DIR) a scrape reads each + # process's mmap files, not the values, so cache operations record into the series directly instead. + _PER_THREAD = _prometheus_values.ValueClass is getattr(_prometheus_values, "MutexValue", None) except ImportError: PROMETHEUS_AVAILABLE = False # type: ignore[misc] + _PER_THREAD = False # type: ignore[misc] REGISTRY = None # type: ignore[assignment] # Mock classes for when Prometheus is not available @@ -110,23 +119,74 @@ def remove(self, *labelvalues): # One lock per process, keyed by pid: a C-level fork skips the at-fork hooks below, so a # child must never take a lock it inherited, possibly held by a thread that is gone. _metrics_locks: dict[int, threading.Lock] = {} +# Guards the per-thread cells' registries (see _Shards and _ThreadState), keyed by pid for the same reason. cachekit +# takes no other lock under it except the value locks of a series it is first installing views into; a finalizer that +# runs under it can take more (a new bind's labels() takes prometheus_client's metric lock). Taken after the metric-cache +# lock, never before it, so the at-fork hooks hold both in that order. +# Reentrant: Python code runs under it, so a garbage collection there can run a finalizer on the same thread, and a +# finalizer that records a cache operation, reads a series or drops a collector takes it again. Every holder iterates +# snapshots, so such a nested call cannot change a dict under an outer loop, and a nested call that would move counts +# (an exited thread's retirement, a rebind, a dropped collector) is deferred: see _locked(). +_shard_locks: dict[int, threading.RLock] = {} +_RLOCK = threading.RLock # bound here: some tests swap this module's threading for a namespace without it -def _metrics_cache_lock() -> threading.Lock: - """Return this process's metric-cache lock, creating it on first use.""" +_L = TypeVar("_L") + + +def _pid_lock(locks: dict[int, _L], factory: Callable[[], _L]) -> _L: + """Return this process's lock in ``locks``, creating it with ``factory`` on first use.""" pid = os.getpid() - lock = _metrics_locks.get(pid) + lock = locks.get(pid) if lock is None: # setdefault is atomic, so threads racing here in a new child all get one lock. - lock = _metrics_locks.setdefault(pid, threading.Lock()) + lock = locks.setdefault(pid, factory()) return lock -def _acquire_metrics_lock() -> None: +def _metrics_cache_lock() -> threading.Lock: + """Return this process's metric-cache lock, creating it on first use.""" + return _pid_lock(_metrics_locks, threading.Lock) + + +def _shard_lock() -> threading.RLock: + """Return this process's per-thread-cell lock, an ``RLock``, creating it on first use.""" + return _pid_lock(_shard_locks, _RLOCK) + + +# Count moves requested while their thread already held _shard_lock(): by a finalizer that a garbage collection ran +# inside another section, such as a cyclic collection freeing an exited thread's state mid-scrape. Moving counts there +# would let the outer section read them twice or not at all. Until the next section that is not nested applies them, +# the counts stay where they were, and every reader still counts them once. +_deferred: list[Callable[[], None]] = [] + + +def _defer_if_nested(move: Callable[[], None]) -> bool: + """Queue ``move`` and return True if this thread already holds ``_shard_lock()``; otherwise return False.""" + if _shard_lock()._is_owned(): # type: ignore[attr-defined] # RLock's ownership test, as Condition uses it + _deferred.append(move) + return True + return False + + +@contextmanager +def _locked() -> Iterator[None]: + """Hold ``_shard_lock()``; an outermost holder first applies the count moves deferred until it.""" + lock = _shard_lock() + nested = lock._is_owned() # type: ignore[attr-defined] + with lock: + while not nested and _deferred: + _deferred.pop(0)() + yield + + +def _acquire_fork_locks() -> None: _metrics_cache_lock().acquire() + _shard_lock().acquire() -def _release_metrics_lock() -> None: +def _release_fork_locks() -> None: + _shard_lock().release() _metrics_cache_lock().release() @@ -153,7 +213,9 @@ def _reset_metric_locks() -> None: ``tests/unit/test_async_metrics_mode_switch.py`` fail if a release renames them. ``_CacheOpSeries`` also reads two private attributes, a labelled metric's ``_metrics`` and a child's ``_labelvalues``, to tell whether a bound child is still exported; ``test_prometheus_private_attributes_the_series_cache_reads`` in - ``tests/unit/test_hot_path_telemetry.py`` fails if a release renames either. Only cachekit's own metrics + ``tests/unit/test_hot_path_telemetry.py`` fails if a release renames either. The per-thread path also replaces a + child's ``_value``, ``_sum`` and ``_buckets`` with ``_ShardView``s, reads a histogram child's ``_upper_bounds``, and + reads ``values.ValueClass`` and ``values.MutexValue`` to tell multiprocess mode; the same test pins those. Only cachekit's own metrics and the default registry are covered, not an application's metrics. In multiprocess mode (``PROMETHEUS_MULTIPROC_DIR``) values share one lock that this cannot reach, so a child forked while a parent thread holds it can still hang. @@ -200,16 +262,17 @@ def _after_fork_in_child() -> None: """Repair the child of a fork made from Python, which runs this while it is single-threaded.""" global _hooked_fork_pid _metrics_locks.clear() # only frees the parent's entry: keyed by its PID, it is never looked up here + _shard_locks.clear() _hooked_fork_pid = os.getpid() _reset_metric_locks() if hasattr(os, "register_at_fork"): - # Hold the lock across fork (as the logging module does) so no thread is between - # registering a metric and caching it. + # Hold the locks across fork (as the logging module does) so no thread is between registering a metric and + # caching it, or between moving an exited thread's counts into a series and dropping its cells. os.register_at_fork( - before=_acquire_metrics_lock, - after_in_parent=_release_metrics_lock, + before=_acquire_fork_locks, + after_in_parent=_release_fork_locks, after_in_child=_after_fork_in_child, ) @@ -221,7 +284,7 @@ def _get_shared_metric( Module-level so ``circuit_breaker_gauge()`` shares the collectors' metric cache without creating a collector. ``buckets`` reach a ``Histogram`` only on first creation. It is a named - parameter, not ``**kwargs``: every recorded cache operation calls this up to three times, and + parameter, not ``**kwargs``: every series bind (``_bind``) calls it up to three times, and forwarding ``**kwargs`` cost about 2,500 instructions per call. Raises: @@ -266,46 +329,29 @@ def circuit_breaker_gauge() -> Any: class _CacheOpSeries: """The three cache-operation series of one (operation, namespace, success, serializer) tuple, bound once. - ``Metric.labels()`` sorts its arguments, validates them and takes a lock on every call: three of them cost - about 5 us per recorded operation. Each series binds on its first update, so a tuple that never counts, or - never observes a duration or size, gets no empty series, exactly as per-call ``labels()`` left it. A series - detached by the parent's ``clear()`` or ``remove()`` binds again on its next update, as per-call ``labels()`` - would: each update checks that the child is still the one its parent exports, inline because a helper call - costs about 1% per operation. The parent metrics are kept so ``_cache_op_series`` can tell when - ``_metrics_cache`` holds a different metric under a name (tests replace them), and rebind instead of recording - into an orphan. + ``Metric.labels()`` sorts its arguments, validates them and takes a lock on every call. Each series binds on + its first update, so a tuple that never counts, or never observes a duration or size, gets no empty series, + exactly as per-call ``labels()`` left it. A series detached by the parent's ``clear()`` or ``remove()`` binds + again on its next update, as per-call ``labels()`` would. ``metrics`` is kept so ``_cache_op_series`` can tell + when ``_metrics_cache`` holds a different metric under a name (tests replace them), and rebind instead of + recording into an orphan. Index 0 is the counter, 1 the duration histogram and 2 the size histogram. """ - __slots__ = ("counter_metric", "duration_metric", "size_metric", "_success", "_labels", "_counter", "_duration", "_size") + __slots__ = ("metrics", "_labels", "_children") - def __init__(self, counter_metric: Any, duration_metric: Any, size_metric: Any, key: _CacheOpKey): + def __init__(self, metrics: tuple[Any, Any, Any], key: _CacheOpKey): operation, namespace, success, serializer = key - self.counter_metric = counter_metric - self.duration_metric = duration_metric - self.size_metric = size_metric - self._success = str(success) - self._labels = {"operation": operation, "namespace": namespace, "serializer": serializer} - self._counter: Any = None - self._duration: Any = None - self._size: Any = None - - def inc(self, amount: int = 1) -> None: - child, parent = self._counter, self.counter_metric - if child is None or (child is not parent and parent._metrics.get(child._labelvalues) is not child): - child = self._counter = self.counter_metric.labels(success=self._success, **self._labels) - child.inc(amount) - - def observe_duration(self, duration_ms: float) -> None: - child, parent = self._duration, self.duration_metric - if child is None or (child is not parent and parent._metrics.get(child._labelvalues) is not child): - child = self._duration = self.duration_metric.labels(**self._labels) - child.observe(duration_ms) - - def observe_size(self, size_bytes: float) -> None: - child, parent = self._size, self.size_metric + self.metrics = metrics + histogram_labels = {"operation": operation, "namespace": namespace, "serializer": serializer} + self._labels = ({**histogram_labels, "success": str(success)}, histogram_labels, histogram_labels) + self._children: list[Any] = [None, None, None] + + def child(self, i: int) -> Any: + """Return series ``i``'s child, binding it if it is unbound or its parent no longer exports it.""" + child, parent = self._children[i], self.metrics[i] if child is None or (child is not parent and parent._metrics.get(child._labelvalues) is not child): - child = self._size = self.size_metric.labels(**self._labels) - child.observe(size_bytes) + child = self._children[i] = parent.labels(**self._labels[i]) + return child # Process-wide, like the metrics whose series it holds. One entry per label tuple, the same cardinality @@ -320,32 +366,270 @@ def _cache_op_series(operation: str, namespace: str, success: bool, serializer: series = _cache_op_series_cache.get(key) if ( series is None - or series.counter_metric is not _metrics_cache.get(_COUNTER) - or series.duration_metric is not _metrics_cache.get(_DURATION) - or series.size_metric is not _metrics_cache.get(_SIZE) + or series.metrics[0] is not _metrics_cache.get(_COUNTER) + or series.metrics[1] is not _metrics_cache.get(_DURATION) + or series.metrics[2] is not _metrics_cache.get(_SIZE) ): series = _cache_op_series_cache[key] = _CacheOpSeries( - _get_shared_metric(_COUNTER, Counter, "Total cache operations", ["operation", "namespace", "success", "serializer"]), - _get_shared_metric( - _DURATION, Histogram, "Cache operation duration", ["operation", "namespace", "serializer"], DURATION_BUCKETS_MS - ), - _get_shared_metric( - _SIZE, Histogram, "Cache operation size", ["operation", "namespace", "serializer"], SIZE_BUCKETS_BYTES + ( + _get_shared_metric( + _COUNTER, Counter, "Total cache operations", ["operation", "namespace", "success", "serializer"] + ), + _get_shared_metric( + _DURATION, + Histogram, + "Cache operation duration", + ["operation", "namespace", "serializer"], + DURATION_BUCKETS_MS, + ), + _get_shared_metric( + _SIZE, Histogram, "Cache operation size", ["operation", "namespace", "serializer"], SIZE_BUCKETS_BYTES + ), ), key, ) return series +def _add_exact(partials: list[float], x: float) -> None: + """Add ``x`` to the exact sum that ``partials`` holds as non-overlapping floats (Shewchuk's algorithm). + + ``math.fsum`` of the partials and any other floats is then the correctly rounded total, however the values were + grouped. A plain float total would round differently once an exited thread's cell joined it, and could read one + ulp lower than the scrape before: a decrease Prometheus takes for a counter reset. + + The algorithm holds only for finite values: with ``inf`` it stores ``nan``. Once a term is not finite the total is + not either, exactness no longer matters, and ``math.fsum`` collapses the terms (``inf + -inf`` gives ``nan``). + """ + if not math.isfinite(x) or not all(map(math.isfinite, partials)): + try: + partials[:] = [math.fsum([*partials, x])] + except ValueError: # fsum raises on inf + -inf + partials[:] = [math.nan] + return + i = 0 + for y in partials: + if abs(x) < abs(y): + x, y = y, x + hi = x + y + lo = y - (hi - x) + if lo: + partials[i] = lo + i += 1 + x = hi + partials[i:] = [x] + + +class _Shards: + """The per-thread cells behind one Prometheus child, and the exact total of every cell already retired. + + A counter child has one value (its count); a histogram child has its sum, then one count per bucket. Each + thread that records into the child owns one cell, a list of floats only that thread writes. Readers sum the + cells under ``_shard_lock()`` and see each value monotonic: a cell only grows, and retiring one moves it into + ``retired`` and drops it in one step under that lock. + """ + + __slots__ = ("retired", "cells") + + def __init__(self, values: list[float]): + self.retired = [[v] for v in values] + self.cells: dict[int, list[float]] = {} # id(cell) -> cell; the dict keeps each cell, so its id is unique + + def add_cell(self) -> list[float]: + """Return a new zeroed cell that the calling thread owns. Caller holds ``_shard_lock()``.""" + cell = [0.0] * len(self.retired) + self.cells[id(cell)] = cell + return cell + + def retire(self, cell: list[float]) -> None: + """Move ``cell``'s counts into ``retired``. Caller holds ``_shard_lock()``; no thread may write ``cell`` again.""" + if self.cells.pop(id(cell), None) is not None: + for partials, value in zip(self.retired, cell, strict=True): + _add_exact(partials, value) + + def total(self, i: int) -> float: + """Return value ``i`` summed over every cell, live and retired. Caller holds ``_shard_lock()``.""" + cells = list(self.cells.values()) # a snapshot: see _shard_locks + return math.fsum([*self.retired[i], *(cell[i] for cell in cells)]) + + +class _ShardView: + """Stands in for one prometheus_client value of a cache-operation child, reading its total from ``_Shards``. + + The child's own ``inc()``, ``observe()``, ``set()`` and ``reset()`` still work: they change the retired total. + Only the per-thread record path writes the cells. + """ + + __slots__ = ("shards", "i", "_exemplar") + + def __init__(self, shards: _Shards, i: int, exemplar: Any = None): + self.shards = shards + self.i = i + self._exemplar = exemplar + + def inc(self, amount: float) -> None: + with _locked(): + _add_exact(self.shards.retired[self.i], amount) + + def set(self, value: float, timestamp: Optional[float] = None) -> None: + with _locked(): + partials = [float(value)] + for cell in list(self.shards.cells.values()): + _add_exact(partials, -cell[self.i]) + self.shards.retired[self.i] = partials + + def get(self) -> float: + with _locked(): + return self.shards.total(self.i) + + def set_exemplar(self, exemplar: Any) -> None: + self._exemplar = exemplar + + def get_exemplar(self) -> Any: + return self._exemplar + + +def _shards_of(child: Any, i: int) -> _Shards: + """Return the ``_Shards`` behind ``child``, installing views over its values first. Caller holds ``_shard_lock()``. + + Index 0 is a counter child; 1 and 2 are histogram children. The values a child already holds become the + retired total, so nothing recorded into it before is lost. + """ + if i == 0: + value = child._value + if isinstance(value, _ShardView): + return value.shards + shards = _Shards([value.get()]) + child._value = _ShardView(shards, 0, value.get_exemplar()) + return shards + if isinstance(child._sum, _ShardView): + return child._sum.shards + buckets = child._buckets + shards = _Shards([child._sum.get(), *(b.get() for b in buckets)]) + child._sum = _ShardView(shards, 0) + child._buckets = [_ShardView(shards, j + 1, b.get_exemplar()) for j, b in enumerate(buckets)] + return shards + + +class _Slot: + """One thread's cell in one bound series, with what the record path checks to tell the series is still live.""" + + __slots__ = ("metric", "child", "cell", "bounds", "shards") + + def __init__(self, metric: Any, child: Any, cell: list[float], bounds: Any, shards: Optional[_Shards]): + self.metric = metric + self.child = child + self.cell = cell + self.bounds = bounds # a histogram's bucket upper bounds, +Inf last; None for the counter + self.shards = shards # None for a metric the host owns, whose records go nowhere + + +class _ThreadState: + """One thread's cache-operation cells and per-collector operation counts. + + Usually only ``_tls`` refers to it, so it is freed when its thread exits, and ``__del__`` then folds its counts + into the totals the thread leaves behind: a collector's ``_retired_ops`` entry and each series' retired total. + That keeps an exited thread's counts and bounds the cells to the live threads. A traceback in a reference cycle + can keep it alive past the exit; a cyclic collection then frees it on another thread, possibly inside a section + of that thread's, and the move is deferred (``_defer_if_nested``). + """ + + __slots__ = ("slots", "ops") + + def __init__(self) -> None: + self.slots: dict[_CacheOpKey, list[Optional[_Slot]]] = {} + self.ops: dict[object, int] = {} # collector token -> operations this thread recorded + with _locked(): + _thread_ops[id(self.ops)] = self.ops # keyed by the dict, which a deferred retirement keeps alive + + def new_slots(self, key: _CacheOpKey) -> list[Optional[_Slot]]: + """Return this thread's empty slots for ``key``: counter, duration, size. ``_bind`` fills each on first use.""" + slots: list[Optional[_Slot]] = [None, None, None] + self.slots[key] = slots + return slots + + def __del__(self) -> None: + move = partial(_retire_thread, self.ops, list(self.slots.values())) + if not _defer_if_nested(move): + with _locked(): + move() + + +def _retire_thread(ops: dict[object, int], slots: list[list[Optional[_Slot]]]) -> None: + """Fold an exited thread's counts into the retired totals. Caller holds ``_shard_lock()`` and is not nested. + + One step under the lock, so a reader sums each count once: before its move, or after it. + """ + _thread_ops.pop(id(ops), None) + for token, count in list(ops.items()): + _retired_ops[token] = _retired_ops.get(token, 0) + count + for series in slots: + for slot in series: + if slot is not None and slot.shards is not None: + slot.shards.retire(slot.cell) + + +_tls = threading.local() +# Every live thread's operation counts, keyed by id() of its _ThreadState's ops dict, and the counts of threads that +# exited, per collector token. Both guarded by _shard_lock(). A forked child keeps its parent's threads' counts here. +_thread_ops: dict[int, dict[object, int]] = {} +_retired_ops: dict[object, int] = {} + + +def _new_thread_state() -> _ThreadState: + state = _tls.state = _ThreadState() + return state + + +def _forget_collector(token: object) -> None: + """Drop a collector's counts once the collector is gone, so building and dropping collectors leaks nothing.""" + move = partial(_forget_token, token) + if not _defer_if_nested(move): + with _locked(): + move() + + +def _forget_token(token: object) -> None: + """Drop ``token`` from every count. Caller holds ``_shard_lock()`` and is not nested.""" + _retired_ops.pop(token, None) + for ops in list(_thread_ops.values()): + ops.pop(token, None) # the collector is gone, so no thread can be adding to this entry + + +def _bind(slots: list[Optional[_Slot]], i: int, key: _CacheOpKey) -> _Slot: + """Give this thread a cell in series ``i`` of ``key``, retiring any cell it held in a series no longer live.""" + # Off the hot path, so a child of a fork made from C replaces any lock a parent thread held before labels() and + # the values below take it. + _reset_metric_locks_once() + series = _cache_op_series(*key) + metric, child = series.metrics[i], series.child(i) + old = slots[i] + noop = isinstance(metric, _NoopMetric) + bounds = None if i == 0 else (math.inf,) if noop else child._upper_bounds + if old is not None and old.shards is not None: + # This thread stops writing the old cell now; its counts move once no section of this thread is open. + retire = partial(old.shards.retire, old.cell) + if not _defer_if_nested(retire): + with _locked(): + retire() + with _locked(): + shards = None if noop else _shards_of(child, i) + # A host-owned name records nowhere: a scratch cell wide enough for a count, or a sum and the one +Inf bucket. + cell = [0.0, 0.0] if shards is None else shards.add_cell() + slot = slots[i] = _Slot(metric, child, cell, bounds, shards) + return slot + + class AsyncMetricsCollector: """High-performance metrics collector with sync/async modes. Features: - - Sync mode: Direct Prometheus calls for low-frequency operations - - Async mode: Non-blocking batched updates for high-frequency scenarios + - Cache operations: recorded into per-thread cells in either mode, so the hot path takes no shared lock and + a scrape reads every operation recorded before it + - Sync mode: Direct Prometheus calls for low-frequency generic metrics + - Async mode: Non-blocking batched updates of generic metrics for high-frequency scenarios - Automatic mode detection based on usage patterns - - Zero-copy thread communication with memory pools - - Graceful overflow handling (drops metrics under extreme load) + - Graceful overflow handling (drops batched metrics under extreme load) Examples: Create collector in sync mode: @@ -391,7 +675,8 @@ def __init__( batch_size: Number of metrics to batch before flushing (async mode only) flush_interval: Maximum time between flushes in seconds (async mode only) max_queue_size: Maximum queue size before dropping metrics (async mode only) - sync_mode: Force sync mode (True) or async mode (False). None for auto-detect. Whatever the mode, a + sync_mode: Force sync mode (True) or async mode (False). None for auto-detect. The mode decides how + generic counters and histograms record; cache operations record per thread in both. Whatever the mode, a collector records synchronously after ``shutdown()``. So does one inherited by a forked child, until a mode check starts a worker of the child's own, which never happens with auto-detect off. In the child of a fork made from C, every collector, inherited or built there, records @@ -405,8 +690,11 @@ def __init__( self.max_queue_size = max_queue_size self.auto_detect_mode = auto_detect_mode - # Performance tracking for mode detection - self._operation_count = 0 + # Performance tracking for mode detection. Each thread counts this collector's operations under this token in + # its own _ThreadState; _operation_count sums them. The offset is what assigning _operation_count adds. + self._token = object() + self._ops_offset = 0 + weakref.finalize(self, _forget_collector, self._token).atexit = False self._start_time = time.time() self._last_mode_check = time.time() @@ -456,18 +744,86 @@ def record_cache_operation( ): """Record cache operation metric. - Automatically uses sync or async mode based on configuration. + In either mode, the operation lands in this thread's cells, which its series sum when read: the call writes + no state another thread writes, and a scrape right after it counts it. In prometheus_client's multiprocess + mode it records into the series directly instead (``_PER_THREAD``). Each series check below is inline, + because a helper call costs about 1% per operation; it rebinds a series whose metric was replaced or whose + child was detached (see ``_CacheOpSeries``). """ - self._operation_count += 1 + try: + state = _tls.state + except AttributeError: + state = _new_thread_state() + ops, token = state.ops, self._token + ops[token] = ops.get(token, 0) + 1 # Auto-detect mode switch if enabled if self.auto_detect_mode and self._should_check_mode(): self._maybe_switch_mode() - if self._sync_mode or not self._may_enqueue(): - self._record_cache_operation_sync(operation, namespace, success, duration_ms, serializer, size_bytes) - else: - self._record_cache_operation_async(operation, namespace, success, duration_ms, serializer, size_bytes) + if not _PER_THREAD: + self._record_cache_operation_direct(operation, namespace, success, duration_ms, serializer, size_bytes) + return + + key = (operation, namespace, success, serializer) + slots = state.slots.get(key) + if slots is None: + slots = state.new_slots(key) + s = slots[0] + if ( + s is None + or s.metric is not _metrics_cache.get(_COUNTER) + or (s.child is not s.metric and s.metric._metrics.get(s.child._labelvalues) is not s.child) + ): + s = _bind(slots, 0, key) + s.cell[0] += 1 + if duration_ms > 0: + s = slots[1] + if ( + s is None + or s.metric is not _metrics_cache.get(_DURATION) + or (s.child is not s.metric and s.metric._metrics.get(s.child._labelvalues) is not s.child) + ): + s = _bind(slots, 1, key) + cell = s.cell + cell[0] += duration_ms + cell[bisect_left(s.bounds, duration_ms) + 1] += 1 # the first bucket whose bound is >= the value + if size_bytes > 0: + s = slots[2] + if ( + s is None + or s.metric is not _metrics_cache.get(_SIZE) + or (s.child is not s.metric and s.metric._metrics.get(s.child._labelvalues) is not s.child) + ): + s = _bind(slots, 2, key) + cell = s.cell + cell[0] += size_bytes + cell[bisect_left(s.bounds, size_bytes) + 1] += 1 + + def _count_operation(self) -> None: + """Count one operation of this collector in this thread's state.""" + try: + state = _tls.state + except AttributeError: + state = _new_thread_state() + ops, token = state.ops, self._token + ops[token] = ops.get(token, 0) + 1 + + def _operation_total(self) -> int: + """Return this collector's operations summed over every thread. Caller holds ``_shard_lock()``.""" + token = self._token + return self._ops_offset + _retired_ops.get(token, 0) + sum(ops.get(token, 0) for ops in list(_thread_ops.values())) + + @property + def _operation_count(self) -> int: + """Operations recorded so far, summed over every thread, live or exited. Never decreases unless assigned.""" + with _locked(): + return self._operation_total() + + @_operation_count.setter + def _operation_count(self, value: int) -> None: + with _locked(): + self._ops_offset += value - self._operation_total() def record_circuit_breaker_state(self, namespace: str, state: str, transitions: int = 0): """Deprecated: records nothing. @@ -500,7 +856,7 @@ def record_counter(self, metric_name: str, labels: Optional[dict[str, Any]] = No label names differ from those the metric was first recorded with. OverflowError: If ``value`` is too large for a float. """ - self._operation_count += 1 + self._count_operation() if self.auto_detect_mode and self._should_check_mode(): self._maybe_switch_mode() @@ -528,7 +884,7 @@ def record_histogram(self, metric_name: str, value: float, labels: Optional[dict label names differ from those the metric was first recorded with. OverflowError: If ``value`` is too large for a float. """ - self._operation_count += 1 + self._count_operation() if self.auto_detect_mode and self._should_check_mode(): self._maybe_switch_mode() @@ -597,24 +953,12 @@ def _flush_batch(self, batch: list[dict[str, Any]]): return # Group metrics by type for efficient processing - cache_ops: dict[_CacheOpKey, int] = defaultdict(int) - # Every record's own duration and size, in record order, so the histograms get what sync mode observes. - cache_observations: list[_CacheObservation] = [] counters = defaultdict(lambda: defaultdict(float)) # {name: {labels_key: value}} histograms = defaultdict(list) # {name: [(value, labels_key)]} for metric in batch: try: - if metric["type"] == "cache_operation": - operation, namespace, serializer = metric["operation"], metric["namespace"], metric["serializer"] - # Compared here, before anything is counted: a value that cannot be compared rejects only this - # record, where in the update below it would end the batch's remaining observations. - duration_ms = metric["duration_ms"] if metric["duration_ms"] > 0 else None - size_bytes = metric["size_bytes"] if metric["size_bytes"] > 0 else None - cache_ops[(operation, namespace, metric["success"], serializer)] += 1 - cache_observations.append((operation, namespace, metric["success"], serializer, duration_ms, size_bytes)) - - elif metric["type"] == "counter": + if metric["type"] == "counter": value = self._check_generic_metric(metric["name"], metric["labels"], metric["value"]) name = metric["name"] labels_key = tuple(sorted(metric["labels"].items())) @@ -637,7 +981,7 @@ def _flush_batch(self, batch: list[dict[str, Any]]): # unforeseen: most worker call sites (the shutdown drain among them) have no handler above them, so an # escaping exception would end the worker and strand every record still queued. try: - self._update_prometheus_metrics(cache_ops, cache_observations, counters, histograms) # type: ignore[arg-type] + self._update_prometheus_metrics(counters, histograms) # type: ignore[arg-type] except Exception as e: logger.error(f"Failed to update metrics batch: {redact_error_for_log(e)}") @@ -673,26 +1017,8 @@ def _check_generic_metric(name: Any, labels: dict[Any, Any], value: Any) -> floa raise TypeError("metric value must be a number") return float(value) - def _update_prometheus_metrics( - self, - cache_ops: dict[_CacheOpKey, int], - cache_observations: list[_CacheObservation], - counters: dict[str, dict[str, Any]], - histograms: dict[str, list[Any]], - ): + def _update_prometheus_metrics(self, counters: dict[str, dict[str, Any]], histograms: dict[str, list[Any]]): """Update Prometheus metrics in batch.""" - # Batch update cache metrics - for (operation, namespace, success, serializer), count in cache_ops.items(): - _cache_op_series(operation, namespace, success, serializer).inc(count) - - # One observation per record, skipping zero values as the sync path does. - for operation, namespace, success, serializer, duration_ms, size_bytes in cache_observations: - series = _cache_op_series(operation, namespace, success, serializer) - if duration_ms is not None: - series.observe_duration(duration_ms) - if size_bytes is not None: - series.observe_size(size_bytes) - # Update generic counters for name, label_values in counters.items(): # Extract label names from first entry @@ -753,11 +1079,12 @@ def get_dropped_metrics_count(self) -> int: def get_stats(self) -> dict[str, Any]: """Get comprehensive performance statistics.""" elapsed = time.time() - self._start_time - ops_per_second = self._operation_count / elapsed if elapsed > 0 else 0 + total = self._operation_count # read once: the sum takes a lock, and both fields must agree + ops_per_second = total / elapsed if elapsed > 0 else 0 return { "mode": "sync" if self._sync_mode else "async", - "total_operations": self._operation_count, + "total_operations": total, "dropped_metrics": self._dropped_metrics, "uptime_seconds": elapsed, "ops_per_second": ops_per_second, @@ -807,12 +1134,7 @@ def _init_async_mode(self) -> bool: def _mode_lock(self) -> threading.Lock: """Return this process's mode-switch lock, creating it on first use.""" - pid = os.getpid() - lock = self._mode_locks.get(pid) - if lock is None: - # setdefault is atomic, so threads racing here in a new child all get one lock. - lock = self._mode_locks.setdefault(pid, threading.Lock()) - return lock + return _pid_lock(self._mode_locks, threading.Lock) def _take_over_if_forked(self) -> None: """In a forked child, replace the batching state inherited from the parent and fall back to sync mode. @@ -870,14 +1192,15 @@ def _should_check_mode(self) -> bool: def _maybe_switch_mode(self): """Switch between sync and async mode based on usage patterns.""" - if self._operation_count < 10: # Need minimum sample size + count = self._operation_count # summed over every thread + if count < 10: # Need minimum sample size return elapsed = time.time() - self._start_time if elapsed < 1.0: # Need at least 1 second of data return - ops_per_second = self._operation_count / elapsed + ops_per_second = count / elapsed # Both switches touch the batching state: one reuses the queue, the other sets the stop event. self._take_over_if_forked() @@ -916,8 +1239,8 @@ def _return_to_pool(self, metric_data: dict[str, Any]): if len(self._metric_pool) < 100: # Limit pool size self._metric_pool.append(metric_data) - # Sync mode implementations (direct Prometheus calls) - def _record_cache_operation_sync( + # Direct recording: cache operations in multiprocess mode, and the generic sync-mode metrics below + def _record_cache_operation_direct( self, operation: str, namespace: str, @@ -926,16 +1249,16 @@ def _record_cache_operation_sync( serializer: str, size_bytes: int, ): - """Record cache operation directly to Prometheus (sync mode).""" + """Record cache operation directly to Prometheus: the path for multiprocess mode, which has no per-thread cells.""" if not PROMETHEUS_AVAILABLE: return series = _cache_op_series(operation, namespace, success, serializer) - series.inc() + series.child(0).inc() if duration_ms > 0: - series.observe_duration(duration_ms) + series.child(1).observe(duration_ms) if size_bytes > 0: - series.observe_size(size_bytes) + series.child(2).observe(size_bytes) def _record_counter_sync(self, metric_name: str, labels: dict[str, Any], value: float): """Record counter directly to Prometheus (sync mode).""" @@ -956,38 +1279,6 @@ def _record_histogram_sync(self, metric_name: str, value: float, labels: dict[st self._series(histogram, labels).observe(value) # Async mode implementations (queued processing) - def _record_cache_operation_async( - self, - operation: str, - namespace: str, - success: bool, - duration_ms: float, - serializer: str, - size_bytes: int, - ): - """Record cache operation to queue for async processing.""" - assert self._queue is not None, "Async metrics not initialized" - - metric_data = self._get_pooled_metric_data() - metric_data.update( - { - "type": "cache_operation", - "operation": operation, - "namespace": namespace, - "success": success, - "duration_ms": duration_ms, - "serializer": serializer, - "size_bytes": size_bytes, - "timestamp": time.time(), - } - ) - - try: - self._queue.put_nowait(metric_data) - except queue.Full: - self._dropped_metrics += 1 - self._return_to_pool(metric_data) # Return to pool if failed - def _record_counter_async(self, metric_name: str, labels: dict[str, Any], value: float): """Record counter to queue for async processing.""" assert self._queue is not None, "Async metrics not initialized" diff --git a/tests/performance/ir_baselines.json b/tests/performance/ir_baselines.json index 68f94cda..015660b2 100644 --- a/tests/performance/ir_baselines.json +++ b/tests/performance/ir_baselines.json @@ -2,47 +2,47 @@ "interpreters": { "cpython-3.12-x86_64": { "budgets": { - "byte_storage_store_40b": 2656, - "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 + "byte_storage_store_40b": 2658, + "byte_storage_store_64k": 720667, + "file_set": 75887, + "l1_hit": 11614, + "l2_hit": 171988, + "l2_hit_async_metrics": 171370, + "minimal_l1_hit": 10496, + "miss": 211434, + "object_cache_put_full": 164514, + "secure_l1_hit": 139943, + "serializer_arrow": 1972713, + "serializer_arrow_usgs": 61750733, + "serializer_auto": 65971, + "serializer_auto_usgs": 61806031, + "serializer_default": 58661, + "serializer_default_records": 1422079, + "serializer_encrypted": 100164, + "serializer_orjson": 22085 }, "python": "3.12.12" }, "cpython-3.14-x86_64": { "budgets": { - "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 + "byte_storage_store_40b": 2530, + "byte_storage_store_64k": 720663, + "file_set": 74132, + "l1_hit": 10904, + "l2_hit": 167408, + "l2_hit_async_metrics": 167447, + "minimal_l1_hit": 9788, + "miss": 203521, + "object_cache_put_full": 160511, + "secure_l1_hit": 140926, + "serializer_arrow": 1982226, + "serializer_arrow_usgs": 61321806, + "serializer_auto": 66624, + "serializer_auto_usgs": 61327994, + "serializer_default": 59595, + "serializer_default_records": 1422120, + "serializer_encrypted": 101927, + "serializer_orjson": 22677 }, "python": "3.14.3" } diff --git a/tests/performance/ir_workload.py b/tests/performance/ir_workload.py index 4c63cafc..80e4bd1c 100644 --- a/tests/performance/ir_workload.py +++ b/tests/performance/ir_workload.py @@ -83,11 +83,11 @@ def batched_metrics(decorate: Callable[[], Callable[..., dict]]) -> Callable[... """Decorate with the metrics collector in batched mode, where a busy long-lived process runs. A collector starts synchronous. At a mode check (every 5 s) that sees more than 100 - records/s it switches to batched mode: each call puts its record on a queue and a worker - thread updates Prometheus. Pinned clocks never let 5 s pass, so this variant starts the - collector batched. It then stops the worker, which drains the queue on real time and would - otherwise change the main thread's count with host load. The budget covers the caller's - side, putting the record on the queue; the worker's Prometheus update is not budgeted. + records/s it switches to batched mode, which queues generic metrics for a worker thread. + Cache operations record into per-thread cells in either mode, so this path must cost what + ``l2_hit`` costs. Pinned clocks never let 5 s pass, so this variant starts the collector + batched. It then stops the worker, which wakes on real time and would otherwise change the + main thread's count with host load. """ made: list[Any] = [] @@ -106,19 +106,9 @@ def __init__(self, **kwargs: Any) -> None: collector._worker_thread.join() queued = collector._queue.qsize() fn(42, "user-profile") - if collector._sync_mode or collector._queue.qsize() <= queued: - raise RuntimeError("l2_hit_async_metrics: the call did not queue its metric, so batched mode is not measured") - - def call(*args: Any) -> dict: - result = fn(*args) - # Do what the worker does with a record: take it off the queue and return its dict to - # the pool. Queue and pool then hold one item each, as a keeping-up worker leaves them, - # so every call takes the pool-hit branch a busy process takes and the heap stops - # growing (a growing queue widened the layout spread). Under 1.6k Ir of the figure. - collector._metric_pool.append(collector._queue.queue.popleft()) - return result - - return call + if collector._sync_mode or collector._queue.qsize() != queued: + raise RuntimeError("l2_hit_async_metrics: the collector left batched mode, or a cache operation queued") + return fn l2 = lambda: cache(backend=DictBackend(), l1_enabled=False, ttl=300)(body) # noqa: E731 # Every metrics path records through the same AsyncMetricsCollector.record_cache_operation, once diff --git a/tests/unit/test_async_metrics_histogram_buckets.py b/tests/unit/test_async_metrics_histogram_buckets.py index a3e845c8..e35b0715 100644 --- a/tests/unit/test_async_metrics_histogram_buckets.py +++ b/tests/unit/test_async_metrics_histogram_buckets.py @@ -152,11 +152,14 @@ def test_default_max_value_size_has_a_finite_size_bucket() -> None: assert SIZE_BUCKETS_BYTES[-1] >= CachekitConfig.model_fields["max_value_size"].default -def test_malformed_batched_record_is_skipped_alone() -> None: - # One record the flush cannot compare must not cost the batch's other observations. +@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "batched"]) +def test_malformed_record_raises_and_costs_no_other_observation(sync_mode: bool) -> None: + # Cache operations record on the caller's thread in both modes, so a duration that cannot be compared reaches the + # caller, after the count and before any observation, as sync mode always did. Later records are unaffected. namespace = f"buckets-malformed-{uuid.uuid4().hex}" - collector = AsyncMetricsCollector(sync_mode=False, auto_detect_mode=False) - collector.record_cache_operation("get", namespace, True, "invalid", serializer="rust", size_bytes=4) # type: ignore[arg-type] + collector = AsyncMetricsCollector(sync_mode=sync_mode, auto_detect_mode=False) + with pytest.raises(TypeError): + collector.record_cache_operation("get", namespace, True, "invalid", serializer="rust", size_bytes=4) # type: ignore[arg-type] collector.record_cache_operation("get", namespace, True, 55.0, serializer="rust", size_bytes=300) collector.shutdown() @@ -165,4 +168,4 @@ def test_malformed_batched_record_is_skipped_alone() -> None: assert _samples(DURATION, namespace)[("cache_operation_duration_ms_sum", labels)] == 55.0 assert _samples(SIZE, namespace)[("cache_operation_size_bytes_count", labels)] == 1 counter = _metrics_cache["cache_operations_total"] - assert counter.labels(operation="get", namespace=namespace, success="True", serializer="rust")._value.get() == 1 + assert counter.labels(operation="get", namespace=namespace, success="True", serializer="rust")._value.get() == 2 diff --git a/tests/unit/test_async_metrics_mode_switch.py b/tests/unit/test_async_metrics_mode_switch.py index e8834c73..3d119601 100644 --- a/tests/unit/test_async_metrics_mode_switch.py +++ b/tests/unit/test_async_metrics_mode_switch.py @@ -112,13 +112,12 @@ def test_no_new_worker_starts_while_the_previous_one_is_alive(monkeypatch): assert collector._sync_mode assert collector._worker_thread is old_worker - before = _flushed(namespace) - q.put_nowait({"type": "cache_operation", "operation": "get", "namespace": namespace, "success": True, - "duration_ms": 1.0, "serializer": "unknown", "size_bytes": 0}) # fmt: skip + counter = _metric_name(namespace) + q.put_nowait({"type": "counter", "name": counter, "labels": {}, "value": 1.0}) old_worker.join(timeout=5.0) assert not old_worker.is_alive() # The old worker's exit drain flushed the record it was woken with. - assert _flushed(namespace) - before == 1 + assert _sample(f"{counter}_total") == 1 _steer(collector, 500) _record(collector, namespace) @@ -347,24 +346,28 @@ def child(): collector.shutdown() -def _prometheus_lock(held: str, namespace: str) -> Any: - """Return the prometheus_client lock named by ``held``, for the series a cache operation in ``namespace`` updates.""" +def _prometheus_lock(held: str) -> Any: + """Return the lock named by ``held``, one that the cache-operation series or their registration take.""" from prometheus_client import REGISTRY if held == "registry": return REGISTRY._lock - counter = async_metrics._metrics_cache["cache_operations_total"] - if held == "counter": - return counter._lock # taken by labels() - if held == "counter-series": - return counter.labels(operation="get", namespace=namespace, success="True", serializer="unknown")._value._lock - duration = async_metrics._metrics_cache["cache_operation_duration_ms"] - series = duration.labels(operation="get", namespace=namespace, serializer="unknown") - return series._sum._lock if held == "histogram-sum" else series._buckets[0]._lock - - -@pytest.mark.parametrize("fork_holding", FORKS) -@pytest.mark.parametrize("held", ["counter", "counter-series", "histogram-sum", "histogram-bucket", "registry"]) + if held == "series": + # Every cache-operation series value reads and retires per-thread cells under this one lock (_ShardView). + return async_metrics._shard_lock() + return async_metrics._metrics_cache["cache_operations_total"]._lock # taken by labels() + + +# The at-fork hook takes the series lock itself, so only a fork made from C can inherit it held. +PROMETHEUS_LOCKS = [ + pytest.param(held, fork.values[0], id=f"{held}-{fork.id}", marks=fork.marks) + for held in ("counter", "series", "registry") + for fork in FORKS + if not (held == "series" and fork.id == "fork") +] + + +@pytest.mark.parametrize(("held", "fork_holding"), PROMETHEUS_LOCKS) def test_child_records_past_a_prometheus_lock_held_at_fork(held, fork_holding): namespace = f"fork-prometheus-{held}-{uuid.uuid4().hex}" # Create the series first, as a parent that has recorded before has; its worker updates them under these locks. @@ -381,12 +384,12 @@ def child(): return _flushed(namespace), _sample(f"{_metric_name(namespace)}_total") # prometheus_client resets none of its locks after a fork, so the child must not use the parent's copies. - assert fork_holding(_prometheus_lock(held, namespace), child) == (2, 1) + assert fork_holding(_prometheus_lock(held), child) == (2, 1) collector.shutdown() @needs_fork -@pytest.mark.parametrize("held", ["counter", "counter-series", "histogram-sum", "histogram-bucket", "registry"]) +@pytest.mark.parametrize("held", ["counter", "registry"]) # the series lock: see PROMETHEUS_LOCKS def test_child_of_a_sync_parent_records_past_a_prometheus_lock_held_at_fork(held): namespace = f"fork-prometheus-sync-{held}-{uuid.uuid4().hex}" collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) @@ -398,7 +401,7 @@ def child(): collector.record_counter(_metric_name(namespace)) return _flushed(namespace), _sample(f"{_metric_name(namespace)}_total") - assert _in_child_forked_holding(_prometheus_lock(held, namespace), child) == (2, 1) + assert _in_child_forked_holding(_prometheus_lock(held), child) == (2, 1) @needs_fork @@ -502,7 +505,7 @@ def child(): return len(_UnstartedThread.started), modes, _flushed(namespace) # A parent thread holds the series lock, so a record into it hangs unless the child's locks are fresh. - held = _prometheus_lock("counter-series", namespace) + held = _prometheus_lock("series") assert _in_c_forked_child_holding(held, child) == (0, [(True, True), (True, True)], 3) @@ -549,7 +552,8 @@ def _batched_collector_as_a_child_inherits_it() -> AsyncMetricsCollector: return collector -@pytest.mark.parametrize("entry", ENTRY_POINTS) +# A cache operation never queues, so it never takes over the batching state; a mode check does that for it. +@pytest.mark.parametrize("entry", [e for e in ENTRY_POINTS if e != "cache-operation"]) def test_take_over_replaces_the_inherited_batching_state(entry): record, read, after_batched = ENTRY_POINTS[entry] namespace = f"take-over-{entry}-{uuid.uuid4().hex}" diff --git a/tests/unit/test_async_metrics_per_thread.py b/tests/unit/test_async_metrics_per_thread.py new file mode 100644 index 00000000..277954a1 --- /dev/null +++ b/tests/unit/test_async_metrics_per_thread.py @@ -0,0 +1,456 @@ +"""Cache operations record into per-thread cells that their Prometheus series sum when read. + +A free-threaded build runs threads in parallel, so a record that writes shared state (a lock, a queue, a shared +counter) serialises them. Each thread writes only cells of its own instead. What a scrape, ``get_stats()`` and the mode +switch read must still be exact, immediate and monotonic, survive a thread's exit and a fork, and stay bounded under +thread churn. +""" + +from __future__ import annotations + +import gc +import math +import multiprocessing +import os +import queue +import random +import threading +import time +import uuid +from collections.abc import Callable + +import pytest + +from cachekit.reliability import async_metrics +from cachekit.reliability.async_metrics import PROMETHEUS_AVAILABLE, AsyncMetricsCollector + +pytestmark = pytest.mark.skipif( + not (PROMETHEUS_AVAILABLE and async_metrics._PER_THREAD), reason="per-thread cells need in-process prometheus values" +) +needs_fork = pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") + +THREADS = 16 + + +def _record(collector: AsyncMetricsCollector, namespace: str, duration_ms: float = 1.5, size_bytes: int = 100) -> None: + collector.record_cache_operation( + operation="get", namespace=namespace, success=True, duration_ms=duration_ms, serializer="rust", size_bytes=size_bytes + ) + + +def _sample(name: str, namespace: str) -> float: + from prometheus_client import REGISTRY + + labels = {"operation": "get", "namespace": namespace, "serializer": "rust"} + if name == "cache_operations_total": + labels["success"] = "True" + return REGISTRY.get_sample_value(name, labels) or 0.0 + + +def _series_samples(namespace: str) -> tuple[float, ...]: + """Every cumulative value one label tuple exports: count, then each histogram's count and sum.""" + return tuple( + _sample(name, namespace) + for name in ( + "cache_operations_total", + "cache_operation_duration_ms_count", + "cache_operation_duration_ms_sum", + "cache_operation_size_bytes_count", + "cache_operation_size_bytes_sum", + ) + ) + + +def _in_threads(target: Callable[[int], None], threads: int = THREADS) -> None: + """Run ``target(i)`` on ``threads`` threads released together, and wait for all of them to exit.""" + barrier = threading.Barrier(threads) + + def run(i: int) -> None: + barrier.wait() + target(i) + + workers = [threading.Thread(target=run, args=(i,)) for i in range(threads)] + for worker in workers: + worker.start() + for worker in workers: + worker.join() + + +def _cells(namespace: str) -> int: + """Return how many live per-thread cells the label tuple's three series hold.""" + total = 0 + for name, labels in ( + ("cache_operations_total", {"success": "True"}), + ("cache_operation_duration_ms", {}), + ("cache_operation_size_bytes", {}), + ): + child = async_metrics._metrics_cache[name].labels(operation="get", namespace=namespace, serializer="rust", **labels) + value = child._value if name == "cache_operations_total" else child._sum + total += len(value.shards.cells) + return total + + +@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "batched"]) +def test_a_scrape_right_after_calls_on_16_threads_counts_every_one(sync_mode): + namespace = f"per-thread-exact-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=sync_mode, auto_detect_mode=False) + per_thread = 250 + live = threading.Event() + done = threading.Barrier(THREADS + 1) + + def record(_: int) -> None: + for _ in range(per_thread): + _record(collector, namespace) + done.wait() # still alive, so the scrape reads live cells, not retired ones + live.wait() + + threads = threading.Thread(target=_in_threads, args=(record,)) + threads.start() + done.wait() + n = THREADS * per_thread + try: + assert _series_samples(namespace) == (n, n, n * 1.5, n, n * 100.0) + assert collector.get_stats()["total_operations"] == n + finally: + live.set() + threads.join() + collector.shutdown() + assert _series_samples(namespace) == (n, n, n * 1.5, n, n * 100.0) # and once every thread has exited + + +def test_each_value_lands_in_the_bucket_prometheus_would_pick(): + namespace = f"per-thread-buckets-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + durations = [0.01, 0.010001, 0.5, 2.5, 7.0, 1000.0, 1e9, math.inf] + for duration_ms in durations: + _record(collector, namespace, duration_ms=duration_ms, size_bytes=0) + + from prometheus_client import REGISTRY, CollectorRegistry, Histogram + + expected = Histogram("reference", "", buckets=async_metrics.DURATION_BUCKETS_MS, registry=CollectorRegistry()) + for duration_ms in durations: + expected.observe(duration_ms) + want = {s.labels["le"]: s.value for s in expected.collect()[0].samples if s.name.endswith("_bucket")} + labels = {"operation": "get", "namespace": namespace, "serializer": "rust"} + got = {le: REGISTRY.get_sample_value("cache_operation_duration_ms_bucket", {**labels, "le": le}) for le in want} + assert got == want + assert _sample("cache_operation_size_bytes_count", namespace) == 0.0 # a zero size observes nothing + + +def test_get_stats_keeps_its_keys_and_sums_every_thread(): + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + namespace = f"per-thread-stats-{uuid.uuid4().hex}" + _in_threads(lambda i: [_record(collector, namespace) for _ in range(10 + i)]) + collector.record_counter(f"per_thread_stats_{uuid.uuid4().hex}") + + stats = collector.get_stats() + assert set(stats) == { + "mode", + "total_operations", + "dropped_metrics", + "uptime_seconds", + "ops_per_second", + "auto_detect_enabled", + "pool_size", + } + assert stats["total_operations"] == sum(10 + i for i in range(THREADS)) + 1 + + +def test_mode_switch_reads_the_count_summed_over_threads(): + collector = AsyncMetricsCollector(flush_interval=0.05) + namespace = f"per-thread-mode-{uuid.uuid4().hex}" + collector._last_mode_check = time.time() # no check while the threads record + _in_threads(lambda _: [_record(collector, namespace) for _ in range(20)]) # 20 per thread, 320 in all + assert collector._sync_mode + + # Over 2 s, each thread's 20 alone is 10/s, below the 100/s switch; the 320 together are 160/s. + collector._start_time = time.time() - 2.0 + collector._last_mode_check = 0.0 + _record(collector, namespace) + assert not collector._sync_mode + collector.shutdown() + + +def test_a_steady_record_writes_no_shared_state(monkeypatch): + collector = AsyncMetricsCollector(sync_mode=False, auto_detect_mode=False) + namespace = f"per-thread-no-shared-{uuid.uuid4().hex}" + _record(collector, namespace) # binds this thread's cells, the one time it takes a lock + snapshot = dict(vars(collector)) + + def forbidden(*args, **kwargs): + raise AssertionError("a steady record took a shared lock or queued") + + monkeypatch.setattr(async_metrics, "_shard_lock", forbidden) + monkeypatch.setattr(async_metrics, "_metrics_cache_lock", forbidden) + monkeypatch.setattr(async_metrics, "_get_shared_metric", forbidden) + monkeypatch.setattr(type(collector._queue), "put_nowait", forbidden) + monkeypatch.setattr(AsyncMetricsCollector, "_get_pooled_metric_data", forbidden) + for _ in range(100): + _record(collector, namespace) + + monkeypatch.undo() + assert vars(collector) == snapshot + assert _sample("cache_operations_total", namespace) == 101 + collector.shutdown() + + +def test_reads_never_decrease_while_threads_record_and_exit(): + namespace = f"per-thread-monotonic-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + stop = threading.Event() + decreases: list[str] = [] + + def record(i: int) -> None: + # Durations whose float sum rounds, so a retired cell would show any re-grouping. Short-lived threads + # retire their cells while the readers run. + rng = random.Random(i) + while not stop.is_set(): + for _ in range(rng.randint(1, 200)): + _record(collector, namespace, duration_ms=rng.uniform(0.001, 3.0), size_bytes=rng.randint(1, 10**6)) + if rng.random() < 0.3: + return + + def churn(i: int) -> None: + while not stop.is_set(): + worker = threading.Thread(target=record, args=(i,)) + worker.start() + worker.join() + + def read() -> None: + last = (_series_samples(namespace), collector.get_stats()["total_operations"]) + while not stop.is_set(): + now = (_series_samples(namespace), collector.get_stats()["total_operations"]) + if any(b < a for a, b in zip(last[0], now[0], strict=True)) or now[1] < last[1]: + decreases.append(f"{last} -> {now}") + last = now + + threads = [threading.Thread(target=churn, args=(i,)) for i in range(THREADS)] + threads += [threading.Thread(target=read) for _ in range(2)] + for thread in threads: + thread.start() + time.sleep(2.0) + stop.set() + for thread in threads: + thread.join() + assert decreases == [] + assert _sample("cache_operations_total", namespace) == collector._operation_count > 0 + + +def test_an_exited_threads_counts_survive_and_its_cells_go(): + namespace = f"per-thread-exit-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _in_threads(lambda _: [_record(collector, namespace) for _ in range(7)]) + + assert _series_samples(namespace) == (7 * THREADS, 7 * THREADS, 7 * THREADS * 1.5, 7 * THREADS, 7 * THREADS * 100.0) + assert collector._operation_count == 7 * THREADS + assert _cells(namespace) == 0 + + +def test_cells_stay_bounded_across_10000_thread_lifetimes(): + namespace = f"per-thread-churn-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(collector, namespace) # this thread's cells stay live + states = len(async_metrics._thread_ops) + for _ in range(10_000): + thread = threading.Thread(target=_record, args=(collector, namespace)) + thread.start() + thread.join() + + assert len(async_metrics._thread_ops) <= states + assert _cells(namespace) == 3 + assert _sample("cache_operations_total", namespace) == collector._operation_count == 10_001 + + +def test_dropped_collectors_leave_no_counts_behind(): + namespace = f"per-thread-drop-{uuid.uuid4().hex}" + _record(AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False), namespace) # this thread's state exists + gc.collect() + retired, own = len(async_metrics._retired_ops), len(async_metrics._tls.state.ops) + for _ in range(500): + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(collector, namespace) + thread = threading.Thread(target=_record, args=(collector, namespace)) # retires its count on exit + thread.start() + thread.join() + del collector, thread + gc.collect() + + assert len(async_metrics._retired_ops) <= retired + assert len(async_metrics._tls.state.ops) <= own + assert _sample("cache_operations_total", namespace) == 1001 # the series keep every operation + + +def test_a_finalizer_that_runs_under_the_series_lock_does_not_deadlock(): + # A garbage collection can run a finalizer while its thread holds the lock (inside a scrape, a bind or a + # thread's exit). A finalizer that records, reads or drops a collector then takes the lock again on that thread. + namespace = f"per-thread-reenter-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(collector, namespace) + results: list[object] = [] + + def nested() -> None: + with async_metrics._shard_lock(): + _record(collector, f"{namespace}-new") # binds a new label tuple + dropped = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(dropped, namespace) + del dropped + gc.collect() # finalizes the dropped collector under the lock + results.append((collector.get_stats()["total_operations"], _sample("cache_operations_total", namespace))) + + thread = threading.Thread(target=nested, daemon=True) + thread.start() + thread.join(10) + assert not thread.is_alive(), "deadlocked on the series lock" + assert results == [(2, 2)] + + +def test_an_infinite_duration_stays_infinite_after_its_thread_exits(): + namespace = f"per-thread-inf-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _in_threads(lambda i: _record(collector, namespace, duration_ms=math.inf if i == 0 else 1.0), threads=2) + + assert _sample("cache_operation_duration_ms_sum", namespace) == math.inf + assert _sample("cache_operation_duration_ms_count", namespace) == 2 + + +def test_a_retirement_inside_an_open_section_waits_until_the_section_closes(): + # A traceback in a reference cycle can keep an exited thread's state alive until a cyclic collection frees it, + # on whatever thread runs that collection, possibly in the middle of a read there. + namespace = f"per-thread-defer-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + kept: list[object] = [] + + def record_and_keep_state() -> None: + _record(collector, namespace) + kept.append(async_metrics._tls.state) + + thread = threading.Thread(target=record_and_keep_state) + thread.start() + thread.join() + (state,) = kept + ops = state.ops # type: ignore[attr-defined] + del state + + with async_metrics._locked(): + before = (collector._operation_count, _sample("cache_operations_total", namespace)) + kept.clear() # frees the state here, inside the section + assert len(async_metrics._deferred) == 1 and id(ops) in async_metrics._thread_ops # nothing moved yet + assert (collector._operation_count, _sample("cache_operations_total", namespace)) == before == (1, 1) + + assert (collector._operation_count, _sample("cache_operations_total", namespace)) == (1, 1) # applied by that read + assert async_metrics._deferred == [] and id(ops) not in async_metrics._thread_ops + assert _cells(namespace) == 0 + + +def test_retiring_a_cell_never_changes_a_total(): + # A plain float total rounds differently once a cell joins it, and can read one ulp lower than before. + rng = random.Random(7117) + for _ in range(200): + shards = async_metrics._Shards([rng.uniform(0, 1e6)]) + cells = [shards.add_cell() for _ in range(rng.randint(2, 30))] + for cell in cells: + cell[0] = sum(rng.uniform(0.001, 3.0) for _ in range(rng.randint(1, 50))) + before = shards.total(0) + for cell in rng.sample(cells, len(cells)): + shards.retire(cell) + assert shards.total(0) == before + + +def test_counter_reset_and_direct_updates_still_reach_the_series(): + namespace = f"per-thread-direct-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(collector, namespace) + child = async_metrics._metrics_cache["cache_operations_total"].labels( + operation="get", namespace=namespace, success="True", serializer="rust" + ) + child.inc(2) + assert _sample("cache_operations_total", namespace) == 3 + child.reset() + _record(collector, namespace) + assert _sample("cache_operations_total", namespace) == 1 + + +def test_multiprocess_mode_records_into_the_series_directly(monkeypatch): + monkeypatch.setattr(async_metrics, "_PER_THREAD", False) + namespace = f"per-thread-multiproc-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(collector, namespace) + + child = async_metrics._metrics_cache["cache_operations_total"].labels( + operation="get", namespace=namespace, success="True", serializer="rust" + ) + assert not isinstance(child._value, async_metrics._ShardView) + assert _series_samples(namespace) == (1, 1, 1.5, 1, 100.0) + assert collector._operation_count == 1 + + +@needs_fork +def test_a_forked_child_starts_from_its_parents_counts(): + namespace = f"per-thread-fork-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(collector, namespace) + hold = threading.Event() + held = threading.Event() + + def record_and_hold() -> None: # a live parent thread with cells of its own; the child does not inherit it + _record(collector, namespace) + held.set() + hold.wait() + + other = threading.Thread(target=record_and_hold) + other.start() + held.wait() + + ctx = multiprocessing.get_context("fork") + results = ctx.Queue() + + def child() -> None: + first = (_series_samples(namespace), collector._operation_count) + _record(collector, namespace) + results.put((first, _series_samples(namespace), collector._operation_count)) + + try: + process = ctx.Process(target=child, daemon=True) + process.start() + try: + got = results.get(timeout=10) + except queue.Empty: + process.kill() + pytest.fail("the forked child hung") + process.join(timeout=5) + finally: + hold.set() + other.join() + assert got == (((2, 2, 3.0, 2, 200.0), 2), (3, 3, 4.5, 3, 300.0), 3) + assert _series_samples(namespace) == (2, 2, 3.0, 2, 200.0) # the child's record stayed in the child + + +@needs_fork +def test_fork_waits_for_a_thread_mid_read_and_the_child_does_not_hang(): + namespace = f"per-thread-fork-held-{uuid.uuid4().hex}" + collector = AsyncMetricsCollector(sync_mode=True, auto_detect_mode=False) + _record(collector, namespace) + entered, release = threading.Event(), threading.Event() + + def hold_the_series_lock() -> None: + with async_metrics._shard_lock(): + entered.set() + release.wait() + + holder = threading.Thread(target=hold_the_series_lock) + holder.start() + entered.wait() + threading.Timer(0.2, release.set).start() # the at-fork hook waits for the reader to finish + + ctx = multiprocessing.get_context("fork") + results = ctx.Queue() + process = ctx.Process(target=lambda: results.put((_record(collector, namespace), _series_samples(namespace))[1])) + process.start() + try: + assert results.get(timeout=10) == (2, 2, 3.0, 2, 200.0) + except queue.Empty: + process.kill() + pytest.fail("the forked child hung on the series lock") + finally: + process.join(timeout=5) + holder.join() diff --git a/tests/unit/test_hot_path_telemetry.py b/tests/unit/test_hot_path_telemetry.py index 6b8d9491..f4faf25d 100644 --- a/tests/unit/test_hot_path_telemetry.py +++ b/tests/unit/test_hot_path_telemetry.py @@ -218,3 +218,15 @@ def test_prometheus_private_attributes_the_series_cache_reads(kind): child = metric.labels(a="x", b=1) metric.clear() assert metric._metrics.get(child._labelvalues) is None + + # The per-thread path swaps these values for views and bisects the bounds; it records directly unless the value + # class is MutexValue. + from prometheus_client import values + + assert values.ValueClass is values.MutexValue + if kind == "Counter": + assert isinstance(child._value, values.MutexValue) + else: + assert isinstance(child._sum, values.MutexValue) + assert all(isinstance(b, values.MutexValue) for b in child._buckets) + assert len(child._buckets) == len(child._upper_bounds) and child._upper_bounds[-1] == float("inf") diff --git a/tests/unit/test_ir_budget.py b/tests/unit/test_ir_budget.py index e2c4baab..92468bb5 100644 --- a/tests/unit/test_ir_budget.py +++ b/tests/unit/test_ir_budget.py @@ -420,29 +420,30 @@ def test_performance_docs_state_the_committed_budgets() -> None: assert table == {path: tuple(interpreters[key]["budgets"][path] for key in columns) for path in PATHS} -def test_batched_metrics_path_queues_every_call_and_never_records_synchronously(monkeypatch) -> None: - """``l2_hit_async_metrics`` claims to budget batched mode; a collector change must not turn it synchronous.""" +def test_batched_metrics_path_records_every_call_per_thread_and_queues_none(monkeypatch) -> None: + """``l2_hit_async_metrics`` budgets a batched collector, whose cache operations record per thread, not on the queue.""" pytest.importorskip("pandas") # the workload builder makes the Arrow path's DataFrame too + from cachekit.reliability import async_metrics from cachekit.reliability.async_metrics import AsyncMetricsCollector - calls: Counter[str] = Counter() + made: list[AsyncMetricsCollector] = [] + real_init = AsyncMetricsCollector.__init__ - def counting(mode: str): - real = getattr(AsyncMetricsCollector, f"_record_cache_operation_{mode}") + def tracking_init(self, *args, **kwargs): + real_init(self, *args, **kwargs) + made.append(self) - def record(self, *args, **kwargs): - calls[mode] += 1 - return real(self, *args, **kwargs) - - return record - - for mode in ("sync", "async"): - monkeypatch.setattr(AsyncMetricsCollector, f"_record_cache_operation_{mode}", counting(mode)) + direct: Counter[str] = Counter() + monkeypatch.setattr(AsyncMetricsCollector, "__init__", tracking_init) + monkeypatch.setattr(AsyncMetricsCollector, "_record_cache_operation_direct", lambda *a, **k: direct.update(["direct"])) op = ir_workload.build_workload("l2_hit_async_metrics") - calls.clear() + (collector,) = made + before = collector._operation_count for _ in range(20): op() - assert calls == {"async": 20} + assert async_metrics._PER_THREAD + assert (collector._sync_mode, collector._queue.qsize(), collector._operation_count - before) == (False, 0, 20) + assert direct == {} def test_pinned_clocks_tick_per_read_on_the_main_thread_only(monkeypatch) -> None: diff --git a/tests/unit/test_single_metrics_record.py b/tests/unit/test_single_metrics_record.py index 1df0a945..bee77af9 100644 --- a/tests/unit/test_single_metrics_record.py +++ b/tests/unit/test_single_metrics_record.py @@ -102,25 +102,16 @@ def compute() -> dict[str, int]: @pytest.mark.unit @pytest.mark.parametrize("sync_mode", [True, False], ids=["sync-sink", "batched-sink"]) def test_collector_counts_one_record_once(sync_mode: bool) -> None: - """Both collector modes turn one record into one counter increment, with no hit field queued.""" + """Both collector modes turn one record into one counter increment, recorded at once and never queued.""" namespace = f"single-record-sink-{sync_mode}" collector = AsyncMetricsCollector(sync_mode=sync_mode, auto_detect_mode=False) - if not sync_mode: - # Stop the batched worker first and flush by hand: a running worker could take the record before this - # reads it. Not via shutdown(), which also returns the collector to sync mode. - assert collector._stopped is not None and collector._worker_thread is not None - collector._stopped.set() - collector._worker_thread.join(5) collector.record_cache_operation( operation="get", namespace=namespace, success=True, duration_ms=1.0, serializer="rust", size_bytes=8 ) if not sync_mode: - assert collector._queue is not None - batch = [collector._queue.get_nowait()] - assert collector._queue.empty() - assert "hit" not in batch[0] - collector._flush_batch(batch) + assert collector._queue is not None and collector._queue.empty() + collector.shutdown() counter = _metrics_cache["cache_operations_total"] # process-wide: shared by every collector assert counter.labels(operation="get", namespace=namespace, success="True", serializer="rust")._value.get() == 1