利用有序数据流中的数据并行性

摘要

许多计算密集型应用涉及将有序输入数据复杂地转换为有序输出数据。例如声音和视频转码、无损数据压缩和地震数据处理。虽然这些转换中使用的算法通常是并行的，但管理I/O顺序依赖性可能是一个挑战。本文识别了其中一些挑战，并说明了在保持并行性能的同时解决这些挑战的策略。

本文是大型系列“Intel 多线程应用程序开发指南”的一部分，该系列为 Intel® 平台上的高效多线程应用程序开发提供了指导方针。

背景

考虑一个视频压缩引擎的线程化问题，该引擎旨在对来自实时视频源的未压缩视频进行实时处理，并将其写入磁盘或网络客户端。显然，利用多处理器的强大功能是满足此类应用实时要求的关键。

MPEG2和MPEG4等视频压缩标准是为通过不可靠链路进行流传输而设计的。因此，很容易将单个视频流视为一系列更小的独立流。通过并行处理这些更小的流，可以实现显著的加速。利用多线程实现这种并行性的一些挑战包括：

• 定义不重叠的问题子集并将其分配给线程
• 确保输入数据只读取一次且按正确顺序读取
• 以正确顺序输出块，无论实际处理完成的顺序如何，且无显著性能损失
• 在事先不知道输入数据实际范围的情况下执行上述操作

在其他情况下，例如无损数据压缩，通常可以预先确定输入数据大小，并将数据明确划分为独立的输入块。本文概述的技术同样适用于这种情况。

通知

人们可能会尝试建立一个生产者和消费者链，但这种方法不可伸缩，并且容易出现负载不平衡。相反，本文通过数据分解解决了上述每个挑战，以实现更具可伸缩性的设计。

这里采用的方法是创建一个线程组，每个线程读取一个视频块，对其进行编码，并将其输出到重排缓冲区。每个块完成后，线程返回读取并处理下一个视频块，依此类推。这种动态的工作分配最大限度地减少了负载不平衡。重排缓冲区确保编码视频块按正确顺序写入，无论其完成顺序如何。

原始视频编码算法可能采用这种形式

inFile = OpenFile ()
outFile == InitializeOutputFile ()
WriteHeader (outFile)
outputBuffer = AllocateBuffer (bufferSize)
while (frame = ReadNextFrame (inFile))
{
  EncodeFrame (frame, outputBuffer)
  if (outputBuffer size > bufferThreshold)
    FlushBuffer(outputBuffer, outFile)
}
FlushBuffer (outputBuffer, outFile)

首要任务是用基于块的算法替换读取和编码帧序列，为跨线程组分解问题做好准备

WriteHeader (outFile)
while (block = ReadNextBlock (inFile))
{
  while(frame = ReadNextFrame (block))
  {
    EncodeFrame (frame, outputBuffer)
    if (outputBuffer size > bufferThreshold)
      FlushBuffer (outputBuffer, outFile)
  }
  FlushBuffer (outputBuffer, outFile)
}

数据块的定义因应用而异，但在视频流的情况下，自然的块边界可能是输入中检测到场景变化的第一个帧，但受最小和最大块大小的约束。基于块的处理需要分配一个输入缓冲区，并对源代码进行少量更改以在处理前填充缓冲区。同样，readNextFrame 方法必须更改为从缓冲区而不是文件读取。

下一步是更改输出缓冲策略，以确保整个块作为一个单元写入。这种方法大大简化了输出重排，因为只需确保块按正确顺序输出即可。以下代码反映了基于块的输出更改

WriteHeader (outFile)
while (block = ReadNextBlock (inFile))
{
  while (frame = ReadNextFrame (block))
  {
    EncodeFrame (frame, outputBuffer)
  }
  FlushBuffer (outputBuffer, outFile)
}

根据最大块大小，可能需要更大的输出缓冲区。

由于每个块都独立于其他块，因此每个输出块通常以特殊标头开头。在MPEG视频流的情况下，此标头位于完整帧（称为I帧）之前，后续帧相对于I帧定义。因此，标头移到块循环内部

while (block = ReadNextBlock (inFile))
{
  WriteHeader (outputBuffer)
  while (frame = ReadNextFrame (block))
  {
    EncodeFrame (frame, outputBuffer)
  }
  FlushBuffer (outputBuffer, outFile)
}

