Python 3 源码分析

主题

动手：用 Python 写一个迷你 Python 虚拟机 ​

读完前面六部分，我们把 CPython 的对象、编译、虚拟机、运行时、内存管理都拆开看了一遍。但「看懂」和「写得出」之间还隔着一层。这一章是全书的实战 capstone：我们用大约 300 行 Python，亲手实现一个迷你 Python 虚拟机——它能把一小段 Python 编译成字节码，再用一个求值循环逐条执行，连函数调用与递归（帧栈）都支持。

更妙的是,它就在你的浏览器里跑:页面底部的交互组件通过 WebAssembly（Pyodide）把真正的 Python 运行环境搬进了网页,你写的代码会被我们的迷你虚拟机单步执行,求值栈、局部变量、调用栈的每一步变化都看得见。整章没有服务端,纯静态。

总览：一条和 CPython 同构的流水线 ​

我们的迷你虚拟机走的是和 CPython 一模一样的路子，只是每一环都简化到「够教学」的程度：

迷你虚拟机的流水线

  1. 源码 → AST：借用 Python 自带的 ast.parse，把源码解析成抽象语法树（第三部分讲过 CPython 也是这么干的）；
  2. AST → 玩具字节码：我们写一个编译器遍历 AST，吐出自定义的指令序列；
  3. 字节码 → 求值循环：我们写一个虚拟机，用一个大循环 + 帧栈逐条执行指令、操作求值栈；
  4. 输出：print 的结果。

唯一「偷懒」的是第 1 步借用了 ast——因为词法/语法分析不是本书重点。从 AST 往后的编译与执行，全是我们自己写的，也正是前几部分的核心。

设计指令集：玩具版对照 CPython ​

先定指令集。我们刻意模仿 CPython 的栈式指令，但只保留最核心的十几条。为了直观，玩具字节码把名字和常量直接内联在参数里（真实 CPython 用下标去 co_consts/co_varnames 查表，第三部分见过）：

玩具指令集对照表

每条指令就是一个 (op, arg) 二元组，承载它的容器是 Code——对应 CPython 的 code object：

python
# minivm.py —— 一段可执行的字节码（模块体，或一个函数体）
class Code:
    def __init__(self, name, params=()):
        self.id = _new_id()
        self.name = name
        self.params = list(params)   # 形参名
        self.instrs = []             # [[op, arg], ...]

    def emit(self, op, arg=None):
        self.instrs.append([op, arg])
        return len(self.instrs) - 1  # 返回这条指令的位置（跳转回填要用）

编译器：把 AST 翻译成字节码 ​

编译器是一个遍历 AST 的访问器。表达式编译成「把值压上求值栈」的指令，语句编译成「产生副作用」的指令。先看表达式——这正是第四部分求值栈那一套的逆向（生成端）：

python
# minivm.py —— 表达式编译（节选）
def compile_expr(self, e):
    if isinstance(e, ast.Constant):
        self.code.emit("LOAD_CONST", e.value)        # 常量 → 压栈
    elif isinstance(e, ast.Name):
        self.load_name(e.id)                          # 变量 → 压栈
    elif isinstance(e, ast.BinOp):
        self.compile_expr(e.left)                     # 先算左
        self.compile_expr(e.right)                    # 再算右
        self.code.emit("BINARY_OP", BINOPS[type(e.op)])  # 弹二压一
    elif isinstance(e, ast.Compare):
        self.compile_expr(e.left)
        self.compile_expr(e.comparators[0])
        self.code.emit("COMPARE_OP", CMPOPS[type(e.ops[0])])
    elif isinstance(e, ast.Call):
        self.compile_call(e)
    ...

a + b * 2 会被编译成 LOAD a / LOAD b / LOAD_CONST 2 / BINARY_OP * / BINARY_OP +——后缀顺序，正好喂给栈式机求值。赋值语句则是「算出值，再存进名字」：

