NeuralBlitz What I do
- Implement caching at multiple levels
- Choose appropriate cache strategies
- Handle cache invalidation
- Prevent cache stampede
- Use distributed caching effectively
- Monitor cache performance
- Implement cache warming
When to use me
When implementing caching solutions or optimizing cache performance.
Multi-Level Caching
┌─────────────────────────────────────────────────────────────┐
│ Request │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ Browser Cache (HTTP) │
│ - Cache-Control headers │
│ - ETag/Last-Modified │
│ - LocalStorage/SessionStorage │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ CDN / Edge Cache │
│ - CloudFront, Cloudflare │
│ - Static assets, API responses │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ Application Cache (Memory) │
│ - In-memory caches (LRU, LFU) │
│ - Per-process or shared memory │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ Distributed Cache (Redis/Memcached) │
│ - Shared across instances │
│ - High availability │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ Database Cache │
│ - Query cache │
│ - Buffer pool │
└─────────────────────────────────────────────────────────────┘
Cache Strategies
import redis
import json
from typing import Optional, Callable, Any
from functools import wraps
from datetime import timedelta
class CacheService:
"""Multi-level cache service with TTL support."""
def __init__(self, redis_url: str) -> None:
self.redis = redis.from_url(redis_url)
self.local_cache = {}
def get(self, key: str) -> Optional[Any]:
"""Get from cache hierarchy."""
# Check local cache first
if key in self.local_cache:
value, expires = self.local_cache[key]
if expires > datetime.utcnow():
return value
del self.local_cache[key]
# Check distributed cache
cached = self.redis.get(key)
if cached:
value = json.loads(cached)
# Populate local cache
self.local_cache[key] = (
value,
datetime.utcnow() + timedelta(seconds=60)
)
return value
return None
def set(
self,
key: str,
value: Any,
ttl_seconds: int = 3600,
local_ttl_seconds: int = 60
) -> None:
"""Set in both cache layers."""
# Set distributed cache
self.redis.setex(key, ttl_seconds, json.dumps(value))
# Set local cache with shorter TTL
self.local_cache[key] = (
value,
datetime.utcnow() + timedelta(seconds=local_ttl_seconds)
)
def delete(self, key: str) -> None:
"""Delete from both cache layers."""
self.redis.delete(key)
self.local_cache.pop(key, None)
def invalidate_pattern(self, pattern: str) -> None:
"""Delete keys matching pattern."""
keys = self.redis.keys(pattern)
if keys:
self.redis.delete(*keys)
# Clear local cache
for key in list(self.local_cache.keys()):
if pattern.replace('*', '') in key:
del self.local_cache[key]
def cached(
ttl: int = 3600,
key_prefix: str = ''
):
"""Decorator for caching function results."""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
cache = get_cache_service()
key = f"{key_prefix}:{func.__name__}:{args}:{kwargs}"
cached_value = cache.get(key)
if cached_value is not None:
return cached_value
result = func(*args, **kwargs)
cache.set(key, result, ttl)
return result
return wrapper
return decorator
Cache Stampede Prevention
import asyncio
import async_timeout
class CacheStampedePreventer:
"""Prevent cache stampede with distributed locking."""
def __init__(
self,
cache: CacheService,
lock_ttl: int = 30
) -> None:
self.cache = cache
self.lock_ttl = lock_ttl
async def get_or_compute(
self,
key: str,
compute_fn: Callable,
ttl: int = 3600
) -> Any:
"""Get from cache or compute with stampede prevention."""
# Try cache first
cached = self.cache.get(key)
if cached is not None:
return cached
# Acquire distributed lock
lock_key = f"lock:{key}"
lock_acquired = self.cache.redis.set(
lock_key,
"locked",
nx=True,
ex=self.lock_ttl
)
if not lock_acquired:
# Wait for other process to compute
return await self._wait_for_computation(key, compute_fn)
try:
# Compute and cache
result = await compute_fn()
self.cache.set(key, result, ttl)
return result
finally:
# Release lock
self.cache.redis.delete(lock_key)
async def _wait_for_computation(
self,
key: str,
compute_fn: Callable
) -> Any:
"""Wait for computation to complete."""
max_wait = 10 # seconds
for _ in range(max_wait * 10): # Poll every 100ms
cached = self.cache.get(key)
if cached is not None:
return cached
await asyncio.sleep(0.1)
# If still not cached, compute ourselves
return await compute_fn()
class ProbabilisticEarlyRecomputation:
"""Prevent expired cache with early recomputation."""
def __init__(
self,
cache: CacheService,
delta_percent: int = 10
) -> None:
self.cache = cache
self.delta_percent = delta_percent
def get(self, key: str, compute_fn: Callable) -> Any:
"""Get with early recomputation."""
cached = self.cache.get(key)
if cached is not None:
return cached
ttl = self.cache.redis.ttl(key)
if ttl == -1: # Key doesn't exist
return compute_fn()
# If within delta period, recompute in background
threshold = ttl * (100 - self.delta_percent) / 100
if ttl < threshold:
# Could trigger background task here
pass
return compute_fn()
HTTP Caching
from fastapi import Request, Response
from typing import Optional
class CacheControlMiddleware:
"""Add appropriate cache control headers."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope['type'] != 'http':
await self.app(scope, receive, send)
return
async def send_wrapper(message):
if message['type'] == 'http.response.start':
request = Request(scope)
response = Response(
content=b'',
status_code=message['status'],
headers=dict(message.get('headers', []))
)
# Set cache headers based on route
if '/api/' in request.url.path:
# API endpoints - no cache or short cache
if '/static/' not in request.url.path:
response.headers['Cache-Control'] = 'no-cache'
else:
# Static content - long cache
response.headers['Cache-Control'] = 'public, max-age=31536000'
message['headers'] = list(response.headers.items())
await send(message)
await self.app(scope, receive, send_wrapper)
Cache Invalidation Patterns
# Event-driven cache invalidation
class CacheInvalidator:
def __init__(self, cache: CacheService) -> None:
self.cache = cache
self.subscriptions = {}
def subscribe(self, event_type: str, callback: Callable) -> None:
"""Subscribe to invalidation events."""
if event_type not in self.subscriptions:
self.subscriptions[event_type] = []
self.subscriptions[event_type].append(callback)
def invalidate(self, event_type: str, data: dict) -> None:
"""Trigger invalidation based on event."""
if event_type in self.subscriptions:
for callback in self.subscriptions[event_type]:
pattern = callback(data)
if pattern:
self.cache.invalidate_pattern(pattern)
def on_user_update(self, user_id: str) -> None:
"""Invalidate user-related caches."""
patterns = [
f"user:{user_id}:*",
f"users:list:*",
f"stats:user:{user_id}",
]
for pattern in patterns:
self.cache.invalidate_pattern(pattern)
Cache Monitoring
class CacheMonitor:
"""Monitor cache performance."""
def __init__(self, redis_client: redis.Redis) -> None:
self.redis = redis_client
def get_stats(self) -> dict:
"""Get cache statistics."""
info = self.redis.info('stats')
memory = self.redis.info('memory')
return {
'hits': info.get('keyspace_hits', 0),
'misses': info.get('keyspace_misses', 0),
'hit_rate': self._calculate_hit_rate(info),
'used_memory': memory.get('used_memory_human'),
'connected_clients': self.redis.info('clients').get('connected_clients', 0),
'uptime_seconds': self.redis.info('server').get('uptime_in_seconds', 0),
}
def _calculate_hit_rate(self, info: dict) -> float:
hits = info.get('keyspace_hits', 0)
misses = info.get('keyspace_misses', 0)
total = hits + misses
if total == 0:
return 0.0
return round(hits / total * 100, 2)