Bufferbloat: a bigger buffer buys latency, not throughput

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Bufferbloat: a bigger buffer buys latency, not throughput.

A producer at 1,200 items/s and a consumer at 1,000 items/s with a bounded
queue (full: drop the newest). An exact discrete-event simulation, so the
timings are not blurred by thread scheduling. Then deadline-aware dropping:
skip items that waited more than 1 s, and count goodput.
"""
import argparse, json
from collections import deque

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--produce", type=float, default=1200, help="Items produced per second.")
p.add_argument("--consume", type=float, default=1000, help="Items the consumer serves per second.")
p.add_argument("--seconds", type=float, default=1500, help="Simulated seconds (long enough for the largest queue to fill).")
p.add_argument("--sizes", type=lambda s: [int(v) for v in s.split(",")], default=[100, 1000, 10000, 100000], help="Queue sizes.")
p.add_argument("--deadline", type=float, default=1.0, help="Seconds after which an item is stale.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()


def run(qsize, drop_stale):
    q, waits, good, dropped_full, dropped_stale = deque(), [], 0, 0, 0
    next_arrival, next_service, t = 0.0, 0.0, 0.0
    while t < args.seconds:
        if next_arrival <= next_service:
            t = next_arrival
            if len(q) < qsize: q.append(t)
            else: dropped_full += 1
            next_arrival += 1 / args.produce
        else:
            t = next_service
            while q:
                w = t - q.popleft()
                if drop_stale and w > args.deadline: dropped_stale += 1; continue
                waits.append(w); good += w <= args.deadline
                break
            next_service += 1 / args.consume
    waits = sorted(waits[len(waits) // 2:])                            # steady state: the second half of the run
    pct = lambda x: waits[min(len(waits) - 1, int(len(waits) * x))] if waits else 0
    return {"queue": qsize, "deadline_drop": "yes" if drop_stale else "no", "served_per_s": len(waits) / (args.seconds / 2),
            "p50_s": pct(0.5), "p99_s": pct(0.99), "goodput_per_s": sum(w <= args.deadline for w in waits) / (args.seconds / 2), "dropped_stale": dropped_stale}


rows = []
for drop in (False, True):
    for n in args.sizes:
        r = run(n, drop); rows.append(r)
        print(f"queue {n:>7,} deadline drop {r['deadline_drop']:<3}  served {r['served_per_s']:6.0f}/s  p50 {r['p50_s']:7.3f} s  "
              f"p99 {r['p99_s']:7.3f} s  goodput (<= {args.deadline:g} s) {r['goodput_per_s']:6.0f}/s")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"runs": rows}}, f)