Spaces:
Sleeping
Sleeping
File size: 2,260 Bytes
29ca14e | 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 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """
Thread-safe prediction cache with TTL expiry.
Uses double-checked locking to prevent duplicate computation
under concurrent requests while avoiding lock contention on hits.
"""
import threading
import time
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass
@dataclass
class CacheEntry:
value: Any
created_at: float
ttl_seconds: float
def is_valid(self) -> bool:
return (time.monotonic() - self.created_at) < self.ttl_seconds
class ThreadSafeCache:
"""
Thread-safe prediction cache with TTL expiry.
Uses double-checked locking to prevent duplicate computation
under concurrent requests while avoiding lock contention on hits.
"""
def __init__(self):
self._store: Dict[str, CacheEntry] = {}
self._locks: Dict[str, threading.Lock] = {}
self._meta_lock = threading.Lock()
def get(self, key: str) -> Optional[Any]:
entry = self._store.get(key)
if entry and entry.is_valid():
return entry.value
return None
def _get_key_lock(self, key: str) -> threading.Lock:
with self._meta_lock:
if key not in self._locks:
self._locks[key] = threading.Lock()
return self._locks[key]
def get_or_compute(self, key: str, compute_fn: Callable, ttl_seconds: float = 300) -> Any:
# Fast path — no lock needed on cache hit
result = self.get(key)
if result is not None:
return result
# Slow path — per-key lock prevents duplicate computation
lock = self._get_key_lock(key)
with lock:
# Re-check after acquiring lock (double-checked locking)
result = self.get(key)
if result is not None:
return result
value = compute_fn()
self._store[key] = CacheEntry(
value=value,
created_at=time.monotonic(),
ttl_seconds=ttl_seconds,
)
return value
def invalidate(self, key: str) -> None:
self._store.pop(key, None)
def clear(self) -> None:
with self._meta_lock:
self._store.clear()
self._locks.clear()
|