check_dyadic_minimum_frontier.py

#!/usr/bin/env python3
"""Ideal ranked-frontier audit, not a native priority implementation.

Integer ranks, tuple selection/copying and cancellation tests are explicit
frontier oracles. Endpoint comparisons use counted least-significant digits.
The model charges promotion work but does not implement coherent root repair.
"""
from dataclasses import dataclass, replace
from random import Random
import json


@dataclass(frozen=True)
class Task:
    name: str
    value: int
    rank: int = 0
    attained: int = 0
    mode: str = 'screen'
    repair: int = 0
    left: int = 0
    right: int = 0
    order: int = 0
    base: int = 0


@dataclass(frozen=True)
class State:
    tasks: tuple = ()
    best: int | None = None
    best_rank: int | None = None
    steps: int = 0

    def release(self, name, value):
        return replace(self, tasks=self.tasks + (Task(name, value),))

    def step(self, kappa=3):
        if not self.tasks:
            return self
        # Oracle: minimum rank and continuation of same-rank comparisons.
        i = min(range(len(self.tasks)), key=lambda i:
                (self.tasks[i].rank, self.tasks[i].mode not in ('compare', 'ready'), i))
        t = self.tasks[i]
        best, br = self.best, self.best_rank
        nxt = t
        if t.mode == 'screen':
            a = t.attained + 1
            if a == t.value.bit_length():
                nxt = replace(t, attained=a, mode='ready')
            elif a == (1 << (t.rank+1))-1:
                nxt = replace(t, attained=a, mode='repair', repair=kappa*(t.rank+1))
            else:
                nxt = replace(t, attained=a)
        elif t.mode == 'repair':
            nxt = replace(t, repair=t.repair-1)
            if nxt.repair == 0:
                nxt = replace(nxt, rank=t.rank+1, mode='screen')
        elif t.mode == 'ready':
            if br is None or t.rank < br:
                best, br, nxt = t.value, t.rank, None
            elif br < t.rank:
                nxt = None
            else:
                nxt = replace(t, mode='compare', left=t.value, right=best, base=best)
        else:
            assert t.mode == 'compare'
            if br < t.rank:
                nxt = None  # Lower class makes both old operands irrelevant.
            else:
                # Assertion checks the invariant; it is not an algorithmic
                # address, identity or equality observation.
                assert br == t.rank and best == t.base
                a, b = t.left & 1, t.right & 1
                order = (1 if a > b else -1) if a != b else t.order
                left, right = t.left >> 1, t.right >> 1
                if left == right == 0:
                    best = t.value if order < 0 else t.base
                    nxt = None
                else:
                    nxt = replace(t, left=left, right=right, order=order)
        tasks = self.tasks[:i] + (() if nxt is None else (nxt,)) + self.tasks[i+1:]
        return State(tasks, best, br, self.steps+1)

    def serve(self, n):
        s = self
        for _ in range(n):
            if not s.tasks:
                break
            s = s.step()
        return s

    def drain(self, rank):
        s = self
        while any(t.rank <= rank for t in s.tasks):
            s = s.step()
        return s


def cap_work(j, kappa=3):
    # All digit blocks, all promotion repairs, one ready transition,
    # one comparison plus its possible cancellation transition.
    return 2*((1 << (j+1))-1) + kappa*(j+1)*(j+2)//2 + 3


def supply_bound(c, q):
    return c + sum((1 << d)*(c-q*d) for d in range(1, (c-1)//q+1))


def main():
    rng = Random(193)
    checks = drains = answers = interruptions = 0
    # A is a generous joint per-job linear envelope, including constants.
    A, q = 12, 8
    phase_checks = 0
    for k in range(4, 513):
        j = k.bit_length()-1
        assert (1 << (j+1))-1 >= k
        assert supply_bound(cap_work(j), 64*A) < 1024*A*(1 << (k-4))
        phase_checks += 1
    for j in range(80):
        assert cap_work(j) <= A*(1 << (j+1))
        checks += 1
    for trial in range(70):
        s = State()
        rows = []
        saved = []
        names = [bin(i+2)[2:] for i in range(40)]
        rng.shuffle(names)
        for name in names:
            value = rng.choice([2, 3, 7, 9, 17, 1 << rng.randrange(5, 240)])
            s = s.serve(q*len(name)).release(name, value)
            rows.append(value)
            if rng.randrange(4) == 0:
                saved.append((s, tuple(rows)))
        for old, values in saved:
            for j in range(5):
                t = old.drain(j)
                assert t.steps-old.steps <= supply_bound(cap_work(j), q)
                cutoff = 1 << ((1 << (j+1))-1)
                for n in (1, 2, 3, 6, cutoff-1):
                    if n >= cutoff:
                        continue
                    assert (t.best is not None and t.best <= n) == (min(values) <= n)
                    answers += 1
                assert all(x.attained >= (1 << (j+1))-1 for x in t.tasks if x.mode == 'screen')
                drains += 1
    # Interrupt a live same-class comparison at every digit position. Lower
    # classes cancel it; equal-class arrivals must wait for its commit.
    base, candidate = (1 << 30)+27, (1 << 30)+19
    init = State(best=base, best_rank=4).release('0', candidate)
    while not any(t.mode == 'compare' for t in init.tasks):
        init = init.step()
    for delay in range(31):
        old = init.serve(delay)
        for value in (2, (1 << 30)+3, (1 << 31)+1):
            s = old.release('1', value).drain(6)
            assert s.best == min(base, candidate, value)
            assert old.best == base
            interruptions += 1
    # Many equal-small arrivals and retained versions do not need a completed
    # set scan; the one saved best is the actual reduced value.
    s = State().release('0', 1 << 10000)
    for i in range(2048):
        name = '1'+format(i, '011b')
        s = s.serve(q*len(name)).release(name, 2)
    s = s.drain(1)
    assert s.best == 2
    # Symbolic nonpreemptible repair witness: the rank is unbounded while
    # service from the late short name through its short query is fixed.
    repair_witnesses = []
    for deadline_service in range(1, 33):
        rank = deadline_service + 1
        remaining_repair = rank
        assert remaining_repair > deadline_service
        repair_witnesses.append((rank, deadline_service))
    # Literal stale root replacement loses a post-snapshot arrival.
    old = ((), (), ('old',))
    arrival = (('new-small',),) + old[1:]
    rebuilt = ((), (), (), ('old',))
    assert 'new-small' in sum(arrival, ())
    assert 'new-small' not in sum(rebuilt, ())
    # Epoch concatenation represented as ((old,new),new) has a linear oldest
    # descent even though construction allocates only one branch per merge.
    tree = 'old'
    for i in range(1000):
        tree = (tree, i)
    depth = 0
    while isinstance(tree, tuple):
        tree = tree[0]
        depth += 1
    assert depth == 1000
    print(json.dumps(dict(linear_envelopes=checks, retained_rank_drains=drains,
        threshold_answers=answers, comparison_interruptions=interruptions,
        phase_inequalities=phase_checks, stress_service_per_name_bit=q,
        equal_small_releases=2048, deep_repair_witnesses=len(repair_witnesses),
        stale_root_loses_arrival=True,
        concatenation_oldest_depth=depth, native_frontier_implemented=False), indent=2))


if __name__ == '__main__':
    main()