From 70924d407305ad933b3dd2ed00355ab8d86352b1 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 07:23:02 +0200 Subject: [PATCH 1/2] test: define dominance, SSA and def-use behavior --- tests/test_dominance.py | 182 ++++++++++++++++++++++++++ tests/test_emit_ir.py | 91 +++++++++++++ tests/test_ssa.py | 275 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 548 insertions(+) create mode 100644 tests/test_dominance.py create mode 100644 tests/test_ssa.py diff --git a/tests/test_dominance.py b/tests/test_dominance.py new file mode 100644 index 0000000..e197f80 --- /dev/null +++ b/tests/test_dominance.py @@ -0,0 +1,182 @@ +"""Acceptance tests for dominance (``docs/architecture.md`` §5). + +``dominator_tree(cfg)`` computes immediate dominators and dominance frontiers over the +reachable blocks. ``post_dominator_tree(cfg)`` does the same on the reversed graph with a +virtual exit, which yields control dependence as the post-dominance frontier. + +Expected to remain red until ``coretrace_python.cfg.dominance`` exists. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python.analysis import AnalysisManager +from coretrace_python.cfg import CFG, BlockId, Branch, CFGAnalysis, build_cfg +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.source import SourceManager + +try: + from coretrace_python.cfg.dominance import ( + EXIT, + DominanceAnalysis, + DominatorTree, + PostDominanceAnalysis, + dominator_tree, + post_dominator_tree, + ) +except ImportError as error: # pragma: no cover - red until dominance lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_dominance() -> None: + if MISSING is not None: + pytest.fail(f"dominance is not implemented yet: {MISSING}") + + +def function(source_text: str) -> nodes.Function: + module = build_hir(SourceManager().add_source("dom.py", source_text)) + return next(s for s in module.body if isinstance(s, nodes.Function)) + + +def cfg_for(source_text: str) -> CFG: + return build_cfg(function(source_text)) + + +def b(name: str) -> BlockId: + return BlockId(name) + + +IF_ELSE = "def f(c, x):\n if c:\n x = 1\n else:\n x = 2\n return x\n" +WHILE = "def f(n):\n while n > 0:\n n = n - 1\n return n\n" +EARLY_RETURN = "def f(a):\n if a:\n return 1\n return 0\n" + + +# --------------------------------------------------------------------------- dominators + + +def test_entry_has_no_dominator_and_dominates_everything() -> None: + tree = dominator_tree(cfg_for(IF_ELSE)) + + assert isinstance(tree, DominatorTree) + assert tree.root == b("entry") + assert tree.idom(b("entry")) is None + for block in ("then_1", "else_1", "merge_1"): + assert tree.dominates(b("entry"), b(block)) + assert tree.dominates(b("entry"), b("entry")) + + +def test_if_else_immediate_dominators_and_frontiers() -> None: + tree = dominator_tree(cfg_for(IF_ELSE)) + + assert tree.idom(b("then_1")) == b("entry") + assert tree.idom(b("else_1")) == b("entry") + assert tree.idom(b("merge_1")) == b("entry") + assert tree.children(b("entry")) == (b("then_1"), b("else_1"), b("merge_1")) + assert tree.frontier(b("then_1")) == frozenset({b("merge_1")}) + assert tree.frontier(b("else_1")) == frozenset({b("merge_1")}) + assert tree.frontier(b("merge_1")) == frozenset() + assert not tree.dominates(b("then_1"), b("merge_1")) + + +def test_loop_header_dominates_its_body_and_exit() -> None: + tree = dominator_tree(cfg_for(WHILE)) + + assert tree.idom(b("loop_1")) == b("entry") + assert tree.idom(b("body_1")) == b("loop_1") + assert tree.idom(b("exit_1")) == b("loop_1") + assert tree.frontier(b("body_1")) == frozenset({b("loop_1")}) + assert tree.frontier(b("loop_1")) == frozenset({b("loop_1")}) + assert tree.dominates(b("loop_1"), b("exit_1")) + + +def test_unreachable_blocks_are_outside_the_tree() -> None: + tree = dominator_tree(cfg_for("def f():\n return 1\n x = 2\n")) + + assert tree.blocks == (b("entry"),) + assert not tree.dominates(b("entry"), b("dead_1")) + with pytest.raises(KeyError): + tree.idom(b("dead_1")) + + +def test_blocks_are_listed_in_dominator_tree_preorder() -> None: + tree = dominator_tree(cfg_for(WHILE)) + assert tree.blocks == (b("entry"), b("loop_1"), b("body_1"), b("exit_1")) + + +def test_iterated_frontier_is_the_phi_placement_set() -> None: + tree = dominator_tree(cfg_for("def f(c, d):\n if c:\n x = 1\n if d:\n x = 2\n return x\n")) + + assert tree.iterated_frontier(frozenset({b("then_1")})) == frozenset({b("merge_1")}) + assert tree.iterated_frontier(frozenset({b("then_1"), b("then_2")})) == frozenset( + {b("merge_1"), b("merge_2")} + ) + + +# --------------------------------------------------------------------------- post-dominators + + +def test_post_dominators_join_at_the_merge() -> None: + tree = post_dominator_tree(cfg_for(IF_ELSE)) + + assert tree.root == EXIT + assert tree.idom(b("merge_1")) == EXIT + assert tree.idom(b("then_1")) == b("merge_1") + assert tree.idom(b("else_1")) == b("merge_1") + assert tree.idom(b("entry")) == b("merge_1") + assert tree.dominates(b("merge_1"), b("entry")) + + +def test_early_returns_post_dominate_through_the_virtual_exit() -> None: + tree = post_dominator_tree(cfg_for(EARLY_RETURN)) + + assert tree.idom(b("then_1")) == EXIT + assert tree.idom(b("merge_1")) == EXIT + assert tree.idom(b("entry")) == EXIT + assert not tree.dominates(b("merge_1"), b("entry")) + + +def test_control_dependence_is_the_post_dominance_frontier() -> None: + tree = post_dominator_tree(cfg_for(EARLY_RETURN)) + + assert tree.frontier(b("then_1")) == frozenset({b("entry")}) + assert tree.frontier(b("merge_1")) == frozenset({b("entry")}) + assert tree.frontier(b("entry")) == frozenset() + + +def test_guard_before_sink_is_detected_by_dominance() -> None: + # The §24 refutation example: the sink only runs when the guard passed. + cfg = cfg_for( + "def f(value):\n" + " if not value.isdigit():\n" + " return None\n" + " sink(value)\n" + " return value\n" + ) + entry = cfg.block(cfg.entry).terminator + assert isinstance(entry, Branch) + tree = dominator_tree(cfg) + + assert tree.idom(entry.else_block) == cfg.entry + assert post_dominator_tree(cfg).frontier(entry.else_block) == frozenset({cfg.entry}) + + +# --------------------------------------------------------------------------- analyses + + +def test_dominance_analyses_require_the_cfg() -> None: + module = build_hir(SourceManager().add_source("dom.py", IF_ELSE)) + manager = AnalysisManager(module) + manager.register(CFGAnalysis, DominanceAnalysis, PostDominanceAnalysis) + target = next(s for s in module.body if isinstance(s, nodes.Function)) + + assert CFGAnalysis in DominanceAnalysis.requires + assert CFGAnalysis in PostDominanceAnalysis.requires + assert DominanceAnalysis.name == "cfg.dominance" + assert PostDominanceAnalysis.name == "cfg.post_dominance" + assert manager.get(DominanceAnalysis, target).idom(b("merge_1")) == b("entry") + assert manager.get(PostDominanceAnalysis, target).idom(b("entry")) == b("merge_1") diff --git a/tests/test_emit_ir.py b/tests/test_emit_ir.py index c141f9b..7f4cae3 100644 --- a/tests/test_emit_ir.py +++ b/tests/test_emit_ir.py @@ -222,3 +222,94 @@ def test_emit_ir_keeps_unreachable_code(tmp_path, capsys) -> None: assert main(["--emit-ir", str(source)]) == 0 output = capsys.readouterr().out assert "dead_1:\n %1 = const 2\n" in output + + +# --------------------------------------------------------------------------- SSA +# ``--emit-ir --ssa`` prints the SSA form derived from the CFG (docs/architecture.md §7). +# Expected to remain red until SSA construction lands. + + +def test_emit_ssa_for_the_doc_example(tmp_path, capsys) -> None: + # The example of §7: ``x = a; if cond: x = b; use(x)``. + source = tmp_path / "phi.py" + source.write_text( + "def f(a, b, cond):\n" + " x = a\n" + " if cond:\n" + " x = b\n" + " use(x)\n", + encoding="utf-8", + ) + + assert main(["--emit-ir", "--ssa", str(source)]) == 0 + assert capsys.readouterr().out == ( + "func @f(%0, %1, %2) {\n" + "entry:\n" + " branch %2, then_1, merge_1\n" + "then_1:\n" + " jump merge_1\n" + "merge_1:\n" + ' %3 = phi "x", [%0, entry], [%1, then_1]\n' + " %4 = global 'use'\n" + " %5 = call %4(%3)\n" + " return\n" + "}\n" + ) + + +def test_emit_ssa_for_a_for_loop(tmp_path, capsys) -> None: + source = tmp_path / "sum.py" + source.write_text( + "def f(items):\n" + " total = 0\n" + " for item in items:\n" + " total = total + item\n" + " return total\n", + encoding="utf-8", + ) + + assert main(["--emit-ir", "--ssa", str(source)]) == 0 + assert capsys.readouterr().out == ( + "func @f(%0) {\n" + "entry:\n" + " %1 = const 0\n" + " %2 = get_iter %0\n" + " jump loop_1\n" + "loop_1:\n" + ' %3 = phi "total", [%1, entry], [%5, body_1]\n' + ' %4 = for_next %2 -> "item", body_1, exit_1\n' + "body_1:\n" + " %5 = binary.add %3, %4\n" + " jump loop_1\n" + "exit_1:\n" + " return %3\n" + "}\n" + ) + + +def test_emit_ssa_shows_undefined_reads(tmp_path, capsys) -> None: + source = tmp_path / "maybe.py" + source.write_text("def f(c):\n if c:\n x = 1\n return x\n", encoding="utf-8") + + assert main(["--emit-ir", "--ssa", str(source)]) == 0 + assert capsys.readouterr().out == ( + "func @f(%0) {\n" + "entry:\n" + ' %1 = undefined "x"\n' + " branch %0, then_1, merge_1\n" + "then_1:\n" + " %2 = const 1\n" + " jump merge_1\n" + "merge_1:\n" + ' %3 = phi "x", [%1, entry], [%2, then_1]\n' + " return %3\n" + "}\n" + ) + + +def test_ssa_flag_requires_emit_ir(tmp_path, capsys) -> None: + source = tmp_path / "x.py" + source.write_text("", encoding="utf-8") + + assert main(["--ssa", str(source)]) == 2 + assert "no action selected" in capsys.readouterr().err diff --git a/tests/test_ssa.py b/tests/test_ssa.py new file mode 100644 index 0000000..d7bbc96 --- /dev/null +++ b/tests/test_ssa.py @@ -0,0 +1,275 @@ +"""Acceptance tests for SSA construction and def-use chains (``docs/architecture.md`` §7). + +``SSAAnalysis`` derives a new immutable ``FunctionIR`` from the non-SSA PyIR: locals +disappear, every value is defined exactly once, merges get ``Phi`` instructions, a +``for_next`` terminator defines the loop variable, and a local read before any +definition on some path reads an explicit ``Undefined`` value. Values are renumbered +densely in block order. ``DefUseAnalysis`` indexes definitions and uses over that IR. + +Expected to remain red until ``coretrace_python.ir.ssa`` and ``coretrace_python.ir.defuse`` +exist. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python.analysis import AnalysisManager +from coretrace_python.cfg import BlockId, CFGAnalysis +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.ir.lowering import PyIRAnalysis +from coretrace_python.ir.model import ( + Branch, + Call, + ForNext, + FunctionIR, + Global, + LoadLocal, + Return, + StoreLocal, + Value, +) +from coretrace_python.semantic import SEMANTIC_ANALYSES +from coretrace_python.source import SourceManager + +try: + from coretrace_python.cfg.dominance import DominanceAnalysis + from coretrace_python.ir.defuse import Definition, DefUse, DefUseAnalysis, Use + from coretrace_python.ir.model import Phi, Undefined + from coretrace_python.ir.ssa import SSAAnalysis, to_ssa +except ImportError as error: # pragma: no cover - red until SSA lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_ssa() -> None: + if MISSING is not None: + pytest.fail(f"SSA is not implemented yet: {MISSING}") + + +DOC_EXAMPLE = "def f(a, b, cond):\n x = a\n if cond:\n x = b\n use(x)\n" + + +def analyses(source_text: str) -> tuple[AnalysisManager, nodes.Function]: + module = build_hir(SourceManager().add_source("ssa.py", source_text)) + manager = AnalysisManager(module) + manager.register( + *SEMANTIC_ANALYSES, + CFGAnalysis, + DominanceAnalysis, + PyIRAnalysis, + SSAAnalysis, + DefUseAnalysis, + ) + return manager, next(s for s in module.body if isinstance(s, nodes.Function)) + + +def ssa(source_text: str) -> FunctionIR: + manager, target = analyses(source_text) + return manager.get(SSAAnalysis, target) + + +def block(function: FunctionIR, name: str): # type: ignore[no-untyped-def] + return next(b for b in function.blocks if b.id == BlockId(name)) + + +def defined_values(function: FunctionIR) -> list[Value]: + values = list(function.parameters) + for basic_block in function.blocks: + for instruction in basic_block.instructions: + if instruction.result is not None: + values.append(instruction.result) + if isinstance(basic_block.terminator, ForNext) and basic_block.terminator.result: + values.append(basic_block.terminator.result) + return values + + +# --------------------------------------------------------------------------- the §7 example + + +def test_doc_example_merges_with_one_phi() -> None: + function = ssa(DOC_EXAMPLE) + merge = block(function, "merge_1") + + phi = merge.instructions[0] + assert isinstance(phi, Phi) + assert phi.name == "x" + assert phi.incoming == ( + (function.parameters[0], BlockId("entry")), + (function.parameters[1], BlockId("then_1")), + ) + call = merge.instructions[2] + assert isinstance(call, Call) + assert call.arguments == (phi.result,) + + +def test_ssa_has_no_locals_and_defines_each_value_once() -> None: + function = ssa(DOC_EXAMPLE) + + for basic_block in function.blocks: + assert not any(isinstance(i, LoadLocal | StoreLocal) for i in basic_block.instructions) + values = defined_values(function) + assert len(values) == len(set(values)) + + +def test_values_are_renumbered_densely_in_block_order() -> None: + function = ssa(DOC_EXAMPLE) + assert [value.id for value in defined_values(function)] == list(range(6)) + + +def test_trivial_stores_do_not_create_instructions() -> None: + # ``x = a`` needs no copy: the local simply refers to the parameter value. + function = ssa(DOC_EXAMPLE) + assert block(function, "entry").instructions == () + assert block(function, "then_1").instructions == () + + +# --------------------------------------------------------------------------- loops + + +def test_loop_header_phi_takes_the_back_edge_value() -> None: + function = ssa("def f(n):\n while n > 0:\n n = n - 1\n return n\n") + header = block(function, "loop_1") + body = block(function, "body_1") + + phi = header.instructions[0] + assert isinstance(phi, Phi) + assert phi.name == "n" + assert phi.incoming[0] == (function.parameters[0], BlockId("entry")) + assert phi.incoming[1] == (body.instructions[-1].result, BlockId("body_1")) + assert isinstance(header.terminator, Branch) + exit_block = block(function, "exit_1") + assert isinstance(exit_block.terminator, Return) + assert exit_block.terminator.value == phi.result + + +def test_for_next_defines_the_loop_variable() -> None: + function = ssa( + "def f(items):\n total = 0\n for item in items:\n total = total + item\n return total\n" + ) + header = block(function, "loop_1") + body = block(function, "body_1") + + assert isinstance(header.terminator, ForNext) + assert header.terminator.result is not None + assert header.terminator.target == "item" + add = body.instructions[0] + assert add.operands() == (header.instructions[0].result, header.terminator.result) # type: ignore[union-attr] + + +def test_loop_variable_after_the_loop_is_a_phi_at_the_header() -> None: + function = ssa("def f(items):\n for item in items:\n pass\n return item\n") + header = block(function, "loop_1") + exit_block = block(function, "exit_1") + + phi = header.instructions[0] + assert isinstance(phi, Phi) + assert phi.name == "item" + assert isinstance(header.terminator, ForNext) + assert phi.incoming[1] == (header.terminator.result, BlockId("body_1")) + assert isinstance(exit_block.terminator, Return) + assert exit_block.terminator.value == phi.result + + +# --------------------------------------------------------------------------- undefined + + +def test_possibly_unassigned_local_reads_an_explicit_undefined_value() -> None: + function = ssa("def f(c):\n if c:\n x = 1\n return x\n") + entry = block(function, "entry") + merge = block(function, "merge_1") + + undefined = entry.instructions[0] + assert isinstance(undefined, Undefined) + assert undefined.name == "x" + phi = merge.instructions[0] + assert isinstance(phi, Phi) + assert phi.incoming == ( + (undefined.result, BlockId("entry")), + (block(function, "then_1").instructions[0].result, BlockId("then_1")), + ) + + +def test_undefined_is_only_introduced_when_needed() -> None: + function = ssa(DOC_EXAMPLE) + assert not any(isinstance(i, Undefined) for b in function.blocks for i in b.instructions) + + +# --------------------------------------------------------------------------- invariants + + +def test_phi_incoming_edges_match_the_predecessors() -> None: + manager, target = analyses( + "def f(items, flag):\n" + " x = 0\n" + " for item in items:\n" + " if flag:\n" + " continue\n" + " x = x + item\n" + " return x\n" + ) + cfg = manager.get(CFGAnalysis, target) + function = manager.get(SSAAnalysis, target) + + for basic_block in function.blocks: + for instruction in basic_block.instructions: + if isinstance(instruction, Phi): + assert tuple(b for _, b in instruction.incoming) == cfg.predecessors(basic_block.id) + + +def test_ssa_is_a_derived_immutable_ir() -> None: + manager, target = analyses(DOC_EXAMPLE) + + original = manager.get(PyIRAnalysis, target) + derived = manager.get(SSAAnalysis, target) + + assert derived is not original + assert any(isinstance(i, StoreLocal) for b in original.blocks for i in b.instructions) + assert manager.get(SSAAnalysis, target) is derived + assert SSAAnalysis.name == "ir.ssa" + assert {PyIRAnalysis, DominanceAnalysis} <= SSAAnalysis.requires + + +def test_to_ssa_is_a_pure_function() -> None: + manager, target = analyses(DOC_EXAMPLE) + pyir = manager.get(PyIRAnalysis, target) + dominance = manager.get(DominanceAnalysis, target) + + assert to_ssa(pyir, dominance) == to_ssa(pyir, dominance) + + +# --------------------------------------------------------------------------- def-use + + +def test_def_use_indexes_definitions_and_uses() -> None: + manager, target = analyses(DOC_EXAMPLE) + function = manager.get(SSAAnalysis, target) + chains = manager.get(DefUseAnalysis, target) + merge = block(function, "merge_1") + phi, callee, call = merge.instructions + + assert isinstance(chains, DefUse) + assert chains.definition(function.parameters[2]) == Definition(None, None) + assert chains.definition(phi.result) == Definition(BlockId("merge_1"), 0) + assert chains.uses(function.parameters[2]) == (Use(BlockId("entry"), None),) + assert chains.uses(function.parameters[0]) == (Use(BlockId("merge_1"), 0),) + assert chains.uses(phi.result) == (Use(BlockId("merge_1"), 2),) + assert chains.uses(call.result) == () + assert isinstance(callee, Global) + + +def test_def_use_records_terminator_definitions() -> None: + manager, target = analyses("def f(items):\n for item in items:\n use(item)\n return 0\n") + function = manager.get(SSAAnalysis, target) + chains = manager.get(DefUseAnalysis, target) + header = block(function, "loop_1") + assert isinstance(header.terminator, ForNext) + assert header.terminator.result is not None + + assert chains.definition(header.terminator.result) == Definition(BlockId("loop_1"), None) + assert chains.uses(header.terminator.result) == (Use(BlockId("body_1"), 1),) + assert DefUseAnalysis.name == "ir.defuse" + assert SSAAnalysis in DefUseAnalysis.requires From e6a062f2a8c2d38a016b0952bfa764fe210ef2e4 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 3 Sep 2026 07:23:02 +0200 Subject: [PATCH 2/2] feat: add dominance, SSA construction and def-use chains --- README.md | 4 +- src/coretrace_python/cfg/__init__.py | 14 ++ src/coretrace_python/cfg/dominance.py | 183 +++++++++++++++++ src/coretrace_python/cli.py | 7 +- src/coretrace_python/ir/defuse.py | 74 +++++++ src/coretrace_python/ir/lowering.py | 65 +++--- src/coretrace_python/ir/model.py | 107 ++++++++-- src/coretrace_python/ir/printer.py | 10 +- src/coretrace_python/ir/ssa.py | 271 ++++++++++++++++++++++++++ 9 files changed, 689 insertions(+), 46 deletions(-) create mode 100644 src/coretrace_python/cfg/dominance.py create mode 100644 src/coretrace_python/ir/defuse.py create mode 100644 src/coretrace_python/ir/ssa.py diff --git a/README.md b/README.md index 846941a..5f38672 100644 --- a/README.md +++ b/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. diff --git a/src/coretrace_python/cfg/__init__.py b/src/coretrace_python/cfg/__init__.py index ea91118..4f31890 100644 --- a/src/coretrace_python/cfg/__init__.py +++ b/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", ] diff --git a/src/coretrace_python/cfg/dominance.py b/src/coretrace_python/cfg/dominance.py new file mode 100644 index 0000000..83ae5ba --- /dev/null +++ b/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("") + + +@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)) diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index 2947daf..9ce54fb 100644 --- a/src/coretrace_python/cli.py +++ b/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, diff --git a/src/coretrace_python/ir/defuse.py b/src/coretrace_python/ir/defuse.py new file mode 100644 index 0000000..24187bf --- /dev/null +++ b/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)) diff --git a/src/coretrace_python/ir/lowering.py b/src/coretrace_python/ir/lowering.py index a01ebbd..923149e 100644 --- a/src/coretrace_python/ir/lowering.py +++ b/src/coretrace_python/ir/lowering.py @@ -259,6 +259,30 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> FunctionIR: return lowerer.function(function) +def top_level_functions(module: nodes.Module) -> tuple[nodes.Function, ...]: + """The functions a module lowers to, rejecting module syntax outside the subset.""" + + functions: list[nodes.Function] = [] + for statement in module.body: + if isinstance(statement, nodes.Function): + functions.append(statement) + elif isinstance(statement, (nodes.Import, nodes.ImportFrom)): + continue + elif ( + isinstance(statement, nodes.ExpressionStatement) + and isinstance(statement.expression, nodes.Constant) + and isinstance(statement.expression.value, str) + ): + # Permit module docstrings. + continue + else: + raise LoweringError( + f"{statement.span.display()}: " + f"unsupported module syntax: {type(statement).__name__}" + ) + return tuple(functions) + + class ModuleIRAnalysis(Analysis[ModuleIR]): """Assemble the PyIR of every top-level function of the module.""" @@ -267,30 +291,23 @@ class ModuleIRAnalysis(Analysis[ModuleIR]): @classmethod def compute(cls, ctx: AnalysisContext) -> ModuleIR: - functions: list[FunctionIR] = [] - for statement in ctx.module.body: - if isinstance(statement, nodes.Function): - functions.append(ctx.get(PyIRAnalysis, statement)) - elif isinstance(statement, (nodes.Import, nodes.ImportFrom)): - continue - elif ( - isinstance(statement, nodes.ExpressionStatement) - and isinstance(statement.expression, nodes.Constant) - and isinstance(statement.expression.value, str) - ): - # Permit module docstrings. - continue - else: - raise LoweringError( - f"{statement.span.display()}: " - f"unsupported module syntax: {type(statement).__name__}" - ) - return ModuleIR(tuple(functions)) - - -def lower_module(module: nodes.Module) -> ModuleIR: - """Lower a whole module through a fresh analysis manager.""" + return ModuleIR( + tuple(ctx.get(PyIRAnalysis, function) for function in top_level_functions(ctx.module)) + ) + + +def lower_module(module: nodes.Module, *, ssa: bool = False) -> ModuleIR: + """Lower a whole module through a fresh analysis manager, optionally to SSA form.""" manager = AnalysisManager(module) manager.register(*SEMANTIC_ANALYSES, CFGAnalysis, PyIRAnalysis, ModuleIRAnalysis) - return manager.get(ModuleIRAnalysis) + if not ssa: + return manager.get(ModuleIRAnalysis) + # Imported here because the SSA pass is built on top of this module's analyses. + from coretrace_python.cfg.dominance import DominanceAnalysis + from coretrace_python.ir.ssa import SSAAnalysis + + manager.register(DominanceAnalysis, SSAAnalysis) + return ModuleIR( + tuple(manager.get(SSAAnalysis, function) for function in top_level_functions(module)) + ) diff --git a/src/coretrace_python/ir/model.py b/src/coretrace_python/ir/model.py index a129fa3..f093e01 100644 --- a/src/coretrace_python/ir/model.py +++ b/src/coretrace_python/ir/model.py @@ -1,7 +1,8 @@ from __future__ import annotations -from dataclasses import dataclass -from typing import TypeAlias +from collections.abc import Callable +from dataclasses import dataclass, fields, replace +from typing import Any, TypeAlias, TypeVar from coretrace_python.cfg import BlockId from coretrace_python.semantic.symbols import SymbolId @@ -13,29 +14,75 @@ class Value: id: int +class Operands: + """Generic operand access for instructions and terminators.""" + + def operands(self) -> tuple[Value, ...]: + found: list[Value] = [] + for field_info in fields(self): # type: ignore[arg-type] + if field_info.name == "result": + continue + value = getattr(self, field_info.name) + if isinstance(value, Value): + found.append(value) + elif isinstance(value, tuple): + for item in value: + if isinstance(item, Value): + found.append(item) + elif isinstance(item, tuple) and item and isinstance(item[0], Value): + found.append(item[0]) + return tuple(found) + + +N = TypeVar("N", bound=Operands) + + +def substitute(node: N, mapping: Callable[[Value], Value], *, include_result: bool = False) -> N: + """Copy ``node`` with every operand (and optionally its result) passed through ``mapping``.""" + + changes: dict[str, Any] = {} + for field_info in fields(node): # type: ignore[arg-type] + value = getattr(node, field_info.name) + if field_info.name == "result": + if include_result and isinstance(value, Value): + changes["result"] = mapping(value) + elif isinstance(value, Value): + changes[field_info.name] = mapping(value) + elif isinstance(value, tuple): + changes[field_info.name] = tuple( + mapping(item) + if isinstance(item, Value) + else (mapping(item[0]), *item[1:]) + if isinstance(item, tuple) and item and isinstance(item[0], Value) + else item + for item in value + ) + return replace(node, **changes) # type: ignore[type-var] + + @dataclass(frozen=True) -class Constant: +class Constant(Operands): result: Value location: SourceSpan value: object @dataclass(frozen=True) -class Global: +class Global(Operands): result: Value location: SourceSpan name: str @dataclass(frozen=True) -class Symbol: +class Symbol(Operands): result: Value location: SourceSpan symbol_id: SymbolId @dataclass(frozen=True) -class BinaryOp: +class BinaryOp(Operands): result: Value location: SourceSpan operator: str @@ -44,7 +91,7 @@ class BinaryOp: @dataclass(frozen=True) -class UnaryOp: +class UnaryOp(Operands): result: Value location: SourceSpan operator: str @@ -52,7 +99,7 @@ class UnaryOp: @dataclass(frozen=True) -class Compare: +class Compare(Operands): result: Value location: SourceSpan operator: str @@ -61,7 +108,7 @@ class Compare: @dataclass(frozen=True) -class Call: +class Call(Operands): result: Value location: SourceSpan callee: Value @@ -69,7 +116,7 @@ class Call: @dataclass(frozen=True) -class GetAttr: +class GetAttr(Operands): result: Value location: SourceSpan object: Value @@ -77,7 +124,7 @@ class GetAttr: @dataclass(frozen=True) -class GetItem: +class GetItem(Operands): result: Value location: SourceSpan object: Value @@ -85,21 +132,40 @@ class GetItem: @dataclass(frozen=True) -class GetIter: +class GetIter(Operands): result: Value location: SourceSpan iterable: Value @dataclass(frozen=True) -class LoadLocal: +class LoadLocal(Operands): + result: Value + location: SourceSpan + name: str + + +@dataclass(frozen=True) +class Phi(Operands): + """Merge of the values of local ``name`` arriving from each predecessor (SSA only).""" + + result: Value + location: SourceSpan + name: str + incoming: tuple[tuple[Value, BlockId], ...] + + +@dataclass(frozen=True) +class Undefined(Operands): + """The value of local ``name`` on a path where it was never assigned (SSA only).""" + result: Value location: SourceSpan name: str @dataclass(frozen=True) -class StoreLocal: +class StoreLocal(Operands): result: None location: SourceSpan name: str @@ -118,6 +184,8 @@ class StoreLocal: | GetItem | GetIter | LoadLocal + | Phi + | Undefined ) EffectInstruction: TypeAlias = StoreLocal Instruction: TypeAlias = ValueInstruction | EffectInstruction @@ -127,13 +195,13 @@ class StoreLocal: @dataclass(frozen=True) -class Return: +class Return(Operands): location: SourceSpan value: Value | None @dataclass(frozen=True) -class Branch: +class Branch(Operands): location: SourceSpan condition: Value then_block: BlockId @@ -141,19 +209,19 @@ class Branch: @dataclass(frozen=True) -class Jump: +class Jump(Operands): location: SourceSpan target: BlockId @dataclass(frozen=True) -class Raise: +class Raise(Operands): location: SourceSpan exception: Value | None @dataclass(frozen=True) -class ForNext: +class ForNext(Operands): """Store the next item of ``iterator`` into local ``target`` and enter ``body``, or ``exit``.""" location: SourceSpan @@ -161,6 +229,7 @@ class ForNext: target: str body: BlockId exit: BlockId + result: Value | None = None Terminator: TypeAlias = Return | Branch | Jump | Raise | ForNext diff --git a/src/coretrace_python/ir/printer.py b/src/coretrace_python/ir/printer.py index a41f9f4..06d695a 100644 --- a/src/coretrace_python/ir/printer.py +++ b/src/coretrace_python/ir/printer.py @@ -15,12 +15,14 @@ Jump, LoadLocal, ModuleIR, + Phi, Raise, Return, StoreLocal, Symbol, Terminator, UnaryOp, + Undefined, Value, ) @@ -70,6 +72,11 @@ def _instruction(instruction: Instruction) -> str: return f'{_value(instruction.result)} = load_local "{instruction.name}"' if isinstance(instruction, StoreLocal): return f'store_local "{instruction.name}", {_value(instruction.value)}' + if isinstance(instruction, Phi): + incoming = ", ".join(f"[{_value(v)}, {b}]" for v, b in instruction.incoming) + return f'{_value(instruction.result)} = phi "{instruction.name}", {incoming}' + if isinstance(instruction, Undefined): + return f'{_value(instruction.result)} = undefined "{instruction.name}"' raise TypeError(f"unknown instruction: {instruction!r}") @@ -86,8 +93,9 @@ def _terminator(terminator: Terminator) -> str: if isinstance(terminator, Raise): return "raise" if terminator.exception is None else f"raise {_value(terminator.exception)}" if isinstance(terminator, ForNext): + prefix = "" if terminator.result is None else f"{_value(terminator.result)} = " return ( - f'for_next {_value(terminator.iterator)} -> "{terminator.target}", ' + f'{prefix}for_next {_value(terminator.iterator)} -> "{terminator.target}", ' f"{terminator.body}, {terminator.exit}" ) raise TypeError(f"unknown terminator: {terminator!r}") diff --git a/src/coretrace_python/ir/ssa.py b/src/coretrace_python/ir/ssa.py new file mode 100644 index 0000000..dd0a46c --- /dev/null +++ b/src/coretrace_python/ir/ssa.py @@ -0,0 +1,271 @@ +"""SSA construction (architecture §7). + +``to_ssa`` derives a new immutable ``FunctionIR`` from the non-SSA PyIR of a function: +local loads and stores disappear, merges get ``Phi`` instructions placed on the iterated +dominance frontier of the definitions and pruned by liveness, a ``for_next`` terminator +defines the loop variable, and a local read before any definition on some path reads +an explicit ``Undefined`` value created once in the entry block. Only reachable blocks +are kept, and values are renumbered densely in block order. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import replace +from typing import ClassVar + +from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis +from coretrace_python.cfg import BlockId +from coretrace_python.cfg.dominance import DominanceAnalysis, DominatorTree +from coretrace_python.hir import nodes +from coretrace_python.ir.lowering import PyIRAnalysis +from coretrace_python.ir.model import ( + BasicBlock, + Branch, + ForNext, + FunctionIR, + Instruction, + Jump, + LoadLocal, + Phi, + StoreLocal, + Terminator, + Undefined, + Value, + substitute, +) + + +class _Renamer: + def __init__(self, function: FunctionIR, tree: DominatorTree) -> None: + self.function = function + self.tree = tree + self.blocks = {block.id: block for block in function.blocks if block.id in tree.blocks} + self.next_id = max((v.id for v in _all_values(function)), default=-1) + 1 + self.stacks: dict[str, list[Value]] = {} + self.aliases: dict[Value, Value] = {} + self.undefined: dict[str, Undefined] = {} + self.phis: dict[BlockId, dict[str, Phi]] = {} + self.instructions: dict[BlockId, list[Instruction]] = {} + self.terminators: dict[BlockId, Terminator] = {} + self.predecessors: dict[BlockId, list[BlockId]] = {b: [] for b in self.blocks} + for block in self.blocks.values(): + for successor in _targets(block.terminator): + self.predecessors[successor].append(block.id) + self.loop_variables: dict[BlockId, tuple[str, BlockId]] = {} + for block in self.blocks.values(): + if isinstance(block.terminator, ForNext): + self.loop_variables[block.terminator.body] = (block.terminator.target, block.id) + + # ------------------------------------------------------------------ helpers + + def new_value(self) -> Value: + value = Value(self.next_id) + self.next_id += 1 + return value + + def current(self, name: str) -> Value: + stack = self.stacks.get(name) + if stack: + return stack[-1] + if name not in self.undefined: + self.undefined[name] = Undefined(self.new_value(), self.function.location, name) + return self.undefined[name].result + + def resolve(self, value: Value) -> Value: + return self.aliases.get(value, value) + + # ------------------------------------------------------------------ phi placement + + def place_phis(self) -> None: + definitions: dict[str, set[BlockId]] = {} + for block in self.blocks.values(): + for instruction in block.instructions: + if isinstance(instruction, StoreLocal): + definitions.setdefault(instruction.name, set()).add(block.id) + if isinstance(block.terminator, ForNext): + definitions.setdefault(block.terminator.target, set()).add(block.terminator.body) + live_in = _live_in(self.blocks, self.predecessors, self.loop_variables) + for name, blocks in definitions.items(): + for block_id in sorted(self.tree.iterated_frontier(blocks), key=self.order): + if name in live_in[block_id]: + self.phis.setdefault(block_id, {})[name] = Phi( + self.new_value(), self.blocks[block_id].terminator.location, name, () + ) + + def order(self, block_id: BlockId) -> int: + return self.tree.blocks.index(block_id) + + # ------------------------------------------------------------------ renaming + + def rename(self, block_id: BlockId) -> None: + pushed: list[str] = [] + + def push(name: str, value: Value) -> None: + self.stacks.setdefault(name, []).append(value) + pushed.append(name) + + if block_id in self.loop_variables: + name, header = self.loop_variables[block_id] + header_terminator = self.terminators[header] + assert isinstance(header_terminator, ForNext) and header_terminator.result is not None + push(name, header_terminator.result) + for name, phi in self.phis.get(block_id, {}).items(): + push(name, phi.result) + + kept: list[Instruction] = [] + for instruction in self.blocks[block_id].instructions: + if isinstance(instruction, LoadLocal): + self.aliases[instruction.result] = self.current(instruction.name) + elif isinstance(instruction, StoreLocal): + push(instruction.name, self.resolve(instruction.value)) + else: + kept.append(substitute(instruction, self.resolve)) + self.instructions[block_id] = kept + + terminator = substitute(self.blocks[block_id].terminator, self.resolve) + if isinstance(terminator, ForNext): + terminator = replace(terminator, result=self.new_value()) + self.terminators[block_id] = terminator + + for successor in _targets(terminator): + for name, phi in self.phis.get(successor, {}).items(): + incoming = (*phi.incoming, (self.current(name), block_id)) + self.phis[successor][name] = replace(phi, incoming=incoming) + + for child in self.tree.children(block_id): + self.rename(child) + for name in reversed(pushed): + self.stacks[name].pop() + + # ------------------------------------------------------------------ assembly + + def build(self) -> FunctionIR: + self.place_phis() + self.rename(self.tree.root) + blocks: list[BasicBlock] = [] + for block in self.function.blocks: + if block.id not in self.blocks: + continue + instructions: list[Instruction] = [] + if block.id == self.function.entry: + instructions.extend(self.undefined.values()) + instructions.extend(self.ordered_phis(block.id)) + instructions.extend(self.instructions[block.id]) + blocks.append(BasicBlock(block.id, tuple(instructions), self.terminators[block.id])) + return _renumber(replace(self.function, blocks=tuple(blocks))) + + def ordered_phis(self, block_id: BlockId) -> list[Phi]: + phis = self.phis.get(block_id, {}) + ordered_incoming = [] + for phi in phis.values(): + by_block = {b: v for v, b in phi.incoming} + incoming = tuple((by_block[p], p) for p in self.predecessors[block_id]) + ordered_incoming.append(replace(phi, incoming=incoming)) + return ordered_incoming + + +def _targets(terminator: Terminator) -> tuple[BlockId, ...]: + if isinstance(terminator, Branch): + return (terminator.then_block, terminator.else_block) + if isinstance(terminator, Jump): + return (terminator.target,) + if isinstance(terminator, ForNext): + return (terminator.body, terminator.exit) + return () + + +def _all_values(function: FunctionIR) -> list[Value]: + values = list(function.parameters) + for block in function.blocks: + for instruction in block.instructions: + if instruction.result is not None: + values.append(instruction.result) + return values + + +def _live_in( + blocks: Mapping[BlockId, BasicBlock], + predecessors: Mapping[BlockId, list[BlockId]], + loop_variables: Mapping[BlockId, tuple[str, BlockId]], +) -> dict[BlockId, frozenset[str]]: + """Backward liveness of local names over the non-SSA blocks.""" + + uses: dict[BlockId, set[str]] = {} + defs: dict[BlockId, set[str]] = {} + for block_id, block in blocks.items(): + used: set[str] = set() + defined: set[str] = set() + if block_id in loop_variables: + defined.add(loop_variables[block_id][0]) + for instruction in block.instructions: + if isinstance(instruction, LoadLocal) and instruction.name not in defined: + used.add(instruction.name) + elif isinstance(instruction, StoreLocal): + defined.add(instruction.name) + uses[block_id], defs[block_id] = used, defined + + successors: dict[BlockId, list[BlockId]] = {b: [] for b in blocks} + for block_id, preds in predecessors.items(): + for predecessor in preds: + successors[predecessor].append(block_id) + + live_in: dict[BlockId, frozenset[str]] = {b: frozenset() for b in blocks} + changed = True + while changed: + changed = False + for block_id in blocks: + live_out: set[str] = set() + for successor in successors[block_id]: + live_out |= live_in[successor] + new = frozenset(uses[block_id] | (live_out - defs[block_id])) + if new != live_in[block_id]: + live_in[block_id] = new + changed = True + return live_in + + +def _renumber(function: FunctionIR) -> FunctionIR: + mapping: dict[Value, Value] = {} + + def fresh(value: Value) -> None: + mapping[value] = Value(len(mapping)) + + for parameter in function.parameters: + fresh(parameter) + for block in function.blocks: + for instruction in block.instructions: + if instruction.result is not None: + fresh(instruction.result) + if isinstance(block.terminator, ForNext) and block.terminator.result is not None: + fresh(block.terminator.result) + + def renamed(value: Value) -> Value: + return mapping[value] + + blocks = tuple( + BasicBlock( + block.id, + tuple(substitute(i, renamed, include_result=True) for i in block.instructions), + substitute(block.terminator, renamed, include_result=True), + ) + for block in function.blocks + ) + return replace( + function, parameters=tuple(renamed(p) for p in function.parameters), blocks=blocks + ) + + +def to_ssa(function: FunctionIR, dominance: DominatorTree) -> FunctionIR: + return _Renamer(function, dominance).build() + + +class SSAAnalysis(FunctionAnalysis[FunctionIR]): + """SSA form of a function, derived from its PyIR and dominator tree.""" + + name: ClassVar[str] = "ir.ssa" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({PyIRAnalysis, DominanceAnalysis}) + + @classmethod + def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> FunctionIR: + return to_ssa(ctx.get(PyIRAnalysis, function), ctx.get(DominanceAnalysis, function))