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()