课题18：指令调度（基本块内列表调度）

难度：中 | 类型：项目实战 | 源文件：scratchv/backend/inst_scheduler.py | 行数：~450 状态：✅ 已完成

概述

在基本块内重排RISC-V指令，减少数据冒险引起的流水线停顿，提高指令级并行性。

---

理解背景

是什么？

指令调度器在基本块内部重排RISC-V指令的执行顺序。它通过分析指令间的数据依赖关系（RAW/WAR/WAW），构建依赖有向图（DAG），然后用列表调度算法（List Scheduling）按关键路径优先级选择指令发射，找出能最大化指令级并行性（ILP）的顺序。

原始顺序:                    调度后:
  lw t0, 0(a0)         →      lw t0, 0(a0)
  add t1, t0, t2       →      lw t3, 4(a0)       ← 提前，不依赖 t0
  lw t3, 4(a0)         →      add t1, t0, t2     ← t0 此时已就绪

为什么？

现代RISC-V处理器采用流水线设计，指令并非严格串行执行。当一条指令的结果还没算完，下一条指令就要用这个结果时（RAW数据冒险），流水线必须停顿（stall）等待。

• lw 需要 2 个周期才能把数据从内存读到寄存器
• 如果 lw t0, 0(sp) 之后紧跟 add t1, t0, t2，ALU 必须等 lw 完成
• 这 1 个周期的空等就是流水线气泡（pipeline bubble）

类比：你有一台洗衣机和一台烘干机。如果"洗→烘→洗→烘"串行，烘干机在洗衣时闲置。更好的安排是：洗第一批时，把第二批准备好；第一批进烘干机时，第二批进洗衣机。指令调度做的事就是这种"重排"。

核心概念

1. 数据冒险类型

冒险类型	全称	说明	是否引起停顿
RAW	Read After Write	先写后读：B 要用 A 刚写的值，但 A 还没写完	是（真依赖）
WAR	Write After Read	先读后写：B 要写，但 A 还没读完（反依赖）	否（寄存器重命名可消除）
WAW	Write After Write	写后写：两条指令写同一寄存器，顺序不能乱	否（输出依赖）

2. 依赖有向图（DAG）

节点 = 指令
边   = 依赖关系（A → B 表示 B 依赖 A）
边权 = A 的延迟周期数

示例:
  lw t0, 0(a0) ──(2)──→ add t1, t0, t2
     延迟2周期            需要 t0 的值

优先级 = 从该节点到叶子节点的最长路径长度

3. RISC-V 延迟模型

指令类	操作码	延迟（周期）
整数ALU	add, sub, sll, or, and, slt 等	1
乘法	mul, mulh	3
除法	div, rem	16
访存（读）	lw, lh, lb, lbu, lhu	2
访存（写）	sw, sh, sb	0（非阻塞）
分支	beq, bne, blt, bge 等	1
跳转	j, jal, jalr, ret	0

4. 列表调度算法（List Scheduling）

1. 构建依赖 DAG，计算每个节点的优先级（关键路径长度）
2. 维护就绪队列：所有前驱已调度的节点
3. 循环：
   a. 从就绪队列中选择优先级最高的节点
   b. 发射该节点（scheduled = True）
   c. 更新时钟（+ 该指令的延迟）
   d. 将新就绪的节点加入队列
4. 输出调度后的指令序列

优先级计算：从叶子节点（无后继）向根节点反向传播，每个节点的优先级 = 自身延迟 + max(后继优先级 + 边延迟)。

---

详细任务

1. 定义RISC-V指令延迟模型（如lw延迟2周期，算术指令1周期）。
2. 构建依赖有向图：节点为指令，边为RAW/WAR/WAW依赖，边权为延迟周期。
3. 实现列表调度算法：维护就绪队列（所有前驱已调度），按优先级（最长路径长度）选择指令发射。
4. 输出调度后的指令序列，并计算预估的总时钟周期数（相比原始顺序的改善）。
5. 集成到后端，添加--schedule选项。

---

交付产物

• inst_scheduler.py模块
• 测试用例及调度前后对比
• 文档：延迟模型、算法说明

---

代码走读

数据结构

@dataclass
class SchedInst:
    """调度器使用的指令节点"""
    id: int              # 在输入列表中的唯一索引
    opcode: str          # 指令助记符
    operands: list[str]  # 操作数列表
    defines: set[str]    # 该指令写入的寄存器集合
    uses: set[str]       # 该指令读取的寄存器集合
    raw_line: str        # 原始汇编行

