出于排序的目的，原子读-修改-写是一种操作还是两种操作？

Question

考虑一个原子读-修改-写操作，例如x.exchange(..., std::memory_order_acq_rel)。出于对其他对象的加载和存储进行排序的目的，这是否被视为：

1. 具有获取-释放语义的单个操作？

2. 或者，作为一个获取加载，然后是一个释放存储，附加保证其他加载和存储x将同时观察它们或两者都不观察？

如果它是 #2，那么尽管在加载之前或存储之后不能对同一线程中的其他操作进行重新排序，但仍然存在在两者之间重新排序的可能性。

作为一个具体的例子，考虑：

std::atomic<int> x, y;

void thread_A() {
    x.exchange(1, std::memory_order_acq_rel);
    y.store(1, std::memory_order_relaxed);
}

void thread_B() {
    // These two loads cannot be reordered
    int yy = y.load(std::memory_order_acquire);
    int xx = x.load(std::memory_order_acquire);
    std::cout << xx << ", " << yy << std::endl;
}

可以thread_B输出0, 1吗？

如果x.exchange()换成了x.store(1, std::memory_order_release);那么thread_B肯定能输出0, 1。是否应该exchange()排除额外的隐式负载？

cppreference听起来像 #1 是这种情况并且0, 1被禁止：

具有此内存顺序的读-修改-写操作既是获取操作又是释放操作。当前线程中的任何内存读取或写入都不能在此存储之前或之后重新排序。

但是我在标准中找不到任何明确的内容来支持这一点。实际上，该标准对原子读-修改-写操作几乎没有说明，除了 N4860 中的 31.4 (10) 这只是读取必须读取写入之前写入的最后一个值的明显属性。所以虽然我不想质疑 cppreference，但我想知道这是否真的正确。

我也在研究它是如何在 ARM64 上实现的。gcc 和 clangthread_A本质上都编译为

ldaxr [x]
stlxr #1, [x]
str #1, [y]

（参见 Godbolt。）根据我对 ARM64 语义的理解和一些测试（使用负载y而不是存储），我认为str [y]可以在 之前变得可见stlxr [x]（尽管当然不是在 之前ldaxr）。这将使thread_B观察成为可能0, 1。所以如果#1 是真的，那么 gcc 和 clang 似乎都是错误的，我不敢相信。

最后，据我所知，替换memory_order_acq_relwithseq_cst不会改变此分析的任何内容，因为它只添加了与其他seq_cst操作相关的语义，而我们这里没有。

---

我发现C++ 内存模型中的哪些确切规则可以防止在获取操作之前重新排序？如果我理解正确，这似乎同意#2 是正确的，并且0, 1可以观察到。我仍然很感激您的确认，以及检查 cppreference 引用是否实际上是错误的或者我是否误解了它。

Answer 1

从标准的角度来看，RMW 操作是单个操作。我不确定它是否在任何地方明确说明，但它的名称（单数）和一些相关的措辞似乎暗示了这一点。

纯粹从标准的角度来看，您的代码可以打印0, 1.

首先，该标准不是按照操作重新排序来措辞，而是按照释放和获取操作之间的同步关系来措辞。

由于y.load(acquire)没有匹配的发行版或更强的商店（没有任何可同步的内容），因此它与y.load(relaxed).

由于x.exchange(1, acq_rel)只有匹配的负载要同步，但没有存储，因此它的“获取”部分不会做任何有用的事情（实际上是放松的），因此可以将其替换为x.store(1, release).

然后我们只有在存储和加载之间存在潜在的同步x，但由于存储之前和加载之后（在各自的线程中）没有任何操作，因此该同步也不会执行任何操作。

所以两个负载都可以返回0或1。

Answer 2

不是语言标准层面的答案，而是一些证据表明，在实践中，答案可以是“两个”。正如我在问题中猜测的那样，即使 RMW 是 ，这种情况也可能发生seq_cst。

我无法像最初的问题一样观察到存储被重新排序，但这里有一个示例，显示原子seq_cstRMW 的存储通过以下relaxed负载重新排序。

下面的程序是 Peterson 算法的实现，改编自 LWimsey 的示例，其中获取释放内存顺序与顺序一致性不同的实际示例是什么？。正如那里所解释的，该算法的正确版本涉及