python
# minivm.py —— 赋值语句
if isinstance(s, ast.Assign):
    self.compile_expr(s.value)                # 算出右边的值（压栈）
    self.store_name(s.targets[0].id)          # 弹栈，存进左边的名字

控制流：跳转回填 ​

if 和 while 编译成条件跳转 + 无条件跳转（第四部分控制流章的核心）。难点在于：编译 if 的条件时，我们还不知道 else 分支在哪——它的地址要等后面的指令都生成完才确定。办法是先 emit 一条占位的跳转，记下它的位置，等目标地址确定后再回填：

if / while 的跳转回填

python
# minivm.py —— 编译 while（跳转回填）
def compile_while(self, s):
    start = len(self.code.instrs)                       # 循环顶部
    self.compile_expr(s.test)                           # 算条件
    jmp_end = self.code.emit("POP_JUMP_IF_FALSE", None) # 占位：条件假→跳出
    self.compile_stmts(s.body)                          # 循环体
    self.code.emit("JUMP_ABSOLUTE", start)              # 往回跳，重来一轮
    self.code.instrs[jmp_end][1] = len(self.code.instrs)  # 回填：循环出口地址

注意末尾那行——循环体编译完了，循环出口的地址（len(self.code.instrs)）才确定，这时回头把占位的 None 改成真实地址。if/else 同理，只是回填两处（else 落点 + 汇合点）。

函数与帧栈：CALL / RETURN ​

最有意思的部分来了——函数。def 编译成「造一个函数对象、存进名字」；调用编译成「压函数、压实参、CALL_FUNCTION」：

python
# minivm.py —— def 与调用
def compile_funcdef(self, s):
    params = [arg.arg for arg in s.args.args]
    fcode = Code(s.name, params)                        # 函数体单独编译成一个 Code
    local = set(params) | assigned_names(s.body)        # 作用域分析：哪些名字是局部
    Compiler(fcode, "function", local).compile_stmts(s.body)
    fcode.emit("LOAD_CONST", None); fcode.emit("RETURN_VALUE")  # 隐式 return None
    self.code.emit("MAKE_FUNCTION", fcode)              # 造函数对象，压栈
    self.store_name(s.name)

这里藏着第四部分讲过的作用域：函数体里赋值过的名字（含形参）是局部，用 LOAD_FAST/STORE_FAST；其余名字（比如调用别的函数、递归调用自己）是全局，用 LOAD_GLOBAL——一个迷你版的 LEGB。

执行时，每次 CALL_FUNCTION 都新建一个帧压入帧栈，RETURN_VALUE 则弹出帧、把返回值交还调用者。这正是第四部分「帧与求值循环」的核心——当前活动的永远是帧栈顶那个帧：

函数调用的帧栈压入与弹出

python
# minivm.py —— 虚拟机里处理调用与返回（节选）
elif op == "CALL_FUNCTION":
    args = [f.stack.pop() for _ in range(argc)][::-1]   # 弹出实参
    func = f.stack.pop()                                 # 弹出函数对象
    new_locals = dict(zip(func.params, args))            # 形参 ← 实参
    frames.append(Frame(func, new_locals, glob, "function"))  # 压入新帧

elif op == "RETURN_VALUE":
    retval = f.stack.pop()
    frames.pop()                                         # 弹出当前帧
    if frames:
        frames[-1].stack.append(retval)                 # 返回值交还调用者

递归（fact、fib）就这样自然成立——同一个函数的多次调用各有各的帧、各有各的局部变量，在帧栈上层层叠起、又层层退回。

虚拟机：一个大循环 ​

把这一切驱动起来的，是那个我们已经无比熟悉的结构——一个大循环 + 一个大 dispatch，逐条「取指令 → 前进指针 → 执行」，直到帧栈空：

