如何在 C++11 中表达普通存储（导出）和加载（导入）障碍（栅栏）？

Question

以下代码实现了一些无锁（且无原子！）的线程间通信，需要使用存储和加载内存屏障，但 C++11 释放-获取语义不合适，也不保证正确性.实际上，该算法需要一种释放-获取语义的反转，即表示某些操作没有发生，而不是它确实发生了。

volatile bool valid=true;
volatile uint8_t blob[1024] = {/*some values*/};

void zero_blob() {
    valid=false;
    STORE_BARRIER;
    memset(blob,0,1024);
}

int32_t try_get_sum(size_t index_1, size_t index_2) {
    uint8_t res = blob[index_1] + blob[index_2];
    LOAD_BARRIER;
    return valid ? res : -1; 
}

我只需使用本机内存屏障（例如在 Intel 上，这里不需要内存屏障，在 Sparc (RMO) membar #StoreStore 和 membar #LoadLoad 上，在 PowerPC lwsync 上都适用。所以没什么大不了的，代码是使用存储和加载障碍的典型示例。现在，假设我不想将“blob”转换为std::atomic 对象，我应该使用什么 C++11 构造来使代码正确，因为它会使“blob”成为保护对象，变量“有效”成为受保护的对象一个，而反过来。 将变量 'valid' 转换为 std::atomic 对象对我来说是可以的，但没有任何障碍可以保证正确性。为了清楚起见，让我们考虑以下代码：

volatile std::atomic<bool> valid{true};
volatile uint8_t blob[1024] = {/*some values*/};

void zero_blob() {
    valid.store(false, std::memory_order_release);
    memset(blob,0,1024);
}

int32_t try_get_sum(size_t index_1, size_t index_2) {
    uint8_t res = blob[index_1] + blob[index_2];
    return valid.load(std::memory_order_acquire) ? res : -1; 
}

代码不正确，因为屏障放置在错误的位置，因此写入“blob”可以先于写入“valid”或/并且从“valid”加载可以先于从“blob”加载。我认为为了处理这种结构，C++11 提供了std::atomic_thread_fence，代码应该是：

volatile std::atomic<bool> valid{true};
volatile uint8_t blob[1024] = {/*some values*/};

void zero_blob() {
    valid.store(false, std::memory_order_relaxed);
    std::atomic_thread_fence(std::memory_order_release);
    memset(blob,0,1024);
}

int32_t try_get_sum(size_t index_1, size_t index_2) {
    uint8_t res = blob[index_1] + blob[index_2];
    std::atomic_thread_fence(std::memory_order_acquire);
    return valid.load(std::memory_order_relaxed); ? res : -1; 
}

不幸的是 C++11 说：

如果存在，释放栅栏 A 与获取栅栏 B 同步 原子操作 X 和 Y，都对某个原子对象 M 进行操作， 使得 A 在 X 之前排序，X 修改 M，Y 在之前排序 B、Y读取X写入的值或任意一方写入的值 如果它是一个假设的释放序列 X 中的效果 释放操作。

其中明确指出std::atomic_thread_fence 应放置在原子对象操作的相对两侧。

稍后编辑

请在下面找到更多有用的示例：

volatile uint64_t clock=1;
volatile uint8_t blob[1024] = {/*some values*/};

void update_blob(uint8_t vals[1024]) {
    clock++;
    STORE_BARRIER;
    memcpy(blob,vals,1024);
    STORE_BARRIER;
    clock++;
}

int32_t try_get_sum(size_t index_1, size_t index_2) {
    uint64_t snapshot = clock;
    if(snapshot & 0x1) {
        LOAD_BARRIER;
        uint8_t res = blob[index_1] + blob[index_2];
        LOAD_BARRIER;
        if(snapshot == clock)
            return res;
    }
    return -1;
}

Answer 1

根据memory_order article，为了保守安全，您需要在存储之后使用memory_order_release，在加载之前使用memory_order_acquire（两者都是相同的原子变量）。

所以：

 std::atomic<int> var;

 // Writer
 // something important <happens-before> writing 42 in the writer thread
 var.store(42, std::std::memory_order_release);

 // Reader
 auto result = var.load(std::std::memory_order_acquire);
 if (result == 42) {
    // transitively, as the result's new value is observed, the "something important" is here too
 }

