做双路视觉这件事,最折磨人的往往不是模型精度,也不是NPU跑不动,而是你辛辛苦苦把两路yolov5s推理都调通了,画面却在肉眼可见地变卡。香橙派5这类RK3588平台上,我前面几篇已经把单路摄像头取流、yolov5s转rknn、NPU推理、画框这些环节都跑通了,但两路同时一开,延迟就像滚雪球一样越滚越大。这篇记录的是我换到“丢旧帧背压”方案之后的完整做法,包括队列为什么一定会膨胀、单槽位邮箱怎么设计、采集线程和推理线程怎么拆分、实测延迟能压到多少,以及几个我在香橙派5上踩到并且觉得你一定也会踩的坑。适合已经把单路跑通、正卡在双路实时性上的朋友。
1. 双路实时瓶颈:先想明白延迟是从哪来的
1.1 算术账:30帧进、15帧出,队列一定失控
假设每路摄像头是1080p@30fps,帧间隔大约是33ms。yolov5s的INT8模型在RK3588的NPU单核上推理大概要20~25ms,加上画框、缩放、颜色转换这些前后处理,一轮下来约30ms。单路情况下这个数字还能看,30帧的输入能勉强跟得上。但双路同时开工,两个推理任务如果排在同一个NPU核上轮流执行,每路有效吞吐会掉到15fps左右。
这时候如果你还用无限制长度的FIFO队列去接采集帧,算一下就知道了:每秒进来30帧,每秒只取走15帧,队列每秒会多积压14帧左右。跑60秒之后,每路队列里摞了接近900帧。新到的帧要排在900个“前辈”后面等待处理,按15fps的消费速度计算,它得等差不多60秒才能被看到。你屏幕上显示出来的,其实是半分钟以前的画面。
更麻烦的是内存。1080p的BGR图像一帧就要6MB左右,900帧就是5.4GB,香橙派5标配8GB内存也扛不住长时间跑,轻则卡死,重则进程被系统杀掉。这里真正的问题不是推理慢,而是“所有帧都想要”这个目标本身。实时检测系统里,把算力花在旧帧上等于白花钱。
1.2 背压的两种形态:阻塞式和丢弃式
说到背压,做流处理和消息队列的人应该不陌生。经典做法是队列满了就让生产者阻塞,下游处理完一个,上游才被允许继续产出一个,这就是阻塞式背压。像Linux管道、GStreamer的dataflow,走的都是这套。
但嵌入式视觉里,摄像头是个没法真正“暂停”的硬件。传感器一直在采集曝光,驱动层的buffer池也不会因为你处理慢就停止填数据。你不可能让镜头“等一等再曝光”。所以这类场景实用的背压形态变成了丢弃式:让队列长度封顶,新帧来了就把旧帧覆盖掉,消费端永远只拿最新的一帧。丢帧动作本身就是背压的体现——下游消费能力决定了有效的处理速度,队列长度有上限,端到端延迟也就有了上界。
用生活化的话说,这就像驿站货架只有一个格子。快递员(采集线程)每送来一个新件,就把架子上那个旧件扔垃圾桶。取件员(推理线程)每次来,货架上永远只有最新的一件。取件员慢一点不要紧,因为货架永远只有一件,旧件早就被新件盖掉了,你永远不会拿到半年前的快递。
2. 丢旧帧方案的核心设计:单槽位邮箱与最新帧优先
2.1 为什么“丢旧保新”是对的,而不是“丢新保旧”
既然弹性缓冲从“无数个格子”缩成一个格子,总得有帧被丢。丢哪头?如果目的是录像取证,那确实不能丢,甚至应该“丢新保旧”,尽量保留历史数据。但实时检测恰恰相反。
目标在移动。0.5秒之前的画面里,人和车的位置已经变了。模型花几十毫秒去算一个过期的框,结果对控制回路、自动跟随、安全告警都没有意义。实时系统追求的是响应新鲜度,而不是帧的完整覆盖率。香橙派这类边缘设备尤其如此,NPU算力就那么多,硬件资源有限,只能把每一帧宝贵算力花在离当前时刻最近的画面上。
所以这个方案里有一条铁律:新帧到达,无条件覆盖旧帧。旧帧还没被消费?抱歉,直接扔。推理线程永远拿最新,永远不追着队列跑。
2.2 单槽位LatestSlot的数据结构与操作语义
我用Python实现了一个单槽位邮箱,代码放在下面。结构不复杂,核心就是一个条件变量保护的一格缓冲:
import threading import time class LatestSlot: """单槽位邮箱:新帧直接覆盖旧帧,消费端拿到的永远是最近一帧。""" def __init__(self): self._cv = threading.Condition() self._frame = None self._seq = 0 self._ts = 0.0 self._ready = False def push(self, frame, seq): with self._cv: self._frame = frame self._seq = seq self._ts = time.perf_counter() self._ready = True self._cv.notify_all() # 唤醒正在等待的消费者 def pop(self, timeout_ms=500): """拿最新帧并标记为已消费;超时没新帧返回(None, -1)。""" deadline = time.monotonic() + timeout_ms / 1000.0 with self._cv: while not self._ready: remain = deadline - time.monotonic() if remain <= 0: return None, -1 self._cv.wait(remain) frame, seq = self._frame, self._seq self._ready = False return frame, seq def peek(self, timeout_ms=500): """只读最近一帧,不标记消费;适合显示/推流端持续出画面。""" deadline = time.monotonic() + timeout_ms / 1000.0 with self._cv: while not self._ready: remain = deadline - time.monotonic() if remain <= 0: return None, -1 self._cv.wait(remain) return self._frame, self._seqpush是非阻塞的,采集线程把新帧引用交出去就继续去抓下一帧;pop是消费端的,拿了就跑,并且在当前这一帧被消费完之前不会重复处理同一帧。你可能会问,推理线程要是处理得比采集快怎么办?那pop会在没数据时挂起等待,只有push进来才被唤醒,不会空转消耗CPU。
C++那边的写法本质上一样,无非就是mutex加条件变量再加一个shared_ptr交换,语义完全相同,这里就不重复贴了。
2.3 消费端等待策略:什么时候用pop,什么时候用peek
有个细节我一开始没处理好:推理线程调用pop超时返回None之后,要不要拿旧帧将就着再推一遍?我的答案是不要。重复推理同一帧纯粹是浪费NPU,画面没有任何更新,白耗算力。
但显示端和推流端恰恰相反。播放器需要连续的画面,没有新结果时会希望重复显示上一帧,而不是直接黑屏。所以我在输出侧单独挂一个out_slot,用peek去取最近的处理结果,这样画面能一直保持“有东西”,而推理线程那边始终只认新鲜帧。两套语义分开,逻辑就清晰了。
还有一点要注意:一个槽位最多挂一个消费者。如果两个线程都去pop同一个槽位,会出现一个读到帧、另一个苦苦等的情况。实际工程里我把每个LatestSlot的访问控制在单生产者单消费者,输出端另外建独立的槽位,各管各的。
3. 香橙派5上的双路实现:采集线程与推理线程的四种搭法
3.1 整体线程模型
双路丢旧帧方案在香橙派5上的落地,线程分配可以画成四条线:两路摄像头各一个采集线程,两个推理线程,外加一个可选的显示/推流线程。每路的链路都是“采集→push到slot→推理线程pop→处理后push到out_slot→显示/推流端peek”。
采集线程的职责很单纯:从摄像头拿到原始帧,立刻push进槽位,不管槽位里是不是还有上一帧正在被推理。推理线程的职责更单纯:从槽位拿最新帧,跑yolov5s,出框,把结果给输出槽位。两路之间除了共享NPU之外完全没有耦合,哪路出问题都不会拖垮另一路。
3.2 采集线程:拿到帧立刻压槽位
摄像头打开方式取决于你用的是USB还是MIPI CSI。USB摄像头最简单,直接给设备号:
import cv2 import itertools cap0 = cv2.VideoCapture(0) cap1 = cv2.VideoCapture(1) # 如果你的香橙派系统里CSI摄像头已经注册成了video节点,也可以走GStreamer管线, # 比直接v4l2好用的地方在于驱动侧做了缓冲管理: # cap = cv2.VideoCapture( # 'v4l2src device=/dev/video0 ! video/x-raw,format=NV12,width=1920,height=1080,framerate=30/1 ! ' # 'videoconvert ! video/x-raw,format=BGR ! appsink drop=1', # cv2.CAP_GSTREAMER)采集线程的主体就是while循环里面read加push,读不到帧时稍微睡一下防止空转:
stop_event = threading.Event() def capture_worker(name, cap, slot): seq = itertools.count() while not stop_event.is_set(): ok, frame = cap.read() if not ok: time.sleep(0.01) continue slot.push(frame, next(seq))这里有个容易忽略的好习惯:把seq序号带上。后边统计丢帧率、排查卡顿全靠它。
3.3 推理线程:RKNNLite拉取最新帧并出框
推理线程代码里,pop超时我习惯设100ms。没有新帧就continue,有帧就做前处理、推理、后处理。前处理的letterbox函数我沿用之前几篇里那个,顺手贴一下,方便你对照代码完整性:
import numpy as np def letterbox(img, size=(640, 640)): src_h, src_w = img.shape[:2] scale = min(size[0] / src_h, size[1] / src_w) new_w = int(src_w * scale + 0.5) new_h = int(src_h * scale + 0.5) resized = cv2.resize(img, (new_w, new_h)) canvas = np.full((size[0], size[1], 3), 114, dtype=np.uint8) ox = (size[1] - new_w) // 2 oy = (size[0] - new_h) // 2 canvas[oy:oy + new_h, ox:ox + new_w] = resized return canvas, scale, (ox, oy)推理线程主体:
def infer_worker(name, slot, out_slot, rknn, cfg): last_seq = 0 dropped = 0 while not stop_event.is_set(): frame, seq = slot.pop(timeout_ms=100) if frame is None: continue # 统计被丢弃的帧数:seq跳了多少,说明中间有多少旧帧被覆盖了 if last_seq and seq != last_seq + 1: dropped += seq - last_seq - 1 print(f"[{name}] skip {seq - last_seq - 1}, total dropped {dropped}") last_seq = seq img, ratio, (ox, oy) = letterbox(frame, (cfg.in_h, cfg.in_w)) rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) outputs = rknn.inference(inputs=[rgb[None, :, :, :]], data_format='nhwc') boxes = postprocess_yolov5s(outputs, frame.shape[1::-1], ratio, ox, oy, cfg) drawn = draw_detections(frame, boxes) out_slot.push(drawn, seq)postprocess和draw_detections是系列前面已经写好的,逻辑就是从RKNN输出里解析出坐标、置信度,再画上去,这里不重复占用篇幅。
3.4 双RKNN实例与NPU核分配
RKNNLite实例必须一路一个。两个推理线程绝对不是共享同一个实例,那会直接互相踩内部状态,轻则推理结果错乱,重则段错误。每路各加载一遍模型,内存开销也就二三十兆,对双路来说完全可以接受。
NPU核分配方面,RK3588有3个NPU核。跑双路时我建议显式绑定核0和核1,避免rknn驱动调度器把两个任务都塞到同一个核上排队。绑定方式取决于你装的rknn-toolkit2版本:
from rknnlite.api import RKNNLite def load_rknn(path, core_mask=None): rknn = RKNNLite() if rknn.load_rknn(path) != 0: raise RuntimeError("load rknn model failed") if core_mask is not None: rknn.init_runtime(core_mask=core_mask) else: rknn.init_runtime() return rknn # 新版本支持core_mask;旧版本没有这个参数就都不传,让驱动自调度 rknn0 = load_rknn('yolov5s.rknn', core_mask=1) # NPU核0 rknn1 = load_rknn('yolov5s.rknn', core_mask=2) # NPU核1实测下来,不绑核时双路吞吐经常只有绑核的六七成。绑核之后两路可以稳定跑到20fps上下,画面延迟基本恒定。如果你两路模型不一样,比如一路yolov5s另一路yolov5n,绑定更是建议做的,否则调度器大概率会按自己的脾气来,你控制不了算力倾斜。
4. 实测数据:延迟、丢帧率和资源占用
4.1 测试条件
先交代测试环境,方便你对照自己的板子:
| 项目 | 配置 |
|---|---|
| 主板 | 香橙派5(RK3588S,即RK3588平台),被动散热片 |
| 系统 | Ubuntu 22.04/20.04官方镜像均可 |
| 摄像头 | 两路USB UVC 1080p@30fps |
| 模型 | yolov5s INT8,640×640输入,RKNN格式 |
| 运行方式 | 双RKNNLite实例,绑定NPU核0/核1 |
| 单帧耗时 | NPU推理20~25ms,前后处理5~8ms |
如果你用的是MIPI CSI摄像头,驱动侧细节会略有不同,但上层这套LatestSlot逻辑是通用的,测试数据也能作为参考。
4.2 FIFO队列与丢旧帧的延迟对比
同样跑60秒,阶段一那种无上限FIFO队列和阶段二的单槽位丢旧帧方案,表现差异非常大:
| 指标(运行60秒后) | FIFO队列(阶段一) | 丢旧帧背压(阶段二) |
|---|---|---|
| 输入帧率 | 30fps/每路 | 30fps/每路 |
| 有效输出帧率 | 约15fps/每路 | 约15fps/每路 |
| 队列长度 | 持续增长,约900帧/路 | 恒定为1 |
| 新帧端到端延迟 | 约60秒且继续增长 | 60~80ms,稳定 |
| 丢帧率 | 0(全攒着不丢) | 约50% |
丢旧帧方案的端到端延迟构成非常清晰:采集等待一帧的时间(约33ms)加NPU推理时间(约25ms)加前后处理(约8ms),总在60~80ms这个区间来回晃。如果摄像头改成720p,推理时间还能再降,延迟能压到50ms出头。
4.3 用帧序号缺口统计真实丢帧率
代码里那个seq序号,就是用来验证背压到底生效没有的。你可以这样想:采集线程每秒push 30帧,seq从0一路涨到30;推理线程每秒只pop走15帧,pop出来的seq一定不是连续的。中间跳过多少,就说明有多少旧帧被覆盖丢弃了。
跑起来之后,我的日志里每路大概每2秒输出一行:skip 2、skip 1、skip 3……一小时累计丢帧率稳定在50%上下。这50%不是系统丢的,是设计上主动丢的,丢的都是“即便算了也已经过时”的帧,这就是背压在工作。
资源占用方面,两路1080p采集加推理,python进程总CPU占用大概50%到70%(主要花在视频解码、BGR转换和画框上),内存稳定在1.5GB以内。相比阶段一那种无限队列,内存占用低了一个数量级。
5. 翻车清单:丢旧帧方案里几个容易忽略的坑
5.1 底层V4L2队列和上层邮箱是两码事
丢旧帧只管应用层,管不到驱动层。用MIPI CSI摄像头时我踩过一个典型的坑:应用层邮箱只有一个格子,看着一切正常,但推理线程一慢,驱动侧的V4L2 buffer池先爆了。因为ISP驱动就那么几个buffer,采集线程从DQBUF拿到NV12原始帧后如果一直攥着不还,驱动没有空buffer可用,ISP就开始报错丢输出。
解决方法是把“取帧”和“处理帧”分离:采集线程拿到原始帧后立刻拷贝一份或者转成BGR,马上把V4L2 buffer还给驱动,慢速的推理全在上层邮箱做。GStreamer的appsink之所以顺手,就是因为它帮你做了buffer回收这件事。USB摄像头也有类似问题,UVC驱动的环形buffer深度是固定的,长时间不读就会出现画面变成慢动作甚至卡死。
5.2 不要在push里做深拷贝
cv2.read()每次返回的都是新分配的numpy数组,采集线程把数组引用直接交给邮箱就行,根本不需要copy()。如果你强行在push里copy,1080p BGR一帧6MB,两路30fps就是每秒360MB的额外内存写入,RK3588带宽虽然扛得住,但CPU占用会明显上升,推理线程抢不到CPU,延迟反而恶化。
唯一例外是高级的零拷贝/DMA buffer复用场景。如果你让摄像头直接写入一块预分配内存,那块内存下一帧会被覆盖,那push前就必须复制,否则推理线程可能一边读一边被采集线程改写,画面出现撕帧。用OpenCV默认read()的话,这个坑碰不到,不折腾直接传引用就对了。
5.3 NPU核绑定、模型轻量化和散热
双路持续满负荷跑yolov5s,RK3588的发热是实打实的。被动散热片撑不到半小时,NPU温度到80度左右就会降频,有效fps会从20掉到12~15。我处理过热问题的顺序是:先加一个USB小风扇,再把不担纲主检测的某一路换成yolov5n模型,最后才考虑降低输入分辨率。注意换分辨率不是简单改letterbox参数,转模型时要用对应尺寸的数据集重新校准量化,不然mAP会掉得很难看。
另外我试过把其中一路模型换成yolov8的rknn导出,坑位其实和yolov5s一模一样,丢旧帧的逻辑完全不用改。选型判断标准始终是:实时场景一律最新优先。
5.4 输出端也要丢旧帧
推理线程处理得快,只解决了一半问题。显示端、推流端、编码器如果还在用FIFO,旧画面照样会从另一端堆回来。我给输出端挂了同样的单槽位邮箱,推流线程用peek拿最近一帧处理结果去编码。ffmpeg推流时加-fflags nobuffer能降低一部分缓冲延迟,但本质问题还得靠上层控制发送速率,不能指望编码器帮你自动丢帧。
你甚至可以做一个调试开关:正常运行丢旧帧,一旦需要保存事件证据(比如检测到特定目标时)就把最近N帧原始图落盘,这样既保住实时性,又不会漏掉关键事件。
最后说点做这一路下来的体会。实时双路视觉真正难的地方不是把yolov5s塞进NPU,而是你能不能接受“每一帧都要处理”这个想法本身是错的。丢旧帧背压说白了就是承认算力不够,然后把有限的算力全部花在最近的画面上。这个LatestSlot小结构,我后来从摄像头取流一路用到了RTSP拉流、夜间低照度检测、甚至和mpp编码器对接的中间缓冲上,原理全都一样。后边我打算把阶段一和阶段二的方案合并成可配置的通用双路框架,再补上关键帧保序的增强版。如果你也正在香橙派或者RK3588上折腾双路视觉,希望这篇能帮你少走我走过的弯路。