python
# minivm.py —— 求值循环骨架（节选）
while frames:
    f = frames[-1]                          # 当前帧 = 帧栈顶
    if f.pc >= len(f.code.instrs):          # 当前帧跑完了
        ... # 函数帧→返回 None；模块帧→整个程序结束
    record(frames, output)                  # 记录这一步的快照（给可视化用）
    op, arg = f.code.instrs[f.pc]
    f.pc += 1                               # 先前进，跳转指令会再改写
    if   op == "LOAD_CONST":   f.stack.append(arg)
    elif op == "LOAD_FAST":    f.stack.append(f.locals[arg])
    elif op == "BINARY_OP":    b = f.stack.pop(); a = f.stack.pop(); f.stack.append(_binop(arg, a, b))
    elif op == "POP_JUMP_IF_FALSE":
        if not _truthy(f.stack.pop()): f.pc = arg
    ... # 其余指令

是不是和第四部分 _PyEval_EvalFrameDefault 的骨架如出一辙？这就是本书反复强调的：虚拟机的心脏，从来就是「取指令 → 派发 → 操作求值栈 → 再取下一条」。那行 record(...) 是我们额外加的——它把每一步的帧栈状态快照下来，串成一条轨迹，正是下面交互组件能「单步回放」的原因。

跑起来：单步看它执行 ​

下面就是这台迷你虚拟机的活体。选一个示例（或自己改代码），点「编译并运行」，然后用「下一步」单步走——盯着右边：求值栈怎么压怎么弹、局部变量何时写入、调用栈在递归时怎么一层层叠起又退回。

首次点击会从 CDN 加载 Python 运行环境（Pyodide，约数 MB），请稍候片刻；之后即可流畅交互。

建议试试这几件事，把前几部分的知识对应起来：

  • 跑「while 累加」，单步看 JUMP_ABSOLUTE 如何往回跳形成循环（对应控制流章）；
  • 跑「递归阶乘」，看调用栈随 fact(5)→fact(4)→… 一层层压起来，到达基准情形后又一层层退回、把返回值交还上一层（对应帧与函数机制章）；
  • 跑「递归 Fibonacci」，感受同一个函数的不同调用各有独立的帧与局部变量。

Playground：改造虚拟机本身 ​

上面的组件让你跑「用虚拟机执行的代码」；而真正的乐趣，是改虚拟机本身。下面这个 Playground，左边是这台迷你虚拟机的完整源码（可直接编辑），右边是一个 REPL——用你改过的虚拟机来执行，立刻验证效果。

👉 打开 Playground（顶部导航栏也有入口）：左边直接编辑这台虚拟机的源码，右边是一个终端式 REPL，用你改过的虚拟机即时运行验证。几个上手实验：

  • 加一个运算符：在 BINARY_OP 的 _binop 里加一行，让某个符号有新含义；
  • 新增一条指令：定义一个新 op，在编译器某处 emit 它、在虚拟机 run 里加一个分支处理它；
  • 改改报错信息，或在 PRINT 分支里给输出加个前缀——再在 REPL 里看变化。

Playground 的 REPL 和本章 minivm.py 的命令行 REPL 是同一套 execute() 在背后驱动——变量与函数定义在多次输入间保留，「应用并重载 VM」或「清空会话」会重置。

旁注：看看真实的 CPython 字节码 ​

我们的玩具指令集是简化版。真实 CPython 的字节码可以用标准库 dis 直接看——你会发现两者形神俱似：

python
>>> import dis
>>> def f(n):
...     return n * 2
>>> dis.dis(f)
  2           0 LOAD_FAST                0 (n)
              2 LOAD_CONST               1 (2)
              4 BINARY_MULTIPLY
              6 RETURN_VALUE

LOAD_FAST、LOAD_CONST、RETURN_VALUE——这些名字我们刚刚都亲手实现过。区别只在于：真实版用下标（0、1）去 co_varnames/co_consts 查表，而我们为了直观把名字和常量内联了；真实版还有上百条指令、EXTENDED_ARG、以及 3.11+ 的内联缓存等优化。但「栈式 + 取指派发」的内核，与你刚跑通的这台迷你虚拟机完全一致。

