check_name_dominated_suffix.py

#!/usr/bin/env python3
"""Immutable stage/readout audit. Python containers are finite model machinery.

The production argument implements the same bounded transitions with unary
fuel, numeral cursors and persistent trie frames; Python time is not its cost.
"""
from dataclasses import dataclass, replace
from random import Random
from itertools import combinations
import json

@dataclass(frozen=True)
class Row:
    lower: int
    bits: tuple

    @staticmethod
    def make(n):
        return Row(n, tuple(int(x) for x in bin(n)[:1:-1]))

@dataclass(frozen=True)
class Node:
    row: object = None
    left: object = None
    right: object = None


def put(root, name, row):
    path = []
    p = root or Node()
    for bit in name:
        path.append((p, bit))
        p = (p.left if bit == '0' else p.right) or Node()
    p = replace(p, row=row)
    for parent, bit in reversed(path):
        p = replace(parent, **{'left' if bit == '0' else 'right': p})
    return p

@dataclass(frozen=True)
class Worker:
    cap: int
    todo: tuple
    best: object = None
    mode: str = 'tree'
    row: object = None
    cursor: int = 0
    order: int = 0
    spent: int = 0
    done: bool = False

    def step(self):
        assert not self.done
        s = replace(self, spent=self.spent + 1)
        if s.mode == 'tree':
            if not s.todo:
                return replace(s, done=True)
            (node, depth), *rest = s.todo
            todo = tuple(rest)
            if depth:
                # Bounded two-child push. An immutable implementation uses frames.
                for child in (node.right, node.left):
                    if child is not None:
                        todo = ((child, depth - 1),) + todo
            return replace(s, todo=todo, row=node.row, cursor=0,
                           mode='read' if node.row else 'tree')
        if s.mode == 'read':
            if s.cursor == len(s.row.bits):
                if s.best is None:
                    return replace(s, best=s.row, mode='tree')
                return replace(s, mode='compare', cursor=0, order=0)
            if s.cursor == s.cap:
                return replace(s, mode='tree')
            return replace(s, cursor=s.cursor + 1)
        assert s.mode == 'compare'
        i = s.cursor
        a, b = s.row.bits, s.best.bits
        if i == len(a) or i == len(b):
            order = s.order if len(a) == len(b) else (-1 if len(a) < len(b) else 1)
            return replace(s, best=s.row if order < 0 else s.best, mode='tree')
        order = s.order if a[i] == b[i] else (-1 if a[i] < b[i] else 1)
        return replace(s, cursor=i + 1, order=order)

@dataclass(frozen=True)
class Query:
    root: object
    common: int
    c: int = 2
    time: int = 0
    worker: object = None
    candidate: object = None
    remaining: int = 0
    lag: int = 0
    found_at: int = 0
    aligned_at: int = 0
    stages: int = 0
    worker_steps: int = 0

    def tick(self):
        t = self.time + 1
        s = replace(self, time=t)
        if self.candidate is not None:
            if self.lag:
                lag, rem = self.lag + 1, self.remaining
                for _ in range(4):
                    if lag:
                        lag -= 1
                        rem -= 1
                assert rem > 0 or lag == 0
                s = replace(s, lag=lag, remaining=rem,
                            aligned_at=self.aligned_at or (t if not lag else 0))
            else:
                s = replace(s, remaining=self.remaining - 1)
            return s
        if t & (t - 1) == 0:
            assert s.worker is None
            k = t.bit_length() + 3
            todo = ((s.root, k // s.c),) if s.root is not None else ()
            s = replace(s, worker=Worker(k, todo), stages=s.stages + 1)
        worker = s.worker
        if worker:
            before = worker.spent
            for _ in range(512):  # Q=64A, A=8 for the model transitions.
                worker = worker.step()
                if worker.done:
                    break
            s = replace(s, worker_steps=s.worker_steps + worker.spent - before)
            if worker.done:
                bound = 8 * (worker.cap + 1) * 2 ** (worker.cap // s.c + 1)
                assert worker.spent <= bound
                s = replace(s, worker=None)
                if worker.best is not None:
                    lag = max(0, t - 4)
                    rem = worker.best.lower - min(t, 4)
                    assert t < worker.best.lower
                    s = replace(s, candidate=worker.best, lag=lag, remaining=rem,
                                found_at=t, aligned_at=t if not lag else 0)
            else:
                s = replace(s, worker=worker)
        return s

    def answer(self):
        return self.candidate is not None and self.lag == 0 and self.remaining <= 0 and self.time <= self.common


def main():
    rng = Random(189)
    prefixes = stages = steps = forks = delayed = 0
    for trial in range(120):
        c = rng.choice((2, 3, 4))
        root = None
        mapping = {}
        versions = []
        for _ in range(rng.randrange(2, 55)):
            length = rng.randrange(1, 7)
            name = ''.join(rng.choice('01') for _ in range(length))
            b = (1 << (c * length)) + rng.randrange(0, 300)
            # Some payloads remain far longer than every tested stage cap.
            if rng.randrange(7) == 0:
                b = 1 << 100
            mapping[name] = Row.make(b)
            root = put(root, name, mapping[name])
            versions.append((root, dict(mapping)))
        for root, rows in versions[::max(1, len(versions)//3)]:
            common = max(row.lower for row in rows.values()) + 100
            truth = min(row.lower for row in rows.values())
            query = Query(root, common, c)
            saved = []
            for n in range(1, min(common + 2, 1400) + 1):
                if n % 137 == 0:
                    saved.append(query)
                query = query.tick()
                assert query.answer() == (truth <= n <= common)
                prefixes += 1
            if query.aligned_at > query.found_at:
                delayed += 1
            stages += query.stages
            steps += query.worker_steps
            for old in saved:
                newer = old.tick()
                assert old.time + 1 == newer.time
                assert newer.answer() == (truth <= newer.time <= common)
                forks += 1
    # Dense same-length family; a final relevant scan and nontrivial catch-up.
    root = None
    for i in range(128):
        name = format(i, '07b')
        root = put(root, name, Row.make(16384 + i))
    query = Query(root, 17000)
    for n in range(1, 17003):
        query = query.tick()
        assert query.answer() == (16384 <= n <= 17000)
        prefixes += 1
    assert query.aligned_at > query.found_at
    # Producer component update is checked against independent sorted runs.
    producer = 0
    for mask in range(1 << 9):
        widths = {i + 1 for i in range(9) if mask >> i & 1}
        old = sum(j - 1 not in widths for j in widths)
        for n in range(1, 12):
            updated = old if n in widths else old + 1 - int(n - 1 in widths) - int(n + 1 in widths)
            new = widths | {n}
            assert updated == sum(j - 1 not in new for j in new)
            assert (updated == 1) == (len(new) == max(new) - min(new) + 1)
            producer += 1
    # Exact phase inequality, including the first phase and ceiling endpoints.
    for c in range(2, 9):
        for k in range(4, 201):
            w = 8 * (k + 1) * 2 ** (k // c + 1)
            start = 2 ** (k - 4)
            finish = start + (w + 511) // 512 - 1
            assert finish < 2 * start
            if k > 4:
                assert finish + (finish + 2) // 3 < 2 ** (k - 1)
    print(json.dumps(dict(query_prefixes=prefixes, retained_query_forks=forks,
                         stages=stages, worker_steps=steps, delayed_random_alignments=delayed,
                         dense_found_at=query.found_at, dense_aligned_at=query.aligned_at,
                         component_updates=producer, phase_inequalities=7*197), indent=2))

if __name__ == '__main__':
    main()