check_recycled_hole_lanes.py

#!/usr/bin/env python3
"""Finite audit of section 17: retire holes and hand off two envelope readers.

Rows/Hole use the predecessor's immutable oracle primitives. Python tuples,
integers, history logs and numerical comparisons below are audit machinery,
not the claimed bounded native implementation. Cost is normalized to a_r=1.
"""
from dataclasses import dataclass, replace
from itertools import combinations, product
import json
import random
from check_born_hole_lanes import Row, Hole, Receipt, Reader, cost


@dataclass(frozen=True)
class RecycledReceipt(Receipt):
    def global_work(self, steps):
        s = self
        while steps and s.queue:
            debt = s.debt or cost(s.maximum)
            amount = min(debt, steps)
            steps -= amount
            debt -= amount
            if debt:
                s = replace(s, debt=debt)
                continue
            row, tail = s.queue[0], s.queue[1:]
            # Complete all old-position tests before atomic retirement/birth.
            holes = tuple(h for h in s.holes
                          if not h.deleted and not row.contains(h.position))
            births = s.births  # Audit history, not a live native field.
            if row.maximum > s.maximum:
                for m in row.holes:
                    if m > s.maximum:
                        holes += (Hole(m, tail),)
                        births += ((s.maximum, m, len(tail)),)
                        assert len(tail)**4 <= 4096*(s.maximum+1)
                        assert len(tail)*cost(m) <= 64*m
                assert len(holes) <= len(row.holes)
            s = replace(s, maximum=max(s.maximum, row.maximum), queue=tail,
                        debt=0, holes=holes, births=births)
        return s

    def tick(self, global_steps=4, hole_steps=4):
        s = self.global_work(global_steps)
        holes = tuple(h.work(hole_steps) for h in s.holes)
        return replace(s, holes=tuple(h for h in holes if not h.deleted))


@dataclass
class TwoReaders:
    aligned: int
    cursor: int = 0
    pending: int | None = None
    pending_cursor: int = 0
    debt: int = 0
    dirty: bool = False

    def tick(self, n, latest, changed, stats):
        self.cursor += 1
        if self.pending is not None:
            self.debt += 1
            if changed:
                stats['buffered_promotions'] += 1
        self.dirty |= changed
        repairs = 0
        while True:
            if self.pending is None:
                if not self.dirty:
                    break
                self.pending = latest
                self.pending_cursor = 0
                self.debt = n
                self.dirty = False
                stats['reader_captures'] += 1
            take = min(8, self.debt)
            self.pending_cursor += take
            self.debt -= take
            repairs += 1
            if self.debt:
                break
            self.aligned, self.cursor = self.pending, self.pending_cursor
            self.pending = None
            assert self.cursor == n
            stats['reader_adoptions'] += 1
        assert repairs <= 2
        stats['maximum_repair_batches'] = max(stats['maximum_repair_batches'], repairs)
        assert self.cursor == n
        return n <= self.aligned


def query(saved, count, r, stats):
    snapshot = repr(saved)
    s = saved
    envelopes = TwoReaders(s.maximum)
    readers = {h.position: Reader(h.position) for h in s.holes}
    for h in s.holes:
        assert h.residual() <= 96*h.position
    for n in range(1, count+1):
        old_max = s.maximum
        old_positions = set(readers)
        s = s.global_work(128)
        assert not s.queue or s.maximum > 4*n
        stats['progress_checks'] += 1
        envelope = envelopes.tick(n, s.maximum, old_max != s.maximum, stats)
        current = {}
        for h in s.holes:
            fresh = h.position not in old_positions
            reader = Reader(h.position) if fresh else readers[h.position]
            reader.tick(newborn=fresh, position=n)
            if reader.debt:
                assert n < h.position
            current[h.position] = reader
            stats['query_hole_births'] += fresh
        stats['query_retirements'] += len(old_positions-set(current))
        holes = tuple(h.work(384) for h in s.holes)
        s = replace(s, holes=tuple(h for h in holes if not h.deleted))
        readers = {h.position: current[h.position] for h in s.holes}
        assert len(s.holes) <= r
        stats['maximum_live_holes'] = max(stats['maximum_live_holes'], len(s.holes))
        missing = False
        for h in s.holes:
            if h.position == n:
                assert not readers[n].debt and not h.queue
                missing = True
        got = envelope and not missing
        want = any(row.contains(n) for row in saved.selected)
        assert got == want, (n, got, want)
        stats['query_prefixes'] += 1
    assert repr(saved) == snapshot
    stats['retained_query_forks'] += 1


def main():
    stats = dict(query_prefixes=0, retained_query_forks=0, progress_checks=0,
                 query_hole_births=0, query_retirements=0, maximum_live_holes=0,
                 buffered_promotions=0, reader_captures=0, reader_adoptions=0,
                 maximum_repair_batches=0, selected_prefixes=0,
                 maximum_lifetime_births=0, deadline_checks=0)
    rows = [Row(h, e) for h in range(1, 7)
            for k in range(min(2, h-1)+1)
            for e in combinations(range(1, h), k)]
    for x, y in product(rows, repeat=2):
        s = RecycledReceipt().select('0', x).select('1', y)
        query(s, 8, 2, stats)
    rng = random.Random(187)
    for _ in range(100):
        s = RecycledReceipt()
        dictionary = {}
        retained = []
        for i in range(60):
            name = bin(rng.randrange(1, 90))[2:]
            if name not in dictionary:
                h = rng.randrange(1, 130)
                e = tuple(sorted(rng.sample(range(1, h), rng.randrange(min(3, h-1)+1))))
                dictionary[name] = Row(h, e)
            retained.append(s)
            s = s.select(name, dictionary[name])
            assert len(s.holes) <= 3
            stats['selected_prefixes'] += 1
            if i % 10 == 0:
                query(s, 132, 3, stats)
        for old in rng.sample(retained, 3):
            query(old, 132, 3, stats)
    # Arbitrarily extensible rising maxima with two births per promotion.
    s = RecycledReceipt().select('0', Row(4, (1, 3)))
    for i in range(1, 201):
        h = 4+4*i
        s = s.select(bin(i+1)[2:], Row(h, (h-3, h-1)))
        assert len(s.holes) <= 2
        stats['selected_prefixes'] += 1
    stats['maximum_lifetime_births'] = len(s.births)
    query(s, 808, 2, stats)
    # Long numerals give many query-time promotions and buffered handoffs.
    for exponent, total, depth in ((200, 80, 300), (2000, 160, 2800),
                                    (4000, 200, 6500)):
        base = 1 << exponent
        s = RecycledReceipt().select('0', Row(base, (1, 3)))
        for i in range(1, total+1):
            h = base+4*i
            s = s.select(bin(i+1)[2:], Row(h, (h-3, h-1)))
            assert len(s.holes) <= 2
            stats['selected_prefixes'] += 1
        query(s, depth, 2, stats)
        query(s, 65, 2, stats)
    # Ceiling inequalities, including the first-bit case and same-bit handoff.
    def finish(p):
        return p+(max(p-8, 0)+6)//7
    for a in range(1, 20000):
        latest_install = max((finish(p) for p in range(1, (a-1)//4+1)), default=0)
        u = max(latest_install, (a+3)//4)
        assert finish(u) <= a
        stats['deadline_checks'] += 1
    assert stats['maximum_lifetime_births'] > 100
    assert stats['buffered_promotions'] > 0 and stats['maximum_repair_batches'] == 2
    assert stats['query_hole_births'] and stats['query_retirements']
    print(json.dumps(stats, indent=2))


if __name__ == '__main__':
    main()