「编译器架构与多遍设计」

# 三阶段编译器骨架: 每条 pipeline 都接收上一个阶段的产品
class FrontEnd:
    def run(self, source):
        return {"ir": "HIR", "symbols": {"x": "int"}}   # 源码 -> IR

class MiddleEnd:
    def run(self, hir):
        return {"ir": hir["ir"].replace("HIR", "OPT")}  # IR 优化

class BackEnd:
    def run(self, opt_ir, target="x86-64"):
        return f"{target} 汇编: {opt_ir['ir']}"

source_code = "int x = 1 + 2;"
hir = FrontEnd().run(source_code)
opt = MiddleEnd().run(hir)
asm = BackEnd().run(opt)
print(asm)

class HIR:
    def __init__(self, expr): self.expr = expr   # 树形, 带类型

class MIR:
    def __init__(self): self.statements = []      # 三地址码, 带 CFG

class LIR:
    def __init__(self): self.instructions = []    # 伪汇编, 带寄存器

def lower_hir_to_mir(hir):
    m = MIR()
    m.statements.append(("t1 = " + hir.expr, "typed"))
    return m

def lower_mir_to_lir(mir):
    l = LIR()
    for stmt, _ in mir.statements:
        l.instructions.append(stmt.replace("t1 = ", "mov eax, "))
    return l

print(lower_mir_to_lir(lower_hir_to_mir(HIR("1 + 2"))).instructions)

import re

def optimize(ir, passes):
    """按顺序执行一串 pass, 每遍都是 pure 函数."""
    for name, fn in passes:
        ir = fn(ir)
        print(f"  [{name}] 完成, IR 节点数={len(ir['nodes'])}")
    return ir

def const_prop(ir):
    """把已知常量代入使用处 (仅演示, 逐条字符串变换)."""
    out = []
    for n in ir["nodes"]:
        if " = " in n:
            lhs, rhs = n.split(" = ", 1)
            out.append(f"{lhs} = {rhs.replace('x', '3')}")
        else:
            out.append(n)
    return {"nodes": out}

def dead_code(ir):
    """删除写入后从未被读取的赋值."""
    used = set()
    for n in ir["nodes"]:
        rhs = n.split(" = ", 1)[1] if " = " in n else n
        used |= set(re.findall(r"[a-zA-Z_]+", rhs))
    return {"nodes": [n for n in ir["nodes"]
                      if " = " not in n or n.split(" = ", 1)[0] in used]}

ir = {"nodes": ["x = 3", "y = x + 1", "unused = 999", "call(y)"]}
opt = optimize(ir, [("常量传播", const_prop), ("死代码消除", dead_code)])
print("最终:", opt["nodes"])

class AnalysisCache:
    """Pass 之间的分析结果共享: 需要时才计算, 失效即丢弃."""
    def __init__(self):
        self.cache = {}
        self.invalidated = set()
    def get(self, key, compute):
        if key in self.invalidated:
            self.cache.pop(key, None)
            self.invalidated.discard(key)
        if key not in self.cache:
            self.cache[key] = compute()
        return self.cache[key]
    def invalidate(self, key):
        self.invalidated.add(key)

cache = AnalysisCache()
cfganal = cache.get("cfg", lambda: {"blocks": 4})
cache.invalidate("cfg")                       # 某 pass 改了 CFG
cfganal2 = cache.get("cfg", lambda: {"blocks": 7})
print(cfganal, cfganal2)

class Pass:
    def __init__(self, name, requires=(), invalidates=()):
        self.name = name
        self.requires = requires
        self.invalidates = invalidates

# 声明式依赖: 调度器据此排序与失效
PASSES = [
    Pass("内联", requires=["callgraph"], invalidates=["cfg", "loops"]),
    Pass("循环简化", requires=["loops"], invalidates=["cfg"]),
    Pass("寄存器分配", requires=[], invalidates=[]),
]
for p in PASSES:
    print(f"{p.name}: 依赖 {p.requires} 失效 {p.invalidates}")

# 一个迷你中端: 折叠 + 死代码 + 循环不变量外提
def fold_constants(stmts):
    """x = 2; y = x + 3  ->  y = 5 (仅演示简单常量折叠)"""
    env = {}
    out = []
    for s in stmts:
        lhs, rhs = s.split(" = ", 1)
        if rhs.lstrip("-").isdigit():      # 整数字面量记录进环境
            env[lhs] = int(rhs)
            out.append(s)
            continue
        folded = rhs
        for k, v in env.items():           # 用已知常量替换右端
            folded = folded.replace(k, str(v))
        try:
            val = eval(folded)             # 仅演示算术折叠
        except Exception:
            val = folded
        out.append(f"{lhs} = {val}")
    return out

print(fold_constants(["x = 2", "y = x + 3", "z = y * 4"]))

def hoist_invariant(loop_body, loop_invariants):
    """把循环体内不变的赋值外提到循环前."""
    preheader = []
    kept = []
    for s in loop_body:
        if all(v not in s.split(" = ")[0] for v in loop_invariants):
            if s in loop_invariants:
                preheader.append(s)
            else:
                kept.append(s)
        else:
            kept.append(s)
    return preheader, kept

pre, body = hoist_invariant(
    ["t = n", "i = i + 1", "a[i] = a[i] + t"],
    ["t = n"],
)
print("外提到 preheader:", pre)
print("循环内剩余:", body)

# 指令选择: 用目标指令模式覆盖 IR 表达式树
def select(ir_node, arch):
    table = {
        "add": {"x86-64": "add", "riscv": "addi" if ir_node.endswith("const") else "add"},
        "load": {"x86-64": "mov (mem)", "riscv": "lw"},
    }
    op = ir_node.split("(")[0]
    return table.get(op, {}).get(arch, f"unimplemented:{op}")

for node in ["add(a,b)", "load(p)"]:
    print(node, "->", select(node, "x86-64"), "|", select(node, "riscv"))

# 简单指令调度: 把独立的 load 提前, 隐藏访存延迟
def schedule(insts, latency=3):
    loads = [i for i in insts if "load" in i]
    others = [i for i in insts if "load" not in i]
    return loads + others      # 让 load 先发, 后随指令掩盖延迟

insts = ["add eax, 1", "load rbx", "mul eax, ebx"]
print("调度后:", schedule(insts))

# 行号表骨架: 每条指令映射回源码行
class LineTable:
    def __init__(self): self.rows = []   # (address, source_line)
    def add(self, addr, line):
        self.rows.append((addr, line))
    def lookup(self, addr):
        row = [r for r in self.rows if r[0] <= addr][-1]
        return f"地址 {addr} 对应源码行 {row[1]}"

lt = LineTable()
lt.add(0x1000, 5); lt.add(0x1004, 6); lt.add(0x1008, 7)
print(lt.lookup(0x1004))

# 插件注册表: 第三方可以注入自定义 pass
class PluginRegistry:
    def __init__(self): self.passes = []
    def register(self, name, fn): self.passes.append((name, fn))
    def run(self, ir):
        for name, fn in self.passes:
            ir = fn(ir)
        return ir

registry = PluginRegistry()
registry.register("我的死代码加强", lambda ir: [s for s in ir if "dead" not in s])
print(registry.run(["x = 1", "dead = 2", "y = 3"]))