#!/usr/bin/env python3
"""Finite logical lock models, not a DBMS, persistence implementation or fairness certification."""
from collections import defaultdict
from itertools import product,combinations
from copy import deepcopy
import json,random

MODES=('NL','IS','IX','S','SIX','X')
BITS={'NL':0,'IS':1,'IX':3,'S':5,'SIX':7,'X':15}
BYBITS={v:k for k,v in BITS.items()}
COMP={
 'NL':set(MODES),'IS':{'NL','IS','IX','S','SIX'},
 'IX':{'NL','IS','IX'},'S':{'NL','IS','S'},
 'SIX':{'NL','IS'},'X':{'NL'}}
def join(a,b):return BYBITS[BITS[a]|BITS[b]]
def compatible(a,b):return b in COMP[a]

def tree(U):
 assert U>=1 and U&(U-1)==0
 parents={(0,U):None}
 def build(a,b):
  if b-a>1:
   mid=(a+b)//2
   for child in [(a,mid),(mid,b)]:parents[child]=(a,b);build(*child)
 build(0,U);return parents

def ancestors(node,parents):
 out=[];p=parents[node]
 while p is not None:out.append(p);p=parents[p]
 return list(reversed(out))

def covers(U,a,b):
 assert 0<=a<=b<=U
 out=[]
 def visit(l,r):
  if r<=a or b<=l:return
  if a<=l and r<=b:out.append((l,r));return
  mid=(l+r)//2;visit(l,mid);visit(mid,r)
 if a<b:visit(0,U)
 return out

class Hierarchy:
 def __init__(self,U):self.U=U;self.parents=tree(U);self.locks=defaultdict(dict)
 def request(self,T,node,mode):
  assert node in self.parents and mode in MODES and mode!='NL'
  assert node[1]-node[0]>1 or mode in ('S','X')
  old=self.locks[T].get(node,'NL');new=join(old,mode)
  needed=3 if new in ('IX','SIX','X') else 1
  for p in ancestors(node,self.parents):assert BITS[self.locks[T].get(p,'NL')]&needed==needed,('missing ancestor permission',T,node,p,new)
  blocked=[u for u,held in self.locks.items() if u!=T and not compatible(new,held.get(node,'NL'))]
  if blocked:return {'result':'WAIT','blockers':blocked,'old':old,'requested':new}
  self.locks[T][node]=new
  return {'result':'GRANT','old':old,'mode':new}
 def path_request(self,T,node,mode):
  intent='IX' if mode in ('IX','SIX','X') else 'IS';trace=[]
  for p in ancestors(node,self.parents):
   result=self.request(T,p,intent);trace.append({'node':p,**result})
   if result['result']!='GRANT':return trace
  trace.append({'node':node,**self.request(T,node,mode)});return trace
 def finish(self,T):self.locks.pop(T,None)
 def access(self,T,k,write=False):
  assert 0<=k<self.U
  permissions=('X',) if write else ('S','SIX','X')
  return any(l<=k<r and mode in permissions for (l,r),mode in self.locks[T].items())
 def verify(self):
  for T,held in self.locks.items():
   for node,mode in held.items():
    needed=3 if mode in ('IX','SIX','X') else 1
    assert all(BITS[held.get(p,'NL')]&needed==needed for p in ancestors(node,self.parents))
    for U,other in self.locks.items():
     if U!=T:assert compatible(mode,other.get(node,'NL'))
  names=list(self.locks)
  for T,U in combinations(names,2):
   for k in range(self.U):
    assert not (self.access(T,k,True) and self.access(U,k))
    assert not (self.access(U,k,True) and self.access(T,k))

class RangeLocks:
 def __init__(self,U):self.U=U;self.held=[]
 def check(self,a,b,mode):assert 0<=a<=b<=self.U and mode in ('S','X')
 def blockers(self,T,a,b,mode):
  self.check(a,b,mode)
  return list(dict.fromkeys(u for u,c,d,n in self.held if u!=T and max(a,c)<min(b,d) and 'X' in (mode,n)))
 def grant(self,T,a,b,mode):
  assert not self.blockers(T,a,b,mode),'cannot grant while conflicting holder remains'
  if a<b and (T,a,b,mode) not in self.held:self.held.append((T,a,b,mode))
 def release(self,T):self.held=[entry for entry in self.held if entry[0]!=T]
 def has(self,T,a,b,write=False):
  return all(any(u==T and c<=k<d and (not write or n=='X') for u,c,d,n in self.held) for k in range(a,b))
 def verify(self):
  for T,a,b,m in self.held:assert not self.blockers(T,a,b,m),'incompatible grants'

