6

我正在寻找符合这些要求的无锁设计：

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

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

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

4

2 回答 2

0

你在正确的轨道上。

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

如果有一个生产者和一个消费者，则固定大小的环形缓冲区可用于线程、进程或处理器之间的无锁通信。要执行的一些检查：

head变量仅由生产者写入（作为写入后的原子操作）

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

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

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

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

扩展队列

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

应用

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
   }
于 2010-02-25T16:00:51.697 回答
0

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

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 函数中传递出来，则可以将其缩减为一个副本。

于 2010-02-25T18:32:54.797 回答