Pythoncaching_system.py
Copy
1
import time
2
import json
3
import redis
4
import hashlib
5
import threading
6
from functools import wraps
7
from typing import Callable, Any
8
 
9
# ── Redis setup ─────────────────────────────────────────────────────
10
# docker run -d -p 6379:6379 redis:alpine
11
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
12
 
13
# ── L1: In-process cache (dict-based LRU) ───────────────────────────
14
class L1Cache:
15
    def __init__(self, max_size: int = 100):
16
        self._store: dict[str, tuple[Any, float]] = {}
17
        self._max_size = max_size
18
        self._lock = threading.Lock()
19
    
20
    def get(self, key: str) -> tuple[Any, bool]:
21
        with self._lock:
22
            if key in self._store:
23
                value, expires_at = self._store[key]
24
                if expires_at == 0 or time.time() < expires_at:
25
                    return value, True
26
                del self._store[key]
27
            return None, False
28
    
29
    def set(self, key: str, value: Any, ttl: int = 300):
30
        with self._lock:
31
            if len(self._store) >= self._max_size:
32
                # Evict oldest
33
                oldest = min(self._store.keys(), key=lambda k: self._store[k][1])
34
                del self._store[oldest]
35
            expires_at = time.time() + ttl if ttl > 0 else 0
36
            self._store[key] = (value, expires_at)
37
    
38
    def delete(self, key: str):
39
        with self._lock:
40
            self._store.pop(key, None)
41
    
42
    @property
43
    def size(self) -> int:
44
        return len(self._store)
45
 
46
l1 = L1Cache(max_size=50)
47
 
48
# ── Cache-aside pattern (multi-layer) ───────────────────────────────
49
class CacheStats:
50
    def __init__(self):
51
        self.l1_hits = 0
52
        self.l2_hits = 0
53
        self.misses = 0
54
    
55
    @property
56
    def total(self):
57
        return self.l1_hits + self.l2_hits + self.misses
58
    
59
    @property
60
    def hit_rate(self):
61
        if self.total == 0:
62
            return 0.0
63
        return (self.l1_hits + self.l2_hits) / self.total
64
 
65
stats = CacheStats()
66
 
67
def get_cached(key: str) -> Any | None:
68
    """Try L1 → L2 → None."""
69
    # L1 check
70
    value, hit = l1.get(key)
71
    if hit:
72
        stats.l1_hits += 1
73
        return value
74
    
75
    # L2 check
76
    raw = r.get(key)
77
    if raw:
78
        value = json.loads(raw)
79
        l1.set(key, value, ttl=60)  # Populate L1
80
        stats.l2_hits += 1
81
        return value
82
    
83
    stats.misses += 1
84
    return None
85
 
86
def set_cached(key: str, value: Any, ttl_l1: int = 60, ttl_l2: int = 300):
87
    """Write to both L1 and L2."""
88
    l1.set(key, value, ttl=ttl_l1)
89
    r.setex(key, ttl_l2, json.dumps(value))
90
 
91
def invalidate(key: str):
92
    """Remove from both layers."""
93
    l1.delete(key)
94
    r.delete(key)
95
 
96
# ── Simulated DB fetch (slow) ────────────────────────────────────────
97
def fetch_user_from_db(user_id: int) -> dict:
98
    time.sleep(0.05)  # Simulate DB latency
99
    return {"id": user_id, "name": f"User {user_id}", "email": f"user{user_id}@example.com"}
100
 
101
def get_user(user_id: int) -> dict:
102
    """Cache-aside: check cache, miss → DB → populate cache."""
103
    key = f"user:{user_id}"
104
    cached = get_cached(key)
105
    if cached:
106
        return cached
107
    
108
    user = fetch_user_from_db(user_id)
109
    set_cached(key, user, ttl_l1=60, ttl_l2=600)
110
    return user
111
 
112
# ── Cache stampede protection ────────────────────────────────────────
113
_locks: dict[str, threading.Lock] = {}
114
_locks_lock = threading.Lock()
115
 
116
def get_user_protected(user_id: int) -> dict:
117
    """Prevent stampede with per-key locking."""
118
    key = f"user:{user_id}"
119
    cached = get_cached(key)
120
    if cached:
121
        return cached
122
    
123
    with _locks_lock:
124
        if key not in _locks:
125
            _locks[key] = threading.Lock()
126
        lock = _locks[key]
127
    
128
    with lock:
129
        # Double-check after acquiring lock
130
        cached = get_cached(key)
131
        if cached:
132
            return cached
133
        
134
        user = fetch_user_from_db(user_id)
135
        set_cached(key, user)
136
        return user
137
 
138
# ── Benchmark ─────────────────────────────────────────────────────────
139
def benchmark():
140
    print("=== Cache Performance Benchmark ===\n")
141
    
142
    # Warm up with user 1
143
    get_user(1)
144
    
145
    # 100 requests, mostly user 1 (cache hit)
146
    start = time.time()
147
    for i in range(100):
148
        user_id = 1 if i % 5 != 0 else (i % 10 + 1)  # 80% user 1, 20% others
149
        get_user(user_id)
150
    elapsed = time.time() - start
151
    
152
    print(f"100 requests in {elapsed:.3f}s ({elapsed*10:.1f}ms avg)")
153
    print(f"L1 hits: {stats.l1_hits}")
154
    print(f"L2 hits: {stats.l2_hits}")
155
    print(f"DB misses: {stats.misses}")
156
    print(f"Overall hit rate: {stats.hit_rate:.0%}")
157
    print(f"L1 cache size: {l1.size}")
158
 
159
benchmark()
160