File size: 810 Bytes
aac350d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 | """
Global monotonic counters — cache hits/misses, requests, errors.
"""
from __future__ import annotations
import threading
from collections import defaultdict
from typing import Dict
class CounterRegistry:
"""Named integer counters."""
def __init__(self) -> None:
self._lock = threading.RLock()
self._counters: Dict[str, int] = defaultdict(int)
def inc(self, name: str, amount: int = 1) -> None:
with self._lock:
self._counters[name] += amount
def get(self, name: str) -> int:
with self._lock:
return self._counters.get(name, 0)
def snapshot(self) -> Dict[str, int]:
with self._lock:
return dict(self._counters)
def reset(self) -> None:
with self._lock:
self._counters.clear()
|