1"""Chapter 21: Polling vs WebSockets vs push. Tested with Python 3, standard library only.
2
3Running example: grocery order tracking. An order produces a few status updates
4(placed, packed, rider assigned, out for delivery, delivered) over about 20 minutes.
5"""
6import random
7
8STATUSES = ["placed", "packed", "rider assigned", "out for delivery", "delivered"]
9
10
11# ---------- 21.2.1 short polling ----------
12
13def short_polling(event_times, interval, duration):
14 """Client asks every `interval` seconds. Returns (requests, empty, avg_delay)."""
15 requests = empty = 0
16 delays, pending = [], sorted(event_times)
17 t = interval
18 while t <= duration:
19 requests += 1
20 got = [e for e in pending if e <= t]
21 if not got:
22 empty += 1
23 for e in got:
24 delays.append(t - e)
25 pending = [e for e in pending if e > t]
26 t += interval
27 return requests, empty, sum(delays) / len(delays)
28
29
30# ---------- 21.2.2 long polling ----------
31
32def long_polling(event_times, timeout, duration):
33 """Server holds each request until an event or `timeout`. Returns (requests, empty, avg_delay)."""
34 requests = empty = 0
35 pending = sorted(event_times)
36 t = 0.0
37 while t < duration:
38 requests += 1
39 nxt = pending[0] if pending else None
40 if nxt is not None and nxt <= t + timeout:
41 t = nxt # answered the moment the event happens
42 pending.pop(0)
43 else:
44 empty += 1
45 t += timeout
46 return requests, empty, 0.0
47
48
49# ---------- 21.2.3 SSE wire format ----------
50
51def sse_format(event_id, event, data):
52 return f"id: {event_id}\nevent: {event}\ndata: {data}\n\n"
53
54
55def sse_parse(stream):
56 out = []
57 for block in stream.strip().split("\n\n"):
58 fields = dict(line.split(": ", 1) for line in block.splitlines())
59 out.append((int(fields["id"]), fields["event"], fields["data"]))
60 return out
61
62
63# ---------- 21.4 missed messages: replay from the last id ----------
64
65class OrderEvents:
66 """Server side: keep recent events per order so reconnecting clients can catch up."""
67
68 def __init__(self):
69 self.log = [] # (id, status)
70
71 def publish(self, status):
72 self.log.append((len(self.log) + 1, status))
73
74 def since(self, last_event_id):
75 return [e for e in self.log if e[0] > last_event_id]
76
77
78class BuggyTracker:
79 """On reconnect, only listens for new events: anything sent while offline is lost."""
80
81 def __init__(self):
82 self.seen = []
83
84 def receive(self, events):
85 self.seen.extend(s for _, s in events)
86
87 def reconnect(self, server):
88 pass # bug: no catch-up
89
90
91class FixedTracker(BuggyTracker):
92 def __init__(self):
93 super().__init__()
94 self.last_id = 0
95
96 def receive(self, events):
97 for i, s in events:
98 if i > self.last_id: # ignore duplicates
99 self.seen.append(s)
100 self.last_id = i
101
102 def reconnect(self, server):
103 self.receive(server.since(self.last_id)) # like the Last-Event-ID header
104
105
106# ---------- 21.4 reconnect storms ----------
107
108def reconnect_peak(clients, base, jitter, seed=21):
109 """Server restarts; every client reconnects after `base` s (+ random jitter). Peak per second."""
110 rnd = random.Random(seed)
111 per_second = {}
112 for _ in range(clients):
113 t = int(base + (rnd.uniform(0, jitter) if jitter else 0))
114 per_second[t] = per_second.get(t, 0) + 1
115 return max(per_second.values())
116
117
118def backoff_delays(attempts, base=1.0, cap=60.0, seed=7):
119 """Exponential backoff with 'full jitter': random between 0 and min(cap, base * 2^n)."""
120 rnd = random.Random(seed)
121 return [rnd.uniform(0, min(cap, base * 2 ** n)) for n in range(attempts)]
122
123
124# ---------- 21.4 heartbeats ----------
125
126def detect_dead(last_seen, now, interval, missed_allowed):
127 """Connections silent for longer than interval * missed_allowed are treated as dead."""
128 return sorted(c for c, t in last_seen.items() if now - t > interval * missed_allowed)
129
130
131# ---------- checks ----------
132
133def run_checks():
134 # one 20-minute order, 5 status updates
135 events = [3, 247, 431, 618, 1197] # seconds after ordering
136 duration = 1200
137
138 req5, empty5, delay5 = short_polling(events, 5, duration)
139 assert req5 == 240 and empty5 == 235 and 2 < delay5 < 3 # 98% wasted; ~interval/2 late
140 req30, _, delay30 = short_polling(events, 30, duration)
141 assert req30 == 40 and 15 < delay30 < 18 # fewer requests, slower news
142
143 lreq, lempty, ldelay = long_polling(events, 30, duration)
144 assert lreq == 45 and lempty == 40 and ldelay == 0.0
145
146 # 100,000 users tracking orders at once
147 users = 100_000
148 assert users / 5 == 20_000 # short polling every 5 s: requests/second
149 assert users / 30 < 3_400 # long polling with 30 s timeouts: ~3,333/s
150
151 # SSE round trip
152 stream = "".join(sse_format(i + 1, "status", s) for i, s in enumerate(STATUSES))
153 assert sse_parse(stream)[-1] == (5, "status", "delivered")
154
155 # missed messages: phone goes into a tunnel after "packed"
156 server = OrderEvents()
157 buggy, fixed = BuggyTracker(), FixedTracker()
158 for s in STATUSES[:2]:
159 server.publish(s)
160 buggy.receive(server.log[-1:])
161 fixed.receive(server.log[-1:])
162 for s in STATUSES[2:4]: # sent while offline
163 server.publish(s)
164 buggy.reconnect(server)
165 fixed.reconnect(server)
166 server.publish(STATUSES[4])
167 buggy.receive(server.log[-1:])
168 fixed.receive(server.log[-1:])
169 fixed.receive(server.log[-1:]) # duplicate delivery
170 assert buggy.seen == ["placed", "packed", "delivered"] # missed two updates
171 assert fixed.seen == STATUSES # caught up, no duplicates
172
173 # reconnect storm after a server restart
174 storm = reconnect_peak(100_000, base=1, jitter=0)
175 spread = reconnect_peak(100_000, base=1, jitter=30)
176 assert storm == 100_000 and spread < 3_700
177
178 delays = backoff_delays(8)
179 assert all(0 <= d <= min(60, 2 ** n) for n, d in enumerate(delays))
180
181 # heartbeats every 30 s; dead after 3 missed
182 seen = {"asha": 1050, "ravi": 900, "kiran": 1080} # last heartbeat times
183 assert detect_dead(seen, now=1100, interval=30, missed_allowed=3) == ["ravi"]
184
185 # connection memory (assumed 20 KB per idle connection) and servers (assumed 100k each)
186 conns, per_conn_kb, per_server = 1_000_000, 20, 100_000
187 assert conns * per_conn_kb / 1_000_000 == 20 # GB
188 assert conns // per_server == 10
189
190 return {"short5": (req5, empty5, delay5), "short30": (req30, delay30),
191 "long": (lreq, lempty), "storm": storm, "spread": spread}
192
193
194if __name__ == "__main__":
195 print("All checks pass.", run_checks())