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

Question

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

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

或者，作为获取加载，然后是释放存储，附加保证其他加载和存储到 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是禁止的:

A read-modify-write operation with this memory order is both an acquire operation and a release operation. No memory reads or writes in the current thread can be reordered before or after this store.

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

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

( See on godbolt. ) 根据我对 ARM64 语义的理解，以及一些测试(负载为 y 而不是存储)，我认为 str [y]可以在 stlxr [x] 之前变得可见(当然不是在 ldaxr 之前)。这将使 thread_B 成为可能观察 0, 1 .所以如果 #1 是真的，那么 gcc 和 clang 似乎都是错误的，我不敢相信。
最后，据我所知，更换 memory_order_acq_rel与 seq_cst不会改变此分析的任何内容，因为它仅添加了与其他 seq_cst 相关的语义。操作，我们这里没有。

我找到了 What exact rules in the C++ memory model prevent reordering before acquire operations?如果我理解正确的话，它似乎同意 #2 是正确的，并且 0, 1可以观察到。我仍然希望得到确认，以及检查 cppreference 引用是否实际上是错误的或者我是否误解了它。

Answer 1

不是语言标准层面的答案，而是一些证据表明，在实践中，答案可以是“两个”。正如我在问题中猜测的那样，即使 RMW 是 seq_cst，也可能发生这种情况。 .
我无法像原始问题一样观察到存储被重新排序，但这里有一个示例显示了原子 seq_cst 的存储RMW 被重新排序为以下 relaxed加载。
下面的程序是根据 What's are practical example where acquire release memory order differs from sequential consistency? 中 LWimsey 的示例改编的 Peterson 算法的实现。 .正如那里所解释的，该算法的正确版本涉及

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
}

理论上，由于交换操作已获得语义，因此在交换完成后，尤其是在 true 存储后，负载必须变为可见。至 me已经变得可见。
但实际上在 ARMv8-a 上，使用 gcc 或 clang，这样的代码经常失败。看来，实际上， exchange确实包含一个获取加载和一个释放存储，以及 other.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 行。 (gcc 是相同的，但隐藏在库函数中。)根据 ARM64 内存排序语义，release-store stlxrb可以通过以下加载和存储重新排序。它是排他性的这一事实并没有改变这一点。
为了使重新排序更频繁地发生，我们安排存储的数据依赖于之前错过缓存的加载，我们通过使用 dc civac 驱逐该行来确保这一点。 .我们还需要放置两个标志 me和 other在单独的缓存行上。否则，据我所知，即使线程 A 在存储之前进行加载，线程 B 也必须等到 A 的存储完成后才开始其 RMW，尤其是在 A 的存储可见之前不会进行加载。
在多核 Cortex A72 (Raspberry Pi 4B) 上，断言通常会在几千次迭代后失败，这几乎是瞬时的。
代码需要用 -O2构建.我怀疑如果为 ARMv8.2 或更高版本构建，它将无法工作，其中 swpalb可用。

// Based on https://stackoverflow.com/a/41859912/634919 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;
}