1"""Chapter 40 (case study: video upload and streaming, YouTube) deep-dive code.
21. Upload pipeline: resumable chunked upload.
32. Transcoding: split into segments and encode in parallel; storage per bitrate ladder.
43. Adaptive bitrate: a simple buffer-aware player vs a fixed-quality player on a bumpy network.
54. CDN delivery: how concentrated views are (popular videos stay hot).
65. View counts: sharded counters, de-duplicated per viewer, flushed in batches.
7Run: python3 ch40_video.py"""
8import math
9import random
10from collections import defaultdict
11
12
13# ---- 1. resumable upload -----------------------------------------------------------
14
15class ResumableUpload:
16 def __init__(self, size, chunk):
17 self.size, self.chunk = size, chunk
18 self.received = set()
19
20 def put(self, index):
21 self.received.add(index)
22
23 def missing(self):
24 n = math.ceil(self.size / self.chunk)
25 return [i for i in range(n) if i not in self.received]
26
27
28# ---- 2. transcoding -----------------------------------------------------------------
29
30LADDER = [("240p", 0.4), ("360p", 0.8), ("480p", 1.4), ("720p", 2.8), ("1080p", 5.0)] # Mbps
31
32
33def transcode_minutes(video_min, encode_speed=0.5, workers=1, segment_s=6):
34 """encode_speed: minutes of video encoded per minute of one worker, for the whole ladder."""
35 total_work = video_min / encode_speed
36 segments = math.ceil(video_min * 60 / segment_s)
37 parallel = min(workers, segments)
38 return total_work / parallel
39
40
41def ladder_gb_per_hour():
42 return sum(mbps for _, mbps in LADDER) * 3600 / 8 / 1000
43
44
45# ---- 3. ABR ------------------------------------------------------------------------
46
47def play(bandwidth_trace, choose, segment_s=4, start_buffer=0):
48 """Each step downloads one segment. Returns (rebuffer seconds, average bitrate)."""
49 buffer, rebuffer, rates = start_buffer, 0.0, []
50 est = bandwidth_trace[0]
51 for bw in bandwidth_trace:
52 rate = choose(est, buffer)
53 download_s = rate * segment_s / bw
54 if download_s > buffer:
55 rebuffer += download_s - buffer
56 buffer = 0
57 else:
58 buffer -= download_s
59 buffer += segment_s
60 est = 0.7 * est + 0.3 * bw # smoothed throughput estimate
61 rates.append(rate)
62 return round(rebuffer, 1), round(sum(rates) / len(rates), 2)
63
64
65def fixed_1080(est, buffer):
66 return 5.0
67
68
69def adaptive(est, buffer):
70 if buffer < 4:
71 return LADDER[0][1] # nearly empty: grab a small segment fast
72 safety = 0.6 if buffer < 8 else 0.85 # be careful when the buffer is low
73 ok = [mbps for _, mbps in LADDER if mbps <= est * safety]
74 return ok[-1] if ok else LADDER[0][1]
75
76
77# ---- 4. popularity -----------------------------------------------------------------
78
79def share_of_views_from_top(n_videos, top_fraction, s=1.0):
80 weights = [1 / (i + 1) ** s for i in range(n_videos)]
81 top = int(n_videos * top_fraction)
82 return sum(weights[:top]) / sum(weights)
83
84
85# ---- 5. view counts ----------------------------------------------------------------
86
87class ViewCounter:
88 def __init__(self, shards=16, window_s=30 * 60):
89 self.shards = [defaultdict(int) for _ in range(shards)]
90 self.window = window_s
91 self.last_counted = {} # (video, viewer) -> time
92 self.total = defaultdict(int)
93
94 def view(self, video, viewer, now, watched_s):
95 if watched_s < 30: # assumed rule: count only real watches
96 return False
97 key = (video, viewer)
98 if key in self.last_counted and now - self.last_counted[key] < self.window:
99 return False # replay or refresh spam within the window
100 self.last_counted[key] = now
101 self.shards[hash((video, viewer, now)) % len(self.shards)][video] += 1
102 return True
103
104 def flush(self):
105 for s in self.shards:
106 for v, c in s.items():
107 self.total[v] += c
108 s.clear()
109
110
111# ---- checks ------------------------------------------------------------------------
112
113def run_checks():
114 # 1. a 2 GB upload in 8 MB chunks loses the connection after 180 chunks
115 up = ResumableUpload(2_000_000_000, 8_000_000)
116 for i in range(180):
117 up.put(i)
118 assert len(up.missing()) == 70 # resume sends only the last 70
119
120 # 2. a 10-minute video: one worker vs 100 workers on 6-second segments
121 assert transcode_minutes(10) == 20.0
122 assert transcode_minutes(10, workers=100) == 0.2 # 12 seconds
123 assert round(ladder_gb_per_hour(), 2) == 4.68 # all five renditions, per hour of video
124 # Step 2: 500 hours a minute of uploads
125 hours_per_day = 500 * 60 * 24
126 assert hours_per_day == 720_000
127 assert round(hours_per_day * ladder_gb_per_hour() / 1e6, 2) == 3.37 # PB a day of renditions
128
129 # 3. ABR on a bumpy mobile connection (Mbps per segment)
130 rnd = random.Random(40)
131 trace = [rnd.choice([1.5, 2, 3, 6, 8, 10]) for _ in range(60)]
132 fixed = play(trace, fixed_1080)
133 adapt = play(trace, adaptive)
134 assert fixed[0] > 100 and adapt[0] < 15 # 4-minute clip: fixed stalls 144 s, adaptive 10 s
135 assert adapt[1] < fixed[1] # at a lower average bitrate
136
137 # 4. 10M videos: the top 1% get most views
138 share = share_of_views_from_top(10_000_000, 0.01)
139 assert share > 0.6
140
141 # 5. view counts
142 vc = ViewCounter()
143 assert vc.view("v1", "asha", 0, 45)
144 assert not vc.view("v1", "asha", 60, 45) # refresh within 30 min: not counted
145 assert not vc.view("v1", "ravi", 70, 5) # 5 seconds: not a view
146 assert vc.view("v1", "ravi", 80, 120)
147 assert vc.view("v1", "asha", 1_900, 45) # back after the window
148 vc.flush()
149 assert vc.total["v1"] == 3
150
151 return {"transcode_min": (20.0, 0.2), "abr_fixed": fixed, "abr_adaptive": adapt,
152 "top1pct_share": round(share, 2)}
153
154
155if __name__ == "__main__":
156 print("All checks pass.", run_checks())