#!/usr/bin/env python3
"""A brute-force linearizability checker rejects a history that "looks consistent", DDIA 2e Ch. 10.

A register x starts at 0. We record a HISTORY of operations on it -- each op is
(process, kind in {write, read, cas}, arg(s), returned value, start_time,
end_time). The checker searches for SOME sequential (total) order of the ops
that (a) respects real time -- if op X finishes before op Y starts, X must come
before Y -- and (b) is a legal register trace -- each read returns the current
value, and cas(old, new) succeeds only when the current value is old.

The surprise: History 1 is DDIA's Figure 10-4. Every read returns a value that
was genuinely written, and no single operation contradicts another. A careful
reader calls it consistent. The checker calls it NOT linearizable -- because
client A already read the new value 4 before client B's read began, so B is
forbidden from returning the older 2. That violation is a cross-client RECENCY
constraint that is invisible if you only ask, op by op, "did this read return a
value that was written?"

History 2 is the same ops with B's last read returning 4 -- and the checker
finds a legal order, proving it is not a rubber stamp. History 3 is Figure 10-6:
a quorum read with w + r > n that still returns stale, and it fails too.

Pure standard library, deterministic -> reproducible.
"""
import itertools
from collections import namedtuple

# An operation in a recorded history. Times are integers on a shared wall clock.
#   kind == 'write': arg = value written,        ret = 'ok'
#   kind == 'read' : arg = None,                 ret = value the read returned
#   kind == 'cas'  : arg = (old, new),           ret = 'ok' | 'fail'
Op = namedtuple("Op", ["proc", "kind", "arg", "ret", "start", "end"])


def w(proc, value, start, end):
    return Op(proc, "write", value, "ok", start, end)


def r(proc, value, start, end):
    return Op(proc, "read", None, value, start, end)


def cas(proc, old, new, ret, start, end):
    return Op(proc, "cas", (old, new), ret, start, end)


# --- The checker ---------------------------------------------------------

def check_linearizable(history, initial=0):
    """Brute-force search for a legal sequential order.

    Returns (True, witness_order) if some total order of `history` respects the
    real-time partial order AND replays as a legal register trace; otherwise
    (False, None). Exponential in len(history) -- only tractable for small
    histories, which is the whole point of the boundary discussion.
    """
    n = len(history)

    # Real-time partial order: op i must precede op j if i finishes strictly
    # before j starts. Overlapping ops are unordered (either may take effect
    # first), which is exactly the freedom a linearization is allowed to use.
    must_before = [[history[i].end < history[j].start for j in range(n)]
                   for i in range(n)]

    for perm in itertools.permutations(range(n)):
        pos = {op_i: p for p, op_i in enumerate(perm)}

        # (a) respect real time: never place i after j when i must precede j.
        if any(must_before[i][j] and pos[i] > pos[j]
               for i in range(n) for j in range(n)):
            continue

        # (b) replay the order as a register trace and check every op is legal.
        val = initial
        legal = True
        for op in (history[k] for k in perm):
            if op.kind == "write":
                val = op.arg
            elif op.kind == "read":
                if op.ret != val:
                    legal = False
                    break
            else:  # cas(old, new)
                old, new = op.arg
                succeeds = (val == old)
                if op.ret == "ok":
                    if not succeeds:
                        legal = False
                        break
                    val = new
                else:  # recorded as failing: it must actually fail here
                    if succeeds:
                        legal = False
                        break
        if legal:
            return True, [history[k] for k in perm]

    return False, None


# --- Diagnosis: why a rejected history fails -----------------------------

def _version_ranks(history, initial=0):
    """Rank each value by how late it is in the write/cas version chain.

    initial value has rank 0; a successful cas(old, new) makes `new` one step
    newer than `old`; a write advances from the initial value. Rank lets us say
    "4 is newer than 2" and spot a read that travelled backwards in time.
    """
    values = {initial}
    edges = {}  # value -> set of strictly-newer values
    for op in history:
        if op.kind == "write":
            values.add(op.arg)
            if op.arg != initial:
                edges.setdefault(initial, set()).add(op.arg)
        elif op.kind == "cas":
            old, new = op.arg
            values.update((old, new))
            if op.ret == "ok":
                edges.setdefault(old, set()).add(new)
        else:
            values.add(op.ret)

    rank = {v: (0 if v == initial else None) for v in values}
    changed = True
    while changed:  # longest-path relaxation over the (acyclic) version graph
        changed = False
        for u, succs in edges.items():
            if rank.get(u) is None:
                continue
            for v in succs:
                if rank.get(v) is None or rank[v] < rank[u] + 1:
                    rank[v] = rank[u] + 1
                    changed = True
    return rank


