multithreading - 无锁有界MPMC环形缓冲区故障

Question

我一直在无锁的多生产者多消费者环形缓冲区上猛烈抨击(我的尝试)。这个想法的基础是使用无符号字符型和无符号短类型的先天溢出，将元素缓冲区固定为这些类型中的任何一个，然后您可以自由循环回到环形缓冲区的开头。

问题是-我的解决方案不适用于多个生产者(尽管适用于N个消费者，也适用于单个生产者单个消费者)。

#include <atomic>

template<typename Element, typename Index = unsigned char> struct RingBuffer
{
  std::atomic<Index> readIndex;
  std::atomic<Index> writeIndex;
  std::atomic<Index> scratchIndex;
  Element elements[1 << (sizeof(Index) * 8)];

  RingBuffer() :
    readIndex(0),
    writeIndex(0),
    scratchIndex(0)
  {
    ;
  }

  bool push(const Element & element)
  {
    while(true)
    {
      const Index currentReadIndex = readIndex.load();
      Index currentWriteIndex = writeIndex.load();
      const Index nextWriteIndex = currentWriteIndex + 1;
      if(nextWriteIndex == currentReadIndex)
      {
        return false;
      }

      if(scratchIndex.compare_exchange_strong(
        currentWriteIndex, nextWriteIndex))
      {
        elements[currentWriteIndex] = element;
        writeIndex = nextWriteIndex;
        return true;
      }
    }
  }

  bool pop(Element & element)
  {
    Index currentReadIndex = readIndex.load();

    while(true)
    {
      const Index currentWriteIndex = writeIndex.load();
      const Index nextReadIndex = currentReadIndex + 1;
      if(currentReadIndex == currentWriteIndex)
      {
        return false;
      }

      element = elements[currentReadIndex];

      if(readIndex.compare_exchange_strong(
        currentReadIndex, nextReadIndex))
      {
        return true;
      }
    }
  }
};

编写的主要思想是使用临时索引'scratchIndex'，该索引充当伪锁，以便在更新writeIndex并允许任何其他生产者进行更新之前，仅一次允许一个生产者在任何时候将其复制构造到元素缓冲区中。 。在我被称为异教徒以暗示我的方法是“无锁的”之前，我意识到这种方法并不是完全无锁的，但是在实践中(如果可行的话)它比使用普通互斥锁要快得多!

我在 http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue处知道了一个(更复杂的)MPMC环形缓冲区解决方案，但是我确实在尝试我的想法，然后与该方法进行比较，找出每种方法的优势(或者实际上我的方法是否完全失败了!)。

我尝试过的东西；

使用compare_exchange_weak

使用更精确的std::memory_order匹配我想要的行为

我有

的各种索引之间添加缓存行填充

使元素std::atomic而不只是元素数组

我敢肯定，这归结为我的脑海中关于如何使用原子访问来使用互斥锁进行回避的基本段错误，对于那些指出哪些神经元在我的脑中急剧失灵的人，我将不胜感激! :)

Answer 1

这是A-B-A problem的一种形式。一个成功的制作人看起来像这样:

加载currentReadIndex

加载currentWriteIndex

cmpxchg商店scratchIndex = nextWriteIndex

商店element

商店writeIndex = nextWriteIndex

如果某个生产者由于某种原因在步骤2和3之间停顿了足够长的时间，则其他生产者有可能产生整个队列中的数据，然后回绕到完全相同的索引，从而使步骤3中的compare-exchange成功(因为scratchIndex恰好再次等于currentWriteIndex)。

就其本身而言，这不是问题。停滞不前的生产者完全有能力增加 scratchIndex来锁定队列，即使检测到ABA的神奇cmpxchg拒绝了商店，生产者也将再次尝试，重新加载完全相同的 currentWriteIndex并正常进行。

实际的问题是步骤2和步骤3之间的 nextWriteIndex == currentReadIndex检查。如果是 currentReadIndex == currentWriteIndex，则队列在逻辑上是空的，因此存在此检查是为了确保没有任何生产者超前以至于它覆盖了尚未弹出任何使用者的元素。在顶部进行一次此检查似乎是安全的，因为所有使用者都应“困在”所观察到的 currentReadIndex和所观察到的 currentWriteIndex之间。

除了其他生产者可以出现并破坏 writeIndex之外，这使消费者摆脱了陷阱。如果生产者在第2步和第3步之间停顿，当它唤醒时， readIndex的存储值可能是绝对任何值。

这是一个示例，从一个空队列开始，显示了发生的问题:

生产者运行步骤1和2。两个已加载的索引均为0。队列为空。

生产者B 中断并产生一个元素。

使用者弹出一个元素。两个索引均为1。

生产者B 可再生产255个元素。写索引回绕到0，读索引仍为1。

生产者从沉睡中醒来。它先前已将读取和写入索引都加载为0(空队列!)，因此它尝试执行步骤3。因为另一个生产者恰巧在索引0上暂停，所以比较交换成功，并且存储继续进行。完成时，生产者使用writeIndex = 1，现在两个存储的索引均为1，并且队列在逻辑上为空。现在，将完全忽略队列中所有元素的值(value)。

(我应该提一下，我无法谈论“停滞”和“唤醒”的唯一原因是所使用的所有原子都是顺序一致的，因此我可以假装我们处于单线程环境中。)

请注意，您使用 scratchIndex保护并发写入的方式本质上是一种锁定。成功完成cmpxchg的任何人都将获得对该队列的总写访问权限，直到释放该锁为止。解决此故障的最简单方法是仅用自旋锁替换 scratchIndex-它不会遭受A-B-A的折磨，而这实际上正在发生。