小结与扩展 ​

这一章我们用 ~300 行 Python 把全书的主线亲手走了一遍：

  • 编译：ast.parse 得到 AST，编译器把表达式翻成「压栈」指令、语句翻成「副作用」指令，if/while 用跳转回填生成条件/无条件跳转；
  • 执行：一个大循环 + dispatch逐条执行，操作求值栈；
  • 函数：MAKE_FUNCTION/CALL_FUNCTION/RETURN_VALUE + 帧栈支撑了调用与递归，作用域分析区分 LOAD_FAST/LOAD_GLOBAL；
  • 这套结构与 CPython 的 compile.c + ceval.c 同构，只是处处从简。

想继续深入，这些都是很好的练习（难度递增）：

  1. 字符串与列表：让 LOAD_CONST 支持字符串，新增 BUILD_LIST/BINARY_SUBSCR；
  2. break/continue：在 while 里用跳转实现（回顾控制流章的 block 栈思路）；
  3. and/or 短路：编译成条件跳转；
  4. 闭包：让函数能捕获外层局部变量——这要引入 cell 与 LOAD_DEREF，正是第四部分「函数机制：闭包」讲的那套；
  5. 异常：try/except 与栈展开，对应「异常机制」章。

每一项，回到对应章节都能找到 CPython 的「标准答案」。至此，从读源码到写实现，这趟旅程画上句号——愿你眼中的 Python，已经从一门「会用的语言」，变成了一台「看得见内部齿轮转动的机器」。

完整源码 ​

上面的片段都摘自同一个文件 minivm.py，也正是交互组件里实际运行的那份代码（单一事实来源）。它还带一个命令行入口，把文件下载下来即可在本地直接运行：

console
$ python minivm.py              # 进入交互式 REPL，边敲边执行
迷你 Python 虚拟机 · 交互式 REPL
>>> 1 + 2 * 3
7
>>> def sq(n):
...     return n * n
...
>>> print(sq(9))
81

$ python minivm.py demo.py      # 直接运行一个脚本文件

完整源码列在这里：

python
"""minivm.py —— 一个用 Python 写成的迷你 Python 虚拟机。

流水线：源码 → AST（ast.parse）→ 玩具字节码 → 求值循环（带帧栈）。

对外只暴露 run_trace(src) -> JSON 字符串：把一段源码编译并「带轨迹地」执行，
返回每一步的帧栈快照，供浏览器里的可视化组件单步播放。

为了直观，玩具字节码把名字与常量直接内联在指令参数里
（真实 CPython 用下标去 co_consts / co_varnames 查表）。
"""
import ast
import json

_counter = 0
def _new_id():
    global _counter
    _counter += 1
    return "c%d" % _counter


# ============ 字节码：一条指令就是 (op, arg) ============

class Code:
    """一段可执行的字节码：模块体，或一个函数体。"""
    def __init__(self, name, params=()):
        self.id = _new_id()
        self.name = name
        self.params = list(params)   # 形参名
        self.instrs = []             # [[op, arg], ...]

    def emit(self, op, arg=None):
        self.instrs.append([op, arg])
        return len(self.instrs) - 1

    def listing(self):
        """生成给人看的反汇编文本。"""
        out = []
        for op, arg in self.instrs:
            if op == "MAKE_FUNCTION":
                a = arg.name
            elif arg is None:
                a = ""
            elif op == "LOAD_CONST":
                a = repr(arg)
            else:
                a = str(arg)
            out.append((op + " " + a).strip())
        return out


# ============ 编译器：AST → 玩具字节码 ============

BINOPS = {ast.Add: "+", ast.Sub: "-", ast.Mult: "*", ast.Div: "/",
          ast.FloorDiv: "//", ast.Mod: "%", ast.Pow: "**"}