def recency_violation(history, initial=0):
    """Find a completed read of a newer value that forbids a later, staler read.

    Returns (earlier_read, later_read, rank_earlier, rank_later) or None.
    """
    ranks = _version_ranks(history, initial)
    reads = [op for op in history if op.kind == "read"]
    for a in reads:
        for b in reads:
            if a.end < b.start:  # a finished before b even started
                ra, rb = ranks.get(a.ret), ranks.get(b.ret)
                if ra is not None and rb is not None and ra > rb:
                    return a, b, ra, rb
    return None


# --- Formatting ----------------------------------------------------------

def _op_str(op):
    if op.kind == "write":
        return f"{op.proc}: write({op.arg}) = ok"
    if op.kind == "read":
        return f"{op.proc}: read() = {op.ret}"
    old, new = op.arg
    return f"{op.proc}: cas({old}, {new}) = {op.ret}"


def print_history(history):
    for op in history:
        print(f"    {_op_str(op):<24} [t {op.start:>2}..{op.end:<2}]")


def print_replay(order, initial=0):
    val = initial
    for i, op in enumerate(order, 1):
        if op.kind == "write":
            print(f"    {i}. {op.proc}: write({op.arg})=ok     x: {val} -> {op.arg}")
            val = op.arg
        elif op.kind == "read":
            print(f"    {i}. {op.proc}: read()={op.ret}        x = {val}  (matches)")
        else:
            old, new = op.arg
            if op.ret == "ok":
                print(f"    {i}. {op.proc}: cas({old},{new})=ok     x: {val} -> {new}")
                val = new
            else:
                print(f"    {i}. {op.proc}: cas({old},{new})=fail   x = {val}  "
                      f"(current {val} != {old}, cas fails)")


def report(title, note, history):
    print("=" * 60)
    print(title)
    print("=" * 60)
    print_history(history)
    if note:
        print()
        print(note)
    print()

    ok, witness = check_linearizable(history)
    if ok:
        print("  VERDICT: linearizable.")
        print("  A legal, real-time-respecting order exists. Replaying it on x (start 0):")
        print_replay(witness)
    else:
        print("  VERDICT: NOT linearizable.")
        v = recency_violation(history)
        if v:
            a, b, ra, rb = v
            print(f"  Reason (real-time recency): {a.proc}'s read returned {a.ret} "
                  f"(version rank {ra}) and finished at t={a.end};")
            print(f"  {b.proc}'s read began later at t={b.start} but returned {b.ret} "
                  f"(version rank {rb}).")
            print(f"  Once a completed read has observed the newer value, no later read may")
            print(f"  return an older one -- and no write reverts it back. No sequential")
            print(f"  order can respect both real time and the register semantics.")
        else:
            print("  Reason: no total order respects both real time and register semantics.")
    print()


def main():
    print("Linearizability checker -- brute-force search for a legal sequential order.")
    print("Register x starts at 0. A read must return the current value; cas(old,new)")
    print("succeeds only when the current value is old. A valid order must also respect")
    print("real time: if op X finishes before op Y starts, X must come before Y.")
    print()

    # --- History 1: DDIA Figure 10-4. x moves 0 -> 1 -> 2 -> 4 via cas. -----
    # Client A reads 1, then 2, then 4. After A has read 4 (finishing at t=17),
    # client B's read STARTS at t=18 and returns the older 2.
    fig_10_4 = [
        cas("C", 0, 1, "ok",   1, 4),
        r("A", 1,               5, 7),
        cas("B", 1, 2, "ok",   6, 9),
        r("A", 2,              10, 12),
        cas("D", 2, 4, "ok",  11, 14),
        cas("D", 2, 3, "fail", 13, 16),
        r("A", 4,              15, 17),
        r("B", 2,              18, 20),
    ]
    report(
        "History 1 -- DDIA Figure 10-4  (x: 0 -> 1 -> 2 -> 4 via cas)",
        "  Every read returned a value that was genuinely written (0,1,2,4 all appear),\n"
        "  and no single operation contradicts another on its own.",
        fig_10_4,
    )

    # --- History 2: control. Same ops, but B's last read returns 4, not 2. ---
    control = fig_10_4[:-1] + [r("B", 4, 18, 20)]
    report(
        "History 2 -- control: identical ops, but B's last read returns 4",
        "  The only change from History 1 is B's final read: 4 instead of 2.",
        control,
    )

    # --- History 3: DDIA Figure 10-6. Quorum read, w + r > n, still stale. ---
    # A long write of 1 overlaps two reads. B (earlier) reads the new 1;
    # C (strictly later) reads the old 0.
    fig_10_6 = [
        w("W", 1,   1, 12),
        r("B", 1,   4, 6),
        r("C", 0,   8, 10),
    ]
    report(
        "History 3 -- DDIA Figure 10-6  (quorum read, w + r > n, still stale)",
        "  A quorum write of 1 overlaps both reads. B reads the new value, C the old --\n"
        "  yet C's read starts after B's finished.",
        fig_10_6,
    )


if __name__ == "__main__":
    main()