check_banded_interval_reuse.py

#!/usr/bin/env python3
"""Immutable finite model of banded interval reuse.

Tuple band deques and a query-local collections.deque output FIFO model
imported strict persistent primitives, not their
native running time. Explicit digit, height and continuation records model
worker steps. Integers occur in preparation and independent audit oracles.
"""
from dataclasses import dataclass, replace
from random import Random
from collections import deque
import json


@dataclass(frozen=True)
class Link:
    head: object
    tail: object = None


def digits(n):
    p = None
    for c in bin(n)[2:]:
        p = Link(int(c), p)
    return p


def units(n):
    p = None
    for _ in range(n):
        p = Link(None, p)
    return p


@dataclass(frozen=True)
class Branch:
    left: object
    right: object


# Finite EMPTY/FULL labels; no pointer identity observation.
EMPTY, FULL = False, True


def join(a, b):
    return FULL if a is FULL and b is FULL else Branch(a, b)


@dataclass(frozen=True)
class Task:
    node: object
    lo: object
    hi: object
    height: object


@dataclass(frozen=True)
class Frame:
    phase: str
    node: object = None
    hi: object = None
    height: object = None
    side: int = 0
    tail: object = None


@dataclass(frozen=True)
class Insert:
    phase: str
    old: object
    height: object
    lo_cur: object
    hi_cur: object
    lo_rev: object = None
    hi_rev: object = None
    task: object = None
    stack: object = None
    result: object = None


def start_insert(old, height, lo, hi):
    return Insert('reverse', old, height, lo, hi)


def insert_step(s):
    if s.phase == 'reverse':
        if s.lo_cur is not None:
            return replace(s, lo_cur=s.lo_cur.tail,
                           lo_rev=Link(s.lo_cur.head, s.lo_rev))
        if s.hi_cur is not None:
            return replace(s, hi_cur=s.hi_cur.tail,
                           hi_rev=Link(s.hi_cur.head, s.hi_rev))
        lo = None if s.lo_rev is None else s.lo_rev.tail
        hi = None if s.hi_rev is None else s.hi_rev.tail
        return replace(s, phase='descend', task=Task(s.old, lo, hi, s.height))
    if s.phase == 'descend':
        t = s.task
        if t.node is FULL or (t.lo is None and t.hi is None) or t.height is None:
            return replace(s, phase='return', task=None, result=FULL)
        left = t.node.left if isinstance(t.node, Branch) else EMPTY
        right = t.node.right if isinstance(t.node, Branch) else EMPTY
        lb = 0 if t.lo is None else t.lo.head
        hb = 1 if t.hi is None else t.hi.head
        lo = None if t.lo is None else t.lo.tail
        hi = None if t.hi is None else t.hi.tail
        if lb == hb:
            frame = Frame('single', right if lb == 0 else left,
                          side=lb, tail=s.stack)
            task = Task(left if lb == 0 else right, lo, hi, t.height.tail)
        else:
            frame = Frame('right', right, hi, t.height.tail, tail=s.stack)
            task = Task(left, lo, None, t.height.tail)
        return replace(s, task=task, stack=frame)
    if s.phase == 'return':
        f = s.stack
        if f is None:
            return replace(s, phase='done')
        if f.phase == 'single':
            result = join(s.result, f.node) if f.side == 0 else join(f.node, s.result)
            return replace(s, result=result, stack=f.tail)
        if f.phase == 'right':
            return replace(s, phase='descend', task=Task(f.node, None, f.hi, f.height),
                           stack=Frame('join', s.result, tail=f.tail), result=None)
        return replace(s, result=join(f.node, s.result), stack=f.tail)
    return s


@dataclass(frozen=True)
class Job:
    lower: object
    upper: object
    lo_cur: object
    hi_cur: object
    started: bool = False


@dataclass(frozen=True)
class Bucket:
    height: object = None
    tree: object = EMPTY
    jobs: object = None
    insertion: object = None
    continuation: object = None


@dataclass(frozen=True)
class Worker:
    prefix: tuple = ()
    suffix: tuple = ()


def release(w, lower, upper):
    seq = w.prefix + w.suffix  # Imported persistent catenation in native proof.
    b = seq[0] if seq else Bucket()
    job = Job(lower, upper, lower, upper)
    b = replace(b, jobs=Link(job, b.jobs))
    return Worker((), (b,) + seq[1:])


def finish(w, b, cont):
    tail = w.suffix[1:]
    if cont is not None:
        nxt = tail[0] if tail else Bucket(Link(None, b.height))
        nxt = replace(nxt, jobs=Link(cont, nxt.jobs))
        tail = (nxt,) + tail[1:]
    return Worker(w.prefix, (b,) + tail)


def service(w):
    if not w.suffix:
        return w
    b = w.suffix[0]
    if b.insertion is not None:
        ins = insert_step(b.insertion)
        if ins.phase == 'done':
            new = replace(b, tree=ins.result, insertion=None, continuation=None)
            return 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, rest = b.jobs.head, b.jobs.tail
    lo = None if job.lo_cur is None else job.lo_cur.tail
    hi = job.hi_cur.tail
    began = job.started or lo is None
    cont = None if hi is None else replace(job, lo_cur=lo, hi_cur=hi, started=began)
    new = replace(b, jobs=rest)
    if not began:
        return finish(w, new, cont)
    lower = job.lower if not job.started else None
    upper = job.upper if hi is None else None
    if b.tree is FULL or (lower is None and upper is None):
        return finish(w, replace(new, tree=FULL), cont)
    ins = start_insert(b.tree, b.height, lower, upper)
    return Worker(w.prefix, (replace(new, insertion=ins, continuation=cont),) + w.suffix[1:])


