视频图像处理中的错帧同步是怎么实现的？

1 什么是错帧同步？

一般 Android 系统相机的最高帧率在 30 FPS 左右，当帧率低于 20 FPS 时，用户可以明显感觉到相机画面卡顿和延迟。

我们在做相机预览和视频流处理时，对每帧图像处理时间过长（超过 30 ms）就很容易造成画面卡顿，这个场景就需要用到错帧同步方法去提升画面的流畅度。

错帧同步，简单来说就是把当前的几帧缓冲到子线程中处理，主线程直接返回子线程之前的处理结果，属于典型的以空间换时间策略。

错帧同步策略也有不足之处，它不能在子线程中缓冲太多的帧，否则造成画面延迟。

另外，每个子线程分配的任务也要均衡（即每帧在子线程中的处理时间大致相同），不然会因为 CPU 线程调度的时间消耗适得其反。

错帧同步的原理如上图所示，我们开启三个线程：一个主线程，两个工作线程，每一帧图像的处理任务分为 2 步，第一个工作线程完成第一步处理，第二个工作线程完成第二步处理，每一帧都要经过这两步的处理。

当主线程输入第 n + 1 帧到第一个工作线程后，主线程会等待第二个工作线程中第 n 帧的处理结果然后返回，这种情况下你肯定会问第 0 帧怎么办？第 0 帧就直接返回就行了。

这些步骤下来，可以看成第 n+1 帧和第 n 帧在 2 个工作线程中同时处理，若忽略 CPU 线程调度时间，2 线程错帧可以提升一倍的性能（性能提升情况，下面会给出实测数据）。

2 错帧同步的简单实现

错帧同步在实现上类似于“生产者-消费者”模式，我们借助于 C 语言信号量 #include <semaphore.h> 可以很方便的实现错帧同步模型。

C 的信号量常用的几个 API ：

int sem_init(sem_t *sem, int pshared, unsigned int value);    功能：初始化信号量
    参数：
        sem：指定要初始化的信号量
        pshared：0：应用于多线程
                非 0：多进程
        value：指定了信号量的初始值
    返回值：0 成功
          -1 失败

int sem_destroy(sem_t *sem);    功能：销毁信号量
    参数：sem：指定要销毁的信号量
    返回值：0 成功
          -1 错误 

int sem_post(sem_t *sem);    功能：信号量的值加 1 操作
    参数：
        sem：指定的信号量，就是这个信号量加 1
    返回值：0 成功
          -1 错误 

int sem_wait(sem_t *sem);    功能：信号量的值减 1 , 如果信号量的值为 0 , 阻塞等待
    参数：
        sem：指定的信号量, 如果信号量的值为 0, 阻塞等待, 否则信号量的值减 1
    返回值：0 成功
          -1 错误

在这里为了简化代码逻辑，我们用字符串来表示视频帧，每个工作线程对输入的字符串进行标记，表示工作线程对视频帧做了处理，最后的输出（第 0 帧除外）都是经过工作线程标记过的字符串。

//初始化
void AsyncFramework::Init() {
    LOGCATE("AsyncFramework::Init");
    memset(work_buffers, 0, sizeof(work_buffers));
    work_thread_running = true;
    main_thread_running = true;

    index = 0;

    // 初始化 3 个信号量
    sem_init(&main_sem, 0, 0);
    sem_init(&first_thread_sem, 0, 0);
    sem_init(&second_thread_sem, 0, 0);

    // WORK_THREAD_NUM = 2 ，为 2 个工作线程申请 2 块 buffer
    for (int i = 0; i < WORK_THREAD_NUM; ++i) {
        work_buffers[i] = static_cast<char *>(malloc(WORK_BUFFER_SIZE));
    }

    // 启动三个线程
    main_thread = new thread(MainThreadProcess);
    first_thread = new thread(FirstStepProcess);
    second_thread = new thread(SecondStepProcess);

}

// 反初始化
void AsyncFramework::UnInit() {
    LOGCATE("AsyncFramework::UnInit");
    //等待三个线程结束
    main_thread_running = false;
    main_thread->join();
    delete main_thread;
    main_thread = nullptr;

    work_thread_running = false;
    sem_post(&first_thread_sem);
    sem_post(&second_thread_sem);
    first_thread->join();
    second_thread->join();

    delete first_thread;
    first_thread = nullptr;
    delete second_thread;
    second_thread = nullptr;

    //销毁信号量
    sem_destroy(&main_sem);
    sem_destroy(&first_thread_sem);
    sem_destroy(&second_thread_sem);

    //释放缓冲区
    for (int i = 0; i < WORK_THREAD_NUM; ++i) {
        if (work_buffers[i]) {
            free(work_buffers[i]);
            work_buffers[i] = nullptr;
        }
    }

}

主线程的逻辑就是不断地生成“视频帧”，将“视频帧”传给第一个工作线程进行第一步处理，然后等待第二个工作线程的处理结果。

void AsyncFramework::MainThreadProcess() {
    LOGCATE("AsyncFramework::MainThreadProcess start");
    while (main_thread_running) {
        memset(work_buffers[index % WORK_THREAD_NUM], 0, WORK_BUFFER_SIZE);
        sprintf(work_buffers[index % WORK_THREAD_NUM], "FrameIndex=%d ", index);
        //通知第一个工作线程处理
        sem_post(&first_thread_sem);
        if (index == 0) {
            //第 0 帧直接返回，不交给工作线程处理
            LOGCATE("AsyncFramework::MainThreadProcess %s", work_buffers[index % WORK_THREAD_NUM]);
            index++;
            continue;
        } else {
            //等待第二个工作线程的处理结果 
            sem_wait(&main_sem);
        }
        LOGCATE("AsyncFramework::MainThreadProcess %s", work_buffers[(index - 1) % WORK_THREAD_NUM]);
        index++;
        if (index == 100) break;//生成100帧
    }
    LOGCATE("AsyncFramework::MainThreadProcess end");

}

2 个工作线程的处理逻辑类似，第一个工作线程收到主线程发来的信号，然后进行第一步处理，处理完成后通知第二个工作线程进行第二步处理，等到第二步处理完成后再通知主线程结束等待，取出处理结果。

void AsyncFramework::FirstStepProcess() {
    LOGCATE("AsyncFramework::FirstStepProcess start");
    int index = 0;
    while (true) {
        //等待主线程发来的信号
        sem_wait(&first_thread_sem);
        if(!work_thread_running) break;
        LOGCATE("AsyncFramework::FirstStepProcess index=%d", index);
        strcat(work_buffers[index % WORK_THREAD_NUM], "FirstStep ");
        //休眠模拟处理耗时
        this_thread::sleep_for(chrono::milliseconds(200));
        //处理完成后通知第二个工作线程进行第二步处理
        sem_post(&second_thread_sem);
        index++;
    }
    LOGCATE("AsyncFramework::FirstStepProcess end");

}

void AsyncFramework::SecondStepProcess() {
    LOGCATE("AsyncFramework::SecondStepProcess start");
    int index = 0;
    while (true) {
        //等待第一个工作线程发来的信号
        sem_wait(&second_thread_sem);
        if(!work_thread_running) break;
        LOGCATE("AsyncFramework::SecondStepProces index=%d", index);
        strcat(work_buffers[index % WORK_THREAD_NUM], "SecondStep");
        //休眠模拟处理耗时
        this_thread::sleep_for(chrono::milliseconds(200));
        //第二步处理完成后通知主线程结束等待
        sem_post(&main_sem);
        index++;
    }
    LOGCATE("AsyncFramework::SecondStepProcess end");
}