Control Flow Graph ও Dataflow Analysis — প্রবাহের উপর গণিত

IR লেসনের TAC emitter-এর naming-ই চলছে এখানে (TACInstr, BasicBlock, CFG) — এই ফাইলটাই পরের তিন লেসনের build-এর base।

from dataclasses import dataclass, field

TERMINATORS = {'if_>=', 'if_<', 'goto', 'return'}   # ব্লক শেষ করে এমন op

@dataclass(frozen=True)
class TACInstr:                    # toy compiler-এর three-address instruction
    op: str                        # '+', '-', '*', copy, print, if_>=, ...
    dest: str | None = None
    a: str | None = None           # operand ('#' prefix = constant)
    b: str | None = None
    label: str | None = None

    def uses(self):                # পঠিত ভেরিয়েবল (constant না)
        return [x for x in (self.a, self.b) if x and not x.startswith('#')]

@dataclass
class BasicBlock:
    bid: int
    instrs: list                   # [TACInstr]
    succs: set = field(default_factory=set)
    preds: set = field(default_factory=set)

class CFG:
    def __init__(self, blocks):
        self.blocks = blocks
        self.succ = {b.bid: sorted(b.succs) for b in blocks}
        self.pred = {b.bid: sorted(b.preds) for b in blocks}
        self.adjacency = '\n'.join(
            f"  B{b.bid} -> " + (', '.join(f'B{s}' for s in sorted(b.succs)) or '(none)')
            for b in blocks)

def find_leaders(instrs):
    leaders = {0}                                  # নিয়ম ১
    for i, ins in enumerate(instrs):
        if ins.op == 'label':
            leaders.add(i)                         # নিয়ম ২
        if ins.op in TERMINATORS and i + 1 < len(instrs):
            leaders.add(i + 1)                     # নিয়ম ৩
    return leaders

def build_cfg(instrs) -> CFG:
    starts = sorted(find_leaders(instrs))
    bounds = starts + [len(instrs)]
    blocks, label_to_bid, bid_of_index = [], {}, {}
    for bid, lo in enumerate(starts):
        blk = BasicBlock(bid=bid, instrs=instrs[bounds[bid]:bounds[bid+1]])
        head = blk.instrs[0]
        if head.op == 'label':
            label_to_bid[head.label] = bid
        bid_of_index.update({i: bid for i in range(bounds[bid], bounds[bid+1])})
        blocks.append(blk)

    for k, blk in enumerate(blocks):
        last, nxt = blk.instrs[-1], k + 1 if k + 1 < len(blocks) else None
        if last.op == 'goto':
            blk.succs.add(label_to_bid[last.label])
        elif last.op.startswith('if_'):
            blk.succs |= {label_to_bid[last.label]} | ({nxt} if nxt is not None else set())
        elif last.op != 'return' and nxt is not None:
            blk.succs.add(nxt)                     # fall-through
    for blk in blocks:
        for s in blk.succs:
            blocks[s].preds.add(blk.bid)
    return CFG(blocks)

def block_use_def(block):
    use, dfn = set(), set()
    for ins in block.instrs:
        for v in ins.uses():
            if v not in dfn:
                use.add(v)                         # upward-exposed only
        if ins.dest:
            dfn.add(ins.dest)
    return use, dfn

def solve_liveness(cfg: CFG):
    use, dfn = zip(*(block_use_def(b) for b in cfg.blocks))
    live_in = {b.bid: frozenset() for b in cfg.blocks}
    live_out = {b.bid: frozenset() for b in cfg.blocks}
    rnd, changed = 0, True
    while changed:                                 # fixed-point iteration
        rnd += 1
        changed = False
        for b in reversed(cfg.blocks):             # backward: উল্টো ক্রমে
            new_out = set().union(*(live_in[s] for s in b.succs)) if b.succs else set()
            new_in = use[b.bid] | (new_out - dfn[b.bid])
            if new_in != live_in[b.bid] or new_out != live_out[b.bid]:
                live_in[b.bid], live_out[b.bid] = new_in, new_out
                changed = True
        fmt = lambda s: '{' + ','.join(sorted(s)) + '}'
        print(f'round {rnd}: ' + '   '.join(
            f'B{b.bid} IN={fmt(live_in[b.bid])} OUT={fmt(live_out[b.bid])}'
            for b in cfg.blocks))
    return live_in, live_out

program = [
    TACInstr('copy',  's', '#0'),
    TACInstr('copy',  'i', '#0'),
    TACInstr('if_>=', None, 'i', 'n', 'L_done'),
    TACInstr('+',     's', 's', 'i'),
    TACInstr('+',     'i', 'i', '#1'),
    TACInstr('label', None, None, None, 'L_done'),
    TACInstr('return', None, 's'),
]

cfg = build_cfg(program)
print(cfg.adjacency)
live_in, live_out = solve_liveness(cfg)
print(f'max live (register pressure) = {max(len(s) for s in live_in.values())}')

প্রত্যাশিত output — example section-এর টেবিলের সাথে মিলিয়ে দেখুন:

  B0 -> B1
  B1 -> B2, B3
  B2 -> B1
  B3 -> (none)
round 1: B0 IN={n} OUT={i,n,s}   B1 IN={i,n,s} OUT={i,s}   B2 IN={i,s} OUT={}   B3 IN={s} OUT={}
round 2: B0 IN={n} OUT={i,n,s}   B1 IN={i,n,s} OUT={i,n,s}   B2 IN={i,n,s} OUT={i,n,s}   B3 IN={s} OUT={}
round 3: ...same... (no change — fixed point reached after 3 rounds)
max live (register pressure) = 3

Round 1 বনাম Round 2-এর পার্থক্যটাই back edge-এর উপহার — body-র তথ্য header-এ ফিরে এসেছে। নিজে বাড়ান: (১) reaching definitions solver — forward দিক, gen/kill আগে হিসাব করে; (২) worklist version, round-robin- এর recomputation count-এর সাথে তুলনা।