Research source · python

check_mixed_region_budget.py

tools/research/check_mixed_region_budget.py

101 lines. Source is displayed for inspection; it is not executed by this page.

File fingerprint

SHA-256: ea2f7d419846e9f62b0b5c4a459961e5a51c1d71a63bf2524c0f0f4957f6d148

#!/usr/bin/env python3
"""Independent finite audit of reader32.7's streamed signed mixed-region budget.
Uses the immutable row model; a separate integer-set recursion counts regions.
"""
from dataclasses import dataclass
from itertools import permutations
from collections import Counter
import random,json
import check_interval_boundary_composition as I
C=Counter()
@dataclass(frozen=True)
class Signed:
    sign:int=0
    units:tuple=()
    def step(self,delta):
        assert delta in (-1,0,1)
        if not delta:return self
        if not self.sign:return Signed(delta,(None,))
        if delta==self.sign:return Signed(self.sign,(None,)+self.units)
        q=self.units[1:]
        return Signed(self.sign if q else 0,q)
    def value(self):return self.sign*len(self.units)
def charge(budget,root,w):
    baseline=candidate=budget
    for bit in w:
        full,_,_,z,o=I.view(root)
        mixed=not isinstance(root,I.Empty) and not full
        other=z if bit=='1' else o
        baseline=baseline.step(int(mixed)-1)
        if not I.view(other)[0]:candidate=baseline;C['resets']+=1
        else:candidate=candidate.step(int(mixed))
        root=o if bit=='1' else z
        C['charged_path_steps']+=1
    return budget if I.view(root)[0] else candidate

def mixed_count(s,width):
    def walk(lo,size):
        count=sum(lo<=x<lo+size for x in s)
        if count==0 or count==size:return 0
        return 1+walk(lo,size//2)+walk(lo+size//2,size//2)
    return walk(0,1<<width)
def initialized(width,bound):
    b=Signed()
    for _ in range(width):
        for _ in range(bound):b=b.step(1)
    return b

def audit(row,s,width,bound,b):
    p=mixed_count(s,width)
    assert b.value()==bound*width-p
    assert (b.sign<0)==(p>bound*width)
    C['budget_audits']+=1

def main():
    rng=random.Random(14907);saved=[]
    for width in range(1,4):
        for mask in range(1<<(1<<width)):
            s=[x for x in range(1<<width) if mask>>x&1]
            for bound in (1,2,3):
                for seq in (s,list(reversed(s))):
                    row=I.Row();seen=set();b=initialized(width,bound)
                    for x in seq+seq[:2]:
                        w=I.bits(x,width);b=charge(b,row.root,w);row=I.add(row,w);seen.add(x)
                        audit(row,seen,width,bound,b)
                    saved.append((row,frozenset(seen),width,bound,b));C['set_order_bound_cases']+=1
    for seq in permutations(range(8)):
        row=I.Row();seen=set();b=initialized(3,1)
        for x in seq:
            w=I.bits(x,3);b=charge(b,row.root,w);row=I.add(row,w);seen.add(x);audit(row,seen,3,1,b)
        C['cube_permutations']+=1
    for row,s,width,bound,b in saved[::5]:
        x=rng.randrange(1<<width);w=I.bits(x,width)
        newb=charge(b,row.root,w);newrow=I.add(row,w)
        audit(newrow,s|{x},width,bound,newb);audit(row,s,width,bound,b)
        C['retained_budget_forks']+=1
    width,bound=5,2;rows={};raw={};budgets={};bad=0
    for _ in range(5000):
        v=rng.randrange(32);x=rng.randrange(32);old=rows.get(v,I.Row());b=budgets.get(v,initialized(width,bound))
        nb=charge(b,old.root,I.bits(x,width));new=I.add(old,I.bits(x,width))
        bad+=int(nb.sign<0)-int(b.sign<0)
        rows[v]=new;budgets[v]=nb;raw.setdefault(v,set()).add(x)
        assert bad==sum(mixed_count(s,width)>bound*width for s in raw.values());C['global_budget_updates']+=1
    # Growing number of intervals/prefix pieces, degree exactly width, bounded actual splice work.
    for r in range(2,7):
        width=1<<r;bound=3;s=set(range(0,2*width,2));row=I.Row();b=initialized(width,bound)
        seq=list(s);rng.shuffle(seq)
        for x in seq:
            w=I.bits(x,width);b=charge(b,row.root,w);row=I.add(row,w,True)
        assert len(row.components)==width and row.quota==()
        assert mixed_count(s,width)==3*width-r-2
        audit(row,s,width,bound,b)
        other=I.Row()
        for x in range(1,2*width,2):other=I.add(other,I.bits(x,width))
        merged,calls=I.splice(other.root,row.root)
        assert calls<=1+2*bound*width
        for x in range(2*width+2):
            assert I.member(merged,I.bits(x,width))==(x<2*width)
            C['growing_family_queries']+=1
        C['growing_interval_families']+=1
    print(json.dumps({'checks':dict(sorted(C.items())),'native_timing_evidence':False},indent=2))
if __name__=='__main__':main()