me.store(true, std::memory_order_seq_cst);
if (other.load(std::memory_order_seq_cst) == false) 
    // lock taken

重要的是负载在存储后变得可见。

如果 RMW 是用于排序语义的单个操作，我们希望这样做是安全的

me.exchange(true, std::memory_order_seq_cst);
if (other.load(std::memory_order_relaxed) == false) {
    // Ensure critical section doesn't start until we know we have the lock
    std::atomic_thread_fence(std::memory_order_seq_cst);
    // lock taken
}

理论上，由于交换操作具有获取语义，因此负载必须在交换完成后变得可见，特别是在trueto的存储me变得可见之后。

但事实上，在 ARMv8-a 上，使用 gcc 或 clang，此类代码经常会失败。事实上，它似乎确实exchange由获取加载和发布存储组成，并且other.load可能在发布存储之前变得可见。（虽然不是在 的 acquire-load 之前exchange，但这与这里无关。）

clang 生成如下代码：

mov w11, #1
retry:
ldaxrb wzr, [me]
stlxrb w12, w11, [me]
cbnz w12, retry
ldrb w11, [other]

请参阅https://godbolt.org/z/fhjjn7，汇编输出的第 116-120 行。stlxrb（gcc 是相同的，但隐藏在库函数中。）通过 ARM64 内存排序语义，可以使用以下加载和存储对释放存储重新排序。它的排他性这一事实并没有改变这一点。

为了使重新排序更频繁地发生，我们安排存储的数据取决于之前错过缓存的加载，我们通过使用 逐出该行来确保这一点dc civac。我们还需要将两个标志me和other放在单独的缓存行上。否则，据我了解，即使线程 A 在存储之前执行加载，线程 B 也必须等待开始其 RMW，直到 A 的存储完成后，特别是在 A 的存储可见之前不会执行加载。

在多核 Cortex A72（Raspberry Pi 4B）上，断言通常在几千次迭代后失败，这几乎是瞬时的。

该代码需要使用 来构建-O2。我怀疑如果为 ARMv8.2 或更高版本构建（如果可用），它将无法工作swpalb。

// Based on /sf/answers/2930193871/ by LWimsey
#include <thread>
#include <atomic>
#include <cassert>

// size that's at least as big as a cache line
constexpr size_t cache_line_size = 256;

static void take_lock(std::atomic<bool> &me, std::atomic<bool> &other) {
    alignas(cache_line_size) bool uncached_true = true;
    for (;;) {
        // Evict uncached_true from cache.
        asm volatile("dc civac, %0" : : "r" (&uncached_true) : "memory");
        
        // So the release store to `me` may be delayed while
        // `uncached_true` is loaded.  This should give the machine
        // time to proceed with the load of `other`, which is not
        // forbidden by the release semantics of the store to `me`.
        
        me.exchange(uncached_true, std::memory_order_seq_cst);
        if (other.load(std::memory_order_relaxed) == false) {
            // taken!
            std::atomic_thread_fence(std::memory_order_seq_cst);
            return;
        }
        // start over
        me.store(false, std::memory_order_seq_cst);
    }
}

static void drop_lock(std::atomic<bool> &me) {
    me.store(false, std::memory_order_seq_cst);
}

alignas(cache_line_size) std::atomic<int> counter{0};

static void critical_section(void) {
    // We should be the only thread inside here.
    int tmp = counter.fetch_add(1, std::memory_order_seq_cst);
    assert(tmp == 0);
    
    // Delay to give the other thread a chance to try the lock
    for (int i = 0; i < 100; i++)
        asm volatile("");
    
    tmp = counter.fetch_sub(1, std::memory_order_seq_cst);
    assert(tmp == 1);
}    

static void busy(std::atomic<bool> *me, std::atomic<bool> *other)
{
    for (;;) {  
        take_lock(*me, *other);
        std::atomic_thread_fence(std::memory_order_seq_cst); // paranoia
        critical_section();
        std::atomic_thread_fence(std::memory_order_seq_cst); // paranoia
        drop_lock(*me);
    }
}


// The two flags need to be on separate cache lines.
alignas(cache_line_size) std::atomic<bool> flag1{false}, flag2{false};

int main()
{
    std::thread t1(busy, &flag1, &flag2);
    std::thread t2(busy, &flag2, &flag1);
    
    t1.join(); // will never happen
    t2.join();
    return 0;
}