「JIT 编译进阶与热点优化」

# 解释循环: 每条指令都在解释器分派循环里执行
def interpret(bytecode, state):
    ip = 0
    counters = {}                     # 记录每个循环入口的执行次数
    while ip < len(bytecode):
        op, arg = bytecode[ip]
        if op == "LOOP_START":
            counters[arg] = counters.get(arg, 0) + 1
        if op == "CONST":
            state.append(arg)
        elif op == "ADD":
            b, a = state.pop(), state.pop()
            state.append(a + b)
        ip += 1
    return counters

bc = [("LOOP_START", 0), ("CONST", 1), ("CONST", 2), ("ADD", None)] * 1000
counters = interpret(bc, [])
print("循环 0 执行次数:", counters)

# 模拟预热曲线: 编译发生后单次迭代耗时骤降
def warm_up_curve(iterations, compile_at=50_000):
    times = []
    for i in range(iterations):
        if i < compile_at:
            times.append(1.0)          # 解释执行: 慢
        elif i < compile_at + 10_000:
            times.append(0.6)          # 编译进行中
        else:
            times.append(0.1)          # 机器码: 快
    return times

curve = warm_up_curve(200_000)
print("预热前 5 次耗时:", curve[:5])
print("峰值耗时:", min(curve))

class MethodProfile:
    def __init__(self, threshold=10000, loop_threshold=1000):
        self.invocations = 0
        self.backedges = 0
        self.threshold = threshold
        self.loop_threshold = loop_threshold

    def on_call(self):
        self.invocations += 1
        if self.invocations >= self.threshold:
            return "compile_now"
        return "keep_interpreting"

    def on_backedge(self):
        self.backedges += 1
        return "compile_now" if self.backedges >= self.loop_threshold else "keep_interpreting"

m = MethodProfile()
for i in range(10005):
    m.on_call()
print("调用后状态:", m.invocations >= m.threshold)

def choose_tier(profile, has_rich_profile):
    """HotSpot 风格: 解释 -> C1 -> C2 的升级决策."""
    if profile.invocations < 2000:
        return "interpreter"
    if not has_rich_profile:
        return "C1 (轻量编译, 继续采集)"
    return "C2 (深度优化)"

class TieredRuntime:
    def __init__(self):
        self.tier = {}          # 方法 -> 当前层级
    def tick(self, method, profile):
        cur = self.tier.get(method, 0)
        if cur == 0 and profile.invocations > 2000:
            self.tier[method] = 1           # 升到 C1
        elif cur == 1 and profile.invocations > 200000:
            self.tier[method] = 2           # 升到 C2
        return self.tier[method]

rt = TieredRuntime()
print(rt.tick("foo", MethodProfile(threshold=1)))

def osr_decision(backedge_count, threshold=100000):
    """长循环在回边处决定是否 OSR 进入编译版本."""
    if backedge_count >= threshold:
        return "OSR: 当前栈帧替换为编译帧, 从 OSR 入口继续"
    return "继续解释执行"

for i in range(5):
    print(i, osr_decision(i * 50000))

def speculative_call(receiver, impl_cache, interpreter):
    """虚拟调用点: 假设接收者类型与缓存一致则直接调用."""
    cached = impl_cache.get(receiver.__class__)
    if cached is not None:
        return cached(receiver)            # 快路径: 去虚化调用
    return interpreter(receiver)           # 慢路径: 回退解释/虚分派

class A:
    def run(self): return "A"

cache = {A: lambda o: "A (内联)"}
print(speculative_call(A(), cache, lambda o: o.run()))

class DeoptFrame:
    """去优化帧: 记录可恢复的解释器状态."""
    def __init__(self, locals_, stack, pc):
        self.locals = locals_
        self.stack = stack
        self.pc = pc
    def to_interpreter(self):
        return f"恢复解释执行 pc={self.pc} locals={self.locals}"

print(DeoptFrame(["x=1"], [], 42).to_interpreter())

class SafepointManager:
    def __init__(self):
        self.polling_page = bytearray(4096)   # 轮询页
    def sync(self, thread, regs):
        if self.polling_page[0]:              # 轮询点: 检查 GC 请求
            return "safepoint 暂停, 保存寄存器与栈"
        return "继续执行"

# 伪代码: JIT 三阶段流水线
def jit_compile(module, backend="cranelift"):
    hir = build_hir(module)            # 1. 前端: 语言 -> HIR
    mir = lower_to_mir(hir)            # 2. 中端: HIR -> 机器无关 MIR
    machine_code = backend(mir)        # 3. 后端: MIR -> 目标机器码
    return Stub(module, machine_code)  # 用桩占位, 触发时才真正链接

class Stub:
    def __init__(self, module, code):
        self.module, self.code, self.patched = module, code, False
    def call(self, *args):
        if not self.patched:
            self.patched = True        # 首次调用后补丁真实入口
        return self.code(*args)