check_scale_sparse_band_reuse.py

#!/usr/bin/env python3
"""Finite audit of scale-sparse band production and reuse.

Tuple deques and integer clock flags model inherited native primitives.
They are not measurements of Python/native complexity. Producer width events
stand for the separately proved distinct-width dictionary; complete binary
levels are checked independently below. Persistent unit cursors, delimiter
updates and resumable consumer insertion are explicit. Calls to digits(n) at
a delimiter stand for inherited numerals already prepared during the word;
this model does not audit their construction. Complete-level cases check
count/set semantics, not the inherited capacity-counter implementation.
"""
from dataclasses import dataclass, replace
from random import Random
from collections import deque
import json
import check_banded_interval_reuse as base
from check_banded_interval_reuse import Link, units, digits, Bucket, Worker, Reader


def size(s):
    n = 0
    while s is not None:
        n += 1
        s = s.tail
    return n


def pop(s):
    assert s is not None
    return s.tail


def delta(s, d):
    return Link(None, s) if d == 1 else pop(s) if d == -1 else s


def over(s, bound):
    # At most bound+1 observations; bound is a fixed machine parameter.
    for _ in range(bound+1):
        if s is None:
            return False
        s = s.tail
    return True


@dataclass(frozen=True)
class Record:
    flag: str = 'EMPTY'
    missing: object = None
    runs: object = None
    low: object = None
    high: object = None
    low_offset: object = None
    high_offset: object = None


@dataclass(frozen=True)
class Row:
    bands: tuple = ()
    groups: tuple = ()
    bad_groups: object = None
    bad_runs: object = None
    widths: frozenset = frozenset()  # Imported dictionary, not worker lookup.


def produce(row, n, r):
    assert n > 0
    P, S, GP, GS = (), row.bands, (), row.groups
    prefix = offset = lc = hc = None
    j = 0
    for position in range(1, n+1):
        prefix = Link(None, prefix)
        if position & (position-1) == 0:  # Inherited native counter flag.
            if j:
                P, S = P+(S[0],), S[1:]
            j += 1
            if not S:
                S = (Record(missing=prefix),)
            if j & (j-1) == 0:
                if j > 1:
                    GP, GS = GP+(GS[0],), GS[1:]
                if not GS:
                    GS = (None,)
            offset, lc, hc = None, S[0].low_offset, S[0].high_offset
        else:
            offset = Link(None, offset)
            lc = None if lc is None else lc.tail
            hc = None if hc is None else hc.tail
    old = rec = S[0]
    first = n not in row.widths
    if first:
        missing = pop(old.missing)
        pred = n-1 in row.widths and n & (n-1) != 0
        succ = n+1 in row.widths and n & (n+1) != 0
        runs = delta(old.runs, 1-int(pred)-int(succ))
        rec = replace(old, missing=missing, runs=runs,
                      flag='FULL' if missing is None else 'PARTIAL')
    if old.flag == 'EMPTY' or lc is not None:
        rec = replace(rec, low=digits(n), low_offset=offset)
    if old.flag == 'EMPTY' or hc is None:
        rec = replace(rec, high=digits(n), high_offset=offset)
    d = int(rec.flag == 'PARTIAL')-int(old.flag == 'PARTIAL')
    group = delta(GS[0], d)
    bad_groups = delta(row.bad_groups, int(over(group,r))-int(over(GS[0],r)))
    bad_runs = delta(row.bad_runs, int(over(rec.runs,1))-int(over(old.runs,1)))
    return Row(P+(rec,)+S[1:], GP+(group,)+GS[1:], bad_groups,
               bad_runs, row.widths | {n})


def number(p):
    n, weight = 0, 1
    while p is not None:
        n += p.head*weight
        p, weight = p.tail, weight*2
    return n


def audit_row(row, r):
    counts = [0]*len(row.groups)
    bad_runs = 0
    for j, rec in enumerate(row.bands, 1):
        a, z = 1 << (j-1), (1 << j)-1
        points = sorted(row.widths & set(range(a,z+1)))
        flag = 'EMPTY' if not points else 'FULL' if len(points)==a else 'PARTIAL'
        assert rec.flag == flag
        assert size(rec.missing) == a-len(points)
        runs = sum(x-1 not in points for x in points)
        assert size(rec.runs) == runs
        bad_runs += runs > 1
        counts[j.bit_length()-1] += flag == 'PARTIAL'
        if points:
            assert number(rec.low) == points[0] and number(rec.high) == points[-1]
            assert size(rec.low_offset) == points[0]-a
            assert size(rec.high_offset) == points[-1]-a
    assert [size(g) for g in row.groups] == counts
    assert size(row.bad_groups) == sum(c > r for c in counts)
    assert size(row.bad_runs) == bad_runs
    return bad_runs == 0 and all(c <= r for c in counts)


@dataclass(frozen=True)
class Cursor:
    records: tuple


def release(w, row):
    seq = w.prefix+w.suffix
    b = seq[0] if seq else Bucket()
    b = replace(b, jobs=Link(Cursor(row.bands), b.jobs))
    return Worker((), (b,)+seq[1:])