class AgeScheduler:
 def __init__(self,policy,ages,U=32,initial=(8,16)):
  assert policy in ('wait-die','wound-wait') and len(set(ages.values()))==len(ages)
  self.policy=policy;self.ages=dict(ages);self.state={T:'ACTIVE' for T in ages};self.pending={}
  self.locks=RangeLocks(U);self.data={k:str(k) for k in initial};self.private={T:{} for T in ages};self.events=[]
 def wound(self,T):
  assert self.state[T] in ('ACTIVE','WAITING')
  self.state[T]='ABORTING';self.pending.pop(T,None)
 def reconcile_pending(self):
  # Recheck every new holder-to-waiter dependency, before the event returns.
  changes=[]
  while True:
   changed=False
   for T,req in list(self.pending.items()):
    if self.state[T]!='WAITING':continue
    active=[U for U in self.locks.blockers(T,*req) if self.state[U] in ('ACTIVE','WAITING')]
    if self.policy=='wait-die':
     if any(self.ages[U]<self.ages[T] for U in active):
      self.wound(T);changes.append({'victim':T,'reason':'new older holder'});changed=True
    else:
     for U in active:
      if self.ages[T]<self.ages[U] and self.state[U] in ('ACTIVE','WAITING'):
       self.wound(U);changes.append({'victim':U,'requester':T,'reason':'new younger holder'});changed=True
   if not changed:return changes
 def request(self,T,a,b,mode):
  assert self.state[T] in ('ACTIVE','WAITING')
  req=(a,b,mode)
  if T in self.pending:assert self.pending[T]==req,'a waiting transaction cannot issue a new request'
  blockers=self.locks.blockers(T,*req)
  active=[U for U in blockers if self.state[U] in ('ACTIVE','WAITING')]
  wounded=[]
  if self.policy=='wait-die' and any(self.ages[U]<self.ages[T] for U in active):
   self.wound(T);result='ABORT_SELF'
  else:
   if self.policy=='wound-wait':
    for U in active:
     if self.ages[T]<self.ages[U]:self.wound(U);wounded.append(U)
   if blockers:self.state[T]='WAITING';self.pending[T]=req;result='WAIT'
   else:self.locks.grant(T,*req);self.state[T]='ACTIVE';self.pending.pop(T,None);result='GRANT'
  rechecks=self.reconcile_pending()
  if self.state[T]=='ABORTING' and result!='ABORT_SELF':result='ABORT_BY_RECHECK'
  event={'rechecks':rechecks,'T':T,'request':req,'blockers':blockers,'wounded':wounded,'result':result,'states':dict(self.state)}
  self.events.append(event);self.verify();return event
 def complete_abort(self,T):
  assert self.state[T]=='ABORTING'
  self.private[T].clear();self.locks.release(T);self.state[T]='ABORTED';self.verify()
 def restart(self,T):
  assert self.state[T]=='ABORTED' and not self.private[T]
  self.state[T]='ACTIVE' # Preserve the age and discard every old result.
 def read(self,T,a,b):
  assert self.state[T]=='ACTIVE' and self.locks.has(T,a,b)
  visible=dict(self.data);visible.update(self.private[T])
  return sorted(k for k,v in visible.items() if a<=k<b and v is not None)
 def write(self,T,k,value):
  assert self.state[T]=='ACTIVE' and self.locks.has(T,k,k+1,True)
  self.private[T][k]=value
 def commit(self,T):
  assert self.state[T]=='ACTIVE'
  for k,value in self.private[T].items():
   if value is None:self.data.pop(k,None)
   else:self.data[k]=value
  self.private[T].clear();self.state[T]='COMMITTED';self.locks.release(T);self.verify()
 def edges(self):
  return [(T,U) for T,req in self.pending.items() for U in self.locks.blockers(T,*req)]
 def verify(self):
  self.locks.verify()
  edges=self.edges()
  for T,U in edges:
   if self.state[U]=='ABORTING':continue
   assert self.state[T]=='WAITING' and self.state[U] in ('ACTIVE','WAITING')
   assert ((self.ages[T]<self.ages[U]) if self.policy=='wait-die' else (self.ages[T]>self.ages[U])),(self.policy,T,U,self.state,self.events[-10:])
  adj=defaultdict(list)
  for T,U in edges:adj[T].append(U)
  colors={}
  def dfs(T):
   assert colors.get(T)!=1,'wait cycle'
   if colors.get(T)==2:return
   colors[T]=1
   for U in adj[T]:dfs(U)
   colors[T]=2
  for T in self.state:dfs(T)
  for T,s in self.state.items():
   assert (T in self.pending)==(s=='WAITING')
   if s in ('ABORTING','ABORTED','COMMITTED'):assert T not in self.pending