更一般地说，根据你需要达到什么效果以及你的目标架构支持什么，你可以不那么保守地去做。

您通常更喜欢std::atomic_flag 而不是std::atomic<bool>，因为与后者不同，前者保证是无锁的。

最后 - 为什么不从受互斥锁保护的关键部分开始，或者更好的是，通过无锁环形缓冲区将更新推送给消费者，这样他们就不会共享任何东西？

Answer 2

请在下面找到更多有用的示例：

请允许我重述一下您实际上在做什么。

您的代码正在从内存中读取。该内存可能随时由其他线程更新。但是您不想在读取器和写入器之间强加 执行 同步（即某种互斥锁）。因此，您构建了一个系统，允许您检测您已经执行的读取是否可能已被其他线程覆盖。如果是，您只需忽略您读取的值。

C++ 不允许你这样做。

如果您正在读取一个可能由其他线程写入的非原子对象，而这两个操作之间没有适当的执行和内存同步，那么您就有了数据竞争。数据竞争的存在并不意味着您可能会读取不正确的值；这意味着您的代码有未定义的行为。

您无法撤消 UB。一旦你的程序进入未定义行为领域，所有的赌注都被取消了。至少，就标准而言。

当然，您可以重写代码以符合标准。但它必须在可能发生写入时阻止读取的执行，而不仅仅是事后检查读取是否正常。如果写入始终为 1KB，则基于原子的自旋锁可能适合写入功能，如果原子锁不可用，则读取器只需返回 -1。

您也可以使用 C++ 原子来编写这种系统（显然称为“SeqLock”），并完全了解它调用 UB 就标准而言。只要您复制的类型是微不足道的，它就可以正常工作。

请注意，C++ 并发 TS 2 将 include a feature that will allow SeqLock implementations。这有望看到 C++23 的完整标准。

Answer 3

你的例子是正确的。关于如何处理非原子，我认为标准不是很清楚。

std::atomic<bool> valid{true};           // removed volatile
uint8_t blob[1024] = {/*some values*/};  // removed volatile

void zero_blob() {
    valid.store(false, std::memory_order_relaxed);            // A)
    std::atomic_thread_fence(std::memory_order_release);      // B)
    memset(blob,0,1024);                                      // C)
}                                                              

int32_t try_get_sum(size_t index_1, size_t index_2) {          
    uint8_t res = blob[index_1] + blob[index_2];              // D)
    std::atomic_thread_fence(std::memory_order_acquire);      // E)
    return valid.load(std::memory_order_relaxed) ? res : -1;  // F)
}

步骤 B) 中的栅栏保证程序顺序会得到遵守，标准并没有真正说明这一点，但应在同一线程中的任何后续写入之前传播对 valid 的存储。

步骤 E) 中的栅栏保证程序顺序得到遵守。

所以 A) 线程间发生在 C) 和 D) 发生在 F) 之前

如果 F) 认为 valid 是 true，那么，假设没有 ABA 问题，F) 发生在 A) 之前。

如果 F) 发生在 A) 之前，那么 D) 发生在 A) 之前。

如果 F 认为 valid 是 false，则 A) 发生在 F) 之前。这并不一定意味着 C) 发生在 D) 之前，但我们会丢弃结果以防万一。丢弃确实很重要，因为blob 中的值可能完全无效。 （虽然使用uint8_t 可以防止部分读取从转换为void*。它是缓存对齐的，但它在技术上不是字大小的，这使原子性受到质疑。IIRC 从理论上讲，它也可能容易受到真正的部分读取的影响在一些不太可能的架构上。）

这些推论基于 Linux 内核内存一致性模型 (LKMM)，而不是 C++ 内存模型。根据我在 cppreference.com 上读到的内容，当他们谈论 C++ 内存模型中的传播顺序时，他们使用诸如“线程 A 中 X 之前的所有更改将在线程 B 中的 Y 之后可见”之类的短语，但通常仅在原子操作的上下文。我认为期望 C++ 编译器发出保证传播顺序的指令的风险相对较低，至少在消费级 CPU 上是这样。您仍应检查组件并查阅 CPU 文档。多亏了 TSO，您在 x86 上肯定会很好，但我从未研究过弱序架构上可用的传播顺序原语。

最后，如果您要重新使用blob，应该有另一个原子变量来指示memset 已完成。