def service(w):
    if not w.suffix:
        return w
    b = w.suffix[0]
    if b.insertion is not None:
        ins = base.insert_step(b.insertion)
        if ins.phase == 'done':
            new = replace(b, tree=ins.result, insertion=None, continuation=None)
            return base.finish(w, new, b.continuation)
        return Worker(w.prefix, (replace(b, insertion=ins),)+w.suffix[1:])
    if b.jobs is None:
        return Worker(w.prefix+(b,), w.suffix[1:])
    job = b.jobs.head
    rec, tail = job.records[0], job.records[1:]
    cont = Cursor(tail) if tail else None
    new = replace(b, jobs=b.jobs.tail)
    if rec.flag == 'EMPTY':
        return base.finish(w, new, cont)
    if rec.flag == 'FULL' or b.tree is True:
        return base.finish(w, replace(new, tree=True), cont)
    ins = base.start_insert(b.tree,b.height,rec.low,rec.high)
    return Worker(w.prefix,(replace(new,insertion=ins,continuation=cont),)+w.suffix[1:])


def drain(w,k):
    steps = 0
    while len(w.prefix)<k and w.suffix:
        w = service(w)
        steps += 1
        assert steps < 1000000
    return steps


def query(w,count,r):
    ready, pending = w.prefix, Worker((),w.suffix)
    reader, output, answers = Reader(), deque(), []
    for _ in range(count):
        for _ in range(1024*256*(r+1)):
            if not pending.suffix:
                break
            pending = service(pending)
            if pending.prefix:
                ready += pending.prefix
                pending = Worker((),pending.suffix)
        for _ in range(1024):
            if reader.stack is None:
                if not ready:
                    break
                b, ready = ready[0], ready[1:]
                reader = replace(reader,stack=Link((b.tree,b.height)))
            reader = base.read_step(reader)
            if reader.emission is not None:
                output.append(reader.emission)
        if output:
            answers.append(output.popleft())
        else:
            assert not ready and not pending.suffix and reader.stack is None
            answers.append(False)
    return answers


def valid_row(rng,r,max_j=7):
    points, partial = [], {}
    for j in range(1,max_j+1):
        a,z = 1 << (j-1),(1 << j)-1
        group = j.bit_length()
        kind = rng.randrange(3)
        if kind == 0:
            continue
        if kind == 1 or partial.get(group,0) >= r:
            lo,hi = a,z
        else:
            lo = rng.randrange(a,z+1)
            hi = rng.randrange(lo,z+1)
            if (lo,hi)!=(a,z):
                partial[group] = partial.get(group,0)+1
        points.extend(range(lo,hi+1))
    if not points:
        points = [1]
    rng.shuffle(points)
    row = Row()
    for n in points:
        row = produce(row,n,r)
    assert audit_row(row,r)
    return row


def main():
    rng = Random(199)
    producer_steps = retained = releases = prefixes = inequalities = 0
    # Arbitrary order, duplicates, temporary invalidity, and repair.
    for r in range(4):
        for trial in range(35):
            row, versions = Row(), []
            order = list(range(1,64)) + [rng.randrange(1,64) for _ in range(30)]
            rng.shuffle(order)
            for n in order:
                if rng.randrange(8)==0:
                    versions.append(row)
                row = produce(row,n,r)
                audit_row(row,r)
                producer_steps += 1
            assert audit_row(row,r)
            for old in versions:
                audit_row(old,r)
                retained += 1
    # Independent complete-level semantics, including duplicates.
    level_cases = 0
    for mask in range(1 << 6):
        words = ['0','1','00','01','10','11']
        selected = {w for i,w in enumerate(words) if mask >> i & 1}
        counts = {n:sum(len(w)==n for w in selected) for n in (1,2)}
        gate = all(c in (0,1 << n) for n,c in counts.items())
        direct = all({w for w in words if len(w)==len(v)} <= selected for v in selected)
        assert gate == direct
        assert selected | selected == selected
        level_cases += 1
    # Generalized consumer with interrupted releases and retained query forks.
    for r in range(4):
        for trial in range(15):
            w, covered, versions = Worker(), set(), []
            for _ in range(9):
                row = valid_row(rng,r)
                for _ in range(rng.randrange(40)):
                    w = service(w)
                k = rng.randrange(1,9)
                before = drain(w,k)
                new = release(w,row)
                assert drain(new,k) <= before+256*(r+1)*k
                versions.append((w,frozenset(covered)))
                w = new
                covered.update(row.widths)
                out = query(w,140,r)
                assert out == [n in covered for n in range(1,141)]
                prefixes += len(out)
                releases += 1
            for old,points in versions[::3]:
                assert query(old,140,r) == [n in points for n in range(1,141)]
                prefixes += 140
                retained += 1
    # Unbounded-family witness at increasingly many scales (finite samples).
    for t in range(1,12):
        js = [1 << a for a in range(1,t+1)]
        for k in js:
            assert sum(j+1 for j in js if j<=k) < 4*k
    for r in range(4):
        for k in range(4,513):
            c = 256*(r+1)
            C,q,Q = c*k,64*c,1024*c
            B = C+sum((1<<d)*(C-q*d) for d in range(1,(C-1)//q+1))
            assert B < Q*(1 << (k-4))
            inequalities += 1
    print(json.dumps(dict(producer_updates=producer_steps,retained_forks=retained,
        complete_level_cases=level_cases,releases_and_envelope_checks=releases,
        query_prefixes=prefixes,deadline_inequalities=inequalities)))


if __name__ == '__main__':
    main()