def reservation(policy):
 s=AgeScheduler(policy,{'T1':10,'T2':20})
 for T in ['T1','T2']:assert s.request(T,10,13,'S')['result']=='GRANT';assert s.read(T,10,13)==[]
 e=s.request('T1',11,12,'X');assert e['result']=='WAIT'
 if policy=='wait-die':assert s.request('T2',12,13,'X')['result']=='ABORT_SELF'
 else:assert s.state['T2']=='ABORTING'
 # Before cleanup, the young transaction still owns the shared range.
 blocked_before=s.locks.blockers('T1',11,12,'X');assert blocked_before==['T2']
 rejected=False
 try:s.locks.grant('T1',11,12,'X')
 except AssertionError:rejected=True
 assert rejected
 s.complete_abort('T2');assert s.request('T1',11,12,'X')['result']=='GRANT'
 s.write('T1',11,'booked by T1');s.commit('T1');s.restart('T2')
 assert s.ages['T2']==20;assert s.request('T2',10,13,'S')['result']=='GRANT'
 retry_result=s.read('T2',10,13);assert retry_result==[11];s.commit('T2')
 return {'events':s.events,'before_cleanup_blockers':blocked_before,'early_grant_rejected':rejected,'retry_result':retry_result,'final_keys':sorted(s.data),'retry_age':s.ages['T2']}

def hierarchy_tests():
 # Directly enumerate static labelings of a four-leaf tree, independently of grant().
 p=tree(4);nodes=list(p);choices=[MODES if b-a>1 else ('NL','S','X') for a,b in nodes];valid=[]
 for labels in product(*choices):
  held=dict(zip(nodes,labels))
  if all(mode=='NL' or all(BITS[held[z]]& (3 if mode in ('IX','SIX','X') else 1)==(3 if mode in ('IX','SIX','X') else 1) for z in ancestors(node,p)) for node,mode in held.items()):
   reads={k for k in range(4) if any(a<=k<b and mode in ('S','SIX','X') for (a,b),mode in held.items())}
   writes={k for k in range(4) if any(a<=k<b and mode=='X' for (a,b),mode in held.items())}
   valid.append((held,reads,writes))
 pairs=0
 for left in valid:
  for right in valid:
   if all(compatible(left[0][n],right[0][n]) for n in nodes):
    assert not left[2]&right[1] and not right[2]&left[1];pairs+=1
 for a,b in product(MODES,repeat=2):
  z=join(a,b);assert BITS[z]&BITS[a]==BITS[a] and BITS[z]&BITS[b]==BITS[b]
  for upper in MODES:
   if BITS[upper]&BITS[a]==BITS[a] and BITS[upper]&BITS[b]==BITS[b]:assert BITS[upper]&BITS[z]==BITS[z]
 h=Hierarchy(32);trace={}
 for T in ['T1','T2']:
  trace[T]=[h.path_request(T,n,'S') for n in covers(32,10,13)];h.verify()
 failed=h.path_request('T1',(11,12),'X');assert failed[-1]['result']=='WAIT' and failed[-1]['node']==(10,12)
 assert h.locks['T1'][(10,12)]=='S';h.finish('T2')
 resumed=h.path_request('T1',(11,12),'X');assert resumed[-1]['result']=='GRANT';h.verify()
 assert h.locks['T1'][(10,12)]=='SIX' and h.access('T1',11,True)
 return {'valid_four_leaf_labelings':len(valid),'compatible_pairs_projected':pairs,'mode_joins':36,'range_cover':covers(32,10,13),'failed_conversion':failed,'resumed_conversion':resumed}

