第 18 / 30 课。 前置:上一课。本篇的运行范围以实验说明和验证记录为准,构造样例不代表真实任务精度或板端性能。
帧率正常,为什么画面仍然滞后
摄像头每 33ms 产生一帧,模型每帧需要 50ms。若全部排队,服务率低于到达率,等待时间持续增长;即使推理耗时稳定,输出也越来越旧。扩大队列只能推迟溢出,不能消除供需差异。
本课比较 FIFO 与只有一个待处理槽的最新帧策略。后者会丢弃等待中的旧帧,限制帧龄,但无法恢复被跳过的快速事件。产品需要明确更看重完整记录还是实时反应。
时间戳穿过全部模块
帧至少携带 frame_id、采集时刻、推理起止时刻和输出时刻。端到端延迟从采集算起,不能只用推理结束减推理开始。设备间时钟不一致时需先解决同步;本地耗时应采用单调时钟。
多工作线程可能乱序完成。下游不能把较旧结果覆盖较新状态;跟踪器需要按采集顺序更新,或设计明确的乱序处理。NPU 异步调用还需等待实际完成后再计时和复用缓冲。
可复现实验
脚本不依赖 sleep,使用构造的到达时间和固定服务时间执行离散事件模拟,因此结果可复现。它报告处理数、丢帧数和 P95,计入最后一帧排空队列的时间,不把未完成帧偷偷丢掉。
动手任务
把服务时间改为 20ms,解释为什么两种策略趋于一致。再让每十帧有一帧耗时 100ms,观察平均值与尾延迟的差别。参考结论:足够快时没有积压;偶发慢帧可显著影响 P95,平均值无法描述尾部。
真实验收还需记录摄像头缓存、颜色转换、传输、后处理和显示,不能用本模拟的毫秒数作为板端性能。
配套实验与运行记录
实验:视频系统:队列、时间戳与异步流水线
查看完整代码:main.py
"""用离散事件回放解释吞吐、帧龄、丢帧和结果顺序。"""
import math
def simulate(latest, service=50):
arrivals = list(range(0, 3300, 33))
i = 0
now = 0
latencies = []
dropped = 0
while i < len(arrivals):
now = max(now, arrivals[i])
chosen = i
if latest:
while chosen + 1 < len(arrivals) and arrivals[chosen + 1] <= now:
chosen += 1
dropped += chosen - i
now += service
latencies.append(now - arrivals[chosen])
i = chosen + 1
return {
"processed": len(latencies),
"dropped": dropped,
"p95_ms": sorted(latencies)[math.ceil(0.95 * len(latencies)) - 1],
"last_finish_ms": now,
}
def main():
fifo = simulate(False)
latest = simulate(True)
print("FIFO", fifo, "latest slot", latest)
assert (
latest["p95_ms"] < fifo["p95_ms"]
and latest["processed"] + latest["dropped"] == 100
)
assert simulate(False, 20) == simulate(True, 20)
print("fast service has no queue-policy difference")
if __name__ == "__main__":
main()
在项目根目录执行;下载包解压后,在解压目录执行 python3 main.py。
cd 'content/posts/AI-深度学习/嵌入式视觉模型进阶/18-视频系统/code' && python3 main.py
在本篇代码目录执行 python3 main.py。依赖安装、可选入口见包内 README.md;实际运行结果见 verification.md。下载包包含本课使用的辅助模块与样例。
参考与进一步阅读
原始论文或官方文档用于核对结构、接口和适用范围。本篇代码是围绕教学目标编写的最小实验;除明确注明的完整入口外,不视为对应论文模型的完整复现。