feat(dataflow): stage 2 — post-dominators and control dependence

Cooper–Harper–Kennedy iterative post-dominators over the reverse CFG
(unique root EXIT, guaranteed by stage 1's synthetic escape edges) and
Ferrante–Ottenstein–Warren control dependence with ENTRY as the region
root. Gate tests pin exact hand-computed CDG sets for the fixture's
if/loop/early-return functions. (#67)

Author: rahlk

codeanalyzer/dataflow/dominance.py

@@ -0,0 +1,140 @@
+################################################################################
+# Copyright IBM Corporation 2025
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+"""Stage 2 of the level-3 dataflow ladder: dominance and control dependence.
+
+Post-dominators are computed with the Cooper–Harper–Kennedy iterative
+algorithm over the reverse CFG. Infinite loops are already normalized by the
+CFG builder (synthetic escape edge to EXIT), so the post-dominator tree always
+has the unique root EXIT.
+
+Control dependence follows Ferrante–Ottenstein–Warren: for each CFG edge
+``(a, b)`` where ``b`` does not post-dominate ``a``, every node on the
+post-dominator-tree path from ``b`` up to (but not including) ``a``'s
+immediate post-dominator is control-dependent on ``a``.
+
+Nodes with no branch-node control dependence are control-dependent on ENTRY —
+the conventional region root, which keeps every statement anchored in the PDG
+and gives interprocedural traversals a path from a callee's ENTRY to its
+unconditional statements.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, Set, Tuple
+
+from codeanalyzer.dataflow.cfg import ControlFlowGraph
+
+
+def _postorder(adj: Dict[int, List[int]], root: int) -> List[int]:
+    """Iterative DFS postorder over ``adj`` from ``root``."""
+    order: List[int] = []
+    visited: Set[int] = set()
+    stack: List[Tuple[int, int]] = [(root, 0)]
+    visited.add(root)
+    while stack:
+        node, i = stack.pop()
+        children = adj.get(node, [])
+        if i < len(children):
+            stack.append((node, i + 1))
+            child = children[i]
+            if child not in visited:
+                visited.add(child)
+                stack.append((child, 0))
+        else:
+            order.append(node)
+    return order
+
+
+def post_dominators(cfg: ControlFlowGraph) -> Dict[int, int]:
+    """Immediate post-dominator of every node, as ``{node: ipdom}``.
+
+    EXIT is its own post-dominator (the tree root). Cooper–Harper–Kennedy
+    ("A Simple, Fast Dominance Algorithm") run on the reverse CFG.
+    """
+    # Reverse CFG: successors of n are the CFG predecessors of n.
+    radj: Dict[int, List[int]] = {n.id: [] for n in cfg.nodes}
+    rpred: Dict[int, List[int]] = {n.id: [] for n in cfg.nodes}
+    for e in cfg.edges:
+        if e.source == e.target:
+            continue  # self-loops carry no dominance information
+        radj[e.target].append(e.source)
+        rpred[e.source].append(e.target)
+
+    root = cfg.exit_id
+    post = _postorder(radj, root)
+    number = {n: i for i, n in enumerate(post)}  # postorder number
+    rpo = list(reversed(post))  # reverse postorder: root first
+
+    ipdom: Dict[int, int] = {root: root}
+
+    def intersect(a: int, b: int) -> int:
+        while a != b:
+            while number[a] < number[b]:
+                a = ipdom[a]
+            while number[b] < number[a]:
+                b = ipdom[b]
+        return a
+
+    changed = True
+    while changed:
+        changed = False
+        for node in rpo:
+            if node == root:
+                continue
+            preds = [p for p in rpred[node] if p in ipdom]
+            if not preds:
+                continue
+            new = preds[0]
+            for p in preds[1:]:
+                new = intersect(new, p)
+            if ipdom.get(node) != new:
+                ipdom[node] = new
+                changed = True
+
+    return ipdom
+
+
+def control_dependence(cfg: ControlFlowGraph) -> List[Tuple[int, int]]:
+    """CDG edges ``(branch_node, dependent_node)`` per Ferrante–Ottenstein–
+    Warren, plus ENTRY-region edges for nodes with no other controller."""
+    ipdom = post_dominators(cfg)
+
+    deps: Set[Tuple[int, int]] = set()
+    for e in cfg.edges:
+        a, b = e.source, e.target
+        if a == b:
+            continue
+        # b post-dominates a iff b is an ancestor of a in the pdom tree.
+        runner = b
+        stop = ipdom.get(a)
+        # Walk from b up the post-dominator tree to (not including) ipdom(a).
+        while runner != stop and runner != a:
+            deps.add((a, runner))
+            nxt = ipdom.get(runner)
+            if nxt is None or nxt == runner:
+                break
+            runner = nxt
+
+    # ENTRY as the region root for otherwise-uncontrolled nodes.
+    controlled = {t for (_, t) in deps}
+    for n in cfg.nodes:
+        if n.id in (cfg.entry_id, cfg.exit_id):
+            continue
+        if n.id not in controlled:
+            deps.add((cfg.entry_id, n.id))
+
+    return sorted(deps)

test/test_dataflow_dominance.py

@@ -0,0 +1,104 @@
+"""Stage-2 gate: post-dominators and control dependence.
+
+Contract assertions:
+- the post-dominator tree is a tree with unique root EXIT (infinite loops
+  included, thanks to the CFG's synthetic escape edge);
+- hand-computed control dependences for the fixture's if / loop /
+  early-return functions match exactly.
+"""
+
+import ast
+from pathlib import Path
+
+from codeanalyzer.dataflow.cfg import ControlFlowGraph, build_cfg
+from codeanalyzer.dataflow.dominance import control_dependence, post_dominators
+
+FIXTURE = Path(__file__).parent / "fixtures" / "single_functionalities" / "dataflow"
+
+
+def _cfg_of(file_name: str, func_name: str) -> ControlFlowGraph:
+    tree = ast.parse((FIXTURE / file_name).read_text())
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == func_name:
+            return build_cfg(node)
+    raise AssertionError(f"{func_name} not found")
+
+
+def _all_fixture_cfgs():
+    cfgs = {}
+    for file_name in ("main.py", "pipeline.py", "state.py"):
+        tree = ast.parse((FIXTURE / file_name).read_text())
+        for node in ast.walk(tree):
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                cfgs[f"{file_name}::{node.name}"] = build_cfg(node)
+    return cfgs
+
+
+def _by_line(cfg: ControlFlowGraph):
+    """id ↔ line helpers for hand-computed expectations."""
+    return {n.start_line: n.id for n in cfg.nodes if n.kind not in ("entry", "exit")}
+
+
+def test_post_dominator_tree_is_rooted_at_exit_for_every_function():
+    for name, cfg in _all_fixture_cfgs().items():
+        ipdom = post_dominators(cfg)
+        ids = {n.id for n in cfg.nodes}
+        assert set(ipdom) == ids, f"{name}: some node has no post-dominator"
+        assert ipdom[cfg.exit_id] == cfg.exit_id, name
+        # Tree: walking up from any node terminates at EXIT without cycles.
+        for n in ids:
+            seen = set()
+            cur = n
+            while cur != cfg.exit_id:
+                assert cur not in seen, f"{name}: ipdom cycle at {cur}"
+                seen.add(cur)
+                cur = ipdom[cur]
+
+
+def test_branchy_control_dependence_exact():
+    cfg = _cfg_of("main.py", "branchy")
+    line = _by_line(cfg)
+    header, then_s, else_s, ret = line[12], line[13], line[15], line[16]
+    expected = {
+        (cfg.entry_id, header),
+        (cfg.entry_id, ret),
+        (header, then_s),
+        (header, else_s),
+    }
+    assert set(control_dependence(cfg)) == expected
+
+
+def test_looped_control_dependence_exact():
+    cfg = _cfg_of("main.py", "looped")
+    line = _by_line(cfg)
+    s_total, s_i, header, s_add, s_inc, ret = (
+        line[20], line[21], line[22], line[23], line[24], line[25],
+    )
+    expected = {
+        (cfg.entry_id, s_total),
+        (cfg.entry_id, s_i),
+        (cfg.entry_id, header),
+        (cfg.entry_id, ret),
+        (header, s_add),
+        (header, s_inc),
+    }
+    assert set(control_dependence(cfg)) == expected
+
+
+def test_early_exit_control_dependence_exact():
+    cfg = _cfg_of("main.py", "early_exit")
+    line = _by_line(cfg)
+    header, ret1, s_y, ret2 = line[29], line[30], line[31], line[32]
+    expected = {
+        (cfg.entry_id, header),
+        (header, ret1),
+        (header, s_y),
+        (header, ret2),
+    }
+    assert set(control_dependence(cfg)) == expected
+
+
+def test_infinite_loop_post_dominance_well_formed():
+    cfg = _cfg_of("main.py", "infinite")
+    ipdom = post_dominators(cfg)
+    assert set(ipdom) == {n.id for n in cfg.nodes}