CMPOPS = {ast.Lt: "<", ast.LtE: "<=", ast.Gt: ">", ast.GtE: ">=",
          ast.Eq: "==", ast.NotEq: "!="}


class CompileError(Exception):
    pass


def assigned_names(body):
    """收集函数体里被赋值过的名字（连同形参，就是这个函数的局部变量）。"""
    names = set()
    for stmt in body:
        for node in ast.walk(stmt):
            if isinstance(node, ast.Assign):
                for t in node.targets:
                    if isinstance(t, ast.Name):
                        names.add(t.id)
            elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name):
                names.add(node.target.id)
    return names


class Compiler:
    def __init__(self, code, scope, localnames):
        self.code = code
        self.scope = scope            # 'module' | 'function'
        self.localnames = localnames  # 函数作用域内的局部名字集合

    def compile_stmts(self, stmts):
        for s in stmts:
            self.compile_stmt(s)

    # ---- 语句 ----
    def compile_stmt(self, s):
        if isinstance(s, ast.Assign):
            if len(s.targets) != 1 or not isinstance(s.targets[0], ast.Name):
                raise CompileError("只支持 `名字 = 表达式` 形式的赋值")
            self.compile_expr(s.value)
            self.store_name(s.targets[0].id)
        elif isinstance(s, ast.AugAssign):
            if not isinstance(s.target, ast.Name):
                raise CompileError("只支持对名字做 += 这类增量赋值")
            if type(s.op) not in BINOPS:
                raise CompileError("暂不支持的运算符")
            self.load_name(s.target.id)
            self.compile_expr(s.value)
            self.code.emit("BINARY_OP", BINOPS[type(s.op)])
            self.store_name(s.target.id)
        elif isinstance(s, ast.If):
            self.compile_if(s)
        elif isinstance(s, ast.While):
            self.compile_while(s)
        elif isinstance(s, ast.Return):
            if self.scope != "function":
                raise CompileError("return 只能用在函数里")
            if s.value is None:
                self.code.emit("LOAD_CONST", None)
            else:
                self.compile_expr(s.value)
            self.code.emit("RETURN_VALUE")
        elif isinstance(s, ast.FunctionDef):
            self.compile_funcdef(s)
        elif isinstance(s, ast.Expr):
            self.compile_expr(s.value)
            self.code.emit("POP_TOP")   # 表达式语句：算完把结果丢弃
        elif isinstance(s, ast.Pass):
            pass
        else:
            raise CompileError("暂不支持的语句：%s" % type(s).__name__)

    def compile_if(self, s):
        self.compile_expr(s.test)
        jmp_else = self.code.emit("POP_JUMP_IF_FALSE", None)   # 占位，待回填
        self.compile_stmts(s.body)
        if s.orelse:
            jmp_end = self.code.emit("JUMP_ABSOLUTE", None)
            self.code.instrs[jmp_else][1] = len(self.code.instrs)   # 回填 else 落点
            self.compile_stmts(s.orelse)
            self.code.instrs[jmp_end][1] = len(self.code.instrs)    # 回填汇合点
        else:
            self.code.instrs[jmp_else][1] = len(self.code.instrs)

    def compile_while(self, s):
        start = len(self.code.instrs)
        self.compile_expr(s.test)
        jmp_end = self.code.emit("POP_JUMP_IF_FALSE", None)
        self.compile_stmts(s.body)
        self.code.emit("JUMP_ABSOLUTE", start)                  # 往回跳，形成循环
        self.code.instrs[jmp_end][1] = len(self.code.instrs)    # 回填循环出口

    def compile_funcdef(self, s):
        if self.scope != "module":
            raise CompileError("迷你版只支持在模块顶层定义函数（不支持嵌套 / 闭包）")
        a = s.args
        if (a.vararg or a.kwarg or a.kwonlyargs or a.defaults or a.kw_defaults):
            raise CompileError("函数暂只支持简单的位置参数")
        params = [arg.arg for arg in a.args]
        fcode = Code(s.name, params)
        local = set(params) | assigned_names(s.body)
        Compiler(fcode, "function", local).compile_stmts(s.body)
        fcode.emit("LOAD_CONST", None)    # 函数体跑到尽头：隐式 return None
        fcode.emit("RETURN_VALUE")
        self.code.emit("MAKE_FUNCTION", fcode)
        self.store_name(s.name)

    # ---- 表达式 ----
    def compile_expr(self, e):
        if isinstance(e, ast.Constant):
            if not isinstance(e.value, (int, float)) and e.value is not None:
                raise CompileError("迷你版只支持数字 / 布尔 / None 常量")
            self.code.emit("LOAD_CONST", e.value)
        elif isinstance(e, ast.Name):
            self.load_name(e.id)
        elif isinstance(e, ast.BinOp):
            if type(e.op) not in BINOPS:
                raise CompileError("暂不支持的运算符")
            self.compile_expr(e.left)
            self.compile_expr(e.right)
            self.code.emit("BINARY_OP", BINOPS[type(e.op)])
        elif isinstance(e, ast.UnaryOp) and isinstance(e.op, ast.USub):
            self.compile_expr(e.operand)
            self.code.emit("UNARY_NEG")
        elif isinstance(e, ast.Compare):
            if len(e.ops) != 1:
                raise CompileError("暂不支持连续比较（如 a < b < c）")
            if type(e.ops[0]) not in CMPOPS:
                raise CompileError("暂不支持的比较运算符")
            self.compile_expr(e.left)
            self.compile_expr(e.comparators[0])
            self.code.emit("COMPARE_OP", CMPOPS[type(e.ops[0])])
        elif isinstance(e, ast.Call):
            self.compile_call(e)
        else:
            raise CompileError("暂不支持的表达式：%s" % type(e).__name__)

    def compile_call(self, e):
        if e.keywords:
            raise CompileError("函数调用暂不支持关键字参数")
        if isinstance(e.func, ast.Name) and e.func.id == "print":
            if len(e.args) != 1:
                raise CompileError("迷你版的 print 只接受一个参数")
            self.compile_expr(e.args[0])
            self.code.emit("PRINT")            # 输出，并压入 None（print 返回 None）
            return
        self.compile_expr(e.func)              # 先把函数对象压栈
        for a in e.args:                       # 再依次压入实参
            self.compile_expr(a)
        self.code.emit("CALL_FUNCTION", len(e.args))

    # ---- 名字的载入 / 存储：迷你 LEGB ----
    def load_name(self, name):
        if self.scope == "function":
            if name in self.localnames:
                self.code.emit("LOAD_FAST", name)     # 局部变量
            else:
                self.code.emit("LOAD_GLOBAL", name)   # 模块级（如调用别的函数）
        else:
            self.code.emit("LOAD_NAME", name)

    def store_name(self, name):
        if self.scope == "function":
            self.code.emit("STORE_FAST", name)
        else:
            self.code.emit("STORE_NAME", name)