通过这些更改，可以使用线程库（即Pthreads或Win32线程API）或OpenMP引入并行性。

// Create a team of threads with private
// copies of outputBuffer, block, and frame
/ and shared copies of inFile and outFile
while (AcquireLock,
       block = ReadNextBlock (inFile),
       ReleaseLock, block)
{
  WriteHeader (outputBuffer)
  while (frame = ReadNextFrame (block))
  {
    EncodeFrame (frame, outputBuffer)
  }
  FlushBuffer (outputBuffer, outFile)
}

这是一种简单但有效的数据安全且按顺序读取策略。每个线程获取一个锁，读取一个数据块，然后释放锁。共享输入文件确保数据块按顺序且只读取一次。由于就绪线程总是获取锁，因此块以动态的或先到先得的方式分配给线程，这通常可以最大限度地减少负载不平衡。

最后一步是确保块安全且按正确顺序输出。一个简单的策略是使用锁和共享输出文件来确保一次只写入一个块。这种方法确保了线程安全，但会允许块以非原始顺序输出。或者，线程可以等到所有先前的块都已写入后才刷新其输出。不幸的是，这种方法会引入效率低下，因为线程会空闲等待其写入轮次。

一种更好的方法是为输出块建立一个循环重排缓冲区。每个块都被分配一个连续的序列号。缓冲区的“尾部”确定要写入的下一个块。如果线程完成处理的块不是尾部指向的块，它只需将其块排入适当的缓冲区位置，然后返回读取并处理下一个可用块。同样，如果线程发现其刚刚完成的块是尾部指向的块，它将写入该块以及以前排队的所有连续块。最后，它更新缓冲区的尾部以指向要输出的下一个块。重排缓冲区允许已完成的块乱序排队，同时确保它们按顺序写入。

图1说明了重排缓冲区的一种可能状态。块0到35已处理并写入，而块37、38、39、40和42已处理并排队等待写入。当处理块36的线程完成时，它会写入块36到40，使重排缓冲区处于图2所示的状态。块42将一直排队，直到块41完成。

当然，需要采取某些预防措施以确保算法健壮和快速

• 共享数据结构在读取或写入时必须加锁。
• 缓冲区中的槽数必须超过线程数。
• 如果缓冲区中没有合适的槽位，线程必须有效地等待。
• 为每个线程预分配多个输出缓冲区。这允许将指针排入缓冲区，并避免了不必要的数据复制和内存分配。

使用输出队列，最终算法如下

inFile = OpenFile ()
outFile == InitializeOutputFile ()
// Create a team of threads with private 
// copies of outputBuffer, block, and frame, shared
// copies of inFile and outFile.
while (AcquireLock,
       block = ReadNextBlock (inFile),
       ReleaseLock, block)
{
  WriteHeader (outputBuffer)
  while (frame = ReadNextFrame (block))
  {
    EncodeFrame (frame, outputBuffer)
  }
  QueueOrFlush (outputBuffer, outFile)
}

此算法允许按序I/O，但仍提供高性能、乱序并行处理的灵活性。

使用指南

在某些情况下，读取和写入数据的时间与处理数据所需的时间相当。在这些情况下，以下技术可能有所帮助

• Linux*和Windows*提供了API来启动读取或写入，并在稍后等待或被通知其完成。在执行其他计算的同时使用这些接口“预取”输入数据和“后写”输出数据可以有效地隐藏I/O延迟。在Windows上，通过提供FILE_FLAG_OVERLAPPED属性打开文件以进行异步I/O。在Linux上，异步操作通过libaio提供的一些aio_*函数实现。
• 当输入数据量很大时，静态分解技术会导致物理磁盘“抖动”，因为硬件试图处理大量并发但不连续的读取。遵循上述共享文件描述符和动态、先到先得调度算法的建议可以强制执行按序、连续读取，这反过来又提高了整个I/O子系统的吞吐量。

仔细选择数据块的大小和数量非常重要。通常，大量的块提供了最大的调度灵活性，这可以减少负载不平衡。另一方面，非常小的块可能会引入不必要的锁开销，甚至阻碍数据压缩算法的有效性。

额外资源

• Intel® 软件网络并行编程社区
• OpenMP* 规范