feat: dominance, SSA and def-use chains

README.md

@@ -36,7 +36,9 @@ Currently supported inside functions: arguments, assignments, names, constants,
 operations, unary operations, comparisons, calls, attribute access, indexing, returns,
 `if`/`elif`/`else`, `while`, `for`, `break`, `continue` and `raise`. Each function is
 emitted as its control-flow graph: one block per basic block, ending in an explicit
-terminator (`branch`, `jump`, `return`, `raise`, `for_next`).
+terminator (`branch`, `jump`, `return`, `raise`, `for_next`). Add `--ssa` to print the
+static single assignment form: locals become numbered values, merges get `phi`
+instructions and unassigned paths read an explicit `undefined` value.
 Unsupported syntax produces a source-located diagnostic and a non-zero exit status.
 
 Names are resolved to canonical symbols through lexical scopes, imports and builtins.

src/coretrace_python/cfg/__init__.py

@@ -1,6 +1,14 @@
 """Per-function control-flow graphs over PyHIR (architecture §5)."""
 
 from coretrace_python.cfg.builder import CFGAnalysis, build_cfg
+from coretrace_python.cfg.dominance import (
+    EXIT,
+    DominanceAnalysis,
+    DominatorTree,
+    PostDominanceAnalysis,
+    dominator_tree,
+    post_dominator_tree,
+)
 from coretrace_python.cfg.model import (
     CFG,
     BasicBlock,
@@ -17,16 +25,22 @@
 
 __all__ = [
     "CFG",
+    "EXIT",
     "BasicBlock",
     "BlockId",
     "Branch",
     "CFGAnalysis",
     "CFGError",
+    "DominanceAnalysis",
+    "DominatorTree",
     "ForEach",
     "Jump",
+    "PostDominanceAnalysis",
     "Raise",
     "Return",
     "Terminator",
     "build_cfg",
+    "dominator_tree",
+    "post_dominator_tree",
     "targets",
 ]

src/coretrace_python/cfg/dominance.py

@@ -0,0 +1,183 @@
+"""Dominance and post-dominance over control-flow graphs (architecture §5).
+
+Both trees come from the same iterative algorithm (Cooper, Harvey and Kennedy) run on
+the forward graph from the entry, or on the reversed graph from a virtual ``EXIT``
+that follows every returning or raising block. The post-dominance frontier is the
+control-dependence relation.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable, Mapping
+from dataclasses import dataclass, field
+from types import MappingProxyType
+from typing import ClassVar
+
+from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis
+from coretrace_python.cfg.builder import CFGAnalysis
+from coretrace_python.cfg.model import CFG, BlockId
+from coretrace_python.hir import nodes
+
+EXIT = BlockId("<exit>")
+
+
+@dataclass(frozen=True)
+class DominatorTree:
+    root: BlockId
+    blocks: tuple[BlockId, ...]
+    _idom: Mapping[BlockId, BlockId | None]
+    _children: Mapping[BlockId, tuple[BlockId, ...]]
+    _frontier: Mapping[BlockId, frozenset[BlockId]]
+    _depth: Mapping[BlockId, int] = field(init=False, repr=False, compare=False)
+
+    def __post_init__(self) -> None:
+        depth: dict[BlockId, int] = {}
+        for block in self.blocks:
+            parent = self._idom[block]
+            depth[block] = 0 if parent is None else depth[parent] + 1
+        object.__setattr__(self, "_depth", MappingProxyType(depth))
+
+    def idom(self, block: BlockId) -> BlockId | None:
+        return self._idom[block]
+
+    def children(self, block: BlockId) -> tuple[BlockId, ...]:
+        return self._children[block]
+
+    def frontier(self, block: BlockId) -> frozenset[BlockId]:
+        return self._frontier[block]
+
+    def dominates(self, dominator: BlockId, block: BlockId) -> bool:
+        if dominator not in self._idom or block not in self._idom:
+            return False
+        current: BlockId | None = block
+        while current is not None and self._depth[current] >= self._depth[dominator]:
+            if current == dominator:
+                return True
+            current = self._idom[current]
+        return False
+
+    def iterated_frontier(self, blocks: Iterable[BlockId]) -> frozenset[BlockId]:
+        result: set[BlockId] = set()
+        pending = list(blocks)
+        while pending:
+            for candidate in self._frontier[pending.pop()]:
+                if candidate not in result:
+                    result.add(candidate)
+                    pending.append(candidate)
+        return frozenset(result)
+
+
+def _build(
+    root: BlockId,
+    successors: Callable[[BlockId], tuple[BlockId, ...]],
+    order: Callable[[BlockId], int],
+) -> DominatorTree:
+    # Reverse postorder over the reachable nodes.
+    postorder: list[BlockId] = []
+    seen: set[BlockId] = set()
+
+    def visit(node: BlockId) -> None:
+        seen.add(node)
+        for successor in successors(node):
+            if successor not in seen:
+                visit(successor)
+        postorder.append(node)
+
+    visit(root)
+    rpo = list(reversed(postorder))
+    index = {node: position for position, node in enumerate(rpo)}
+    predecessors: dict[BlockId, list[BlockId]] = {node: [] for node in rpo}
+    for node in rpo:
+        for successor in successors(node):
+            predecessors[successor].append(node)
+
+    idom: dict[BlockId, BlockId | None] = {root: None}
+
+    def intersect(a: BlockId, b: BlockId) -> BlockId:
+        while a != b:
+            while index[a] > index[b]:
+                a = idom[a]  # type: ignore[assignment]
+            while index[b] > index[a]:
+                b = idom[b]  # type: ignore[assignment]
+        return a
+
+    changed = True
+    while changed:
+        changed = False
+        for node in rpo[1:]:
+            candidates = [p for p in predecessors[node] if p in idom]
+            new_idom = candidates[0]
+            for other in candidates[1:]:
+                new_idom = intersect(other, new_idom)
+            if idom.get(node) != new_idom:
+                idom[node] = new_idom
+                changed = True
+
+    children: dict[BlockId, list[BlockId]] = {node: [] for node in rpo}
+    for node in sorted(rpo, key=order):
+        parent = idom[node]
+        if parent is not None:
+            children[parent].append(node)
+
+    frontier: dict[BlockId, set[BlockId]] = {node: set() for node in rpo}
+    for node in rpo:
+        if node == root or len(predecessors[node]) < 2:
+            continue
+        for predecessor in predecessors[node]:
+            runner: BlockId | None = predecessor
+            while runner is not None and runner != idom[node]:
+                frontier[runner].add(node)
+                runner = idom[runner]
+
+    preorder: list[BlockId] = []
+
+    def walk(node: BlockId) -> None:
+        preorder.append(node)
+        for child in children[node]:
+            walk(child)
+
+    walk(root)
+    return DominatorTree(
+        root=root,
+        blocks=tuple(preorder),
+        _idom=MappingProxyType(dict(idom)),
+        _children=MappingProxyType({k: tuple(v) for k, v in children.items()}),
+        _frontier=MappingProxyType({k: frozenset(v) for k, v in frontier.items()}),
+    )
+
+
+def dominator_tree(cfg: CFG) -> DominatorTree:
+    order = {block_id: position for position, block_id in enumerate(cfg.blocks)}
+    return _build(cfg.entry, cfg.successors, lambda block: order[block])
+
+
+def post_dominator_tree(cfg: CFG) -> DominatorTree:
+    reachable = cfg.reachable()
+    order = {block_id: position for position, block_id in enumerate(cfg.blocks)}
+    order[EXIT] = -1
+    leaves = tuple(b for b in cfg.blocks if b in reachable and not cfg.successors(b))
+
+    def reversed_successors(block: BlockId) -> tuple[BlockId, ...]:
+        if block == EXIT:
+            return leaves
+        return tuple(p for p in cfg.predecessors(block) if p in reachable)
+
+    return _build(EXIT, reversed_successors, lambda block: order[block])
+
+
+class DominanceAnalysis(FunctionAnalysis[DominatorTree]):
+    name: ClassVar[str] = "cfg.dominance"
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({CFGAnalysis})
+
+    @classmethod
+    def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> DominatorTree:
+        return dominator_tree(ctx.get(CFGAnalysis, function))
+
+
+class PostDominanceAnalysis(FunctionAnalysis[DominatorTree]):
+    name: ClassVar[str] = "cfg.post_dominance"
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({CFGAnalysis})
+
+    @classmethod
+    def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> DominatorTree:
+        return post_dominator_tree(ctx.get(CFGAnalysis, function))

src/coretrace_python/cli.py

@@ -23,6 +23,11 @@ def build_parser() -> argparse.ArgumentParser:
         action="store_true",
         help="print the lowered Python intermediate representation",
     )
+    parser.add_argument(
+        "--ssa",
+        action="store_true",
+        help="with --emit-ir, print the static single assignment form",
+    )
     return parser
 
 
@@ -34,7 +39,7 @@ def main(argv: list[str] | None = None) -> int:
 
     try:
         source = SourceManager().load_file(args.path)
-        module = lower_module(build_hir(source))
+        module = lower_module(build_hir(source), ssa=args.ssa)
     except (
         OSError,
         UnicodeError,

src/coretrace_python/ir/defuse.py

@@ -0,0 +1,74 @@
+"""Def-use and use-def chains over SSA values (architecture §7)."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import ClassVar
+
+from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis
+from coretrace_python.cfg import BlockId
+from coretrace_python.hir import nodes
+from coretrace_python.ir.model import ForNext, FunctionIR, Value
+from coretrace_python.ir.ssa import SSAAnalysis
+
+
+@dataclass(frozen=True)
+class Definition:
+    """Where a value is defined: a parameter (no block), an instruction, or a terminator."""
+
+    block: BlockId | None
+    index: int | None
+
+
+@dataclass(frozen=True)
+class Use:
+    """Where a value is used: an instruction index, or ``None`` for the terminator."""
+
+    block: BlockId
+    index: int | None
+
+
+class DefUse:
+    def __init__(
+        self,
+        definitions: Mapping[Value, Definition],
+        uses: Mapping[Value, tuple[Use, ...]],
+    ) -> None:
+        self._definitions = MappingProxyType(dict(definitions))
+        self._uses = MappingProxyType(dict(uses))
+
+    def definition(self, value: Value) -> Definition:
+        return self._definitions[value]
+
+    def uses(self, value: Value) -> tuple[Use, ...]:
+        return self._uses.get(value, ())
+
+
+def def_use(function: FunctionIR) -> DefUse:
+    definitions: dict[Value, Definition] = {}
+    uses: dict[Value, list[Use]] = {}
+    for parameter in function.parameters:
+        definitions[parameter] = Definition(None, None)
+    for block in function.blocks:
+        for index, instruction in enumerate(block.instructions):
+            if instruction.result is not None:
+                definitions[instruction.result] = Definition(block.id, index)
+            for operand in instruction.operands():
+                uses.setdefault(operand, []).append(Use(block.id, index))
+        terminator = block.terminator
+        if isinstance(terminator, ForNext) and terminator.result is not None:
+            definitions[terminator.result] = Definition(block.id, None)
+        for operand in terminator.operands():
+            uses.setdefault(operand, []).append(Use(block.id, None))
+    return DefUse(definitions, {value: tuple(found) for value, found in uses.items()})
+
+
+class DefUseAnalysis(FunctionAnalysis[DefUse]):
+    name: ClassVar[str] = "ir.defuse"
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis})
+
+    @classmethod
+    def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> DefUse:
+        return def_use(ctx.get(SSAAnalysis, function))