使用多线程加速实时视频流处理的思路
2023-08-17 23:41:48 # cpp # 并行计算

再初次接触机器视觉的小伙伴应该都会遇到这种情况，明明摄像头输出的画面是 30 fps 的，但对每一帧图像进行畸变矫正或者是一些图像处理后，视频帧数降低到了不到 10 fps！还伴随出现丢帧或者是播放缓慢，内存占用越来越多。

最近想到了一种利用线程池去并行处理视频流的方法，该方法在处理耗时高于视频帧产生间隔时，能有效提高最终视频输出的帧率。本文会用伪代码来描述整个方法的框架，不提供实际的代码。

基本思路就是将读取、计算和消费三部分分离到不同的线程中，读取视频流的图像为单独一条线程，图像处理的计算在线程池中进行，最后使用结果的部分也是单独一条线程。

下面开始介绍完整的流程细节，先定义一些基本的接口和类型:

1
2
3
4
5
6
7
8
9
10
// 图像数据
type Image
{
    // 图像序号，用于排序结果
    size_t id
    bytes data
}

// 读取视频流到图像的接口
Image NextFrame()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
// 优先队列。作为输出队列，按序号排序
type PriorityQueue<Image>
{
    // 插入新图像
    void Push(Image image)

    // 判断队列是否有值
    bool Empty()

    // 获取第一个队头元素的引用
    Image& Top()

    // 出队队头元素
    Image Pop()
}

// 线程池
type ThreadPool
{
    // 添加后台任务
    void AddTask(Functor task)
}

// 互斥量
type Mutex
{
    void Lock()
    void Unlock()
}

// 条件变量
type ConditionVariable
{
    void Wait(Mutex &m, Predicate pred)
    void Notify()
}

首先就是读取图像线程的部分，该线程负责读取一帧视频图像，然后创建一个图像处理任务提交到线程池中，图像处理完成后，把结果塞到优先队列里。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// 先定义一些全局变量
// 线程池
ThreadPool pool
// 存放结果的优先队列
PriorityQueue<Image> result_buffer
// 线程间共享优先队列的保护锁
Mutex result_buffer_mutex
// 用来通知有新结果产生的条件变量
ConditionVariable notifier

// 读取新图像的工作线程
void Producer()
{
    while
    {
        input_image = NextFrame()
        pool.AddTask(func() {
            Image output_image
            // 进行图像处理操作
            SomeOperation(input_image, output_image)
            //将结果加入到优先队列中
            result_buffer_mutex.Lock()
            result_buffer.Push(output_image)
            result_buffer_mutex.Unlock()
            // 通知消费线程
            notifier.Notify()
        })
    }
}

消费结果的线程逻辑会稍微复杂一点，因为线程池中执行任务的顺序是不确定，我们需要用条件变量来控制输出顺序：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 记录当前应该输出的视频帧的编号
size_t need_output_id = 0;

// 读取视频帧处理结果的线程
void Consumer()
{
    while
    {
        // 使用条件变量等待结果产生的通知
        result_buffer_mutex.Lock()
        notifier.Wait(result_buffer_mutex, func() {
            // 结束等待的条件就是优先队列的顶部是我现在需要输出的序号
            return !result_buffer.Empty() && 
                   (result_buffer.Top().id == need_output_id);
        })
        output_image = result_buffer.Pop()
        result_buffer_mutex.Unlock()

        // 使用结果图像
        DoSomething(output_image);
    }
}

总结

整个框架的思路其实挺简单的，但实际实现的时候需要注意的细节会比较多，这里举例两个：

  1. 多线程的安全问题，用 C/C++ 实现的时候，要尽可能避免图像数据的拷贝，又要保证线程池任务执行和最后读取结果的时候图像数据是有效的；
  2. 最终使用结果图像的部分，由于是并行处理多帧图像，最终得到图像的频率是不稳定的，有可能一瞬间就读读到 N 幅图像，这时候就需要根据需求来修改代码了，例如我最终的图像是要实时显示出来，那就要加上合适的延迟再播放下一帧，不如画面就会出现卡顿一下下，然后瞬间播放好几帧的情况；

(●’◡’●)