# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Measure the Dekker pattern and the latency floor.

Two "regions" with a one-way message delay d. Client A writes x = 1 then
reads y; client B writes y = 1 then reads x. With local reads and writes,
both can read 0, which no single copy could produce; making the write wait
for the other copy removes it, and the latency it costs is at least d.
"""
import argparse, json, threading, time

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--delay", type=float, default=0.08, help="d: one-way message delay between the regions, seconds.")
p.add_argument("--runs", type=int, default=100, help="Runs per variant.")
p.add_argument("--margin", type=float, default=0.005, help="Seconds a waiting write adds to d, so it does not race the message arriving at the same instant.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
D = args.delay


def trial(mode):
    rep = {"A": {"x": 0, "y": 0}, "B": {"x": 0, "y": 0}}
    acked = {"A": threading.Event(), "B": threading.Event()}
    def send(to, frm, k, v):                           # asynchronous replication with an ack back after d
        def deliver():
            rep[to][k] = v
            threading.Timer(D, acked[frm].set).start()
        threading.Timer(D, deliver).start()
    out, lat = {}, {}
    def client(me, other, mine, theirs):
        t0 = time.perf_counter()
        rep[me][mine] = 1; send(other, me, mine, 1)
        if mode == "sync write": acked[me].wait()      # the write returns once the other copy has it (2d round trip)
        if mode == "write waits d": time.sleep(D + args.margin)   # the write returns after one delay
        t1 = time.perf_counter()
        out[me] = rep[me][theirs]
        lat[me] = (t1 - t0, time.perf_counter() - t1)
    a = threading.Thread(target=client, args=("A", "B", "x", "y"))
    b = threading.Thread(target=client, args=("B", "A", "y", "x"))
    a.start(); b.start(); a.join(); b.join()
    time.sleep(2.2 * D)                                # let in-flight messages land before the next trial
    return (out["A"], out["B"]), lat


rows = []
for mode in ("local", "write waits d", "sync write"):
    both_zero, w, r = 0, [], []
    for _ in range(args.runs):
        res, lat = trial(mode)
        both_zero += res == (0, 0)
        for wl, rl in lat.values(): w.append(wl); r.append(rl)
    rows.append({"variant": mode, "both_zero": both_zero, "both_zero_pct": both_zero / args.runs * 100,
                 "write_ms": sum(w) / len(w) * 1000, "read_ms": sum(r) / len(r) * 1000,
                 "r_plus_w_ms": (sum(w) + sum(r)) / len(w) * 1000})
    x = rows[-1]
    print(f"{mode:<14} (0,0) in {both_zero:>3} of {args.runs} runs   write {x['write_ms']:6.1f} ms  read {x['read_ms']:5.2f} ms  "
          f"r+w {x['r_plus_w_ms']:6.1f} ms  (d = {D * 1000:.0f} ms)")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"variants": rows, "summary": [{"d_ms": D * 1000}]}}, f)