每個原子的記憶體順序是否正確？

我有一個環形緩沖區，看起來像：

template<class T>
class RingBuffer {
public:
  bool Publish();
 
  bool Consume(T& value);

  bool IsEmpty(std::size_t head, std::size_t tail);

  bool IsFull(std::size_t head, std::size_t tail);

private:
  std::size_t Next(std::size_t slot);

  std::vector<T> buffer_;
  std::atomic<std::size_t> tail_{0};
  std::atomic<std::size_t> head_{0};
  static constexpr std::size_t kBufferSize{8};
};

此資料結構旨在與兩個執行緒一起使用：發布者執行緒和消費者執行緒。下面列出了沒有將記憶體命令傳遞給原子的兩個主要函式：

bool Publish(T value) {
    const size_t curr_head = head_.load(/* memory order */);
    const size_t curr_tail = tail_.load(/* memory_order */);

    if (IsFull(curr_head, curr_tail)) {
      return false;
    }

    buffer_[curr_tail] = std::move(value);
    tail_.store(Next(curr_tail) /*, memory order */);
    return true;
  }

bool Consume(T& value) {
    const size_t curr_head = head_.load(/* memory order */);
    const size_t curr_tail = tail_.load(/* memory order */);

    if (IsEmpty(curr_head, curr_tail)) {
      return false;
    }

    value = std::move(buffer_[curr_head]);
    head_.store(Next(curr_head) /*, memory order */);
    return true;
  }

我知道，至少，我必須tail_.store()在Publish()函式 withstd::memory_order::release和tail_.load()withstd::memory_order::acquire在Consume()函式中創建發生在write tobuffer_和 read之間的連接之前buffer_。另外，我可以傳遞std::memory_order::relaxed到tail_.load()inPublish()和 to head_.load()inConsume()因為同一個執行緒將看到最后一次寫入原子。現在的功能是這樣的：

 bool Publish(T value) {
     const size_t curr_head = head_.load(/* memory order */);
     const size_t curr_tail = tail_.load(std::memory_order::relaxed);
    
     if (IsFull(curr_head, curr_tail)) {
         return false;
     }
    
     buffer_[curr_tail] = std::move(value);
     tail_.store(Next(curr_tail), std::memory_order::release);
     return true;
 }

 bool Consume(T& value) {
     const size_t curr_head = head_.load(std::memory_order::relaxed);
     const size_t curr_tail = tail_.load(std::memory_order::acquire);
    
     if (IsEmpty(curr_head, curr_tail)) {
         return false;
     }
    
     value = std::move(buffer_[curr_head]);
     head_.store(Next(curr_head) /*, memory order */);
     return true;
 }

The last step is to put memory orders in the remaining pair: head_.load() in Publish() and head_.store() in Consume(). I have to have value = std::move(buffer_[curr_head]); line executed before head_.store() in Consume(), otherwise I will have data races in cases when the buffer is full, so, at least, I must pass std::memory_order::release to that store operation to avoid reorderings. But do I have to put std::memory_order::acquire in head_.load() in the Publish() function? I understand that it will help head_.load() to see head_.store() for reasonable time unlike std::memory_order::relaxed, but if I don't need this guarantee of shorter time to see a side effect of a store operation, can I have a relaxed memory order? If I can't, then why? Completed code:

bool Publish(T value) {
    const size_t curr_head = head_.load(std::memory_order::relaxed); // or acquire?
    const size_t curr_tail = tail_.load(std::memory_order::relaxed);
        
    if (IsFull(curr_head, curr_tail)) {
        return false;
    }
        
    buffer_[curr_tail] = std::move(value);
    tail_.store(Next(curr_tail), std::memory_order::release);
    return true;
}
    
bool Consume(T& value) {
    const size_t curr_head = head_.load(std::memory_order::relaxed);
    const size_t curr_tail = tail_.load(std::memory_order::acquire);
        
    if (IsEmpty(curr_head, curr_tail)) {
        return false;
    }
        
    value = std::move(buffer_[curr_head]);
    head_.store(Next(curr_head), std::memory_order::release);
    return true;
}

Are memory orders for each atomic correct? Am I right about explanation of use of each memory order in each atomic variable?

uj5u.com熱心網友回復：

以前的答案可能有助于作為背景：
c 、std::atomic、std::memory_order 是什么以及如何使用它們？
https://bartoszmilewski.com/2008/12/01/c-atomics-and-memory-ordering/

首先，您描述的系統稱為單一生產者 - 單一消費者佇列。您可以隨時查看此容器的 boost 版本進行比較。我經常會檢查 boost 代碼，即使我在不??允許 boost 的情況下作業。這是因為檢查和理解一個穩定的解決方案會讓你洞察你可能遇到的問題（他們為什么這樣做？哦，我明白了 - 等等）。鑒于您的設計，并且已經撰寫了許多類似的容器，我會說您的設計必須小心區分空的和滿的。如果您使用經典的 {begin,end} 對，則會遇到由于包裝而導致的問題

{開始，開始 大小} == {開始，開始} == 空

好的，回到同步問題。

鑒于該順序僅影響重新排序，在 Publish 中使用 release 似乎是該標志的教科書使用。在容器的大小增加之前，什么都不會讀取值，因此您不關心值本身的寫入順序是否以隨機順序發生，您只關心在增加計數之前必須完全寫入值. 所以我同意，您正確地使用了 Publish 函式中的標志。
我確實質疑消費中是否需要“釋放”，但是如果您要移出佇列，并且這些移動會產生副作用，則可能需要。我會說，如果您追求原始速度，那么可能值得制作第二個版本，該版本專門用于瑣碎的物件，它使用寬松的順序來增加頭部。

您也可以考慮在推送/彈出時就地新建/洗掉。雖然大多數移動會使物件處于空狀態，但標準僅要求移動后它仍處于有效狀態。移動后顯式洗掉物件可能會使您以后避免出現晦澀的錯誤。

您可能會爭辯說，消耗中的兩個原子負載可能是 memory_order_consume。這放寬了限制，說“我不在乎加載它們的順序，只要它們在使用時都已加載”。雖然我懷疑在實踐中它會產生任何收益。我也對這個建議感到緊張，因為當我查看 boost 版本時，它非常接近你所擁有的。 https://www.boost.org/doc/libs/1_66_0/boost/lockfree/spsc_queue.hpp

 template <typename Functor>
    bool consume_one(Functor & functor, T * buffer, size_t max_size)
    {
        const size_t write_index = write_index_.load(memory_order_acquire);
        const size_t read_index  = read_index_.load(memory_order_relaxed);
        if ( empty(write_index, read_index) )
            return false;

        T & object_to_consume = buffer[read_index];
        functor( object_to_consume );
        object_to_consume.~T();

        size_t next = next_index(read_index, max_size);
        read_index_.store(next, memory_order_release);
        return true;
    }

總的來說，盡管您的方法看起來不錯并且與 boost 版本非常相似。但是...如果我是你，我可能會逐行瀏覽 boost 版本，看看它有什么不同。很容易犯錯。

編輯：抱歉，我剛剛注意到您的代碼中 memory_order_acquire/memory_order_relaxed 標志的順序錯誤。您應該讓最后一次寫入“發布”，而第一次讀取“獲取”。這不會顯著影響行為，但它使閱讀更容易。（開始同步，結束同步）

回復評論：正如@user17732522暗示的，copy操作也是寫，所以對trivial object的優化不會同步，對trivial object的優化會引入U/B（媽的很容易出錯）

標籤：c multithreading memory-barriers

上一篇：C 參考包裝器作為函式引數

下一篇：檔案I/O操作是否會釋放Python中的GIL？