def range_tests():
 intervals=[(a,b) for a in range(17) for b in range(a,17)];pairs=0
 for left,right in product(intervals,repeat=2):
  ref=bool(set(range(*left))&set(range(*right)))
  assert (max(left[0],right[0])<min(left[1],right[1]))==ref;pairs+=1
 covers_count=0
 for exponent in range(7):
  U=1<<exponent
  for a in range(U+1):
   for b in range(a,U+1):
    parts=covers(U,a,b);flat=[k for l,r in parts for k in range(l,r)]
    assert flat==list(range(a,b)) and all(n in tree(U) for n in parts);covers_count+=1
 locks=RangeLocks(32);locks.grant('reader',10,13,'S')
 assert locks.blockers('writer',9,10,'X')==[] and locks.blockers('writer',11,12,'X')==['reader']
 assert locks.blockers('writer',13,14,'X')==[] and locks.blockers('writer',10,10,'X')==[]
 # Broken row-only check: both scans return empty and distinct point writes coexist.
 plain={8,16};seen=[k for k in plain if 10<=k<13];assert seen==[]
 plain.add(11);plain.add(12);assert len([k for k in plain if 10<=k<13])==2
 return {'interval_pairs':pairs,'dyadic_covers':covers_count,'old_and_new_key_required':True,'row_only_counterexample':[11,12]}

def random_age_tests():
 rng=random.Random(730071);operations=0;aborts=0
 for policy in ['wait-die','wound-wait']:
  for trial in range(300):
   s=AgeScheduler(policy,{f'T{i}':i+1 for i in range(6)},8,())
   for step in range(120):
    T=rng.choice(list(s.state));state=s.state[T]
    if state=='ABORTING':s.complete_abort(T);aborts+=1
    elif state=='ABORTED':s.restart(T)
    elif state=='COMMITTED':continue
    elif state=='WAITING':s.request(T,*s.pending[T])
    elif rng.random()<.15:s.commit(T)
    else:
     a=rng.randrange(8);b=rng.randrange(a+1,9);s.request(T,a,b,rng.choice(['S','X']))
    s.verify();operations+=1
 # Mixed ages: one old and one young shared holder.
 mixed={}
 for policy in ['wait-die','wound-wait']:
  s=AgeScheduler(policy,{'old':10,'mid':20,'young':30},8,())
  s.request('old',2,3,'S');s.request('young',2,3,'S');event=s.request('mid',2,3,'X')
  if policy=='wait-die':assert event['result']=='ABORT_SELF'
  else:assert event['wounded']==['young'] and s.state['old']=='ACTIVE' and s.state['mid']=='WAITING'
  mixed[policy]=event
 # New compatible holders may create fresh edges to an already waiting X.
 barging={}
 for policy in ['wait-die','wound-wait']:
  s=AgeScheduler(policy,{'old':10,'mid':20,'young':30},8,())
  first,new=('young','old') if policy=='wait-die' else ('old','young')
  assert s.request(first,2,3,'S')['result']=='GRANT'
  assert s.request('mid',2,3,'X')['result']=='WAIT'
  event=s.request(new,2,3,'S')
  victim='mid' if policy=='wait-die' else 'young'
  assert s.state[victim]=='ABORTING' and event['rechecks']
  barging[policy]=event
 return {'new_holder_rechecks':barging,'cases':600,'state_changing_or_checked_steps':operations,'completed_abort_cleanups':aborts,'mixed_holder_examples':mixed}

def main():
 if not __debug__:raise RuntimeError('This checker requires Python assertions; run without -O')
 result={'status':'PASS','scope':'Finite logical-lock models; no DBMS concurrency/durability/fairness certification','hierarchy':hierarchy_tests(),'ranges':range_tests(),'age':random_age_tests(),'wait_die_reservation':reservation('wait-die'),'wound_wait_reservation':reservation('wound-wait')}
 print(json.dumps(result,ensure_ascii=False,indent=2))
if __name__=='__main__':main()