#!/usr/bin/env python3 """SCHED-16 exact replay + independent small-model checks. Python 3, standard library. No network, files modified, packages, or runtime-specific scheduler are required. Prints reproducible JSON. Times are exact fractions until formatting. """ from collections import deque, Counter from fractions import Fraction as F from functools import lru_cache from itertools import product, permutations import json JOBS = {'A': (0,9), 'B': (1,4), 'C': (2,2), 'D': (4,1)} def report(trace, jobs=JOBS): svc=Counter();first={};finish={};overhead=F(0);idle=F(0);end=F(0) for name,start,stop in trace: assert start==end and stop>start, (name,start,stop,end) end=stop if name=='SWITCH': overhead+=stop-start;continue if name=='IDLE': idle+=stop-start;continue assert start>=jobs[name][0] first.setdefault(name,start);svc[name]+=stop-start;finish[name]=stop assert svc=={j:F(b) for j,(a,b) in jobs.items()}, (svc,jobs) assert sum(svc.values())+overhead+idle==end per={j:dict(first=first[j],finish=finish[j],turnaround=finish[j]-a,response=first[j]-a,waiting=finish[j]-a-b) for j,(a,b) in jobs.items()} return dict(trace=trace,per_task=per,mean_turnaround=sum(v['turnaround'] for v in per.values())/F(len(jobs)),mean_response=sum(v['response'] for v in per.values())/F(len(jobs)),mean_waiting=sum(v['waiting'] for v in per.values())/F(len(jobs)),overhead=overhead,end=end,utilization=sum(svc.values())/end) def nonpreemptive(mode,jobs=JOBS): pending=set(jobs);t=F(0);trace=[] while pending: ready=[j for j in pending if jobs[j][0]<=t] if not ready: u=min(F(jobs[j][0]) for j in pending);trace.append(('IDLE',t,u));t=u;continue key=(lambda j:(jobs[j][0],j)) if mode=='FIFO' else (lambda j:(jobs[j][1],jobs[j][0],j)) j=min(ready,key=key);stop=t+jobs[j][1];trace.append((j,t,stop));t=stop;pending.remove(j) return report(trace,jobs) def srpt(jobs=JOBS): # Independent one-unit simulator; retain current task for minimum ties. rem={j:b for j,(a,b) in jobs.items()};t=0;cur=None;trace=[] while any(rem.values()): ready=[j for j in jobs if jobs[j][0]<=t and rem[j]>0] if not ready:j='IDLE';cur=None else: v=min(rem[j] for j in ready) j=cur if cur in ready and rem[cur]==v else min((j for j in ready if rem[j]==v),key=lambda j:(jobs[j][0],j)) rem[j]-=1;cur=j if trace and trace[-1][0]==j:trace[-1]=(j,trace[-1][1],t+1) else:trace.append((j,t,t+1)) t+=1 return report(trace,jobs) def rr(jobs=JOBS,q=F(2),cost=F(0)): arrivals=deque(sorted(jobs,key=lambda j:(jobs[j][0],j)));ready=deque();rem={j:F(b) for j,(a,b) in jobs.items()};t=F(0);trace=[];last=None def admit(): while arrivals and jobs[arrivals[0]][0]<=t:ready.append(arrivals.popleft()) while arrivals or ready: admit() if not ready: u=F(jobs[arrivals[0]][0]);trace.append(('IDLE',t,u));t=u;last=None;admit() j=ready.popleft() # dispatch target fixed across its switching interval if last is not None and last!=j and cost: trace.append(('SWITCH',t,t+cost));t+=cost;admit() delta=min(q,rem[j]);trace.append((j,t,t+delta));t+=delta;rem[j]-=delta admit() # arrivals at boundary BEFORE expired task is requeued if rem[j]:ready.append(j) last=j return report(trace,jobs) def mlfq(jobs=JOBS,boost=12,until=None): # Unit-step reference supports main integer model and periodic boost counterexample. ready=[deque(),deque(),deque()];level={j:0 for j in jobs};used=Counter();slice_used=Counter();rem={j:b for j,(a,b) in jobs.items()};cur=None;trace=[];t=0 while any(rem.values()) and (until is None or t=quota: level[cur]=min(2,lev+1);used[cur]=slice_used[cur]=0;ready[level[cur]].append(cur);cur=None elif slice_used[cur]>=q: slice_used[cur]=0;ready[lev].append(cur);cur=None elif any(ready[h] for h in range(lev)): ready[lev].appendleft(cur);cur=None if cur is None: for q in ready: if q:cur=q.popleft();break name=cur or 'IDLE' if cur: assert rem[cur]>0 and jobs[cur][0]<=t rem[cur]-=1;used[cur]+=1;slice_used[cur]+=1 if trace and trace[-1][0]==name:trace[-1]=(name,trace[-1][1],t+1) else:trace.append((name,t,t+1)) t+=1 if until is not None:return trace,rem return report(trace,jobs) def stride(weights=(3,2,1),steps=12,K=6): ds=[F(K,w) for w in weights];ps=ds[:];counts=[0]*len(weights);rows=[] for k in range(steps): i=min(range(len(weights)),key=lambda i:(ps[i],i));before=ps[:];counts[i]+=1;ps[i]+=ds[i] assert max(ps)-min(ps)<=max(ds) assert all(ps[j]==(counts[j]+1)*ds[j] for j in range(len(weights))) rows.append(dict(step=k+1,before=before,winner=chr(65+i),after=ps[:])) return dict(weights=weights,strides=ds,counts=counts,rows=rows) def optimum_flow(arrivals,bursts): # Enumerates ALL integer non-idling one-unit schedules by dynamic programming. @lru_cache(None) def f(t,rem): if not any(rem):return 0 ready=[i for i,r in enumerate(rem) if r and arrivals[i]<=t] waiting=sum(1 for i,r in enumerate(rem) if r and arrivals[i]<=t) if not ready:return f(t+1,rem) options=[] for i in ready: nex=list(rem);nex[i]-=1;options.append(waiting+f(t+1,tuple(nex))) return min(options) return f(0,tuple(bursts)) def tests(): results={m:nonpreemptive(m) for m in ['FIFO','SJF']};results['SRTF']=srpt();results['RR']=rr();results['MLFQ']=mlfq();results['RR_cost']=rr(cost=F(1,4)) expected={'FIFO':F(23,2),'SJF':F(10),'SRTF':F(13,2),'RR':F(35,4),'MLFQ':F(37,4)} for name,v in expected.items():assert results[name]['mean_turnaround']==v,(name,results[name]) assert results['RR_cost']['end']==F(35,2) assert results['RR_cost']['overhead']==F(3,2) assert results['RR_cost']['utilization']==F(32,35) sjf_cases=0 for bs in product(range(1,5),repeat=4): def cost(order): acc=total=0 for i in order:acc+=bs[i];total+=acc return total optimum=min(cost(p) for p in permutations(range(4))) assert cost(sorted(range(4),key=lambda i:bs[i]))==optimum;sjf_cases+=1 srpt_cases=0 for arrivals in product(range(3),repeat=3): for bursts in product(range(1,4),repeat=3): jobs={chr(65+i):(a,b) for i,(a,b) in enumerate(zip(arrivals,bursts))} out=srpt(jobs);val=sum(r['turnaround'] for r in out['per_task'].values()) assert val==optimum_flow(arrivals,bursts),(arrivals,bursts,val) srpt_cases+=1 draws=[0,5,2,3,1,0,4,2,5,0,1,3] winners=['A' if x<3 else 'B' if x<5 else 'C' for x in draws] assert Counter(winners)==dict(A=7,B=3,C=2) assert F(5,6)**16>F(1,20)>=F(5,6)**17 st=stride();assert st['counts']==[6,4,2] for ws in product(range(1,6),repeat=3):stride(ws,steps=1000,K=60) badboost,left=mlfq({'A':(0,100),'B':(0,100)},boost=1,until=20) assert left=={'A':80,'B':100} fee_boundary=rr({'A':(0,4),'B':(0,2)},cost=F(1,4)) a_segments=[seg for seg in fee_boundary['trace'] if seg[0]=='A'] assert a_segments[1][1]-a_segments[0][2]==F(5,2) # q + 2c, not q + c q1=rr(q=F(1)) results['RR_q1_transfer']=q1 return dict(model='SCHED-16; all figures from exact rational/event or integer-step replay',results=results,lottery=dict(draws=draws,winners=winners,counts=dict(Counter(winners)),p_no_C_12=F(5,6)**12,min_95_percent_one_C=17),stride=st,checks=dict(sjf_all_permutations_cases=sjf_cases,srpt_exhaustive_unit_schedule_cases=srpt_cases,stride_weight_sets=125,stride_steps_per_set=1000,boost_counterexample=dict(trace=badboost,remaining=left),rr_requeue_wait=dict(trace=fee_boundary['trace'],wait=F(5,2),invalid_old_bound=F(9,4)))) def encode(x): if isinstance(x,F):return str(x) if x.denominator!=1 else x.numerator raise TypeError(type(x).__name__) if __name__=='__main__': if not __debug__: raise SystemExit('Run without -O: this checker requires assertions to be enabled.') print(json.dumps(tests(),ensure_ascii=False,indent=2,default=encode))