把输入输出先固定成固定帧率——HelixWorld 的实时交互才谈得上可复现

文章导读
每次跑 HelixWorld 的交互,效果都不一样时,先不要急着怀疑模型。常见情况是:输入采样一会儿按事件触发、一会儿按渲染帧触发,输出侧又没有记录每帧什么时候到,于是两次运行之间差的不只是模型,还有输入的节奏和采到的样本数量。可复现的前提是把输入采集和输出呈现都钉在同一个固定间隔上,记录时间戳和丢帧,再去比较模型或参数的变化。下面的四个动作可以按顺序做:先量出当前输入节奏,再改成固定间隔触发,然
📋 目录
  1. A 记录当前输入是怎么采进来的
  2. B 把输入采集改成固定间隔触发
  3. C 在输出侧记录每帧到达时间与丢帧数
  4. D 用同一段输入跑两次做对照
  5. E 把固定节奏写成默认档
A A

每次跑 HelixWorld 的交互,效果都不一样时,先不要急着怀疑模型。常见情况是:输入采样一会儿按事件触发、一会儿按渲染帧触发,输出侧又没有记录每帧什么时候到,于是两次运行之间差的不只是模型,还有输入的节奏和采到的样本数量。可复现的前提是把输入采集和输出呈现都钉在同一个固定间隔上,记录时间戳和丢帧,再去比较模型或参数的变化。下面的四个动作可以按顺序做:先量出当前输入节奏,再改成固定间隔触发,然后在输出侧埋点,最后用同一段输入跑两次对照,确认节奏已经固定下来。

判断能不能复现,先看节奏不看模型。把输入采集从“事件到达就用”改成固定间隔批量取用,把输出每帧到达时间与丢帧数写进日志,再用同一段输入跑两次对照。适用场景是需要横向比较模型或参数、但每次交互手感都不一致的调试阶段。验证方式是TICK日志的间隔是否稳定、两次运行的输入批次序列与丢帧数是否落在同一量级。边界在于:固定节奏只消除输入抖动这一项差异,模型自身的随机性、随机种子、硬件负载仍需单独控制。

记录当前输入是怎么采进来的

输入节奏不固定的来源通常有这几类:输入回调直接触发处理逻辑,鼠标移动快时一秒处理几十次、慢时几次;输入在渲染循环里顺带读取,渲染一卡输入就跟着稀疏;还有的把输入采样放在异步线程里,却用另一个线程的时钟打时间戳。先不改结构,只加时间戳打印,看日志里相邻两次输入处理的时间差分布,就能判断属于哪一类。

import time

pending = []

# 第一层:输入到达时刻(设备或事件回调里)
def on_input_event(evt):
    t_arrive = time.perf_counter()
    print(f"IN_ARRIVE t={t_arrive:.6f} kind={evt.kind} src={evt.source}")
    pending.append((t_arrive, evt))

# 第二层:输入被真正用掉的时刻(处理逻辑里)
def consume_input():
    t_use = time.perf_counter()
    batch = pending[:]
    pending.clear()
    print(f"IN_USE t={t_use:.6f} n={len(batch)}")
    apply(batch)

把IN_ARRIVE和IN_USE两类日志都打开,跑一段有快有慢的操作。如果IN_USE的间隔忽大忽小,且n的值在0和很大之间跳,说明输入是被“来多少用多少”消费的,这就是要排除的抖动源。这一步只做观测,先别改代码,否则后面没有基线可对比。

把输入采集改成固定间隔触发

改动思路是把“输入到达”和“输入被使用”解耦:回调只负责把事件塞进缓冲队列,处理侧由一个固定间隔的定时器统一取走。这样每次tick拿到的输入批次数量可能不同,但tick本身的发生时刻是均匀的,模型看到的是稳定的采样节奏,而不是设备事件流的随机疏密。

TICK_HZ = 60
TICK_S = 1.0 / TICK_HZ
seq = 0
next_t = time.perf_counter()

