1"""Chapter 28: Consistent hashing. Tested with Python 3, standard library only.
2
3Running example: the bookstore's cache cluster. Book pages are cached across 4 cache
4servers; at the Big Book Sale we add a 5th. How many cached pages suddenly miss?
5"""
6import bisect
7import hashlib
8import statistics
9
10
11def h(s):
12 """A stable 64-bit hash (Python's hash() changes between runs)."""
13 return int.from_bytes(hashlib.md5(s.encode()).digest()[:8], "big")
14
15
16KEYS = [f"book:{i}" for i in range(100_000)]
17
18
19# ---------- 28.1 mod-N ----------
20
21def mod_n(key, n):
22 return h(key) % n
23
24
25def moved_fraction(assign_before, assign_after, keys=KEYS):
26 return sum(assign_before(k) != assign_after(k) for k in keys) / len(keys)
27
28
29# ---------- 28.2-28.5 the ring, with virtual nodes ----------
30
31class Ring:
32 def __init__(self, servers=(), vnodes=1):
33 self.vnodes = vnodes
34 self.points, self.owner = [], {}
35 for s in servers:
36 self.add(s)
37
38 def add(self, server):
39 for v in range(self.vnodes):
40 p = h(f"{server}#{v}")
41 bisect.insort(self.points, p)
42 self.owner[p] = server
43
44 def remove(self, server):
45 for v in range(self.vnodes):
46 p = h(f"{server}#{v}")
47 self.points.remove(p)
48 del self.owner[p]
49
50 def lookup(self, key):
51 """Walk clockwise from the key's position to the first server point."""
52 i = bisect.bisect_right(self.points, h(key)) % len(self.points)
53 return self.owner[self.points[i]]
54
55 # 28.6 replication: the next N *distinct* servers clockwise
56 def replicas(self, key, n):
57 out = []
58 i = bisect.bisect_right(self.points, h(key))
59 while len(out) < n:
60 s = self.owner[self.points[i % len(self.points)]]
61 if s not in out:
62 out.append(s)
63 i += 1
64 return out
65
66 def replicas_buggy(self, key, n):
67 """Takes the next N points, even if two belong to the same server."""
68 i = bisect.bisect_right(self.points, h(key))
69 return [self.owner[self.points[(i + j) % len(self.points)]] for j in range(n)]
70
71
72def load_spread(ring, keys=KEYS):
73 counts = {}
74 for k in keys:
75 s = ring.lookup(k)
76 counts[s] = counts.get(s, 0) + 1
77 vals = list(counts.values())
78 mean = len(keys) / len(vals)
79 return max(vals) / mean, statistics.pstdev(vals) / mean
80
81
82# ---------- 28.8 rendezvous (highest random weight) hashing ----------
83
84def rendezvous(key, servers):
85 return max(servers, key=lambda s: h(f"{key}|{s}"))
86
87
88# ---------- checks ----------
89
90def run_checks():
91 four = [f"cache-{i}" for i in range(4)]
92 five = four + ["cache-4"]
93
94 # mod-N: adding one server moves most keys
95 m = moved_fraction(lambda k: mod_n(k, 4), lambda k: mod_n(k, 5))
96 assert 0.78 < m < 0.82 # ~80% (expected 1 - 1/5)
97
98 # ring: adding one server moves ~1/5 of keys, and only to the new server
99 r4, r5 = Ring(four, vnodes=100), Ring(five, vnodes=100)
100 moved = [k for k in KEYS if r4.lookup(k) != r5.lookup(k)]
101 frac = len(moved) / len(KEYS)
102 assert 0.15 < frac < 0.25
103 assert all(r5.lookup(k) == "cache-4" for k in moved) # nobody else reshuffles
104
105 # removing a server: only its keys move
106 r_rm = Ring(four, vnodes=100)
107 r_rm.remove("cache-2")
108 changed = [k for k in KEYS if r4.lookup(k) != r_rm.lookup(k)]
109 assert all(r4.lookup(k) == "cache-2" for k in changed)
110
111 # virtual nodes even out the load
112 worst1, sd1 = load_spread(Ring(four, vnodes=1))
113 worst100, sd100 = load_spread(Ring(four, vnodes=100))
114 assert worst1 > 1.9 and worst100 < 1.2 and sd100 < sd1 # busiest server: 1.95x vs 1.16x its fair share
115
116 # replication on the ring
117 reps = r5.replicas("book:42", 3)
118 assert len(set(reps)) == 3 and reps[0] == r5.lookup("book:42")
119
120 # rendezvous: also moves ~1/5 on add, only to the new server
121 rv_moved = [k for k in KEYS[:20_000] if rendezvous(k, four) != rendezvous(k, five)]
122 assert 0.17 < len(rv_moved) / 20_000 < 0.23
123 assert all(rendezvous(k, five) == "cache-4" for k in rv_moved)
124
125 # Level 4: replicas on the same physical server
126 dup = sum(1 for k in KEYS[:10_000] if len(set(r5.replicas_buggy(k, 3))) < 3)
127 assert dup > 1_000 # many keys lose real redundancy
128 assert all(len(set(r5.replicas(k, 3))) == 3 for k in KEYS[:10_000])
129
130 # Level 2 arithmetic: 10M cached pages
131 assert round(10_000_000 * (1 - 1 / 5)) == 8_000_000
132 assert round(10_000_000 / 5) == 2_000_000
133
134 return {"mod_n_moved": round(m, 3), "ring_moved": round(frac, 3),
135 "worst_load": (round(worst1, 2), round(worst100, 2)),
136 "buggy_dup_keys": dup}
137
138
139if __name__ == "__main__":
140 print("All checks pass.", run_checks())