@dataclass
class DAGNode:
    """依赖 DAG 中的节点"""
    inst: SchedInst
    predecessors: list[tuple["DAGNode", int]]  # (前驱节点, 延迟)
    successors: list[tuple["DAGNode", int]]    # (后继节点, 延迟)
    scheduled: bool      # 是否已调度
    ready_time: int      # 最早可发射的周期
    priority: int        # 关键路径长度（列表调度优先级）

DAG 构建

def build_dag(self, instructions: list[SchedInst]) -> list[DAGNode]:
    """为基本块内指令构建依赖 DAG"""
    # 1. 创建 DAGNode
    id_to_node: dict[int, DAGNode] = {}
    for inst in instructions:
        node = DAGNode(inst=inst)
        id_to_node[inst.id] = node

    # 2. 构建依赖边
    last_def: dict[str, DAGNode] = {}  # 每个寄存器最后定义的节点
    for inst in instructions:
        node = id_to_node[inst.id]

        # RAW: 每个 use 依赖对应寄存器的上一次定义
        for use_reg in inst.uses:
            if use_reg in last_def:
                pred = last_def[use_reg]
                latency = self._get_latency(pred.inst.opcode)
                node.predecessors.append((pred, latency))
                pred.successors.append((node, latency))

        # WAW: 同一寄存器的后一次定义依赖前一次定义
        for def_reg in inst.defines:
            if def_reg in last_def and last_def[def_reg] is not node:
                pred = last_def[def_reg]
                latency = self._get_latency(pred.inst.opcode)
                node.predecessors.append((pred, latency))
                pred.successors.append((node, latency))
            last_def[def_reg] = node

    # 3. 计算优先级
    self._compute_priorities()
    return self._nodes

优先级计算（关键路径）

def _compute_priorities(self) -> None:
    """从叶子向根反向传播，计算每个节点的关键路径长度"""
    # 拓扑排序（反向遍历用）
    order: list[DAGNode] = []
    visited: set[int] = set()

    def _dfs(n: DAGNode) -> None:
        if n.inst.id in visited:
            return
        visited.add(n.inst.id)
        for succ, _ in n.successors:
            _dfs(succ)
        order.append(n)

    for node in self._nodes:
        _dfs(node)

    # 反向传播计算优先级
    for node in reversed(order):
        max_succ_prio = 0
        for succ, lat in node.successors:
            max_succ_prio = max(max_succ_prio, succ.priority + lat)
        node.priority = max_succ_prio + self._get_latency(node.inst.opcode)

列表调度主循环

def schedule(self, dag: list[DAGNode]) -> list[SchedInst]:
    """列表调度主循环"""
    remaining = {id(node): node for node in self._nodes}
    clock = 0
    result: list[SchedInst] = []

    while remaining:
        # 找出所有前驱已调度的节点
        ready = []
        for node in remaining.values():
            if all(p.scheduled for p, _ in node.predecessors):
                ready.append(node)

        if not ready:
            break  # 不应发生（DAG 无环）

        # 按优先级降序排序，同优先级按原始顺序
        ready.sort(key=lambda n: (-n.priority, n.inst.id))

        # 发射优先级最高的节点
        node = ready[0]
        node.scheduled = True
        node.ready_time = clock
        result.append(node.inst)
        del remaining[id(node)]

        clock += self._get_latency(node.inst.opcode)

    return result

调度报告

def report(self, original: list[SchedInst],
           scheduled: list[SchedInst]) -> str:
    """生成调度前后对比报告"""
    orig_cycles = self.estimate_cycles(original)
    sched_cycles = self.estimate_cycles(scheduled)
    improvement = orig_cycles - sched_cycles
    pct = (improvement / orig_cycles * 100) if orig_cycles > 0 else 0.0

    # 输出: 原始周期数、调度后周期数、改善量、改善百分比
    # 以及调度后的指令顺序

汇编解析器

def parse_instructions(asm_text: str) -> list[SchedInst]:
    """将 RISC-V 汇编文本解析为 SchedInst 列表"""
    # 用正则匹配每行: [label:] [.]opcode operands
    # 自动识别 def/use 寄存器:
    #   - ALU 类指令: 第一个操作数是 def，其余是 use
    #   - 访存写 (sw/sh/sb): 第一个操作数是 use（要存的值）
    #   - 访存读 (lw/lh/lb): 第一个操作数是 def
    #   - 分支指令: 所有操作数都是 use

调度结果回写

