#!/usr/bin/env python3
"""
DDIA Ch. 10 - Lamport timestamps vs vector clocks over ONE event log.

A Lamport timestamp compresses causality into a single integer + node id, so it
imposes a TOTAL order. That order is safe in one direction only: if a -> b
(a happened-before b) then L(a) < L(b). The CONVERSE is false: L(a) < L(b) does
NOT mean a -> b -- the two events may be CONCURRENT, and Lamport orders them
anyway. A vector clock keeps one counter per process, so it reports concurrency
honestly: neither vector dominates the other.

This script runs BOTH clocks over the SAME fixed event log, then:
  1) prints the event table (Lamport ts + vector ts side by side),
  2) exhibits a CONCURRENT pair that Lamport falsely orders but vectors call
     concurrent,
  3) exhibits a CAUSAL pair where both agree,
  4) shows the MECHANISM: drop Lamport's max-on-receive rule and even the
     one-way guarantee (a -> b => L(a) < L(b)) breaks -- a received message gets
     a LOWER timestamp than its send.

Deterministic: the event log is fixed, so the numbers reproduce exactly.
Pure Python 3 standard library.
"""

# --------------------------------------------------------------------------
# The event log. Three processes P1, P2, P3 (vector indices 0, 1, 2).
# Events are listed in ONE valid causal order (a linear extension): every
# receive appears after the send it delivers, so a single left-to-right pass
# can stamp every event. Independent events across processes keep their own
# per-process counters, so their relative listing order does not affect their
# stamps.
#
#   P1:  e1 local ---- e2 send(m1) ----------------------------------.
#                            \                                        |
#   P2:                       e3 recv(m1) - e4 local - e5 local - e6 send(m2)
#                                                                      \
#   P3:  e7 local - e8 local - e9 local ---------------------- e10 recv(m2)
#
# Note there is NO message path from P3's local run (e7,e8,e9) to P2's local
# run (e4,e5): those events are genuinely concurrent.
# --------------------------------------------------------------------------

PROCS = ["P1", "P2", "P3"]
PIDX = {p: i for i, p in enumerate(PROCS)}

# Each event: (id, process, kind, msg)
#   kind = "local" | "send" | "recv"
#   msg  = message label for "send"/"recv" events, else None
EVENT_LOG = [
    ("e1",  "P1", "local", None),
    ("e2",  "P1", "send",  "m1"),
    ("e3",  "P2", "recv",  "m1"),
    ("e4",  "P2", "local", None),
    ("e5",  "P2", "local", None),
    ("e6",  "P2", "send",  "m2"),
    ("e7",  "P3", "local", None),
    ("e8",  "P3", "local", None),
    ("e9",  "P3", "local", None),
    ("e10", "P3", "recv",  "m2"),
]


def run_lamport(log, use_max_on_receive):
    """Stamp every event with a Lamport timestamp.

    Rules:
      local / send:  L[proc] += 1;                stamp = L[proc]
      recv:          L[proc]  = max(L[proc], msg) + 1   (max-on-receive)
                     -- or, if use_max_on_receive is False, the broken rule
                        L[proc] += 1 (ignore the message's timestamp)

    Returns (stamps, sends) where
      stamps[event_id] = integer Lamport timestamp
      sends[msg_label] = the Lamport timestamp carried by that message
    """
    counter = {p: 0 for p in PROCS}
    stamps = {}
    sends = {}
    for eid, proc, kind, msg in log:
        if kind == "recv":
            incoming = sends[msg]
            if use_max_on_receive:
                counter[proc] = max(counter[proc], incoming) + 1
            else:
                counter[proc] = counter[proc] + 1  # BROKEN: ignore the message
        else:  # local or send
            counter[proc] += 1
        stamps[eid] = counter[proc]
        if kind == "send":
            sends[msg] = counter[proc]
    return stamps, sends


def run_vector(log):
    """Stamp every event with a vector clock [P1, P2, P3].

    Rules:
      local / send:  V[proc][proc] += 1
      recv:          V[proc] = elementwise_max(V[proc], msg_vector);
                     V[proc][proc] += 1

    Returns (stamps, sends) where each stamp is a tuple of 3 ints.
    """
    clock = {p: [0, 0, 0] for p in PROCS}
    stamps = {}
    sends = {}
    for eid, proc, kind, msg in log:
        i = PIDX[proc]
        if kind == "recv":
            incoming = sends[msg]
            clock[proc] = [max(a, b) for a, b in zip(clock[proc], incoming)]
            clock[proc][i] += 1
        else:  # local or send
            clock[proc][i] += 1
        stamps[eid] = tuple(clock[proc])
        if kind == "send":
            sends[msg] = list(stamps[eid])
    return stamps, sends


def vec_str(v):
    return "[" + ",".join(str(x) for x in v) + "]"


def dominates(a, b):
    """True if vector a dominates b: a >= b componentwise AND a != b.
    a dominates b  <=>  b happened-before a."""
    return all(x >= y for x, y in zip(a, b)) and a != b


def relation(a, b):
    """Causal relation between two vector timestamps."""
    if a == b:
        return "equal"
    if dominates(a, b):
        return "a-after-b"    # b -> a
    if dominates(b, a):
        return "b-after-a"    # a -> b
    return "concurrent"       # incomparable


def line(ch="-", n=74):
    return ch * n


