diff --git a/.gitignore b/.gitignore index 216a671..b2c2eb1 100644 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,7 @@ benchmark_reports/ .mypy_cache/ .pytest_cache/ .ruff_cache/ +# CFG visualization artifacts (generated, never commit) +*.dot +*.png +.idea/ diff --git a/docs/CFG_Design.md b/docs/CFG_Design.md new file mode 100644 index 0000000..29ec166 --- /dev/null +++ b/docs/CFG_Design.md @@ -0,0 +1,334 @@ +# ScratchV 控制流图（CFG）模块 — 设计文档 + +> **版本**: 1.0 | **读者**: 模块负责人、代码审查者、下游优化 pass 开发者 +> **源文件**: `scratchv/ir/cfg.py`（待实现） | **课题**: 11 — 控制流图生成器 +> **状态**: 设计中 | **最后更新**: 2026-08-02 + +--- + +## 1. 问题定义 + +### 1.1 输入与输出 + +**输入**: ScratchV IR 的 `Program` 对象（经过前端解析的 ONNX 或 DSL）。 + +**输出**: 每个 `Function` 对应一个 `CFG` 对象，包含基本块节点和控制流边。 + +### 1.2 实例 + +给定 DSL 源码： + +``` +if (x > 0): + y = add(x, 1) +else: + y = sub(x, 1) +endif +return y +``` + +经前端解析为 IR 指令后，CFG 模块应产出 4 个基本块和 4 条边： + +``` +entry --BRANCH(true)--> L_then --JUMP--> merge --(return, 出口) + | ^ + +-------BRANCH(false)--> L_else --JUMP--+ +``` + +### 1.3 设计目标 + +| 目标 | 含义 | +|------|------| +| **正确的基本块划分** | 以 label/br/br_if/return 为边界，单趟线性扫描完成 | +| **准确的边类型** | FALLTHROUGH / BRANCH / JUMP / CALL（预留）四类 | +| **基础分析能力** | 不可达消除、支配树、自然循环检测 | +| **可可视化** | 输出 Graphviz DOT，节点着色区分入口/出口/循环头 | +| **零外部依赖** | 仅依赖 Python 标准库 + `scratchv.ir.types` | + +### 1.4 非目标（当前版本不做） + +- **控制依赖图** — 程序切片的前置需求 +- **SSA 构造 / 破坏** — ScratchV IR 已有 SSA 形式 +- **数据流分析**（到达定义、活跃变量）— 留给后续优化 pass +- **跨函数分析**（调用图）— 每个函数的 CFG 独立构建 + +--- + +## 2. 架构定位 + +### 2.1 在 ScratchV 管线中的位置 + +``` + ONNX / DSL + │ + ▼ +┌─────────┐ ┌───────────┐ ┌───────────┐ +│ Frontend│────▶│ IR Program│────▶│ Optimizer │────▶ Backend +│ Parser │ │ types.py │ │ 5 passes │ +└─────────┘ └─────┬─────┘ └─────┬─────┘ + │ │ + ▼ │ + ┌──────────┐ │ + │ CFG │◀───────────┘ + │ Builder │ + └────┬─────┘ + │ + ▼ + ┌────────────────┐ + │ 下游消费者: │ + │ · LICM │ + │ · 死代码消除 │ + │ · 寄存器分配 │ + │ · IR 验证器 │ + └────────────────┘ +``` + +**核心定位**: CFG 是分析基础设施，位于 IR 层之上、优化 pass 之下。 + +### 2.2 模块依赖 + +``` +scratchv/ir/cfg.py + ├── 依赖: scratchv/ir/types.py + ├── 被依赖: scratchv/optimizer/ + ├── 被依赖: scratchv/analysis/ir_verifier.py + └── 外部依赖: 无（仅 Python 标准库） +``` + +--- + +## 3. 数据结构设计 + +### 3.1 EdgeType + +```python +class EdgeType(enum.Enum): + FALLTHROUGH = "fallthrough" # 顺序执行过渡 + BRANCH = "branch" # 条件分支（带 condition 标签） + JUMP = "jump" # 无条件跳转 + CALL = "call" # 函数调用（预留，IR 当前无此指令） +``` + +**设计决策**: 包含 CALL 作为预留类型，与课题 user guide 保持一致。 + +### 3.2 CFGNode + +```python +@dataclass +class CFGNode: + name: str # 块名（唯一标识） + instructions: list[Instruction] # 对原列表的引用（非拷贝），外部修改原 IR 会反映到此列表 + is_entry: bool # 函数入口 + is_exit: bool # 以 return 结尾 + terminator_opcode: str | None # 终止指令类型 +``` + +### 3.3 CFGEdge + +```python +@dataclass +class CFGEdge: + source: str + target: str + edge_type: EdgeType + condition: str | None # "true"/"false"，仅 BRANCH 类型 +``` + +### 3.4 CFG + +```python +@dataclass +class CFG: + function_name: str + nodes: dict[str, CFGNode] # dict 保证 O(1) 按名查找 + edges: list[CFGEdge] # list 保证遍历效率 + entry: str # 入口块名 +``` + +**DOT 可视化样式规范**（与课题 user guide 一致）： + +节点样式: + +| 节点类型 | 颜色 | +|---------|------| +| 入口块 | 绿色 (`#90EE90`) | +| 出口块 (return) | 红色 (`#FF6B6B`) | +| 循环头 | 蓝色 (`#87CEEB`) | +| 普通块 | 浅黄色 (`lightyellow`) | + +边样式: + +| 边类型 | DOT 样式 | +|--------|---------| +| FALLTHROUGH | 黑色实线 | +| BRANCH | 蓝色虚线 + `[true]`/`[false]` 标签 | +| JUMP | 红色实线 | +| CALL（预留） | 紫色点线 | + +### 3.5 NaturalLoop + +```python +@dataclass +class NaturalLoop: + header: str # 循环头块名 + body: set[str] # 循环体中所有块名（含 header） + back_edges: list[tuple[str, str]] # (source, header) 回边列表 + parent: str | None # 外层循环 header + children: list[str] # 内层循环 headers + nesting_depth: int # 嵌套深度（0 = 最外层） +``` + +**嵌套关系构建**: 在所有循环检测完毕后，对每对循环 (outer, inner)，若 `inner.header in outer.body` 且 `inner.body` 是 `outer.body` 的真子集，则 inner 嵌套于 outer。递归计算 nesting_depth = outer.depth + 1。 + +--- + +## 4. 算法设计 + +### 4.1 基本块划分 + +**输入**: `list[Instruction]` **输出**: `list[tuple[name, list[Instruction]]]` + +**算法**: 单趟线性扫描 O(n) + +``` +遍历指令: + 遇到 LABEL → 结束当前块；标签名 = 新块名 + LABEL 指令本身不加入任何块的指令列表，仅用作块名 + 加入指令到当前块 + 遇到 BR / BR_IF / RETURN → 结束当前块 +``` + +**边界情况**: 空函数返回空 CFG；连续 label 产生空块（保留）；BR 后无 label 则自动命名 `b0`, `b1`...。 + +### 4.2 支配集计算 + +**算法**: 迭代不动点 (Iterative Fixed-Point) + +**初始化**: +``` +Dom(entry) = {entry} +Dom(n) = {所有节点} (n != entry) +``` + +**迭代** (重复直到不再变化): +``` +Dom(n) = {n} ∪ 交集{ Dom(p) | p ∈ predecessors(n) } +``` + +**复杂度**: +- 支配集迭代：理论最坏 O(N²) 轮，实际 3-5 轮收敛（ScratchV CFG 节点数 N < 100） +- idom 提取：对每个节点遍历其严格支配者，最坏 O(N²) +- 总体在实际规模下 < 1ms + +**选型理由**: 工业编译器用 Lengauer-Tarjan（近似 O(N log N)），此处选迭代不动点——代码量约 30 行 vs LT 约 150 行，CFG 节点数小，可读性优先。 + +### 4.3 直接支配树 + +从支配集筛选: `idom(n)` = 最接近 n 的严格支配者（在 `Dom(n) - {n}` 中，不被任何其他严格支配者支配的那个）。 + +### 4.4 自然循环检测 + +**回边识别**: +``` +edge(a → b) 且 b ∈ Dom(a) → 回边 +``` + +**循环体收集**: +``` +对每条回边 (source, header): + 从 source 反向 BFS，遇到 header 即停止 + 所有被访问到的节点 + header = body + 多条回边指向同一 header 时，合并所有回边对应的 body 取并集 +``` + +**嵌套检测**: +``` +对每对循环 (outer, inner): + 若 inner.header ∈ outer.body 且 inner.body ⊂ outer.body: + inner.parent = outer.header + inner.nesting_depth = outer.nesting_depth + 1 + outer.children.append(inner.header) +``` + +### 4.5 边构建规则 + +| 块终止指令 | 产生的边 | +|-----------|---------| +| RETURN | 无出边 | +| BR target | 1 条 JUMP → target（不加 FALLTHROUGH） | +| BR_IF → t1, t2 | 2 条 BRANCH: true→t1, false→t2 | +| 无终止符（纯计算块） | 1 条 FALLTHROUGH → 下一个块 | +| BR_IF 只有 1 个 target | true→target + FALLTHROUGH→下一个块 作为 false 路径 | + +--- + +## 5. 接口契约 + +| 函数 | 输入 | 输出 | 不变量 | +|------|------|------|--------| +| `partition_basic_blocks_with_names` | 指令列表 | (名,指令) 列表 | 纯函数；LABEL 不进任何指令列表 | +| `build_cfg_from_instructions` | 指令列表 | CFG | 纯函数；entry 始终是第一个块的名称 | +| `cfg.successors(name)` | 块名 | 后继列表 | 不存在的块返回 [] | +| `eliminate_unreachable` | CFG | 被删块名列表 | 原地修改 CFG（非纯函数）；entry 始终保留 | + +--- + +## 6. 技术债与已知限制 + +| 项 | 说明 | 优先级 | +|----|------|--------| +| idom 提取复杂度 | O(N²) 理论最坏；实际 N<100 可接受 | 低 | +| 回边检测去重 | 多条回边指向同一 header 时合并 body 取并集——算法设计中已规划，实现时需验证 | 中 | +| 空块处理 | 连续 label 间的空块保留但无意义 | 低 | +| 与旧版 `analysis/cfg_builder.py` 的关系 | 本模块是重写版，旧文件应标注 deprecated | 中 | +| CALL 边未实现 | CALL 已定义，构建逻辑待 IR 支持函数调用后实现 | 低 | +| instructions 是引用非拷贝 | 外部修改原 IR 会影响 CFG 节点；CFG 构建期间应冻结 IR | 低 | + +--- + +## 7. 相关文件索引 + +### 现有文件 +| 文件 | 关系 | +|------|------| +| `scratchv/ir/types.py` | 依赖 — Instruction、OpCode、BasicBlock 等 IR 类型 | +| `scratchv/analysis/cfg_builder.py` | 旧版实现（本模块重写后标注 deprecated） | +| `docs/topics/11-控制流图生成器.md` | 课题原始说明 | +| `docs/CFG_Design.md` | 本文档 | +| `docs/CFG_Dev.md` | 配套开发文档 | + +### 待创建文件 +| 文件 | 对应周 | 说明 | +|------|--------|------| +| `scratchv/ir/cfg.py` | W2-W9 | 主实现 | +| `tests/test_cfg.py` | W2-W12 | 单元测试 | +| `examples/cfg/if_else.dsl` | W4 | if/else 示例 | +| `examples/cfg/while_loop.dsl` | W9 | while 循环示例 | +| `examples/cfg/nested_loop.dsl` | W9 | 嵌套循环示例 | +| `examples/cfg/unreachable.dsl` | W7 | 不可达代码示例 | +| `scripts/visualize_cfg.py` | W10 | 可视化脚本 | + +--- + +## 8. 审查清单 + +| 检查项 | 验证方法 | +|--------|---------| +| 基本块划分正确处理 LABEL/BR/BR_IF/RETURN | 提供 if_else.dsl，检查块数和块名 | +| LABEL 不进任何指令列表 | 检查每个块的 instructions 中无 LABEL | +| 边类型与 IR 语义一致 | 对 if_else.dsl 验证 JUMP>=2, BRANCH>=2 | +| DOT 样式与 user guide 一致 | 生成 DOT 后用 Graphviz 渲染，检查颜色 | +| 不可达消除不破坏 entry 可达性 | 构造含 return 后死代码的 DSL，验证 dead 块被移除 | +| 支配集在 10 轮内收敛 | 对所有示例 DSL 运行，断言迭代次数 < 10 | +| 自然循环检测识别回边和嵌套 | while_loop.dsl 检测到循环，nested_loop.dsl 检测到外层 depth=0 内层 depth=1 | +| 所有测试通过 | pytest tests/test_cfg.py -v | + +--- + +## 9. 参考资料 + +- Aho, Lam, Sethi, Ullman — *Compilers: Principles, Techniques, and Tools* (龙书), 第 8.4 / 9.6 节 +- Cooper, Torczon — *Engineering a Compiler*, 第 5 / 9 章 +- Graphviz — [DOT Language Specification](https://graphviz.org/doc/info/lang.html) +- 课题 11 User Guide — `topic11_cfg_builder_guide.md` + diff --git a/docs/CFG_Dev.md b/docs/CFG_Dev.md new file mode 100644 index 0000000..44a4b1c --- /dev/null +++ b/docs/CFG_Dev.md @@ -0,0 +1,176 @@ +# ScratchV 控制流图（CFG）模块 — 开发文档 + +> **配套设计文档**: `CFG_Design.md` | **目标文件**: `scratchv/ir/cfg.py` +> **状态**: 待开发 | **最后更新**: 2026-08-02 + +--- + +## 1. 开发环境 + +| 依赖 | 版本 | 用途 | +|------|------|------| +| Python | >= 3.12 | 运行时 | +| ScratchV | main 分支 | IR 类型 (`scratchv/ir/types.py`)、DSL 解析器 | +| pytest | >= 8.0 | 测试 | +| Graphviz | 任意 | DOT -> PNG 渲染（可选） | + +```bash +cd ScratchV +python3 -m venv .venv && source .venv/bin/activate +pip install -e . +pytest tests/ -v --tb=short +``` + +--- + +## 2. Module Map（符号 -> 职责） + +| 符号 | 职责 | +|------|------| +| `EdgeType` | 边类型枚举（FALLTHROUGH / BRANCH / JUMP / CALL） | +| `CFGNode` | 基本块节点数据类；`instructions` 是对原 IR 列表的引用（非拷贝） | +| `CFGEdge` | 控制流边数据类 | +| `CFG` | 完整控制流图 + successors / predecessors / to_dot | +| `NaturalLoop` | 自然循环结构数据类 | +| `partition_basic_blocks_with_names` | 指令列表 -> 带名称的基本块列表（纯函数） | +| `build_cfg_from_instructions` | 指令列表 -> 完整 CFG（纯函数；串联块划分 + 节点 + 边） | +| `eliminate_unreachable` | 从 entry DFS，删除不可达块及关联边（**原地修改**，非纯函数） | +| `compute_dominators` | 迭代不动点计算支配者集合（纯函数） | +| `compute_dominator_tree` | 提取每个节点的直接支配者（纯函数） | +| `detect_loops` | 基于回边检测自然循环；同 header 多条回边合并 body 取并集（纯函数） | +| `detect_nested_loops` | 计算循环嵌套关系与深度（纯函数） | + +> **Instruction 类型定义**: 参见 `scratchv/ir/types.py`。核心字段：`opcode`（OpCode 枚举）、`dest`（目标 Value）、`operands`（操作数列表）、`target`（跳转目标字符串）。 + +--- + +## 3. 12 周开发路线 + +| 周 | 目标 | 关键产出 | +|----|------|---------| +| W1 | 理论：龙书 8.4/9.6 + 手绘 3 张 CFG | 手绘草图 | +| W2 | `partition_basic_blocks_with_names` | 块划分函数 | +| W3 | `EdgeType`, `CFGNode`, `CFGEdge`, `CFG` | 数据结构 | +| W4 | `build_cfg_from_instructions` | 完整 CFG 构建 | +| W5 | 学习 Graphviz DOT 语法 | 手写 DOT | +| W6 | `CFG.to_dot()` — 样式与 user guide 一致 | DOT 输出 | +| W7 | `eliminate_unreachable` | 不可达消除 | +| W8 | `compute_dominators` + `compute_dominator_tree` | 支配树 | +| W9 | `detect_loops` + `detect_nested_loops` | 循环检测 | +| W10 | `visualize_cfg.py` 独立脚本 | 可视化工具 | +| W11 | 可选：CLI 集成 --cfg | 管线集成 | +| W12 | 测试完善 + 文档 | 35+ 用例 + 交付 | + +--- + +## 4. API 契约与不变量 + +### 4.1 纯函数（只读分析） + +以下函数不修改输入，多次调用结果一致： + +- `partition_basic_blocks_with_names` +- `build_cfg_from_instructions` +- `compute_dominators` +- `compute_dominator_tree` +- `detect_loops` +- `detect_nested_loops` + +### 4.2 原地修改函数 + +- `eliminate_unreachable` — 修改传入的 `CFG` 对象，删除不可达节点和边。调用后应丢弃旧 CFG 引用。 + +### 4.3 build_cfg_from_instructions + +```python +def build_cfg_from_instructions( + instructions: list[Instruction], + function_name: str = "main", + entry_name: str = "entry", +) -> CFG: +``` + +**不变量**: +1. 纯函数 +2. `cfg.entry` 始终是第一个块的名称 +3. 空指令列表返回含空 entry 块的 CFG（`CFGNode(name=entry_name, instructions=[])`），避免调用方依赖 entry 时出错 +4. 每个块的 `instructions` 是对原列表子序列的**引用**（非拷贝）。CFG 构建期间不应修改原 IR + +### 4.4 自动命名策略 + +当 BR/BR_IF/RETURN 后紧跟的指令无 LABEL 时，新块自动命名为 `b0`, `b1`, `b2`...（自增计数器，从 0 开始）。 + +### 4.5 边构建规则 + +| 块类型 | 边 | 说明 | +|--------|-----|------| +| 以 `BR target` 结尾 | 1 条 JUMP -> target | **不加** FALLTHROUGH | +| 以 `BR_IF -> t1,t2` 结尾 | 2 条 BRANCH: true->t1, false->t2 | | +| 以 `RETURN` 结尾 | 无出边 | | +| 其他（无终止指令） | 1 条 FALLTHROUGH -> 下一个块 | 仅当不是最后一个块时 | + +### 4.6 DOT 样式（与 user guide 一致） + +| 节点 | 颜色 | 边类型 | 样式 | +|------|------|--------|------| +| 入口 | 绿色 | FALLTHROUGH | 黑色实线 | +| 出口 | 红色 | BRANCH | 蓝色虚线 + 标签 | +| 循环头 | 蓝色 | JUMP | 红色实线 | +| 普通 | 浅黄色 | CALL（预留） | 紫色点线 | + +--- + +## 5. 常见陷阱 + +| 陷阱 | 表现 | 修复 | +|------|------|------| +| LABEL 被当成普通指令加入块 | 块内出现 `.L1:` | `continue` 跳过 LABEL | +| 最后一个块忘记追加 | CFG 缺最后指令 | 循环后加收尾逻辑 | +| BR 后无 LABEL 导致块名冲突 | 多个块同名 | 自动命名 `b0`, `b1`... | +| 支配集不收敛 | 迭代超过 100 轮 | 检查 entry 是否在 nodes 中 | +| BR 块误加 FALLTHROUGH | 无条件跳转后还有额外边 | BR 块只加 JUMP，不加 FALLTHROUGH | +| DOT 引号未转义 | Graphviz 渲染报错 | 标签中的 `"` 替换为 `\"` | +| visited 用 list 而非 set | 无限循环 | 必须用 set | +| CFG 构建后修改原 IR 列表 | CFG 节点内容异常变化 | `instructions` 是引用非拷贝，构建后应冻结原 IR | + +--- + +## 6. 测试覆盖矩阵 + +| 测试类 | 关键用例 | 断言 | +|--------|---------|------| +| TestBasicBlockPartitioning | 纯计算 / if-else / for | 块数正确 | +| TestEdgeTypes | if-else / while / 线形 | JUMP>=2; BRANCH>=2; 线形无 JUMP | +| TestDOTOutput | 基础 / 高亮循环 | 含 `digraph`、`CFG_main` | +| TestUnreachableElimination | 无线形不可达 / 含孤立节点 | removed 正确 | +| TestDominators | entry 自支配/支配全部/idom=None | 支配集正确 | +| TestLoopDetection | 无线形 / while / 嵌套for | 循环数正确；嵌套 depth 正确 | +| TestEdgeCases | 空函数 / 多函数 / 空指令列表 | 不崩溃；entry 节点存在 | + +--- + +## 7. 修改检查清单 + +**改代码之前**: +- [ ] 新增函数在 Module Map 中有条目？ +- [ ] 纯函数确认不修改输入；原地修改函数在 Module Map 中已标注？ +- [ ] 覆盖了边界情况（空输入、单节点、无出边）？ +- [ ] DOT 样式符合 user guide？ + +**改代码之后**: +- [ ] `pytest tests/test_cfg.py -v` 全通过 +- [ ] `visualize_cfg.py` 对 if_else/while_loop 生成 DOT 正确 +- [ ] Graphviz 渲染图中节点颜色和边样式正确 + +--- + +## 8. 端到端示例：新增一个分析函数 + +以添加 `count_back_edges(cfg) -> int` 为例： + +1. `cfg.py` 中添加函数 -> Module Map 加一行，标注纯函数 +2. 写测试 -> 测试覆盖矩阵加一条 +3. 对 while_loop.dsl 手工验证（回边数应为 1） +4. `pytest tests/test_cfg.py -v` +5. 更新本文档 + diff --git a/examples/cfg/if_else.dsl b/examples/cfg/if_else.dsl new file mode 100644 index 0000000..5ea9cc8 --- /dev/null +++ b/examples/cfg/if_else.dsl @@ -0,0 +1,7 @@ +# if/else - 4个基本块, 2条JUMP + 2条BRANCH +if (x > 0): + y = add(x, 1) +else: + y = sub(x, 1) +endif +return y diff --git a/examples/cfg/nested_loop.dsl b/examples/cfg/nested_loop.dsl new file mode 100644 index 0000000..edc615a --- /dev/null +++ b/examples/cfg/nested_loop.dsl @@ -0,0 +1,7 @@ +# 嵌套循环 - 内层depth=1 +for i = 0, 3 + for j = 0, 2 + sum = add(sum, a) + endfor +endfor +return sum diff --git a/examples/cfg/unreachable.dsl b/examples/cfg/unreachable.dsl new file mode 100644 index 0000000..124cb93 --- /dev/null +++ b/examples/cfg/unreachable.dsl @@ -0,0 +1,5 @@ +# return后死代码 -> 不可达块 +c = add(a, b) +return c +dead = mul(a, b) +return dead diff --git a/examples/cfg/while_loop.dsl b/examples/cfg/while_loop.dsl new file mode 100644 index 0000000..6ef67e9 --- /dev/null +++ b/examples/cfg/while_loop.dsl @@ -0,0 +1,6 @@ +# while - 含回边，检测到1个循环 +while (i < 10): + acc = add(acc, x) + i = add(i, 1) +endwhile +return acc diff --git a/scratchv/ir/cfg.py b/scratchv/ir/cfg.py new file mode 100644 index 0000000..584491a --- /dev/null +++ b/scratchv/ir/cfg.py @@ -0,0 +1,423 @@ +"""控制流图 (CFG) 生成器 — ScratchV 课题 11。 + +从 IR Program 构建控制流图，支撑不可达代码消除、支配树、 +自然循环检测等分析。输出 Graphviz DOT 用于可视化。 + +用法:: + + from scratchv.ir.cfg import CFGBuilder + + builder = CFGBuilder() + cfgs = builder.build(program) + cfg = cfgs["main"] + print(cfg.to_dot()) +""" + +from __future__ import annotations + +import enum +from collections import deque +from dataclasses import dataclass, field + +from scratchv.ir.types import Instruction, OpCode + + +# ============================================================================ +# 边类型 (W3) +# ============================================================================ + +class EdgeType(enum.Enum): + """控制流图中边的类型。""" + FALLTHROUGH = "fallthrough" + BRANCH = "branch" + JUMP = "jump" + CALL = "call" + + +# ============================================================================ +# 数据结构 (W3) +# ============================================================================ + +@dataclass +class CFGNode: + name: str + instructions: list[Instruction] = field(default_factory=list) + is_entry: bool = False + is_exit: bool = False + terminator_opcode: str | None = None + + +@dataclass +class CFGEdge: + source: str + target: str + edge_type: EdgeType = EdgeType.FALLTHROUGH + condition: str | None = None + + +@dataclass +class NaturalLoop: + """自然循环结构 (W9)。""" + header: str + body: set[str] = field(default_factory=set) + back_edges: list[tuple[str, str]] = field(default_factory=list) + parent: str | None = None + children: list[str] = field(default_factory=list) + nesting_depth: int = 0 + + +@dataclass +class CFG: + """单个函数的控制流图。""" + function_name: str + nodes: dict[str, CFGNode] = field(default_factory=dict) + edges: list[CFGEdge] = field(default_factory=list) + entry: str = "entry" + + def successors(self, name: str) -> list[str]: + return [e.target for e in self.edges if e.source == name] + + def predecessors(self, name: str) -> list[str]: + return [e.source for e in self.edges if e.target == name] + + def to_dot(self, loop_headers: set[str] | None = None) -> str: + """生成 Graphviz DOT 字符串 (W6)。""" + lines = [f'digraph "CFG_{self.function_name}" {{'] + lines.append(" rankdir=TB;") + lines.append(" node [shape=box, style=filled];") + + loop_set = loop_headers or set() + + for name, node in self.nodes.items(): + if node.is_entry: + color = "#90EE90" + elif name in loop_set: + color = "#87CEEB" + elif node.is_exit: + color = "#FF6B6B" + else: + color = "lightyellow" + + label_parts = [f"[{name}]"] + for instr in node.instructions[:3]: + label_parts.append(str(instr)[:60]) + if len(node.instructions) > 3: + label_parts.append(f"... (+{len(node.instructions) - 3} more)") + label = "\\n".join(label_parts) + + lines.append(f' "{name}" [fillcolor="{color}", label="{label}"];') + + for edge in self.edges: + if edge.edge_type == EdgeType.FALLTHROUGH: + style = "color=black" + elif edge.edge_type == EdgeType.BRANCH: + cond = f" [{edge.condition}]" if edge.condition else "" + style = f'color=blue, style=dashed, fontcolor=blue, label="{cond}"' + elif edge.edge_type == EdgeType.JUMP: + style = "color=red" + else: + style = "color=purple, style=dotted" + lines.append(f' "{edge.source}" -> "{edge.target}" [{style}];') + + lines.append("}") + return "\n".join(lines) + + +# ============================================================================ +# 基本块划分 (W2) +# ============================================================================ + +def partition_basic_blocks_with_names( + instructions: list[Instruction], + entry_name: str = "entry", +) -> list[tuple[str, list[Instruction]]]: + """将线性 IR 指令切分为带名称的基本块。""" + result: list[tuple[str, list[Instruction]]] = [] + current: list[Instruction] = [] + current_name: str | None = None + auto_id = 0 + + for instr in instructions: + if instr.opcode == OpCode.LABEL: + if current: + result.append((current_name or f"b{auto_id}", current)) + auto_id += 1 + current_name = instr.target or f"L_{auto_id}" + current = [] + continue + + if current_name is None: + current_name = entry_name if not result else f"b{auto_id}" + if result: + auto_id += 1 + + current.append(instr) + + if instr.opcode in (OpCode.BR, OpCode.BR_IF, OpCode.RETURN): + result.append((current_name, current)) + current_name = None + current = [] + + if current: + result.append((current_name or f"b{auto_id}", current)) + + return result + + +# ============================================================================ +# CFG 构建 (W4) +# ============================================================================ + +def _add_fallthrough(cfg: CFG, name: str, names: list[str], idx: int) -> None: + if idx + 1 < len(names): + cfg.edges.append(CFGEdge(name, names[idx + 1], EdgeType.FALLTHROUGH)) + + +def build_cfg_from_instructions( + instructions: list[Instruction], + function_name: str = "main", + entry_name: str = "entry", +) -> CFG: + """从线性指令列表构建完整 CFG。""" + partitioned = partition_basic_blocks_with_names(instructions, entry_name) + + if not partitioned: + cfg = CFG(function_name=function_name, entry=entry_name) + cfg.nodes[entry_name] = CFGNode(name=entry_name) + return cfg + + cfg = CFG(function_name=function_name, entry=partitioned[0][0]) + block_names = [name for name, _ in partitioned] + + for idx, (name, instrs) in enumerate(partitioned): + node = CFGNode(name=name, instructions=instrs, is_entry=(idx == 0)) + if instrs: + last = instrs[-1] + if last.opcode == OpCode.RETURN: + node.is_exit = True + node.terminator_opcode = "return" + elif last.opcode == OpCode.BR: + node.terminator_opcode = "br" + elif last.opcode == OpCode.BR_IF: + node.terminator_opcode = "br_if" + cfg.nodes[name] = node + + for idx, (name, instrs) in enumerate(partitioned): + if not instrs: + _add_fallthrough(cfg, name, block_names, idx) + continue + + last = instrs[-1] + + if last.opcode == OpCode.RETURN: + pass + elif last.opcode == OpCode.BR: + target = last.target or "" + if target: + cfg.edges.append(CFGEdge(name, target, EdgeType.JUMP)) + elif last.opcode == OpCode.BR_IF: + targets = [t.strip() for t in (last.target or "").split(",") if t.strip()] + if len(targets) >= 2: + cfg.edges.append(CFGEdge(name, targets[0], EdgeType.BRANCH, "true")) + cfg.edges.append(CFGEdge(name, targets[1], EdgeType.BRANCH, "false")) + elif len(targets) == 1: + cfg.edges.append(CFGEdge(name, targets[0], EdgeType.BRANCH, "true")) + _add_fallthrough(cfg, name, block_names, idx) + else: + _add_fallthrough(cfg, name, block_names, idx) + else: + _add_fallthrough(cfg, name, block_names, idx) + + return cfg + + +# ============================================================================ +# CFG 构建器 (整合 W7/W8/W9) +# ============================================================================ + +class CFGBuilder: + """控制流图构建器 + 分析工具。""" + + # ---- 构建 ---- + + def build(self, program) -> dict[str, CFG]: + """直接从 IR Function 的 BasicBlock 构建 CFG（不拍平再切分）。 + + parser 已经做好块划分时，直接信任其块结构； + partition_basic_blocks_with_names 保留给纯指令列表的独立调用。 + """ + from scratchv.ir.types import Program + cfgs: dict[str, CFG] = {} + for func in program.functions: + if not func.blocks: + cfgs[func.name] = CFG(function_name=func.name) + continue + + cfg = CFG(function_name=func.name, + entry=func.blocks[0].name) + names = [b.name for b in func.blocks] + + # 创建节点 + for i, block in enumerate(func.blocks): + node = CFGNode( + name=block.name, + instructions=block.instructions, + is_entry=(i == 0), + ) + if block.instructions: + last = block.instructions[-1] + if last.opcode == OpCode.RETURN: + node.is_exit = True + node.terminator_opcode = "return" + elif last.opcode == OpCode.BR: + node.terminator_opcode = "br" + elif last.opcode == OpCode.BR_IF: + node.terminator_opcode = "br_if" + cfg.nodes[block.name] = node + + # 添加边 + for i, block in enumerate(func.blocks): + name = block.name + instrs = block.instructions + if not instrs: + _add_fallthrough(cfg, name, names, i) + continue + + last = instrs[-1] + if last.opcode == OpCode.RETURN: + pass + elif last.opcode == OpCode.BR: + t = last.target or "" + if t: + cfg.edges.append(CFGEdge(name, t, EdgeType.JUMP)) + elif last.opcode == OpCode.BR_IF: + targets = [x.strip() for x in (last.target or "").split(",") if x.strip()] + if len(targets) >= 2: + cfg.edges.append(CFGEdge(name, targets[0], EdgeType.BRANCH, "true")) + cfg.edges.append(CFGEdge(name, targets[1], EdgeType.BRANCH, "false")) + elif len(targets) == 1: + cfg.edges.append(CFGEdge(name, targets[0], EdgeType.BRANCH, "true")) + _add_fallthrough(cfg, name, names, i) + else: + _add_fallthrough(cfg, name, names, i) + else: + _add_fallthrough(cfg, name, names, i) + + cfgs[func.name] = cfg + return cfgs + + # ---- W7: 不可达消除 ---- + + def eliminate_unreachable(self, cfg: CFG) -> list[str]: + reachable = self._dfs(cfg, cfg.entry) + removed = [n for n in list(cfg.nodes) if n not in reachable] + for n in removed: + del cfg.nodes[n] + cfg.edges = [e for e in cfg.edges + if e.source in reachable and e.target in reachable] + return removed + + @staticmethod + def _dfs(cfg: CFG, start: str) -> set[str]: + visited: set[str] = set() + stack = [start] + while stack: + n = stack.pop() + if n in visited or n not in cfg.nodes: + continue + visited.add(n) + stack.extend(cfg.successors(n)) + return visited + + # ---- W8: 支配树 ---- + + def compute_dominators(self, cfg: CFG) -> dict[str, set[str]]: + all_nodes = set(cfg.nodes.keys()) + if not all_nodes: + return {} + dom = {n: all_nodes.copy() for n in all_nodes} + dom[cfg.entry] = {cfg.entry} + + changed = True + while changed: + changed = False + for node in all_nodes: + if node == cfg.entry: + continue + preds = cfg.predecessors(node) + new_dom = {node} + if preds: + new_dom |= set.intersection(*(dom[p] for p in preds if p in dom)) + if new_dom != dom[node]: + dom[node] = new_dom + changed = True + return dom + + def compute_dominator_tree(self, cfg: CFG) -> dict[str, str | None]: + dom_sets = self.compute_dominators(cfg) + idom: dict[str, str | None] = {} + for node in cfg.nodes: + if node == cfg.entry: + idom[node] = None + continue + strict = dom_sets.get(node, set()) - {node} + for d in strict: + if all(d not in (dom_sets.get(o, set()) - {o}) or o == d + for o in strict): + idom[node] = d + break + else: + idom[node] = cfg.entry + return idom + + # ---- W9: 自然循环检测 ---- + + def detect_loops(self, cfg: CFG) -> list[NaturalLoop]: + dom = self.compute_dominators(cfg) + back_edges = [(e.source, e.target) for e in cfg.edges + if e.target in dom.get(e.source, set())] + + loops: list[NaturalLoop] = [] + seen: dict[str, NaturalLoop] = {} + + for source, header in back_edges: + if header in seen: + seen[header].back_edges.append((source, header)) + seen[header].body |= self._loop_body(cfg, source, header) + continue + body = self._loop_body(cfg, source, header) + loop = NaturalLoop(header=header, body=body, + back_edges=[(source, header)]) + loops.append(loop) + seen[header] = loop + return loops + + def _loop_body(self, cfg: CFG, source: str, header: str) -> set[str]: + body: set[str] = set() + queue = deque([source]) + while queue: + n = queue.popleft() + if n == header or n in body: + continue + body.add(n) + for pred in cfg.predecessors(n): + if pred not in body and pred != header: + queue.append(pred) + return body | {header} + + def detect_nested_loops(self, cfg: CFG) -> list[NaturalLoop]: + loops = self.detect_loops(cfg) + for i, outer in enumerate(loops): + for j, inner in enumerate(loops): + if i == j: + continue + if (inner.header in outer.body + and inner.body != outer.body + and inner.body.issubset(outer.body)): + inner.nesting_depth = max( + inner.nesting_depth, outer.nesting_depth + 1) + inner.parent = outer.header + outer.children.append(inner.header) + return loops + + diff --git a/scripts/visualize_cfg.py b/scripts/visualize_cfg.py new file mode 100644 index 0000000..ca0b849 --- /dev/null +++ b/scripts/visualize_cfg.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""CFG 可视化工具 — 从 DSL/ONNX 生成控制流图 PNG/PDF/SVG。 + +用法: + python3 scripts/visualize_cfg.py examples/cfg/if_else.dsl -o cfg.png + python3 scripts/visualize_cfg.py examples/cfg/while_loop.dsl --show-loops +""" + +import argparse +import re +import subprocess +import sys +from pathlib import Path + + +def main() -> int: + ap = argparse.ArgumentParser(description="CFG 可视化工具") + ap.add_argument("input", help="输入文件 (.dsl 或 .onnx)") + ap.add_argument("-o", "--output", help="输出图片 (.png/.pdf/.svg)") + ap.add_argument("--show-loops", action="store_true", help="打印循环信息") + ap.add_argument("--eliminate-unreachable", action="store_true", + help="先消除不可达代码") + ap.add_argument("--format", choices=["png","pdf","svg"], default="png") + ap.add_argument("--extended", action="store_true", + help="使用 ExtendedDSLParser (if/else, while)") + args = ap.parse_args() + + path = Path(args.input) + if not path.exists(): + print(f"Error: {args.input} not found", file=sys.stderr); return 1 + + source = path.read_text(encoding="utf-8-sig") + + # 解析 + needs_extended = ( + args.extended + or path.suffix == ".edsl" + or bool(re.search(r"^\s*(?:if|while)\s*\(", source, re.MULTILINE)) + ) + if needs_extended: + from scratchv.frontend.dsl_extended import ExtendedDSLParser + program = ExtendedDSLParser().parse(source) + elif path.suffix == ".onnx": + from scratchv.frontend.onnx_parser import ONNXParser + program = ONNXParser().parse(str(path)) + else: + from scratchv.frontend.dsl_parser import DSLParser + program = DSLParser().parse(source) + + # 构建 CFG + from scratchv.ir.cfg import CFGBuilder + builder = CFGBuilder() + cfgs = builder.build(program) + + dot_parts = [] + for fname, cfg in cfgs.items(): + if args.eliminate_unreachable: + removed = builder.eliminate_unreachable(cfg) + if removed: + print(f"[{fname}] removed: {removed}", file=sys.stderr) + + if args.show_loops: + loops = builder.detect_nested_loops(cfg) + for loop in loops: + indent = " " * loop.nesting_depth + print(f"{indent}[{fname}] Loop: {loop.header} " + f"depth={loop.nesting_depth} body={sorted(loop.body)}", + file=sys.stderr) + + dot_parts.append(cfg.to_dot( + loop_headers={l.header for l in builder.detect_loops(cfg)})) + + full_dot = "\n\n".join(dot_parts) + + if args.output: + dot_path = Path(args.output).with_suffix(".dot") + dot_path.write_text(full_dot, encoding="utf-8") + print(f"DOT: {dot_path}", file=sys.stderr) + try: + subprocess.run(["dot", f"-T{args.format}", str(dot_path), + "-o", args.output], check=True) + print(f"Image: {args.output}", file=sys.stderr) + except FileNotFoundError: + print("Install Graphviz: apt install graphviz", file=sys.stderr) + else: + print(full_dot) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_cfg.py b/tests/test_cfg.py new file mode 100644 index 0000000..904ea06 --- /dev/null +++ b/tests/test_cfg.py @@ -0,0 +1,257 @@ +"""控制流图 (CFG) 模块测试 — 课题11。 + +覆盖: 基本块划分、CFG构建、边类型、DOT输出、 +不可达消除、支配树、循环检测、边界情况。 +""" + +import pytest +from scratchv.ir.types import Program, Function, BasicBlock, Instruction, OpCode, Value, DataType +from scratchv.ir.cfg import ( + CFGNode, CFGEdge, CFG, EdgeType, NaturalLoop, + CFGBuilder, partition_basic_blocks_with_names, build_cfg_from_instructions, +) +from scratchv.frontend.dsl_parser import DSLParser +from scratchv.frontend.dsl_extended import ExtendedDSLParser + + +# ============================================================================ +# 辅助 +# ============================================================================ + +def _parse(src: str) -> Program: + return DSLParser().parse(src) + +def _parse_ext(src: str) -> Program: + return ExtendedDSLParser().parse(src) + +def _inst(op: OpCode, dest=None, operands=None, target=None) -> Instruction: + return Instruction(opcode=op, + dest=Value(name=dest, dtype=DataType.FLOAT32) if dest else None, + operands=operands or [], target=target) + + +# ============================================================================ +# W2: 基本块划分 +# ============================================================================ + +class TestPartition: + def test_single_block(self): + """无控制流的程序 -> 1个块。""" + instrs = [_inst(OpCode.ADD, "c"), _inst(OpCode.RETURN)] + blocks = partition_basic_blocks_with_names(instrs) + assert len(blocks) == 1 + assert blocks[0][0] == "entry" + + def test_br_splits(self): + """BR 之后切新块。""" + instrs = [ + _inst(OpCode.ADD, "a"), + _inst(OpCode.BR, target="L1"), + _inst(OpCode.LABEL, target="L1"), + _inst(OpCode.RETURN), + ] + blocks = partition_basic_blocks_with_names(instrs) + assert len(blocks) == 2 + assert blocks[0][0] == "entry" + assert blocks[1][0] == "L1" + + def test_label_not_in_instructions(self): + """LABEL 不进任何块的指令列表。""" + instrs = [ + _inst(OpCode.BR, target="L1"), + _inst(OpCode.LABEL, target="L1"), + _inst(OpCode.RETURN), + ] + blocks = partition_basic_blocks_with_names(instrs) + for _, instr_list in blocks: + for i in instr_list: + assert i.opcode != OpCode.LABEL + + def test_auto_naming(self): + """BR 后无 LABEL -> 自动命名 b0, b1。""" + instrs = [ + _inst(OpCode.BR, target="L1"), + _inst(OpCode.RETURN), + ] + blocks = partition_basic_blocks_with_names(instrs) + names = [n for n, _ in blocks] + assert "b0" in names + + +# ============================================================================ +# W3: 数据结构 +# ============================================================================ + +class TestDataStructures: + def test_edge_type_values(self): + assert EdgeType.FALLTHROUGH.value == "fallthrough" + assert EdgeType.BRANCH.value == "branch" + assert EdgeType.JUMP.value == "jump" + assert EdgeType.CALL.value == "call" + + def test_cfg_successors_predecessors(self): + cfg = CFG("test") + cfg.nodes["A"] = CFGNode("A") + cfg.nodes["B"] = CFGNode("B") + cfg.edges = [CFGEdge("A", "B")] + assert cfg.successors("A") == ["B"] + assert cfg.predecessors("B") == ["A"] + + +# ============================================================================ +# W4: CFG 构建 +# ============================================================================ + +class TestCFGBuild: + def test_if_else_blocks(self): + prog = _parse_ext("if (a > b):\n c = add(a, b)\nelse:\n" + " c = mul(a, b)\nendif\nreturn c\n") + cfg = CFGBuilder().build(prog)["main"] + assert len(cfg.nodes) >= 4 + + def test_if_else_edge_types(self): + prog = _parse_ext("if (a > b):\n c = add(a, b)\nelse:\n" + " c = mul(a, b)\nendif\nreturn c\n") + cfg = CFGBuilder().build(prog)["main"] + jumps = sum(1 for e in cfg.edges if e.edge_type == EdgeType.JUMP) + branches = sum(1 for e in cfg.edges if e.edge_type == EdgeType.BRANCH) + assert jumps >= 2 + assert branches >= 2 + + def test_empty_program(self): + cfg = build_cfg_from_instructions([]) + assert cfg.entry in cfg.nodes + assert cfg.nodes[cfg.entry].instructions == [] + + def test_linear_no_jump(self): + instrs = [_inst(OpCode.ADD, "c"), _inst(OpCode.RETURN)] + cfg = build_cfg_from_instructions(instrs) + jumps = sum(1 for e in cfg.edges if e.edge_type == EdgeType.JUMP) + assert jumps == 0 + + +# ============================================================================ +# W6: DOT 输出 +# ============================================================================ + +class TestDOT: + def test_contains_digraph(self): + cfg = build_cfg_from_instructions([_inst(OpCode.RETURN)]) + dot = cfg.to_dot() + assert dot.startswith("digraph") + + def test_node_colors(self): + """2块CFG：entry(绿) + exit(红)。""" + instrs = [ + _inst(OpCode.BR, target="L_end"), + _inst(OpCode.LABEL, target="L_end"), + _inst(OpCode.RETURN), + ] + cfg = build_cfg_from_instructions(instrs) + dot = cfg.to_dot() + assert "#90EE90" in dot # 入口绿色 + assert "#FF6B6B" in dot # 出口红色 + + +# ============================================================================ +# W7: 不可达消除 +# ============================================================================ + +class TestUnreachable: + def test_removes_isolated(self): + cfg = CFG("test") + cfg.entry = "A" + cfg.nodes["A"] = CFGNode("A") + cfg.nodes["B"] = CFGNode("B") + cfg.nodes["C"] = CFGNode("C") # 孤立 + cfg.edges = [CFGEdge("A", "B")] + builder = CFGBuilder() + removed = builder.eliminate_unreachable(cfg) + assert "C" in removed + assert "C" not in cfg.nodes + + def test_all_reachable(self): + prog = _parse("c = add(a, b)\nreturn c\n") + builder = CFGBuilder() + cfg = builder.build(prog)["main"] + removed = builder.eliminate_unreachable(cfg) + assert len(removed) == 0 + + +# ============================================================================ +# W8: 支配树 +# ============================================================================ + +class TestDominators: + def test_entry_self(self): + cfg = build_cfg_from_instructions([_inst(OpCode.RETURN)]) + builder = CFGBuilder() + dom = builder.compute_dominators(cfg) + assert cfg.entry in dom[cfg.entry] + + def test_idom_entry_none(self): + cfg = build_cfg_from_instructions([_inst(OpCode.RETURN)]) + builder = CFGBuilder() + idom = builder.compute_dominator_tree(cfg) + assert idom[cfg.entry] is None + + +# ============================================================================ +# W9: 循环检测 +# ============================================================================ + +class TestLoops: + def test_no_loops_linear(self): + cfg = build_cfg_from_instructions([_inst(OpCode.RETURN)]) + builder = CFGBuilder() + loops = builder.detect_loops(cfg) + assert len(loops) == 0 + + def test_while_loop(self): + prog = _parse_ext("while (i < 10):\n acc = add(acc, x)\nendwhile\n" + "return acc\n") + builder = CFGBuilder() + cfg = builder.build(prog)["main"] + loops = builder.detect_loops(cfg) + assert len(loops) >= 1 + + def test_nested(self): + """嵌套 while：外层 depth=0。""" + prog = _parse_ext( + "while (i < 3):\n" + " while (j < 2):\n" + " c = add(a, b)\n" + " j = add(j, 1)\n" + " endwhile\n" + " i = add(i, 1)\n" + "endwhile\n" + "return c\n") + builder = CFGBuilder() + cfg = builder.build(prog)["main"] + loops = builder.detect_nested_loops(cfg) + depths = [l.nesting_depth for l in loops] + assert 0 in depths + + +# ============================================================================ +# 边界情况 +# ============================================================================ + +class TestEdgeCases: + def test_empty_function(self): + func = Function(name="f") + prog = Program() + prog.add_function(func) + cfgs = CFGBuilder().build(prog) + assert "f" in cfgs + + def test_multi_function(self): + f1, f2 = Function("f1"), Function("f2") + f1.add_block(BasicBlock("entry")) + f2.add_block(BasicBlock("entry")) + prog = Program() + prog.add_function(f1); prog.add_function(f2) + cfgs = CFGBuilder().build(prog) + assert "f1" in cfgs and "f2" in cfgs + +