def machine_instrs_from_scheduled(
        scheduled: list[SchedInst],
) -> list[MachineInstr]:
    """将调度后的 SchedInst 列表转回 MachineInstr 列表

    使得调度器的输出可以被 RegisterAllocator 和 AsmEmitter 消费。
    """

CLI 入口

def main() -> None:
    """命令行入口: python -m scratchv.backend.inst_scheduler input.s --report"""
    # 读取汇编文件 → 解析指令 → 构建 DAG → 调度 → 输出 + 报告

---

动手练习

练习 1: 手算优先级

给定以下 4 条指令，手工画出依赖 DAG，计算每个节点的优先级：

lw t0, 0(a0)     # 延迟 2
add t1, t0, t2    # 延迟 1
mul t3, t1, t1    # 延迟 3
sw t3, 0(a0)      # 延迟 0

验证你的优先级是否与 build_dag 的输出一致。

练习 2: 追踪调度过程

用 --report 模式对一个包含 10 条以上指令的基本块运行调度器，逐步追踪就绪队列的变化：每条指令在哪个周期入队、何时发射。

练习 3: 自定义延迟模型

修改 _DEFAULT_LATENCY 字典，尝试一个不同的延迟模型（如提高访存延迟到 4 周期），观察调度结果的变化。哪种模型下调度器的改善最大？

练习 4: 处理分支指令

修改调度器，确保分支指令（beq, bne 等）始终排在基本块末尾——因为基本块以分支结束是一个硬约束。在 ready.sort 中给分支指令降权。

练习 5: 集成为后端 pass

将 InstructionScheduler 集成为后端的一个 pass（类似 RegisterAllocator），在 scratchv/backend/code_generator.py 中调用，添加 --schedule 选项。测试一个包含 Conv 的完整程序，观察调度前后的性能差异。

---

常见坑

坑	说明
基本块边界	列表调度只在单个基本块内进行，不能跨越分支/跳转指令。跨基本块的调度需要全局调度（trace scheduling）
分支必须排末尾	调度算法可能把分支指令提前到基本块中间，这破坏了控制流语义。必须在调度时约束分支保持在末尾
WAW/WAR 是假依赖	WAW 和 WAR 不引起真正的流水线停顿（可以通过寄存器重命名消除），但当前实现保留了这些边，使依赖图更保守（安全但可能不够优化）
流水线建模简化	当前延迟模型是单指令延迟的简单累加，未建模超标量发射（双发射）、乱序执行、或缓存未命中。实际 CPU 的性能改善可能与此估算不同
优先级同值时	当多条指令优先级相同时，当前按原始顺序选择。可以尝试其他 tie-breaking 策略（如选择释放寄存器更多的指令）
时钟推进方式	当前实现每条指令发射后时钟 +1*延迟，这是一个粗粒度的模型。更精确的做法是按周期推进，模拟每周期能发射几条指令

---

进阶阅读

• 龙书第 10.3 节：Instruction Scheduling
• Hennessy & Patterson: Computer Architecture, Appendix H — Hardware and Software for VLIW and EPIC
• Gibbons & Muchnick (1986): Efficient Instruction Scheduling for a Pipelined Architecture (ACM SIGPLAN)
• RISC-V 指令集手册：Volume 1: Unprivileged ISA
• 相关 topic: 课题17 — 寄存器分配 | 课题13 — 窥孔优化器 | 课题8 — 指令选择

---

12周每周目标

• W1：学习指令调度原理（数据冒险、流水线停顿、列表调度算法）。
• W2：定义RISC-V简单延迟模型（如lw延迟2，算术指令1，分支1）。
• W3：实现依赖分析：为基本块内指令构建有向图，节点为指令索引，边为RAW依赖。
• W4：添加WAR和WAW依赖边（虽不引起真冒险，但影响寄存器分配，先处理RAW）。
• W5：为每个节点计算优先级（最长路径长度到结束节点）。
• W6：实现列表调度核心：维护就绪队列，按优先级选择指令，更新时钟。
• W7：实现调度器，输出调度后的指令序列。忽略分支延迟槽。
• W8：编写模拟器功能：给定原始顺序和调度后顺序，比较预估总周期数。
• W9：处理分支指令：分支必须作为基本块的最后一条，调度时不能将其提前。
• W10：实现寄存器重命名（可选，复杂），先不实现，靠列表调度避免冲突。
• W11：集成到代码生成后端，添加--schedule选项，测试短基本块。
• W12：测试更复杂的循环体，分析调度前后性能提升，撰写文档。