def compile_module(src):
    tree = ast.parse(src)
    code = Code("<module>")
    Compiler(code, "module", set()).compile_stmts(tree.body)
    return code


# ============ 虚拟机：求值循环 + 帧栈 ============

class Frame:
    """执行一段字节码的现场：求值栈、局部变量、指令指针。"""
    def __init__(self, code, local, glob, kind):
        self.code = code
        self.locals = local      # 名字 -> 值
        self.globals = glob
        self.kind = kind         # 'module' | 'function'
        self.stack = []          # 求值栈
        self.pc = 0              # 指令指针


class VMError(Exception):
    pass


MAX_STEPS = 6000
MAX_DEPTH = 60


def _disp(v):
    if isinstance(v, Code):
        return "<function %s>" % v.name
    if v is None:
        return "None"
    if v is True:
        return "True"
    if v is False:
        return "False"
    return repr(v)


def _truthy(v):
    return not (v is None or v is False or v == 0)


def _binop(op, a, b):
    if op == "+":
        return a + b
    if op == "-":
        return a - b
    if op == "*":
        return a * b
    if op == "/":
        if b == 0:
            raise VMError("除以零")
        return a / b
    if op == "//":
        if b == 0:
            raise VMError("除以零")
        return a // b
    if op == "%":
        return a % b
    if op == "**":
        return a ** b
    raise VMError("未知运算符 %s" % op)


