为什么(或不是？)SFENCE + LFENCE相当于MFENCE？

Question

正如我们从之前的回答中所知道的,它是否在处理器x86/x86_64中指示LFENCE？我们不能使用SFENCE而不是MFENCE顺序一致性.

这里的答案表明MFENCE= SFENCE+ LFENCE,即LFENCE没有我们不能提供顺序一致性的东西.

LFENCE 无法重新排序:

SFENCE
LFENCE
MOV reg, [addr]

- 到 - >

MOV reg, [addr]
SFENCE
LFENCE

例如重新排序MOV [addr], reg LFENCE- > LFENCE MOV [addr], reg由机制提供- 存储缓冲区,它重新排序存储 - 负载以提高性能,并且因为LFENCE它不会阻止它.并SFENCE 禁用此机制.

什么机制禁用LFENCE无法重新排序(x86没有机制 - Invalidate-Queue)？

并且只是在理论上或者在现实中重新排序SFENCE MOV reg, [addr]- > MOV reg, [addr] SFENCE可能吗？如果可能,实际上,什么机制,它是如何工作的？

Answer 1

SFENCE + LFENCE不会阻止StoreLoad重新排序,因此它不足以实现顺序一致性.只有mfence(或locked操作,或类似的实际序列化指令cpuid)才能做到这一点.请参阅Jeff Preshing的" 记忆重新排序法案",了解只有完整障碍就足够的情况.

从英特尔的指令集参考手册输入sfence:

在SFENCE全局可见之后,处理器确保SFENCE之前的每个商店在任何商店之前全局可见.

但

它没有针对内存加载或LFENCE指令进行排序.

LFENCE迫使早期的指令"在本地完成"(即从核心的无序部分退出),但对于存储或SFENCE而言,只是意味着将数据或标记放入内存顺序缓冲区,而不是将其刷新商店变得全球可见.即SFENCE"完成"(从ROB退出)不包括刷新存储缓冲区.

这就像Preshing描述的内存障碍就像源控制操作,其中StoreStore障碍不是"即时"的.在那篇文章的后面,他解释了为什么#StoreStore + #LoadLoad + #LoadStore屏障不会加起来#StoreLoad屏障.(x86 LFENCE有一些额外的指令流序列化,但由于它没有刷新存储缓冲区,因此推理仍然有效).

LFENCE没有完全序列化cpuid(与内存屏障mfence或locked指令一样强).它只是LoadLoad + LoadStore屏障,加上一些执行序列化的东西,可能作为一个实现细节开始,但现在作为保证,至少在Intel CPU上.它rdtsc对于避免分支推测以减轻幽灵有用.

顺便说一句,除了NT商店外,SFENCE是无操作的; 它就正常(释放)商店命令他们.但不是关于负载或LFENCE.只有在通常弱订购的CPU上,商店商店的屏障才能做任何事情.

真正令人担忧的是StoreLoad在商店和负载之间重新排序,而不是在商店和障碍之间重新排序,因此您应该查看具有商店,然后是障碍,然后是负载的案例.

mov  [var1], eax
sfence
lfence
mov   eax, [var2]

可以按以下顺序变为全局可见(即提交到L1d缓存):

lfence
mov   eax, [var2]     ; load stays after LFENCE

mov  [var1], eax      ; store becomes globally visible before SFENCE
sfence                ; can reorder with LFENCE

Answer 2

一般来说，MFENCE != SFENCE + LFENCE。例如，下面的代码在使用 编译时-DBROKEN在某些 Westmere 和 Sandy Bridge 系统上失败，但在 Ryzen 上似乎可以运行。事实上，在 AMD 系统上，一个 SFENCE 似乎就足够了。

#include <atomic>
#include <thread>
#include <vector>
#include <iostream>
using namespace std;

#define ITERATIONS (10000000)
class minircu {
        public:
                minircu() : rv_(0), wv_(0) {}
                class lock_guard {
                        minircu& _r;
                        const std::size_t _id;
                        public:
                        lock_guard(minircu& r, std::size_t id) : _r(r), _id(id) { _r.rlock(_id); }
                        ~lock_guard() { _r.runlock(_id); }
                };
                void synchronize() {
                        wv_.store(-1, std::memory_order_seq_cst);
                        while(rv_.load(std::memory_order_relaxed) & wv_.load(std::memory_order_acquire));
                }
        private:
                void rlock(std::size_t id) {
                        rab_[id].store(1, std::memory_order_relaxed);
#ifndef BROKEN
                        __asm__ __volatile__ ("mfence;" : : : "memory");
#else
                        __asm__ __volatile__ ("sfence; lfence;" : : : "memory");
#endif
                }
                void runlock(std::size_t id) {
                        rab_[id].store(0, std::memory_order_release);
                        wab_[id].store(0, std::memory_order_release);
                }
                union alignas(64) {
                        std::atomic<uint64_t>           rv_;
                        std::atomic<unsigned char>      rab_[8];
                };
                union alignas(8) {
                        std::atomic<uint64_t>           wv_;
                        std::atomic<unsigned char>      wab_[8];
                };
};

minircu r;

std::atomic<int> shared_values[2];
std::atomic<std::atomic<int>*> pvalue(shared_values);
std::atomic<uint64_t> total(0);

void r_thread(std::size_t id) {
    uint64_t subtotal = 0;
    for(size_t i = 0; i < ITERATIONS; ++i) {
                minircu::lock_guard l(r, id);
                subtotal += (*pvalue).load(memory_order_acquire);
    }
    total += subtotal;
}

void wr_thread() {
    for (size_t i = 1; i < (ITERATIONS/10); ++i) {
                std::atomic<int>* o = pvalue.load(memory_order_relaxed);
                std::atomic<int>* p = shared_values + i % 2;
                p->store(1, memory_order_release);
                pvalue.store(p, memory_order_release);

                r.synchronize();
                o->store(0, memory_order_relaxed); // should not be visible to readers
    }
}

int main(int argc, char* argv[]) {
    std::vector<std::thread> vec_thread;
    shared_values[0] = shared_values[1] = 1;
    std::size_t readers = (argc > 1) ? ::atoi(argv[1]) : 8;
    if (readers > 8) {
        std::cout << "maximum number of readers is " << 8 << std::endl; return 0;
    } else
        std::cout << readers << " readers" << std::endl;

    vec_thread.emplace_back( [=]() { wr_thread(); } );
    for(size_t i = 0; i < readers; ++i)
        vec_thread.emplace_back( [=]() { r_thread(i); } );
    for(auto &i: vec_thread) i.join();

    std::cout << "total = " << total << ", expecting " << readers * ITERATIONS << std::endl;
    return 0;
}

Answer 3

什么机制禁用 LFENCE 来实现不可能的重新排序（x86 没有机制 - Invalidate-Queue）？

来自英特尔手册第 2A 卷第 3-464 页的说明文档LFENCE：

LFENCE 直到所有前面的指令都在本地完成后才会执行，并且在 LFENCE 完成之前不会开始执行后面的指令

所以是的，指令明确阻止了您的示例重新排序LFENCE。仅涉及指令的第二个示例SFENCE是有效的重新排序，因为SFENCE对加载操作没有影响。