def settled(w, k):
    # Whole-state audit only. Native query consumes the emitted prefix FIFO.
    return len(w.prefix) >= k or not w.suffix


def drain_to(w, k):
    count = 0
    while not settled(w, k):
        w = service(w)
        count += 1
        assert count < 1000000
    return w, count


def tree_bits(node, height):
    # Independent whole-tree audit, outside the worker.
    if isinstance(node, bool):
        return [node] * (1 << height)
    return tree_bits(node.left, height-1) + tree_bits(node.right, height-1)


@dataclass(frozen=True)
class Reader:
    stack: object = None
    emission: object = None


def read_step(r):
    if r.stack is None:
        return r
    (node, height), tail = r.stack.head, r.stack.tail
    if height is None:
        return Reader(tail, node is FULL)
    left = node.left if isinstance(node, Branch) else node
    right = node.right if isinstance(node, Branch) else node
    stack = Link((left, height.tail), Link((right, height.tail), tail))
    return Reader(stack)


def query(w, count, worker_rate=262144, reader_rate=1024):
    ready, pending = w.prefix, Worker((), w.suffix)
    reader, answers, output = Reader(), [], deque()
    # Actual native service uses fixed batches. Stopping an idle worker early
    # below only omits no-ops, and does not transfer unused credit.
    for _ in range(count):
        for _ in range(worker_rate):
            if not pending.suffix:
                break
            pending = service(pending)
            if pending.prefix:
                ready += pending.prefix
                pending = Worker((), pending.suffix)
        for _ in range(reader_rate):
            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 = 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 oracle(rows, n):
    return any(b <= n <= h for b, h in rows)


def main():
    rng = Random(198)
    cases = queries = forks = release_bounds = insertions = step_states = 0
    # Every interval in small bands, retaining old coverage after each insert.
    for k in range(1, 6):
        a, z = 1 << (k-1), (1 << k)-1
        tree, covered = EMPTY, set()
        intervals = [(b,h) for b in range(a,z+1) for h in range(b,z+1)]
        rng.shuffle(intervals)
        for b,h in intervals:
            ins = start_insert(tree, units(k-1), digits(b), digits(h)); cost=0
            while ins.phase != 'done':
                ins=insert_step(ins);cost+=1
            assert cost <= 16*k+16
            tree=ins.result;covered.update(range(b,h+1));insertions+=1
            assert tree_bits(tree,k-1)==[n in covered for n in range(a,z+1)]
    for trial in range(150):
        w, rows, retained = Worker(), [], []
        for j in range(rng.randrange(3,16)):
            # Interleaving includes gaps, nested intervals, equal endpoints,
            # short late rows and enormous unresolved upper numerals.
            b=rng.randrange(1,85);h=b+rng.randrange(0,90)
            if j%7==0:h=(1<<rng.randrange(150,600))+h
            if j%5==4:b=h=rng.randrange(1,8)
            for _ in range(rng.randrange(0,50)):w=service(w);step_states+=1
            k=rng.randrange(1,9)
            _,before=drain_to(w,k)
            nxt=release(w,digits(b),digits(h))
            _,after=drain_to(nxt,k)
            assert after <= before+256*k
            release_bounds+=1
            if j%3==0:retained.append((w,tuple(rows)))
            w=nxt;rows.append((b,h))
            if j%4==0:
                out=query(w,96);queries+=len(out)
                assert out==[oracle(rows,n) for n in range(1,97)]
            cases+=1
        for old,oldrows in retained:
            out=query(old,96);queries+=len(out);forks+=1
            assert out==[oracle(oldrows,n) for n in range(1,97)]
    # Interrupt each phase of an actual large boundary insertion with a short
    # disjoint row; resume and check every retained predecessor independently.
    w=release(Worker(),digits(71),digits(119));rows=[(71,119)]
    interrupted=0
    while w.suffix:
        if w.suffix[0].insertion is not None:
            new=release(w,digits(2),digits(2))
            assert query(new,128)==[oracle(rows+[(2,2)],n) for n in range(1,129)]
            assert query(w,128)==[oracle(rows,n) for n in range(1,129)]
            interrupted+=1;queries+=256
        w=service(w)
    # Many separated components survive, unlike minimum-only cancellation.
    rows=[(2*i,2*i) for i in range(1,65)];rng.shuffle(rows);w=Worker()
    for b,h in rows:w=release(w,digits(b),digits(h))
    out=query(w,140);assert out==[oracle(rows,n) for n in range(1,141)];queries+=140
    assert oracle([(1,1),(4,4)],4) and not oracle([(1,1)],4)
    # Exact integer checks of fixed-rate deadline bounds; no float oracle.
    for k in range(4,513):
        C,q,Q=256*k,16384,262144
        B=C+sum((1<<d)*(C-q*d) for d in range(1,(C-1)//q+1))
        assert B<Q*(1<<(k-4))
    print(json.dumps(dict(interval_insertions=insertions,releases=cases,
        release_increase_checks=release_bounds,retained_query_forks=forks,
        query_prefixes=queries,boundary_phase_interruptions=interrupted,
        random_service_states=step_states,deadline_inequalities=509)))


if __name__=='__main__':main()