def main():
    lam, lam_sends = run_lamport(EVENT_LOG, use_max_on_receive=True)
    vec, vec_sends = run_vector(EVENT_LOG)

    # ---- Event table ----------------------------------------------------
    print(line("="))
    print("ONE event log, two clocks: Lamport timestamp vs vector clock")
    print(line("="))
    print("  P1: e1 local, e2 send(m1)")
    print("  P2: e3 recv(m1), e4 local, e5 local, e6 send(m2)")
    print("  P3: e7 local, e8 local, e9 local, e10 recv(m2)")
    print("  (no message path from P3's e7/e8/e9 to P2's e4/e5)")
    print()
    print(f"  {'event':>5}  {'proc':>4}  {'kind':<9}  {'Lamport':>7}  {'vector [P1,P2,P3]':>17}")
    print(f"  {line('-',5):>5}  {line('-',4):>4}  {line('-',9):<9}  {line('-',7):>7}  {line('-',17):>17}")
    for eid, proc, kind, msg in EVENT_LOG:
        label = kind + (f"({msg})" if msg else "")
        print(f"  {eid:>5}  {proc:>4}  {label:<9}  {lam[eid]:>7}  {vec_str(vec[eid]):>17}")
    print()

    # ---- The concurrent pair Lamport falsely orders --------------------
    A, B = "e9", "e4"   # P3 local (L=3) vs P2 local (L=4)
    print(line())
    print("THE TRAP: a concurrent pair that Lamport puts in a total order")
    print(line())
    print(f"  {A}: {vec_str(vec[A])}  Lamport = {lam[A]}   (P3 local)")
    print(f"  {B}: {vec_str(vec[B])}  Lamport = {lam[B]}   (P2 local)")
    print()
    print(f"  Lamport says  L({A}) = {lam[A]} < {lam[B]} = L({B})")
    print(f"    -> tempting (WRONG) reading: '{A} happened before {B}'.")
    print(f"  Vector clock: {vec_str(vec[A])} vs {vec_str(vec[B])}")
    print(f"    {A} <= {B}?  {'yes' if all(x<=y for x,y in zip(vec[A],vec[B])) else 'no'}"
          f"  ({A}[P3]={vec[A][2]} > {vec[B][2]}={B}[P3])")
    print(f"    {B} <= {A}?  {'yes' if all(x<=y for x,y in zip(vec[B],vec[A])) else 'no'}"
          f"  ({B}[P1]={vec[B][0]} > {vec[A][0]}={A}[P1])")
    print(f"    neither dominates -> vectors report: CONCURRENT")
    print()
    print(f"  Verdict: '{A} before {B}' is FALSE. They are concurrent. Lamport's")
    print(f"  single integer cannot tell 'earlier' from 'concurrent'; the vector can.")
    print()

    # ---- A causal pair where both agree --------------------------------
    C, D = "e2", "e3"   # send m1 -> recv m1
    print(line())
    print("THE CONTRAST: a causal pair (m1 send -> recv) where BOTH agree")
    print(line())
    print(f"  {C}: {vec_str(vec[C])}  Lamport = {lam[C]}   (send m1)")
    print(f"  {D}: {vec_str(vec[D])}  Lamport = {lam[D]}   (recv m1)")
    print(f"  Lamport:  L({C}) = {lam[C]} < {lam[D]} = L({D})   ->  ordered {C} before {D}")
    print(f"  Vector:   {vec_str(vec[D])} dominates {vec_str(vec[C])}  ->  {C} -> {D} (causal)")
    print(f"  Here the total order is RIGHT: {C} really did happen-before {D}.")
    print(f"  (Lamport is safe one way: a -> b  =>  L(a) < L(b). The trap above is")
    print(f"   the converse, which does not hold.)")
    print()

    # ---- Mechanism: drop max-on-receive, the one-way guarantee breaks ---
    broken, broken_sends = run_lamport(EVENT_LOG, use_max_on_receive=False)
    print(line())
    print("THE MECHANISM: why max-on-receive is the minimum for even the one-way rule")
    print(line())
    S, R, MSG = "e2", "e3", "m1"   # send/recv of m1; e3 is P2's first event
    print(f"  Message {MSG}:  sent at {S} (on P1),  received at {R} (on P2).")
    print(f"  Since the send causes the receive, {S} -> {R}, so the one-way")
    print(f"  guarantee REQUIRES L({S}) < L({R}).")
    print()
    print(f"  BROKEN rule (recv just does L += 1, ignoring the message):")
    print(f"    L({S}) = {broken[S]}   L({R}) = {broken[R]}   ->  {broken[S]} "
          f"{'<' if broken[S] < broken[R] else '>=' } {broken[R]}"
          f"   {'ok' if broken[S] < broken[R] else 'VIOLATION: receive <= send'}")
    print(f"    P2's first event {R} is stamped {broken[R]}, but the message it")
    print(f"    delivered was sent at Lamport {broken[S]}. The effect looks EARLIER")
    print(f"    than its cause -- the one thing Lamport promises is broken.")
    print()
    print(f"  FIXED rule (recv does L = max(L, msg) + 1):")
    print(f"    L({S}) = {lam[S]}   L({R}) = {lam[R]}   ->  {lam[S]} < {lam[R]}"
          f"   ok: receive > send, guarantee restored")
    print(f"    max-on-receive drags the receiver's counter past the message's")
    print(f"    timestamp, so no receive can ever undercut its send.")
    print()
    print(line("="))
    print("Bottom line: L(a) < L(b) does NOT imply a -> b (see the trap). The")
    print("compression to a single integer is exactly the causal information the")
    print("vector clock keeps: domination = causality, incomparability = concurrency.")
    print(line("="))


if __name__ == "__main__":
    main()