def _compare(op, a, b):
    return {"<": a < b, "<=": a <= b, ">": a > b, ">=": a >= b,
            "==": a == b, "!=": a != b}[op]


def run(module_code, record, glob=None):
    if glob is None:
        glob = {}
    frames = [Frame(module_code, glob, glob, "module")]
    output = []
    steps = 0

    while frames:
        f = frames[-1]

        # 当前帧跑到尽头
        if f.pc >= len(f.code.instrs):
            if f.kind == "function":         # 函数没显式 return：返回 None
                frames.pop()
                if frames:
                    frames[-1].stack.append(None)
                continue
            break                            # 模块体结束：整个程序结束

        record(frames, output)               # 记录「即将执行 f.pc 这条指令」的快照
        steps += 1
        if steps > MAX_STEPS:
            raise VMError("执行步数过多（可能是死循环）")

        op, arg = f.code.instrs[f.pc]
        f.pc += 1                            # 先前进，跳转指令会再改写

        if op == "LOAD_CONST":
            f.stack.append(arg)
        elif op == "LOAD_FAST":
            if arg not in f.locals:
                raise VMError("局部变量 %s 在赋值前被使用" % arg)
            f.stack.append(f.locals[arg])
        elif op == "STORE_FAST":
            f.locals[arg] = f.stack.pop()
        elif op == "LOAD_NAME":
            if arg not in f.locals:
                raise VMError("名字 %s 未定义" % arg)
            f.stack.append(f.locals[arg])
        elif op == "STORE_NAME":
            f.locals[arg] = f.stack.pop()
        elif op == "LOAD_GLOBAL":
            if arg not in glob:
                raise VMError("名字 %s 未定义" % arg)
            f.stack.append(glob[arg])
        elif op == "BINARY_OP":
            b = f.stack.pop()
            a = f.stack.pop()
            f.stack.append(_binop(arg, a, b))
        elif op == "UNARY_NEG":
            f.stack.append(-f.stack.pop())
        elif op == "COMPARE_OP":
            b = f.stack.pop()
            a = f.stack.pop()
            f.stack.append(_compare(arg, a, b))
        elif op == "POP_JUMP_IF_FALSE":
            if not _truthy(f.stack.pop()):
                f.pc = arg
        elif op == "JUMP_ABSOLUTE":
            f.pc = arg
        elif op == "PRINT":
            output.append(_disp(f.stack.pop()))
            f.stack.append(None)
        elif op == "POP_TOP":
            f.stack.pop()
        elif op == "MAKE_FUNCTION":
            f.stack.append(arg)              # arg 是函数体 Code，直接当函数对象
        elif op == "CALL_FUNCTION":
            argc = arg
            args = [f.stack.pop() for _ in range(argc)][::-1]
            func = f.stack.pop()
            if not isinstance(func, Code):
                raise VMError("不是可调用对象")
            if len(args) != len(func.params):
                raise VMError("%s() 需要 %d 个参数，给了 %d 个"
                              % (func.name, len(func.params), len(args)))
            if len(frames) >= MAX_DEPTH:
                raise VMError("递归过深（可能无限递归）")
            new_locals = dict(zip(func.params, args))
            frames.append(Frame(func, new_locals, glob, "function"))
        elif op == "RETURN_VALUE":
            retval = f.stack.pop()
            frames.pop()
            if frames:
                frames[-1].stack.append(retval)
        else:
            raise VMError("未知指令 %s" % op)

    record(frames, output, done=True)
    return output


