schedule.py

1"""Deterministic disk schedule: ordinary once, theory at three separated windows."""
2import hashlib,json,math
3from pathlib import Path
4import numpy as np
5EXPOSURE=np.dtype([('record','<u8'),('ordinal','u1'),('tie','<u2')])
6META=np.dtype([('source','u1'),('phase','u1'),('kind','u1')])
7def affine(n,seed):
8    if n<1:raise ValueError('empty permutation')
9    a=1+int.from_bytes(hashlib.sha256(f'{seed}:a'.encode()).digest()[:8],'little')%n
10    while math.gcd(a,n)!=1:a=(a+1)%n or 1
11    b=int.from_bytes(hashlib.sha256(f'{seed}:b'.encode()).digest()[:8],'little')%n
12    return a,b
13def validate(meta,offsets):
14    if len(meta)<1 or len(offsets)!=len(meta)+1 or offsets[0]!=0 or np.any(np.diff(offsets.astype(np.int64))<1):raise ValueError('nonempty tie list required')
15    if np.any(meta['source']>1) or np.any(meta['phase']>2) or np.any(meta['kind']>2):raise ValueError('unknown source/phase/label kind')
16    ordinary=meta['kind']==0;counts=[int(np.sum(ordinary&(meta['source']==s))) for s in (0,1)]
17    if counts[0]!=counts[1]:raise ValueError('ordinary source counts must be equal')
18    if np.any(np.diff(offsets)[ordinary]!=1):raise ValueError('ordinary record must have one source target')
19    if len(meta)*123>2_000_000_000 or int(np.sum(~ordinary))*123>200_000_000:raise ValueError('unique slot ceiling')
20    if np.max(np.diff(offsets))>65535:raise ValueError('tie count overflow')
21    return counts[0]
22def build(meta,offsets,path,seed=20261029):
23    """Bounded <~0.5GB indexes at the approved maximum; records are never copied."""
24    n=validate(meta,offsets);ordinary=[np.flatnonzero((meta['kind']==0)&(meta['source']==s)) for s in (0,1)]
25    theory=np.flatnonzero(meta['kind']!=0);t=len(theory);size=2*n+3*t
26    if size==0:raise ValueError('empty curriculum')
27    path=Path(path)
28    with path.open('xb') as f:f.truncate(size*EXPOSURE.itemsize)
29    out=np.memmap(path,dtype=EXPOSURE,mode='r+',shape=(size,));cursor=0;ordinary_cursor=0
30    perms=[affine(n,seed+s) if n else (1,0) for s in (0,1)]
31    # Split ordinary paired decisions into thirds; each theory window is centered
32    # in a third. Repeats cannot collapse across adjacent stage boundaries.
33    for stage in range(3):
34        begin=n*stage//3;end=n*(stage+1)//3;left=(end-begin)//2
35        def emit_ordinary(start,stop):
36            nonlocal cursor
37            for q0 in range(start,stop,65536):
38                qs=np.arange(q0,min(stop,q0+65536),dtype=np.int64);k=len(qs)
39                for s,(a,b) in enumerate(perms):
40                    ids=ordinary[s][(a*qs+b)%n]
41                    out['record'][cursor+s:cursor+2*k:2]=ids
42                out['ordinal'][cursor:cursor+2*k]=0;out['tie'][cursor:cursor+2*k]=0;cursor+=2*k
43        emit_ordinary(begin,begin+left)
44        if t:
45            a,b=affine(t,seed+100) # same theory order preserves exact spacing
46            for q0 in range(0,t,65536):
47                qs=np.arange(q0,min(t,q0+65536),dtype=np.int64);ids=theory[(a*qs+b)%t];k=len(ids)
48                out['record'][cursor:cursor+k]=ids;out['ordinal'][cursor:cursor+k]=stage
49                out['tie'][cursor:cursor+k]=(ids+seed+stage)%(offsets[ids+1]-offsets[ids]);cursor+=k
50        emit_ordinary(begin+left,end)
51    assert cursor==size
52    out.flush();del out
53    return {'unique_states':len(meta),'ordinary_positions_per_source':n,'theory_states':t,'decision_exposures':size,'prediction_targets':3*size,'input_slots':122*size,'unique_serialized_slots':123*len(meta),'schedule_bytes':size*EXPOSURE.itemsize,'seed':seed,'theory_spacing':'one centered window in each ordinary third; fixed per-state order across windows'}
54def counters(schedule,cursor,n):
55    if not 0<=cursor<=len(schedule):raise ValueError('cursor')
56    count=np.zeros(n,dtype=np.uint32)
57    for start in range(0,cursor,65536):
58        part=schedule[start:min(start+65536,cursor)]
59        if np.any(part['record']>=n):raise ValueError('invalid state id')
60        np.add.at(count,part['record'],1)
61    if np.any(count>3):raise ValueError('overexposure')
62    return count.astype(np.uint8)
63def select_targets(rows,exposures,offsets,ties):
64    """Always patch teacher-forced move prefix together with rotated target."""
65    ids=exposures['record'].astype(np.int64);chosen=offsets[ids]+exposures['tie']
66    if np.any(chosen>=offsets[ids+1]):raise ValueError('tie index')
67    y=np.asarray(ties[chosen.astype(np.int64)],dtype=np.uint16)
68    if np.any(y[:,:2]>63) or np.any(y[:,2]<64) or np.any(y[:,2]>68):raise ValueError('invalid move triple')
69    out=rows.copy();out['y']=y;out['x'][:,120:122]=y[:,:2]
70    return out
71class DistributedPlan:
72    def __init__(self,schedule,rank,world,micro):
73        if world!=2 or rank not in (0,1) or micro<1:raise ValueError('two-rank plan')
74        self.schedule=schedule;self.rank=rank;self.world=world;self.micro=micro;self.steps=math.ceil(len(schedule)/(world*micro))
75    def exposures(self,step):
76        if not 0<=step<self.steps:raise IndexError('step')
77        start=(step*self.world+self.rank)*self.micro
78        return self.schedule[start:min(start+self.micro,len(self.schedule))]
79    def global_decisions(self,step):return max(0,min(self.world*self.micro,len(self.schedule)-step*self.world*self.micro))
80