"""Can a byte-shuffling proxy multiplex many clients onto one connection?

    python3 demo.py            # the deterministic run: byte-identical output
    python3 demo.py --race     # the same questions under a real race

Determinism: every step in the default run is sequenced on a byte, request or
reply counter reported by the upstream or the proxy -- never on a sleep. No
elapsed time is printed anywhere. Ports are fixed and never printed, so no
number in the transcript depends on the machine.

--race re-asks three of the questions with no gate and no scripting at all.
Its numbers are distributions, not constants, so they are NOT reproducible and
are kept out of the default output.
"""
import sys
import threading

import tcp_proxy as T

PORT = 8340
KEYS = ["alpha", "beta", "gamma"]


def ports():
    global PORT
    PORT += 2
    return PORT, PORT + 1


def rule(title):
    print("\n" + "=" * 74)
    print(title)
    print("=" * 74)


def show(name, data):
    print("  %-10s got %r" % (name, data))


def load(px_port, n_clients, n_requests, timeout=2.0):
    """n_clients concurrent clients, n_requests each, released by a barrier."""
    got = [None] * n_clients
    barrier = threading.Barrier(n_clients)

    def one(i):
        sock = T.connect(px_port)
        lines = []
        barrier.wait()
        for r in range(n_requests):
            key = KEYS[(i + r) % len(KEYS)]
            sock.sendall(b"GET " + key.encode() + b"\n")
            lines.append((key, T.read_line(sock, timeout=timeout)))
        got[i] = lines
        sock.close()

    threads = [threading.Thread(target=one, args=(i,), daemon=True)
               for i in range(n_clients)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=20)
    return got


def score(got):
    """(right answers, wrong answers) over every request that was asked."""
    ok = bad = 0
    for lines in got or []:
        for key, reply in lines or []:
            want = ("%s=%s\n" % (key, T.VALUES[key])).encode()
            ok, bad = (ok + 1, bad) if reply == want else (ok, bad + 1)
    return ok, bad


# --- 1. the request stream, interleaved ------------------------------------

def interleave():
    rule("1. one upstream connection, two clients, bytes interleaved")
    up_port, px_port = ports()
    up = T.Upstream(up_port)
    px = T.NaiveProxy(px_port, up.addr(), routing="last")

    a, b = T.connect(px_port), T.connect(px_port)
    a.sendall(b"GET alp")           # 7 bytes: a partial request
    up.wait_bytes(7)
    b.sendall(b"GET beta\n")        # 9 bytes: a whole request, in the gap
    up.wait_bytes(16)
    up.wait_requests(1)
    px.wait_routed(1)               # reply 1 is downstream before A writes on
    a.sendall(b"ha\n")              # 3 bytes: the rest of A's request
    up.wait_bytes(19)
    up.wait_requests(2)
    px.wait_routed(2)

    print("  client A wrote  %r  then  %r" % (b"GET alp", b"ha\n"))
    print("  client B wrote  %r  in between" % b"GET beta\n")
    print("\n  what the upstream parsed as requests:")
    for i, (cid, line) in enumerate(up.requests, 1):
        print("    request %d on upstream conn %d: %r" % (i, cid, line))
    print("\n  what each client received back:")
    show("client A", T.drain(a))
    show("client B", T.drain(b))
    print("\n  Two well-formed requests went in; two requests nobody sent came")
    print("  out. The proxy forwarded every byte, in order, exactly once.")
    a.close(); b.close(); px.close(); up.close()


# --- 2. no interleaving, still wrong ---------------------------------------

def misdelivery(routing):
    rule("2%s. no interleaving at all -- routing=%r"
         % ("ab"[routing != "last"], routing))
    up_port, px_port = ports()
    gate = threading.Event()
    up = T.Upstream(up_port, gate=gate.wait)
    px = T.NaiveProxy(px_port, up.addr(), routing=routing)

    a, b = T.connect(px_port), T.connect(px_port)
    a.sendall(b"GET alpha\n")       # 10 bytes, complete and well-formed
    px.wait_forwarded(10)
    up.wait_requests(1)             # the upstream is now holding A's reply
    b.sendall(b"GET beta\n")        # 9 bytes, also complete and well-formed
    px.wait_forwarded(19)           # ...and B is now the last writer
    gate.set()
    up.wait_requests(2)

    print("  client A asked for alpha, client B asked for beta.")
    print("  both requests arrived whole; nothing was interleaved:")
    for i, (cid, line) in enumerate(up.requests, 1):
        print("    request %d on upstream conn %d: %r" % (i, cid, line))
    print("\n  what each client received back:")
    show("client A", T.drain(a))
    show("client B", T.drain(b))
    a.close(); b.close(); px.close(); up.close()


# --- 3. the 1:1 fallback ---------------------------------------------------

