1. 双路视觉方案的整体设计思路
1.1 为什么要在RK3588上做双路视觉
香橙派5搭载的RK3588芯片,CPU部分为4核A76加4核A55的big.LITTLE架构,GPU为Mali-G610 MP4,NPU算力标称6TOPS。这个配置放在边缘计算场景里,单路yolov5s推理跑个十几二十帧问题不大,但一旦要同时处理两路摄像头输入,事情就变得复杂了。
我最初的想法很朴素:两路摄像头各自开一个线程,每个线程里做取流、解码、预处理、推理、后处理、画框、显示这一整套流程。结果实测下来,两路同时跑的时候帧率直接腰斩,NPU利用率忽高忽低,CPU的A76核心倒是跑满了,但整体吞吐量上不去。问题出在资源竞争上——两个线程同时抢NPU、抢内存带宽、抢CPU时间片,谁也没法稳定输出。
后来我把方案改成“两路各一个线程池”的结构,核心思路是:每一路视觉流水线拥有自己独立的线程池,池内线程各司其职,取流归取流、推理归推理、后处理归后处理,通过队列串联。这样做的直接好处是,两路之间互不干扰,单路内部的各个阶段也能并行起来,不会因为某一帧的后处理卡住而阻塞下一帧的采集。
这个方案不是拍脑袋想出来的。RK3588的NPU支持多核调度,但同一时刻一个NPU核心只能处理一个推理任务。如果两路共用一个线程池,任务在队列里排队,先到的先算,后到的等着,延迟会随着负载波动而剧烈变化。而每路独立线程池之后,我可以给每路分配固定的NPU核心绑定策略,让两路推理真正并行起来,而不是伪并行。
1.2 线程池方案对比:共用池 vs 独立池
在确定“两路各一个线程池”之前,我试过几种方案,这里把对比结果列出来,方便你根据自己场景选型。
| 方案 | 结构 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 单线程串行 | 一路一个线程,全流程串行 | 实现简单,无锁竞争 | 帧率低,各阶段互相阻塞 | 单路低帧率场景 |
| 共用线程池 | 两路任务提交到同一个池 | 资源利用率高,线程复用 | 任务排队延迟大,两路互相影响 | 两路负载差异大且不要求实时 |
| 独立线程池 | 每路一个池,池内多线程 | 隔离性好,延迟稳定,可绑核 | 线程数多,内存占用略高 | 双路实时视觉,推荐 |
| 独立线程池+流水线 | 每路池内再分阶段流水线 | 吞吐量最高,延迟最低 | 实现复杂度最高 | 高帧率双路推理 |
我最终选的是第四种,每路一个线程池,池内再按流水线阶段划分任务。取流线程只管从MIPI摄像头或USB摄像头拿数据,拿到就丢进原始帧队列;预处理线程从队列取帧,做resize、归一化、颜色空间转换,处理完丢进推理队列;推理线程调用RKNN的API执行模型,输出丢进后处理队列;后处理线程做NMS、画框、坐标映射,最后交给显示线程。每个阶段之间用阻塞队列连接,队列长度设上限,防止内存无限增长。
1.3 线程池的核心参数怎么定
线程池不是随便开几个线程就完事,参数定不好,要么资源浪费,要么频繁阻塞。我主要关注三个参数:核心线程数、最大线程数、阻塞队列容量。
核心线程数方面,取流阶段1个线程足够,因为摄像头出帧是固定节奏的,多开线程反而增加调度开销。预处理阶段我给了2个线程,因为resize和归一化是纯CPU计算,RK3588的A76核心跑这个很快,两个线程可以交替处理,避免单线程成为瓶颈。推理阶段1个线程,因为NPU同一时刻只能执行一个推理任务,多开线程只会让任务在NPU驱动层排队,不如在应用层控制。后处理阶段2个线程,NMS计算量不大但画框和坐标变换有点耗时,两个线程可以并行处理前后两帧。
阻塞队列容量我设的是4。这个数字是试出来的。设1的话,取流线程刚放一帧就被取走,流水线退化成串行;设8的话,内存里积压的帧太多,延迟肉眼可见地增加,而且RK3588的内存带宽有限,队列太长反而拖慢整体。4帧的缓冲,既能吸收短时抖动,又不会让延迟失控。
注意:队列容量一定要设上限,用有界队列。无界队列在负载突增时会吃光内存,最后OOM被系统杀掉,排查起来很痛苦。
2. 核心细节解析与实操要点
2.1 线程池的阻塞队列选择
阻塞队列的选择直接影响到线程池的行为。C++标准库里有几种队列,但都不是现成的线程安全阻塞队列,需要自己封装。我用的是std::deque加std::mutex加两个条件变量(非空、非满)的实现方式。
为什么不用std::queue?因为std::deque支持两端操作,虽然我这里只用到了尾部插入和头部弹出,但std::deque的内存分配策略比std::queue默认底层容器更适合频繁增删的场景。实际上std::queue默认就是基于std::deque的,所以性能差异不大,我选std::deque主要是为了代码直观。
条件变量的使用有个细节:等待非空的时候,要用while循环判断队列是否为空,而不是if。因为条件变量可能被虚假唤醒,用if的话醒来后发现队列还是空的,继续往下走就会取到空数据。这个坑我踩过,现象是程序偶尔崩溃,查了半天才发现是虚假唤醒导致的。
template<typename T> class BlockingQueue { public: explicit BlockingQueue(size_t capacity) : capacity_(capacity) {} void push(const T& item) { std::unique_lock<std::mutex> lock(mtx_); not_full_.wait(lock, [this] { return queue_.size() < capacity_; }); queue_.push_back(item); not_empty_.notify_one(); } bool pop(T& item) { std::unique_lock<std::mutex> lock(mtx_); not_empty_.wait(lock, [this] { return !queue_.empty() || stop_; }); if (queue_.empty()) return false; item = queue_.front(); queue_.pop_front(); not_full_.notify_one(); return true; } void stop() { std::unique_lock<std::mutex> lock(mtx_); stop_ = true; not_empty_.notify_all(); not_full_.notify_all(); } private: std::deque<T> queue_; size_t capacity_; std::mutex mtx_; std::condition_variable not_empty_; std::condition_variable not_full_; bool stop_ = false; };这个实现里,stop_标志位很重要。程序退出时,如果取流线程还在等队列非空,而生产者已经停了,线程会永远挂起。调用stop()之后,所有等待的线程都会被唤醒,然后检查stop_标志退出循环。
2.2 每路线程池的线程绑定策略
RK3588有8个CPU核心,0-3是A55小核,4-7是A76大核。默认情况下,Linux调度器会把线程在核心之间来回迁移,缓存命中率低,延迟不稳定。对于视觉流水线这种对延迟敏感的任务,我建议做CPU亲和性绑定。
取流线程绑到A55小核,因为取流主要是IO等待,计算量小,放小核省电且不占大核资源。预处理和后处理线程绑到A76大核,这两个阶段是纯计算,大核跑得快。推理线程不绑核,因为NPU是独立硬件,CPU只负责提交任务和等待结果,绑不绑影响不大。
绑核用pthread_setaffinity_np实现:
#include <pthread.h> #include <sched.h> void bind_to_cpu(std::thread& th, int cpu_id) { cpu_set_t cpuset; CPU_ZERO(&cpuset); CPU_SET(cpu_id, &cpuset); int rc = pthread_setaffinity_np(th.native_handle(), sizeof(cpu_set_t), &cpuset); if (rc != 0) { // 处理错误 } }两路视觉的话,第一路的预处理绑到CPU4和CPU5,第二路绑到CPU6和CPU7,这样两路的计算任务不会抢同一个大核。实测下来,绑核之后帧率波动从±5帧降到±1帧,效果很明显。
实操心得:绑核不是万能的。如果你的系统里还有其他任务在跑,绑太死可能导致某个核心过载而其他核心空闲。建议先用
htop观察一段时间,确认没有其他重负载任务再绑。
2.3 两路之间的资源隔离
两路独立线程池最大的好处是隔离,但隔离不等于完全隔离。NPU是共享资源,两路推理都要用。RK3588的NPU驱动支持多进程和多线程访问,但同一时刻只有一个推理任务在执行。如果两路同时提交推理请求,驱动层会串行化处理。
我的做法是在应用层做推理调度。每路的推理线程在提交任务前,先获取一个全局的NPU互斥锁,拿到锁的提交推理,没拿到的等待。这样虽然推理还是串行的,但至少保证了公平性,不会出现一路饿死另一路的情况。
内存带宽也是共享的。两路同时做预处理和后处理时,内存访问量翻倍,如果带宽吃满,NPU取权重和特征图的速度也会变慢。我的经验是,把预处理里的resize操作尽量用硬件加速,比如用RGA(Rockchip Graphics Accelerator)做缩放和格式转换,比CPU软算省带宽。
3. 实操过程与核心环节实现
3.1 环境准备与依赖安装
香橙派5的官方系统是Ubuntu 20.04或22.04,内核版本5.10。RKNN Toolkit2的版本要和NPU驱动匹配,我用的组合是rknn_toolkit2 1.5.2加NPU驱动0.9.2。版本不匹配的话,推理时会报错或者结果不对。
安装依赖:
sudo apt update sudo apt install -y cmake g++ libopencv-dev python3-opencv pip3 install rknn-toolkit2==1.5.2OpenCV用系统自带的就行,版本4.5以上都支持。如果要自己编译OpenCV,记得开启NEON和VFPV3优化,RK3588的A76支持这些指令集,开启后图像处理速度能快不少。
摄像头方面,我用的是MIPI摄像头加USB摄像头各一个。MIPI摄像头在香橙派5上需要配置设备树,官方Wiki有详细步骤。USB摄像头即插即用,但要注意带宽,两路USB摄像头同时跑1080p可能会抢USB总线带宽,建议一路MIPI一路USB,或者两路都走MIPI。
3.2 模型转换与RKNN初始化
yolov5s的PyTorch模型不能直接在NPU上跑,需要用RKNN Toolkit2转换成rknn格式。转换脚本网上很多,我说几个关键点。
首先是输入尺寸。yolov5s默认640x640,但RK3588的NPU对输入尺寸有对齐要求,宽高最好是16的倍数。640是16的倍数,没问题。如果改成416或者320,也要确保是16的倍数。
其次是量化。RKNN支持fp16和int8两种精度。fp16精度高但速度慢,int8速度快但精度有损失。双路视觉场景下,我建议用int8,因为两路同时跑,算力紧张,int8能省一半的NPU时间。量化需要准备校准集,我从COCO数据集里随机抽了200张图做校准,量化后的mAP下降在1%以内,可以接受。
转换命令:
python3 convert.py --weights yolov5s.pt --img-size 640 640 --batch-size 1 --rknn-version 1.5.2转换完成后,在香橙派上初始化RKNN:
rknn_context ctx; int ret = rknn_init(&ctx, model_path, 0, 0, nullptr); if (ret < 0) { // 初始化失败 } rknn_set_core_mask(ctx, RKNN_NPU_CORE_AUTO);RKNN_NPU_CORE_AUTO让驱动自动选择NPU核心。RK3588有3个NPU核心,可以手动指定用哪个,但自动模式在双路场景下表现更均衡。
3.3 双路线程池的完整代码结构
整个程序的入口在main函数里创建两个VisionPipeline对象,每个对象内部有自己的线程池和队列。主线程只负责等待信号,不做具体计算。
class VisionPipeline { public: VisionPipeline(int cam_id, int npu_core) : cam_id_(cam_id), npu_core_(npu_core), raw_queue_(4), pre_queue_(4), infer_queue_(4), post_queue_(4) {} void start() { capture_thread_ = std::thread(&VisionPipeline::capture_loop, this); pre_threads_.emplace_back(&VisionPipeline::preprocess_loop, this); pre_threads_.emplace_back(&VisionPipeline::preprocess_loop, this); infer_thread_ = std::thread(&VisionPipeline::infer_loop, this); post_threads_.emplace_back(&VisionPipeline::postprocess_loop, this); post_threads_.emplace_back(&VisionPipeline::postprocess_loop, this); display_thread_ = std::thread(&VisionPipeline::display_loop, this); } void stop() { raw_queue_.stop(); pre_queue_.stop(); infer_queue_.stop(); post_queue_.stop(); // join所有线程 } private: void capture_loop(); void preprocess_loop(); void infer_loop(); void postprocess_loop(); void display_loop(); int cam_id_; int npu_core_; BlockingQueue<cv::Mat> raw_queue_; BlockingQueue<cv::Mat> pre_queue_; BlockingQueue<cv::Mat> infer_queue_; BlockingQueue<DetectionResult> post_queue_; std::thread capture_thread_; std::vector<std::thread> pre_threads_; std::thread infer_thread_; std::vector<std::thread> post_threads_; std::thread display_thread_; };每个loop函数的逻辑都很直接:从上游队列pop数据,处理,push到下游队列。以preprocess_loop为例:
void VisionPipeline::preprocess_loop() { cv::Mat frame; while (raw_queue_.pop(frame)) { cv::Mat resized; cv::resize(frame, resized, cv::Size(640, 640)); cv::Mat rgb; cv::cvtColor(resized, rgb, cv::COLOR_BGR2RGB); rgb.convertTo(rgb, CV_32FC3, 1.0 / 255.0); pre_queue_.push(rgb); } }这里有个优化点:cv::resize和cvtColor可以合并成一个操作,用cv::dnn::blobFromImage一步到位。但blobFromImage内部还是分步做的,性能差不多,我分开写是为了方便调试,哪一步出问题一眼就能看出来。
3.4 推理线程与NPU交互细节
推理线程从pre_queue_取预处理好的图像,转成RKNN需要的输入格式,提交推理,拿回输出。
void VisionPipeline::infer_loop() { cv::Mat input; while (pre_queue_.pop(input)) { rknn_input inputs[1]; inputs[0].index = 0; inputs[0].type = RKNN_TENSOR_FLOAT32; inputs[0].size = input.total() * input.elemSize(); inputs[0].fmt = RKNN_TENSOR_NHWC; inputs[0].buf = input.data; rknn_inputs_set(ctx_, 1, inputs); rknn_run(ctx_, nullptr); rknn_output outputs[3]; for (int i = 0; i < 3; i++) { outputs[i].want_float = 1; } rknn_outputs_get(ctx_, 3, outputs, nullptr); // 把outputs打包成DetectionResult,push到post_queue_ // ... rknn_outputs_release(ctx_, 3, outputs); } }这里要注意rknn_outputs_get和rknn_outputs_release必须成对出现,否则内存泄漏。我一开始忘了release,跑了几分钟内存就满了,系统开始杀进程。
另外,rknn_run是阻塞调用,会一直等到推理完成才返回。如果两路推理线程同时调用rknn_run,驱动层会串行化,后调用的那个会等待。这就是为什么我在应用层加了NPU互斥锁,让等待发生在应用层,方便做超时控制和日志记录。
4. 常见问题与排查技巧实录
4.1 帧率上不去怎么办
双路跑起来之后,如果帧率明显低于预期,按以下顺序排查。
先看单路帧率。把另一路停掉,只跑一路,看帧率是否正常。如果单路正常,双路不行,说明是资源竞争问题。如果单路也不行,说明是单路流水线本身有瓶颈。
单路瓶颈常见的有:取流线程阻塞在摄像头读取上,检查摄像头驱动是否正常,用v4l2-ctl测试;预处理太慢,检查resize是否用了硬件加速;推理太慢,检查模型是否量化,输入尺寸是否过大;后处理太慢,检查NMS的阈值是否设得太低导致候选框太多。
双路竞争问题,先看CPU占用。如果A76核心跑满,说明计算任务太重,考虑把部分任务挪到A55,或者降低预处理分辨率。如果NPU利用率不高但帧率低,说明任务在排队,检查NPU互斥锁的粒度是否太粗。
避坑技巧:用
perf top看热点函数,能快速定位到哪个阶段最耗时。我当初就是靠这个发现cv::cvtColor占了30%的CPU时间,后来换成RGA硬件转换,直接降到5%。
4.2 内存泄漏与队列积压
内存泄漏在长时间运行的视觉程序里很常见。除了前面说的rknn_outputs_release,还有几个地方容易漏。
OpenCV的cv::Mat是引用计数,一般不会漏,但如果手动调用了release()又继续使用,会出问题。我的习惯是尽量用RAII,让Mat自己管理生命周期。
队列积压的表现是延迟越来越大,画面越来越滞后。原因是生产速度大于消费速度。解决办法有两个:一是提高消费速度,比如增加后处理线程数;二是降低生产速度,比如取流时做丢帧,队列满的时候直接丢弃当前帧而不是阻塞等待。
我选的是丢帧策略。取流线程push的时候用try_push,如果队列满了就丢弃这一帧,继续取下一帧。这样虽然会丢帧,但延迟不会累积,画面始终是实时的。
bool try_push(const T& item) { std::unique_lock<std::mutex> lock(mtx_); if (queue_.size() >= capacity_) { return false; // 队列满,丢弃 } queue_.push_back(item); not_empty_.notify_one(); return true; }4.3 常见问题速查表
| 现象 | 可能原因 | 排查方法 | 解决措施 |
|---|---|---|---|
| 帧率低 | 单路瓶颈 | 停一路测单路 | 定位瓶颈阶段优化 |
| 帧率低 | 资源竞争 | 看CPU/NPU利用率 | 绑核、加锁、隔离 |
| 延迟大 | 队列积压 | 看队列长度 | 丢帧、增加消费者 |
| 内存涨 | 未释放输出 | 看内存曲线 | 补release调用 |
| 画面卡顿 | 取流阻塞 | v4l2-ctl测试 | 换摄像头或驱动 |
| 推理报错 | 版本不匹配 | 查驱动版本 | 对齐版本号 |
| 结果不对 | 量化损失 | 对比fp16 | 重新校准或换fp16 |
| 程序崩溃 | 虚假唤醒 | 看core dump | 条件变量用while |
4.4 双路视觉的扩展思路
这套双路线程池方案跑通之后,可以往几个方向扩展。
一是增加路数。RK3588的NPU算力有限,三路以上yolov5s可能跑不动,但可以降低分辨率或者换更小的模型,比如yolov5n。线程池的结构不用变,只是多创建几个VisionPipeline对象。
二是加入跟踪算法。检测加跟踪是视觉应用的常见组合,可以在后处理阶段后面再加一个跟踪线程池,用ByteTrack或者DeepSORT做目标跟踪。跟踪的计算量比检测小,对整体帧率影响不大。
三是接入RTSP流。现在处理的是本地摄像头,如果换成网络流,取流阶段要换成FFmpeg解码,线程池结构不变,只是capture_loop里的实现要改。
我个人在实际操作中的体会是,线程池方案的核心不在于代码多复杂,而在于对每个阶段的耗时和资源占用有清晰的认知。你得知道每一帧在哪个阶段花了多少时间,才能有针对性地优化。我习惯在代码里加时间戳日志,每处理完一帧就打印各阶段耗时,跑一段时间后统计平均值和最大值,瓶颈一目了然。这个习惯帮我省了很多瞎猜的时间。