使用原子锁定自由的单生产者多消费者数据结构

我最近有如下样例代码(实际代码要复杂得多)。看了Hans Boehm关于atomic的cppcon16演讲后，我有点担心我的代码是否能正常工作。

produce由单个生产者线程调用，consume由多个消费者线程调用。生产者只更新序列号为2,4,6,8，…的数据。，但设置为奇数序列数，如1,3,5,7，…在更新数据之前指示数据可能是脏的。消费者也尝试以相同的顺序(2,4,6，…)获取数据。

消费者在读取后再次检查序列数，以确保数据是正确的(在读取期间不会被生产者更新)。

我认为我的代码在x86_64(我的目标平台)上工作得很好，因为x86_64不会与其他商店重新排序商店，或加载商店或加载，但我怀疑这在其他平台上是错误的。

我是否正确，数据分配(在生产中)可以移动到上面的"存储(n-1)"，所以消费者读取损坏的数据，但t == t2仍然成功?

struct S 
{
    atomic<int64_t> seq;
    // data members of primitive type int, double etc    
    ...
};
S s;
void produce(int64_t n, ...) // ... for above data members
{
    s.seq.store(n-1, std::memory_order_release); // indicates it's working on data members
    // assign data members of s
    ...
    s.seq.store(n, std::memory_order_release); // complete updating
}
bool consume(int64_t n, ...) // ... for interested fields passed as reference
{
    auto t = s.load(std::memory_order_acquire);
    if (t == n)
    {
        // read fields
        ...
        auto t2 = s.load(std::memory_order_acquire);
        if (t == t2)
            return true;
    }        
    return false;
}

当目标是x86时，编译时的重新排序仍然会让您感到困扰，因为编译器会优化以保留程序在c++抽象机器上的行为，而不是任何更强大的依赖于体系结构的行为。因为我们想要避免memory_order_seq_cst，所以允许重新排序。

是的，你们的商店可以按照你们的建议重新订货。您的加载也可以使用t2加载重新排序，因为获取加载只是一个单向屏障。编译器完全优化掉t2检查是合法的。如果重新排序是可能的，则允许编译器决定总是发生的事情，并应用as-if规则来生成更高效的代码。(当前的编译器通常不这样做，但这是当前标准所明确允许的。请参阅关于此的讨论的结论，以及标准建议的链接。

防止重新排序的选项有:

使用释放和获取语义使所有数据成员的存储/加载原子化。(最后一个数据成员的acquire-load会阻止t2的加载被首先完成。)
使用屏障(又名栅栏)将所有非原子存储和非原子负载排序为一个组。
正如Jeff Preshing解释的那样，mo_release 栅栏与mo_release存储不同，它是我们需要的那种双向屏障。Std::atomic只是循环使用Std::mo_名称，而不是为栅栏指定不同的名称。
(顺便说一下，非原子存储/加载应该是真正的原子与mo_relaxed，因为它在技术上是未定义的行为，读取它们，而它们可能在被重写的过程中，即使你决定不看你读的东西。)

void produce(int64_t n, ...) // ... for above data members
{
    /*********** changed lines ************/
    std::atomic_signal_fence(std::memory_order_release);  // compiler-barrier to make sure the compiler does the seq store as late as possible (to give the reader more time with it valid).
    s.seq.store(n-1, std::memory_order_relaxed);          // changed from release
    std::atomic_thread_fence(std::memory_order_release);  // StoreStore barrier prevents reordering of the above store with any below stores.  (It's also a LoadStore barrier)
    /*********** end of changes ***********/
    // assign data members of s
    ...
    // release semantics prevent any preceding stores from being delayed past here
    s.seq.store(n, std::memory_order_release); // complete updating
}

bool consume(int64_t n, ...) // ... for interested fields passed as reference
{
    if (n == s.seq.load(std::memory_order_acquire))
    {
        // acquire semantics prevent any reordering with following loads
        // read fields
        ...
    /*********** changed lines ************/
        std::atomic_thread_fence(std::memory_order_acquire);  // LoadLoad barrier (and LoadStore)
        auto t2 = s.seq.load(std::memory_order_relaxed);    // relaxed: it's ordered by the fence and doesn't need anything extra
        // std::atomic_signal_fence(std::memory_order_acquire);  // compiler barrier: probably not useful on the load side.
    /*********** end of changes ***********/
        if (n == t2)
            return true;
    }
    return false;
}

注意额外的编译器屏障(signal_fence只影响编译时的重排序)，以确保编译器不会将来自一次迭代的第二个存储与来自下一次迭代的第一个存储合并，如果这是在循环中运行的话。或者更一般地说，确保使区域无效的存储尽可能晚地完成，以减少误报。(对于真正的编译器可能不是必需的，并且在调用此函数之间有大量代码。但是signal_fence永远不会编译成任何指令，这似乎是比将第一个store保留为mo_release更好的选择。在释放-存储和线程栅栏都编译成额外指令的体系结构中，宽松的存储避免了有两个单独的屏障指令。

我还担心第一个存储与前一个迭代的发布存储重新排序的可能性。但我认为这是不可能的，因为两家店都在同一个地址。(在编译时，可能标准允许敌对的编译器执行此操作，但任何理智的编译器如果认为其中一个存储可以通过另一个存储，则根本不会执行其中一个存储。)在弱顺序架构的运行时，我不确定相同地址的存储是否会无序地成为全局可见的。这在现实生活中应该不是问题，因为制作人可能不会被背靠背调用

BTW， 你正在使用的同步技术是一个Seqlock，但是只有一个写器。您只有序列部分，而没有锁部分来同步不同的写入器。在多写入器版本中，写入器将在读取/写入序列号和数据之前获取锁。(而不是将seq no作为函数参数，您可以从锁中读取它)。

c++标准讨论文件N4455(关于原子的编译器优化，请参阅我对num++是否为'int num'?的回答的后半部分)使用它作为示例。

他们对写入器中的数据项使用release-store，而不是StoreStore栅栏。(对于原子数据项，正如我提到的，这需要真正正确)。

void writer(T d1, T d2) {
  unsigned seq0 = seq.load(std::memory_order_relaxed);  // note that they read the current value because it's presumably a multiple-writers implementation.
  seq.store(seq0 + 1, std::memory_order_relaxed);
  data1.store(d1, std::memory_order_release);
  data2.store(d2, std::memory_order_release);
  seq.store(seq0 + 2, std::memory_order_release);
}

他们讨论了让读取器的第二次加载序列号可能与以后的操作重新排序，如果编译器这样做是有利可图的，并且在读取器中使用t2 = seq.fetch_add(0, std::memory_order_release)作为获得带有释放语义的加载的潜在方法。对于当前的编译器，我建议而不是;您可能会在x86上获得lock ed操作，而我上面建议的方法没有任何(或任何实际的屏障指令，因为只有全屏障seq_cst栅栏需要x86上的指令)。

相关内容

最新更新

热门标签：