def passthrough():
    rule("3. the honest L4 proxy: one upstream conn per client conn")
    up_port, px_port = ports()
    up = T.Upstream(up_port)
    px = T.PassthroughProxy(px_port, up.addr())
    ok, bad = score(load(px_port, 8, 3))
    print("  clients=8  requests=%d  correct=%d  wrong=%d" % (ok + bad, ok, bad))
    print("  upstream connections opened: %d" % up.conns)
    print("  ratio: 8 client conns : %d upstream conns = %.1f:1"
          % (up.conns, 8 / up.conns))
    print("\n  Every answer is right, and nothing is multiplexed. This is what")
    print("  you fall back to if you refuse to parse.")
    px.close(); up.close()


# --- 4. the pool, once the proxy parses ------------------------------------

def framed(caps=(1, 2, 4, 8, 16), n_clients=8, n_requests=3):
    rule("4. the L7 pool: parse a request, borrow a connection, give it back")
    print("  %-5s %-9s %-8s %-6s %s"
          % ("cap", "upstream", "requests", "wrong", "requests/conn"))
    print("  %-5s %-9s %-8s %-6s %s"
          % ("-" * 5, "-" * 9, "-" * 8, "-" * 6, "-" * 13))
    for cap in caps:
        up_port, px_port = ports()
        need = min(cap, n_clients)
        up = T.Upstream(up_port)
        # Hold every reply until `need` requests have arrived, so the number
        # of connections the pool opens is pinned instead of raced.
        up.gate = lambda n=need: up.wait_requests(n)
        px = T.FramedProxy(px_port, up.addr(), pool_cap=cap)
        ok, bad = score(load(px_port, n_clients, n_requests))
        print("  %-5d %-9d %-8d %-6d %.1f"
              % (cap, up.conns, ok + bad, bad, (ok + bad) / up.conns))
        px.close(); up.close()
    print("\n  8 clients, 24 requests, every answer right at every cap. The")
    print("  pool never exceeds the number of requests actually in flight,")
    print("  so cap=16 opens 8 connections and buys nothing over cap=8.")


# --- 5. the load-bearing line ----------------------------------------------

def release_policy():
    rule("5. the load-bearing line: where _read_reply decides a reply ended")
    print("  Identical proxy, pool (cap=1) and clients. Two ways to decide a")
    print("  reply ended -- at the delimiter, or after 50ms of upstream")
    print("  quiet -- crossed with two upstream write patterns.")
    print("\n  %-9s %-32s %-9s %s"
          % ("release", "upstream write pattern", "correct", "wrong"))
    print("  %-9s %-32s %-9s %s"
          % ("-" * 9, "-" * 32, "-" * 9, "-" * 5))
    for release in ("framed", "timeout"):
        for label, delay, split in (("one write", 0.0, 0),
                                    ("pauses mid-reply", 0.12, 6)):
            up_port, px_port = ports()
            up = T.Upstream(up_port, reply_delay=delay, reply_split=split)
            px = T.FramedProxy(px_port, up.addr(), pool_cap=1,
                               release=release, release_timeout=0.05)
            ok, bad = score(load(px_port, 8, 1))
            print("  %-9s %-32s %-9d %d" % (release, label, ok, bad))
            px.close(); up.close()
    print("\n  Read the last two rows against each other: 8 of 8 becomes 0 of")
    print("  8 with nothing changed but whether the upstream paused. A")
    print("  timeout is a guess about the upstream's scheduling; a delimiter")
    print("  is a fact about the message.")


# --- 6. the boundary -------------------------------------------------------

def boundary():
    rule("6. the boundary: the same byte-shuffler, one request in flight")
    up_port, px_port = ports()
    up = T.Upstream(up_port)
    px = T.NaiveProxy(px_port, up.addr(), routing="last")
    socks = [T.connect(px_port) for _ in range(8)]
    ok = bad = 0
    for r in range(3):
        for i, sock in enumerate(socks):
            key = KEYS[(i + r) % len(KEYS)]
            sock.sendall(b"GET " + key.encode() + b"\n")
            reply = T.read_line(sock)
            want = ("%s=%s\n" % (key, T.VALUES[key])).encode()
            ok, bad = (ok + 1, bad) if reply == want else (ok, bad + 1)
    print("  clients=8  requests=%d  correct=%d  wrong=%d" % (ok + bad, ok, bad))
    print("  upstream connections opened: %d" % up.conns)
    print("  ratio: 8 client conns : %d upstream conn = %.1f:1"
          % (up.conns, 8 / up.conns))
    print("\n  Same protocol-blind relay as sections 1 and 2, and every answer")
    print("  is right. The only thing that changed is that no two requests")
    print("  are ever in flight at once. That mechanism is serialisation.")
    for sock in socks:
        sock.close()
    px.close(); up.close()