while running:
    t_use = time.perf_counter()
    batch = pending[:]
    pending.clear()
    apply(batch)
    print(f"TICK seq={seq} t={t_use:.6f} n_in={len(batch)}")
    seq += 1
    next_t += TICK_S
    delay = next_t - time.perf_counter()
    if delay > 0:
        time.sleep(delay)
    else:
        # 落后时不追帧,只记录,避免一次补多帧把节奏放大
        print(f"TICK_LATE seq={seq - 1} late_ms={-delay * 1000:.2f}")
        next_t = time.perf_counter()

预期日志是连续的TICK行,相邻t的差值接近1/60秒,n_in的取值随操作快慢变化。如果频繁出现TICK_LATE,说明主循环里有阻塞操作(同步IO、模型推理卡住、锁竞争),这类阻塞本身也会改变输入被使用的时间点,需要先定位。这里刻意选择“落后不补帧”:补帧会让一次循环里连续应用多个批次,反而把节奏弄乱,通常更适合调试阶段。

在输出侧记录每帧到达时间与丢帧数

输入固定了,还要能判断输出有没有跟上。输出侧埋点只需要两样:每帧呈现时刻,以及相对期望帧间隔漏了几帧。在渲染完成提交或画面上屏的回调里打点,不要放在“逻辑帧开始”,两者在有队列缓冲时会差出好几帧,容易误判。

OUT_HZ = 60
OUT_S = 1.0 / OUT_HZ
last_present = None
frame_seq = 0
drop_total = 0


def on_frame_present():
    global last_present, frame_seq, drop_total
    t = time.perf_counter()
    if last_present is None:
        print(f"FRAME seq={frame_seq} t={t:.6f} dt=NA drop=0")
    else:
        dt = t - last_present
        missed = max(0, round(dt / OUT_S) - 1)
        if missed:
            drop_total += missed
        print(f"FRAME seq={frame_seq} t={t:.6f} "
              f"dt_ms={dt * 1000:.2f} missed={missed} total_drop={drop_total}")
    last_present = t
    frame_seq += 1

需要结合环境确认一件事:OUT_HZ要和实际屏幕刷新或合成节奏对齐。如果显示器是另一个刷新率,missed会被系统性放大,那不是真丢帧,是常量填错。可以先跑一段静止画面,看dt_ms的稳定值是多少,再反推OUT_S。日志落盘时建议按运行编号分文件,方便后面做对照。

用同一段输入跑两次做对照

把输入录下来(每条事件的时间偏移和内容),回放两次,其余条件不变,包括同一台机器、同一个窗口尺寸、尽量接近的负载状态。回放时输入批次应该完全一致,TICK的seq序列也应该一致,此时剩下的差异才可能来自模型或参数。

对照记录表建议包含这些字段:

  • run_id、开始/结束时间、输入录制文件与其校验值
  • input_tick_hz、output_target_hz、实际生效值
  • TICK总条数、n_in=0的tick条数、TICK_LATE条数
  • FRAME总帧数、total_drop、最大dt_ms
  • 关键状态快照(如首帧、第N个tick时的可见状态或状态哈希)

差异判定分两层:先比输入层,两次的TICK条数、n_in序列、seq是否一致,不一致说明回放本身没对齐,后面的比较没有意义;输入层一致后再比输出层,看total_drop和最大dt_ms是否在同一量级、关键状态快照是否可解释。不建议只凭“看着差不多”下判断,这类主观判断在帧间差异很小的时候基本不可靠。

把固定节奏写成默认档

调试完成后要把这些值落到配置里,而不是靠每次手动改代码。常用做法是配置文件加命令行覆盖,启动时打印实际生效值,避免出现“以为设了其实没读到”的情况。

[timing]
input_tick_hz = 60
output_target_hz = 60
late_warn_ms = 5
replay_input = ./runs/base_input.jsonl
record_frames = ./runs/latest/frames.log

启动参数可以对应成 `--input-tick-hz` 60 `--output-target-hz` 60 `--replay-input` ./runs/base_input.jsonl 这样的形式。生效验证方式有三条:启动日志里回显的数值与配置一致;跑固定时长后TICK行数约等于频率乘时长(按日志自行核对,不预设结论);TICK行里的t间隔分布没有明显长尾。三条都过了,再把固定档定为默认,后续调模型时输入这一项就是常量,出现差异时可以先从模型参数、随机种子和渲染负载去找原因。