Locking ও 2PL — strict two-phase আর deadlock-এর খেলা

লক্ষ্য: engine lock-manager-এর মেরুদণ্ড — multi-granularity compatibility + deadlock detection — নিজের হাতে।

from collections import defaultdict

COMPAT = {   # সব স্তরে একই — intention সংকেত, X একা
    'IS': {'IS':True,'IX':True,'S':True,'SIX':True,'X':False},
    'IX': {'IS':True,'IX':True,'S':False,'SIX':False,'X':False},
    'S':  {'IS':True,'IX':False,'S':True,'SIX':False,'X':False},
    'SIX':{'IS':True,'IX':False,'S':False,'SIX':False,'X':False},
    'X':  {'IS':False,'IX':False,'S':False,'SIX':False,'X':False},
}
ANCESTOR = {'S': ['IS'], 'X': ['IX']}   # row-request-এর ইঙ্গিত-chain

class LockManager:
    def __init__(self, log=print):
        self.held = defaultdict(list)   # res -> [(tx, mode)]
        self.waits = {}                 # tx -> (res, mode)
        self.log = log

    def _ok(self, res, tx, mode):
        return all(COMPAT[mode][m] for t, m in self.held[res] if t != tx)

    def _anc(self, res):
        parts = res.split('/')          # যেমন db/accounts/a
        return ['/'.join(parts[:i]) for i in range(1, len(parts))]

    def acquire(self, tx, res, mode):
        hints = dict(zip(self._anc(res),
                         ANCESTOR[mode]*len(self._anc(res))))
        for anc, hint in hints.items():
            if not self._ok(anc, tx, hint):
                self.log(f"  T{tx} blocked on {anc}[{hint}]")
                return False
            if (tx, hint) not in self.held[anc]:
                self.held[anc].append((tx, hint))
                self.log(f"  T{tx} granted {anc}[{hint}] (intention)")
        if self._ok(res, tx, mode):
            self.held[res].append((tx, mode))
            self.log(f"  T{tx} granted {res}[{mode}]")
            return True
        self.waits[tx] = (res, mode)
        self.log(f"  T{tx} WAITS for {res}[{mode}]")
        return False

    def release_all(self, tx):
        n = 0
        for res in list(self.held):
            n += len([1 for t,_ in self.held[res] if t == tx])
            self.held[res] = [(t,m) for t,m in self.held[res] if t != tx]
        self.waits.pop(tx, None)
        self.log(f"  T{tx} released all ({n} locks)")

    def find_cycle(self):
        edge = defaultdict(set)
        for ta, (res, _) in self.waits.items():
            for tb, _m in self.held[res]:
                if tb != ta: edge[ta].add(tb)
        seen = set()
        def dfs(u, path):
            if u in path: return path[path.index(u):]+[u]
            if u in seen: return None
            seen.add(u)
            for v in sorted(edge[u]):
                c = dfs(v, path+[u])
                if c: return c
        for n_ in sorted(set(edge) | set(self.waits)):
            c = dfs(n_, [])
            if c: return c
        return None

    def resolve_deadlocks(self):
        victims = []
        while (cyc := self.find_cycle()):
            members = cyc[:-1]
            # cost-heuristic: সবচেয়ে কম lock-ধারী = সস্তা undo
            victim = min(members, key=lambda t: sum(
                len([1 for tt,_ in self.held[r] if tt==t])
                for r in self.held))
            self.log(f"  DEADLOCK cycle "
                     f"{'->'.join('T%d'%t for t in cyc)} — victim T{victim}")
            self.release_all(victim)
            victims.append(victim)
        return victims


print("AB-BA interleaving:")
lm = LockManager(log=lambda s: print(s))
lm.acquire(1, 'db/accounts/a', 'X')
lm.acquire(2, 'db/accounts/b', 'X')
lm.acquire(1, 'db/accounts/b', 'X')   # blocks (T2 holds b)
lm.acquire(2, 'db/accounts/a', 'X')   # blocks (T1 holds a) — বৃত্ত!
lm.resolve_deadlocks()
lm.acquire(2, 'db/accounts/a', 'X')   # victim গেল — grant
lm.release_all(2)

print("Reader-ভিড় আর table-X:")
lm2 = LockManager(log=lambda s: print(s))
for r in (1, 2, 3): lm2.acquire(r, 'db/reports/r1', 'S')
lm2.acquire(9, 'db/reports', 'X')     # table-X: সবার সাথে conflict
lm2.resolve_deadlocks()               # cycle নেই — শুধু অপেক্ষা
lm2.release_all(1)

প্রত্যাশিত output (সংক্ষেপে):

AB-BA interleaving:
  T1 granted db[a?]... intentions...
  T2 WAITS for db/accounts/a[X]
  DEADLOCK cycle T1->T2->T1 — victim T1
  T1 released all (N locks)
  T2 granted db/accounts/a[X]

Reader-ভিড় আর table-X:
  T9 WAITS for db/reports[X]     ← cycle নেই, detector নীরব

দুই শিক্ষা output-এই: (১) victim সবসময় “সবচেয়ে সস্তা”; (২) অপেক্ষা deadlock না — detector শুধু বৃত্তে হস্তক্ষেপ করে।

নিজে বাড়ান: (১) per-resource FIFO queue + jump-নিষেধ (fairness); (২) victim-heuristic বদল — youngest-tx, কোনটা কখন ভালো? (৩) ৮+ row-lock ধরলে page-level escalation — concurrency-র কী হারাল দেখুন।