Group commit: break the 1/latency ceiling

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Group commit: break the 1 / fsync-latency ceiling.

A tiny log server with client threads. Version A: each commit does write +
sync under a lock. Version B: a flusher thread takes everything queued,
writes it, syncs once, and wakes the waiters. Commits per second and commit
latency for 1 to 256 clients. Uses F_FULLFSYNC on macOS (a real flush) and
fsync elsewhere.
"""
import argparse, json, os, platform, queue, statistics, tempfile, threading, time

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--clients", type=lambda s: [int(v) for v in s.split(",")], default=[1, 8, 64, 256], help="Client threads.")
p.add_argument("--seconds", type=float, default=5.0, help="Seconds per measurement.")
p.add_argument("--path", default=os.path.join(tempfile.gettempdir(), "group-commit.log"), help="Log file.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
import atexit, shutil
atexit.register(lambda: shutil.rmtree(args.path, ignore_errors=True) if os.path.isdir(args.path) else os.path.exists(args.path) and os.remove(args.path))   # leave nothing behind
FULL = platform.system() == "Darwin"

def sync(fd):
    if FULL:
        import fcntl; fcntl.fcntl(fd, fcntl.F_FULLFSYNC)
    else: os.fsync(fd)

def run(version, clients):
    fd = os.open(args.path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC)
    lock, stop, lat, done = threading.Lock(), threading.Event(), [], [0]
    q = queue.Queue()
    def flusher():
        while not stop.is_set() or not q.empty():
            try: batch = [q.get(timeout=0.05)]
            except queue.Empty: continue
            while not q.empty(): batch.append(q.get_nowait())
            os.write(fd, b"".join(r for r, _ in batch)); sync(fd)
            for _, ev in batch: ev.set()
    def client():
        rec = b"x" * 200
        while not stop.is_set():
            t = time.perf_counter()
            if version == "A":
                with lock: os.write(fd, rec); sync(fd)
            else:
                ev = threading.Event(); q.put((rec, ev)); ev.wait()
            lat.append(time.perf_counter() - t); done[0] += 1
    th = [threading.Thread(target=client) for _ in range(clients)]
    if version == "B": th.append(threading.Thread(target=flusher))
    for t in th: t.start()
    time.sleep(args.seconds); stop.set()
    for t in th: t.join()
    os.close(fd)
    return {"version": "A: sync per commit" if version == "A" else "B: group commit", "clients": clients,
            "commits_per_s": done[0] / args.seconds, "median_ms": statistics.median(lat) * 1000}

rows = []
for v in ("A", "B"):
    for c in args.clients:
        r = run(v, c); rows.append(r)
        print(f"{r['version']:<20} {c:>4} clients: {r['commits_per_s']:9,.0f} commits/s   median latency {r['median_ms']:6.2f} ms")

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