寻找无锁 RT-safe 单读单写结构

Question

我正在寻找符合以下要求的无锁设计：

• 单一写入器写入结构，单一读取器从该结构中读取（该结构已经存在并且可以安全地同时读取/写入）
• 但在某些时候，结构需要由编写器更改，然后它会初始化、切换并写入一个新结构（相同类型但具有新内容）
• 并且在下次阅读器读取时，它切换到这个新结构（如果写入器多次切换到一个新的无锁结构，阅读器丢弃这些结构，忽略它们的数据）。
• 必须重复使用这些结构，即在写入/读取/切换操作期间不允许堆内存分配/释放，以实现 RT 目的。

我目前已经实现了一个包含这些结构的多个实例的环形缓冲区；但是这个实现的缺点是，当编写者使用了环形缓冲区中存在的所有结构时，就没有更多的地方可以改变结构了……但是环形缓冲区的其余部分包含一些不必读取的数据由读者使用，但作者不能重复使用。因此，环形缓冲区不适合此目的。

任何想法（名称或伪实现）无锁设计？感谢您考虑这个问题。

Answer 1

你在正确的轨道上。

线程/进程/处理器之间固定消息的无锁通信

如果有一个生产者和一个消费者，

固定大小环形缓冲区可用于线程、进程或处理器之间的无锁通信。需要执行的一些检查：

head 变量只能由生产者写入（作为写入后的原子动作）

tail 变量仅由消费者写入（作为读取后的原子操作）

陷阱：引入大小变量或缓冲区满/空标志；这些通常由生产者和消费者共同编写，因此会给您带来问题。

我通常为此目的使用环形缓冲区。我学到的最重要的一课是环形缓冲区永远不能包含多个元素。这样一个 head 和 tail 变量分别由生产者和消费者编写。

大/可变尺寸块的扩展 要在实时环境中使用缓冲区，您可以使用内存池（通常在实时操作系统中以优化形式提供）或将分配与使用分离。我相信后者适合这个问题。

如果您需要交换大块，我建议使用带有缓冲区块的池，并使用队列传递指向缓冲区的指针。所以使用带有缓冲区指针的第三个队列。这样，分配可以在应用程序（后台）中完成，您的实时部分可以访问可变数量的内存。

应用程序

while (blockQueue.full != true)
{
    buf = allocate block of memory from heap or buffer pool
    msg = { .... , buf };
    blockQueue.Put(msg)
}

Producer:
   pBuf = blockQueue.Get()
   pQueue.Put()

Consumer
   if (pQueue.Empty == false)
   {
      msg=pQueue.Get()
      // use info in msg, with buf pointer
      // optionally indicate that buf is no longer used
   }

Answer 2

这是一个。关键是有三个缓冲区，阅读器保留它正在读取的缓冲区。写入器写入其他两个缓冲区之一。碰撞的风险很小。另外，这扩大了。只需使您的成员数组比读者数量加上作者数量长一个元素即可。

class RingBuffer
{
  RingBuffer():lastFullWrite(0)
  { 
    //Initialize the elements of dataBeingRead to false
    for(unsigned int i=0; i<DATA_COUNT; i++)
    {
      dataBeingRead[i] = false;
    } 
  }

  Data read()
  {
    // You may want to check to make sure write has been called once here
    // to prevent read from grabbing junk data. Else, initialize the elements
    // of dataArray to something valid.
    unsigned int indexToRead = lastFullWriteIndex;
    Data dataCopy;
    dataBeingRead[indexToRead] = true;
    dataCopy = dataArray[indexToRead];
    dataBeingRead[indexToRead] = false;
    return dataCopy;
  }

  void write( const Data& dataArg )
  {
    unsigned int writeIndex(0);

    //Search for an unused piece of data.
    // It's O(n), but plenty fast enough for small arrays.
    while( true == dataBeingRead[writeIndex] && writeIndex < DATA_COUNT )
    {
      writeIndex++;
    }  

    dataArray[writeIndex] = dataArg;

    lastFullWrite = &dataArray[writeIndex];
  }

private:
  static const unsigned int DATA_COUNT;
  unsigned int lastFullWrite;
  Data dataArray[DATA_COUNT];
  bool dataBeingRead[DATA_COUNT];
};

注意：这里写的方式，有两个副本来读取你的数据。如果您通过引用参数将数据从 read 函数中传递出去，则可以将其缩减为一份。