ios - 异步执行但同步写入

Question

我有这种情况:视频必须逐帧处理，但在处理帧时，输出必须按顺序写入文件。

我想使用 dispatch_async 将异步 block 触发到并发队列以加快处理速度，但因为这个队列是异步的，所以我不知道如何协调将帧串行写入输出。

假设这样一种情况:帧1、2、3、4、5被送到并发队列处理。因为任何 block 都可以随时完成，所以第 4 帧可能最先完成，然后是 5、3、1、2。那么我将如何设法按顺序将帧写入输出？

我有这样的代码:

dispatch_queue_t aQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);

while (true) {

    video >> frame;  // read a frame from the video

    dispatch_async(aQueue, ^{
         processVideo(frame, outputFrame);
         writeToVideo(outputFrame); // this is here just to show what needs to be done
    });

    // bla bla

}

有什么线索吗？

谢谢

Answer 1

我会结合使用串行调度队列和 NSCondition .串行队列确保没有任何写入同时发生，而 NSCondition 确保它们以正确的顺序发生。

来自 NSCondition 文档:

A condition object acts as both a lock and a checkpoint in a given thread. The lock protects your code while it tests the condition and performs the task triggered by the condition. The checkpoint behavior requires that the condition be true before the thread proceeds with its task. While the condition is not true, the thread blocks.

在你的具体情况下，我会做这样的事情......

在您的循环中，您首先声明一个 BOOL(初始设置为 NO)，它指示您的帧是否已被处理，以及一个 NSCondition。然后，dispatch_async 到后台队列处理帧，串行队列写入数据。

当串行队列中的 block 运行时，锁定NSCondition，然后检查BOOL以查看该帧是否已被处理。如果有，继续写入。如果没有，请等待 NSCondition 的信号，并在收到信号时再次检查。完成后，解锁 NSCondition。

当后台队列中的 block 运行时，锁定 NSCondition 并处理帧。当帧被处理时，设置BOOL表示帧被处理。然后发出信号并解锁NSCondition。

注意:重要的是您只访问指示帧已处理的 BOOL 和 outputFrame 在 NSCondition 的锁；锁确保它们在线程之间保持同步。

// Create the background and serial queues
dispatch_queue_t backgroundQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
dispatch_queue_t writeQueue = dispatch_queue_create("writeQueue", DISPATCH_QUEUE_SERIAL);

while (true) { // I'm assuming you have some way to break out of this...
    NSCondition *condition = [[NSCondition alloc] init];

    // These need the __block attribute so they can be changed inside the blocks
    __block BOOL frameProcessed = NO;
    __block FrameType outputFrame = nil;

    // video >> frame;  // read a frame from the video

    // dispatch the frame for processing
    dispatch_async(backgroundQueue, ^{
        [condition lock];

        processVideo(frame, outputFrame);
        frameProcessed = YES;

        [condition signal];
        [condition unlock];
    });

    // dispatch the write
    dispatch_async(writeQueue, ^{
        [condition lock];
        while (!frameProcessed) {
            [condition wait]; // this will block the current thread until it gets a signal
        }

        writeToVideo(outputFrame);

        [condition unlock];
    });
}

注意:在上面的代码中，BOOL frameProcessed 也有一个半微妙的技巧。由于它是在循环内部而不是外部声明的，因此每个 block 都将捕获与其帧关联的那个。

---

更新:同时添加一个 NSCondition 用于阅读。

Because writing to video is slow compared to the parallel execution, zillions of frames are allocated and sit in memory until they are saved to disk.

我会通过使用另一个 NSCondition 限制读取来处理这个问题，如果在您的 writeQueue 中有太多帧等待写入，它会阻止您的读取。这个概念与我们之前添加的 NSCondition 几乎相同，只是条件不同；在此转换中，它将是一个 int，指示有多少帧正在等待写入。

在循环之前，定义一个 readCondition、writeQueueSize 和 maxWriteQueueSize。在循环内部，首先锁定 readCondition，检查是否writeQueueSize >= maxWriteQueueSize。如果不是，则继续读取帧并排队处理和写入。在发送到 writeQueue 之前，增加 writeQueueSize。然后unlock readCondition.

然后，在分派(dispatch)到writeQueue 的 block 内，一旦写入完成，lock readCondition，递减writeQueueSize, signal and unlock readCondition.

这应该确保 writeQueue 中等待的 block 永远不会超过 maxWriteQueueSize。如果有那么多 block 在等待，它会有效地暂停从视频中读取帧，直到 writeQueue 准备好接收更多帧。

// Create the background and serial queues
dispatch_queue_t backgroundQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
dispatch_queue_t writeQueue = dispatch_queue_create("writeQueue", DISPATCH_QUEUE_SERIAL);

NSCondition *readCondition = [[NSCondition alloc] init];
__block int writeQueueSize = 0;
const int maxWriteQueueSize = 10;

while (true) { // I'm assuming you have some way to break out of this...
    NSCondition *writeCondition = [[NSCondition alloc] init];

    // These need the __block attribute so they can be changed inside the blocks
    __block BOOL frameProcessed = NO;
    __block FrameType outputFrame = nil;

    [readCondition lock];
    while (writeQueueSize >= maxWriteQueueSize) {
        [readCondition wait];
    }

    // video >> frame;  // read a frame from the video

    // dispatch the frame for processing
    dispatch_async(backgroundQueue, ^{
        [writeCondition lock];

        processVideo(frame, outputFrame);
        frameProcessed = YES;

        [writeCondition signal];
        [writeCondition unlock];
    });

    // dispatch the write
    writeQueueSize++; // Increment the write queue size here, before the actual dispatch
    dispatch_async(writeQueue, ^{
        [writeCondition lock];
        while (!frameProcessed) {
            [writeCondition wait]; // this will block the current thread until it gets a signal
        }

        writeToVideo(outputFrame);

        [writeCondition unlock];

        // Decrement the write queue size and signal the readCondition that it changed
        [readCondition lock];
        writeQueueSize--;
        [readCondition signal];
        [readCondition unlock];
    });

    [readCondition unlock];
}