def execute(src, glob=None):
    """编译并执行一段源码，返回（输出文本, 全局名字空间）。

    传入并复用同一个 glob，即可在多次调用间保留变量与函数定义——REPL 靠它实现。
    """
    module_code = compile_module(src)
    out = run(module_code, lambda *a, **k: None, glob)
    return "\n".join(out), glob


# ============ 对外接口：带轨迹地跑一遍，返回 JSON ============

def _collect_codes(code, out):
    out[code.id] = {"name": code.name, "listing": code.listing()}
    for op, arg in code.instrs:
        if op == "MAKE_FUNCTION":
            _collect_codes(arg, out)


def run_trace(src):
    global _counter
    _counter = 0
    try:
        module_code = compile_module(src)
    except (SyntaxError, CompileError) as e:
        return json.dumps({"ok": False, "error": "编译错误：%s" % e})

    codes = {}
    _collect_codes(module_code, codes)
    trace = []

    def record(frames, output, done=False):
        trace.append({
            "frames": [
                {
                    "code_id": fr.code.id,
                    "name": fr.code.name,
                    "pc": fr.pc,
                    "stack": [_disp(x) for x in fr.stack],
                    "locals": [[k, _disp(v)] for k, v in fr.locals.items()],
                }
                for fr in frames
            ],
            "output": "\n".join(output),
            "done": done,
        })

    try:
        run(module_code, record)
    except Exception as e:                   # noqa: BLE001 —— 任何运行期错误都友好返回
        return json.dumps({"ok": False, "error": "运行错误：%s" % e,
                           "codes": codes, "trace": trace})

    return json.dumps({"ok": True, "codes": codes, "trace": trace})


# ============ 命令行入口：直接运行文件，或进入交互式 REPL ============

def _maybe_echo(src):
    """REPL 小贴心：若输入是一句纯表达式（且不是 print 调用），自动回显它的值。"""
    try:
        tree = ast.parse(src)
    except SyntaxError:
        return src
    if len(tree.body) == 1 and isinstance(tree.body[0], ast.Expr):
        e = tree.body[0].value
        is_print = (isinstance(e, ast.Call) and isinstance(e.func, ast.Name)
                    and e.func.id == "print")
        if not is_print:
            return "print(" + src.strip() + ")"
    return src


def _run_text(src, glob):
    try:
        text, glob = execute(src, glob)
    except (CompileError, SyntaxError) as e:
        print("编译错误：%s" % e)
    except VMError as e:
        print("运行错误：%s" % e)
    else:
        if text:
            print(text)
    return glob


def _run_file(path):
    with open(path, encoding="utf-8") as fp:
        _run_text(fp.read(), {})


def _repl():
    print("迷你 Python 虚拟机 · 交互式 REPL")
    print("支持：赋值、算术 / 比较、if / while、def / return（含递归）、print")
    print("块语句（def/if/while）输入完后敲一个空行执行；Ctrl-D / Ctrl-C 退出。\n")
    glob = {}
    while True:
        try:
            first = input(">>> ")
            if first.strip() == "":
                continue
            lines = [first]
            # 以冒号结尾说明是块（def/if/while…），继续读到空行为止
            if first.rstrip().endswith(":"):
                while True:
                    cont = input("... ")
                    if cont.strip() == "":
                        break
                    lines.append(cont)
            src = "\n".join(lines)
            glob = _run_text(_maybe_echo(src), glob)
        except (EOFError, KeyboardInterrupt):
            print("\n再见！")
            break


if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1:
        _run_file(sys.argv[1])
    else:
        _repl()

Copyright © FlagGo 2019-2026