# --- appendix: a second mechanism, not part of the argument ----------------

def half_close():
    rule("APPENDIX. bonus mechanism: what the 1:1 relay does with a FIN")
    for keep in (False, True):
        up_port, px_port = ports()
        gate = threading.Event()
        up = T.Upstream(up_port, gate=gate.wait)
        px = T.PassthroughProxy(px_port, up.addr(), half_close=keep)
        sock = T.connect(px_port)
        sock.sendall(b"GET alpha\n")
        up.wait_requests(1)              # the upstream is holding the reply
        sock.shutdown(1)                 # SHUT_WR: "I am done asking"
        px.wait_eofs(1)                  # the relay has reacted to the FIN
        gate.set()
        call = "dst.shutdown(SHUT_WR)" if keep else "dst.close()"
        print("  on client EOF the relay calls %-22s client got %r"
              % (call, T.read_line(sock, timeout=1.0)))
        sock.close(); px.close(); up.close()
    print("\n  One call. A half-close is an ordinary thing for a client to do")
    print("  after its last request, and one of these two answers to it costs")
    print("  the entire response. This is a second mechanism, and it is why")
    print("  tcp_proxy.py is over the repo's 300-line ceiling -- see the")
    print("  commentary's appendix.")


# --- quarantine: real races, not reproducible ------------------------------

def race_boundary(trials=20):
    rule("RACE A. wrong answers vs requests in flight -- NOT reproducible")
    print("  Naive shared-upstream relay, 3 requests per client, %d trials"
          % trials)
    print("  each, no gate and no scripting. The column is a range.")
    print("  %-10s %-10s %-16s %s"
          % ("clients", "requests", "wrong (min-max)", "mean wrong"))
    for n in (1, 2, 3, 4, 8):
        results = []
        for _ in range(trials):
            up_port, px_port = ports()
            up = T.Upstream(up_port)
            px = T.NaiveProxy(px_port, up.addr(), routing="last")
            results.append(score(load(px_port, n, 3, timeout=0.5))[1])
            px.close(); up.close()
        print("  %-10d %-10d %-16s %.1f"
              % (n, n * 3, "%d-%d" % (min(results), max(results)),
                 sum(results) / len(results)))
    print("\n  Perfect on every trial at one client. The effect appears the")
    print("  moment a second request can be in flight: a cliff, not a slope.")


def race_routing(trials=20):
    rule("RACE B. sweep the routing policy -- NOT reproducible")
    print("  8 clients, 3 requests each, %d trials per policy. Both of the" % trials)
    print("  defensible policies, under a real race. There is no third one.")
    print("  %-12s %-18s %s" % ("routing", "correct (min-max)", "mean correct"))
    for routing in ("last", "broadcast"):
        results = []
        for _ in range(trials):
            up_port, px_port = ports()
            up = T.Upstream(up_port)
            px = T.NaiveProxy(px_port, up.addr(), routing=routing)
            results.append(score(load(px_port, 8, 3, timeout=0.5))[0])
            px.close(); up.close()
        print("  %-12s %-18s %.1f"
              % (routing, "%d-%d of 24" % (min(results), max(results)),
                 sum(results) / len(results)))
    print("\n  Neither policy works, and the gap between them is noise. The")
    print("  information needed to route was never on the wire.")


def race_pool(trials=20):
    rule("RACE C. the framed pool with no gate at all -- NOT reproducible")
    print("  %-6s %-12s %-8s %s" % ("cap", "conns seen", "wrong", "trials"))
    for cap in (1, 2, 8):
        conns, wrong = set(), 0
        for _ in range(trials):
            up_port, px_port = ports()
            up = T.Upstream(up_port)
            px = T.FramedProxy(px_port, up.addr(), pool_cap=cap)
            wrong += score(load(px_port, 8, 3, timeout=0.5))[1]
            conns.add(up.conns)
            px.close(); up.close()
        print("  %-6d %-12s %-8d %d"
              % (cap, "%d-%d" % (min(conns), max(conns)), wrong, trials))
    print("\n  The connection count varies with the race; the correctness")
    print("  does not. Section 4's gate exists to pin the left column only.")


def main():
    racing = "--race" in sys.argv
    global PORT
    PORT = 8400 if racing else 8340
    for p in range(PORT + 1, PORT + (700 if racing else 40)):
        if not T.free_port(p):
            sys.exit("demo.py: port %d is busy" % p)
    if racing:
        race_boundary()
        race_routing()
        race_pool()
    else:
        interleave()
        misdelivery("last")
        misdelivery("broadcast")
        passthrough()
        framed()
        release_policy()
        boundary()
        half_close()
    print()


if __name__ == "__main__":
    main()