1"""Chapter 13 practice (Caching at scale):
2- single-flight / request coalescing against a stampede (Level 3, and the Level 4 bug)
3- negative caching against penetration
4- hot-key copies spread across nodes
5- hash-slot sharding of a cache cluster (Redis Cluster style: CRC16 mod 16384)
6- ETag / 304 revalidation
7Standard library only. Run: python3 ch13_cache_at_scale.py"""
8import hashlib
9import threading
10import time
11from collections import Counter
12
13
14class SlowDB:
15 def __init__(self, data, delay=0.05):
16 self.data, self.delay, self.calls = dict(data), delay, 0
17 self.lock = threading.Lock()
18
19 def get(self, key):
20 with self.lock:
21 self.calls += 1
22 time.sleep(self.delay)
23 return self.data.get(key)
24
25
26# ---- Stampede protection: single flight ---------------------------------------------------
27
28class SingleFlightCache:
29 """On a miss, only ONE caller loads from the database; everyone else waiting for the
30 same key gets that caller's result."""
31
32 def __init__(self, db):
33 self.db, self.store = db, {}
34 self.lock = threading.Lock()
35 self.inflight = {} # key -> Event the waiters block on
36
37 def get(self, key):
38 if key in self.store:
39 return self.store[key]
40 with self.lock:
41 if key in self.store: # someone filled it while we waited for the lock
42 return self.store[key]
43 event = self.inflight.get(key)
44 leader = event is None
45 if leader:
46 event = self.inflight[key] = threading.Event()
47 if leader:
48 try:
49 self.store[key] = self.db.get(key)
50 finally:
51 with self.lock:
52 del self.inflight[key]
53 event.set() # wake everyone, even if the load failed
54 return self.store[key]
55 event.wait(timeout=2)
56 return self.store.get(key) # read what the leader stored
57
58
59class BuggySingleFlight(SingleFlightCache):
60 def get(self, key):
61 if key in self.store:
62 return self.store[key]
63 with self.lock:
64 event = self.inflight.get(key)
65 leader = event is None
66 if leader:
67 event = self.inflight[key] = threading.Event()
68 if leader:
69 self.store[key] = self.db.get(key)
70 with self.lock:
71 del self.inflight[key]
72 event.set()
73 return self.store[key]
74 event.wait(timeout=2)
75 return self.db.get(key) # BUG: waiters query the database anyway
76
77
78def hammer(cache, key, n=100):
79 results = []
80 threads = [threading.Thread(target=lambda: results.append(cache.get(key))) for _ in range(n)]
81 for t in threads:
82 t.start()
83 for t in threads:
84 t.join()
85 return results
86
87
88# ---- Penetration: negative caching -----------------------------------------------------------
89
90MISSING = object()
91
92
93class NegativeCache:
94 def __init__(self, db):
95 self.db, self.store = db, {}
96
97 def get(self, key):
98 if key in self.store:
99 v = self.store[key]
100 return None if v is MISSING else v
101 v = self.db.get(key)
102 self.store[key] = MISSING if v is None else v # remember "not found" too (short TTL in real life)
103 return v
104
105
106# ---- Hot keys: spread copies across nodes ----------------------------------------------------
107
108def crc16(data: bytes) -> int:
109 """CRC16-CCITT (XMODEM), the variant Redis Cluster uses for key slots."""
110 crc = 0
111 for b in data:
112 crc ^= b << 8
113 for _ in range(8):
114 crc = ((crc << 1) ^ 0x1021) if crc & 0x8000 else (crc << 1)
115 crc &= 0xFFFF
116 return crc
117
118
119def slot(key: str) -> int:
120 return crc16(key.encode()) % 16384
121
122
123def node_for(key, nodes):
124 """Each node owns an equal range of the 16,384 slots."""
125 per = 16384 // len(nodes)
126 return nodes[min(slot(key) // per, len(nodes) - 1)]
127
128
129def spread_hot_key(key, copies, request_no):
130 """Read one of N copies (product:milk#0 .. #N-1) so the load lands on several nodes."""
131 return f"{key}#{request_no % copies}"
132
133
134# ---- HTTP revalidation --------------------------------------------------------------------------
135
136def etag_of(body: bytes) -> str:
137 return '"' + hashlib.sha256(body).hexdigest()[:16] + '"'
138
139
140def serve(body: bytes, if_none_match=None):
141 tag = etag_of(body)
142 if if_none_match == tag:
143 return 304, b"", tag # "your copy is still good": no body sent
144 return 200, body, tag
145
146
147def run_checks():
148 # Stampede: 100 concurrent misses on one key.
149 db = SlowDB({"product:milk": 52})
150 results = hammer(SingleFlightCache(db), "product:milk")
151 assert results == [52] * 100 and db.calls == 1, db.calls
152
153 db_bad = SlowDB({"product:milk": 52})
154 hammer(BuggySingleFlight(db_bad), "product:milk")
155 assert db_bad.calls > 50, db_bad.calls # most waiters still hit the database
156
157 # Penetration: 1,000 lookups of a product that doesn't exist.
158 db2 = SlowDB({}, delay=0)
159 plain = {}
160 for _ in range(1000):
161 if "product:nope" not in plain:
162 v = db2.get("product:nope")
163 if v is not None:
164 plain["product:nope"] = v
165 assert db2.calls == 1000 # every lookup reached the database
166 db3 = SlowDB({}, delay=0)
167 neg = NegativeCache(db3)
168 for _ in range(1000):
169 neg.get("product:nope")
170 assert db3.calls == 1
171
172 # Redis Cluster slots: known reference value from the Redis Cluster spec.
173 assert crc16(b"123456789") == 0x31C3
174 nodes = ["redis-a", "redis-b", "redis-c", "redis-d"]
175 load = Counter(node_for(f"product:{i}", nodes) for i in range(40_000))
176 assert all(9_000 < c < 11_000 for c in load.values()), load # roughly even
177
178 # Hot key: one key lands on ONE node; 4 copies land on several.
179 one = Counter(node_for("product:milk", nodes) for _ in range(1000))
180 spread = Counter(node_for(spread_hot_key("product:milk", 8, i), nodes) for i in range(1000))
181 assert len(one) == 1 and len(spread) >= 3, spread
182
183 # ETag revalidation
184 body = b"<html>milk page v1</html>"
185 status, payload, tag = serve(body)
186 assert status == 200 and payload == body
187 status2, payload2, _ = serve(body, if_none_match=tag)
188 assert status2 == 304 and payload2 == b""
189 status3, _, _ = serve(b"<html>milk page v2</html>", if_none_match=tag)
190 assert status3 == 200
191
192 print(f"All checks pass. Stampede: single flight 1 DB call, buggy version {db_bad.calls}. "
193 f"Hot key spread over {len(spread)} nodes.")
194
195
196if __name__ == "__main__":
197 run_checks()