无锁单生产者/单消费者循环缓冲区

Question

当我无法弄清楚为什么需要特定的内存屏障时，我一直在查看this website 上的无锁单一生产者/单一消费者循环缓冲区。 我已经仔细阅读了一百次the standard rules about memory order，但我不明白我错过了什么。

有了这个实现，只有一个可以调用push()函数的唯一线程和另一个可以调用pop()函数的唯一线程。

这里是Producer 代码：

bool push(const Element& item)
{       
  const auto current_tail = _tail.load(std::memory_order_relaxed);  //(1)
  const auto next_tail = increment(current_tail);

  if(next_tail != _head.load(std::memory_order_acquire))            //(2)               
  {     
    _array[current_tail] = item;                                    //(3)
    _tail.store(next_tail, std::memory_order_release);              //(4)
    return true;
  }
  return false; // full queue
}

这里是Consumer 代码：

bool pop(Element& item)
{
  const auto current_head = _head.load(std::memory_order_relaxed);    //(1)
  if(current_head == _tail.load(std::memory_order_acquire))           //(2)
    return false; // empty queue

  item = _array[current_head];                                       //(3)
  _head.store(increment(current_head), std::memory_order_release);   //(4)
  return true;
}

我明白为什么Producer (4) 和Consumer (2) 语句是绝对需要的，这是因为我们必须确保所有发生在(4) released storeProducer 之前的写入一旦consumer 看到存储的值，就会出现可见的副作用。

我也明白为什么需要Consumer (4) 语句，这是为了确保在执行Consumer (4) 存储之前执行Consumer (3) 加载。

问题

• 为什么Producer (2) 加载需要acquire semantic（而不是relaxed）执行？是为了防止Producer (3) or (4)在条件之前（编译时还是运行时）被重新排序？

Answer 1

我们需要证明

_array[current_tail] = item; // push(3)

执行在符合 (current_head == current_tail)

item = _array[current_head]; // pop(3)

已完成。我们可以覆盖单元格，只有在它的数据已经复制到项目之后

_head.load(std::memory_order_acquire) // push(2)

同步

_head.store(increment(current_head), std::memory_order_release);   //pop(4)

通过 Release-Acquire 排序：

在_head 上的原子存储释放 ( pop(4) ) 之前发生的所有内存写入 ( pop(3) ) 都成为可见的副作用一旦原子负载获取 ( push(2) ) 在 _head 上完成。

所以push(2)完成后的Producer代码，保证能看到pop(3)的结果。这意味着来自_array[current_head] 的数据被复制到项目中，并且此操作的结果在push(2) 之后对生产者代码可见，因此_array[current_head] 已经空闲。

从memory_order_acquire 加载描述的另一端 - 当前线程中的读取或写入 (push(3) ) 可以在此加载之前重新排序。所以 push(3) 将在 push(2) 加载完成后执行，但此时 pop(3) 已经完成

item = _array[current_head];                                        //pop(3)
_head.store(increment(current_head), std::memory_order_release);    //pop(4)
-----
    _head.load(std::memory_order_acquire);                          //push(2)
    _array[current_tail] = item;                                    //push(3)         

Answer 2

内存屏障阻止 CPU 对未使用互锁的 Element 对象的访问重新排序访问队列结构（此处使用索引实现，但指针同样可行）。

使用您的编号，必须在 (2) 和 (4) 之间执行 (3)，并且内存屏障提供了这一点。

您在 Producer 中询问的 (2)-vs-(3) 的确切情况可防止在队列已满时推测性地覆盖有效数据（建议的站点与有效数据重叠）。如果没有屏障，即使从生产者线程的角度来看，当条件失败时，原始数据也会被恢复，中间值可能对消费者来说是短暂可见的。