原子计数与内存序

原子操作

原子操作在不同的语境下有不同的含义。在CPU中，原子操作是一种不会被打断的操作，CPU的任何一条指令都是原子操作。在编程语言中，原子操作通常用来操作内存中的值，而此时原子形容的是读写内存时以一个整体进行操作，不会因为其他核心或硬件对内存的修改就出现和预期不一致的行为，它可能是一条指令，也可能由多条指令组成。原子操作是多核CPU中多线程同步的必须能力。

例如一个基本的x++语句，在编译后可能就会被拆成多条指令。又或者虽然仅被编译为一条指令，但在多核多线程场景下存在数据竞争导致结果与预期不符。

; ARM64架构
ldr     w8, [x0]
add     w8, w8, #1
str     w8, [x0]
; X86架构
incl    (%rdi)

为了解决这种问题，就需要CPU提供原子操作的能力，x++必须要不被其他东西干扰地完成。X86架构提供了一系列的原子访存指令，包括CAS、原子加法等。而ARM架构通常依赖锁缓存行的方式来实现多指令的原子操作。

不过对于单纯的读写操作，如果一个值小于内存总线的宽度，那么正常的读写指令都是原子的。

但是仅靠原子操作，并不能完美实现锁的功能，这是因为还存在指令重排的情况。

指令重排

指令重排是一种提高运行速度的方法，它可以在不改变单线程执行结果的情况下，对执行指令的顺序进行调整，从而尽可能充分利用CPU中的资源。

编译器重排

编译器在优化代码时，通常会为了提高速度，对指令进行重排，例如下面的代码。

struct add_s {
    int a;
    int b;
    int sum;
};
void add_nz(struct add_s *s) {
    if (s->b == 0) {
        return;
    }
    int a = s->a;
    int b = s->b;
    s->sum = a + b;
}

在优化过程中，编译器看到代码虽然访问了两次s->b，但是两次访问期间并没有修改，所以优化后的代码中两次读取很可能只会将第一次的值加载入寄存器，而第二次直接使用寄存器内部的值。

这种编译器重排，通常可以通过为变量增加volatile修饰符或在必要处增加内联的空汇编指令asm volatile("" ::: "memory")来避免。

CPU重排

除了编译器可能会优化代码导致代码重排外，CPU也会重排代码使其不按顺序执行。相比编译器的重排，CPU的重排是为了最大限度地利用CPU中的各个元件，减少因为某一个指令执行慢而导致整个CPU流水线停顿的情况。

例如，CPU中加法器和除法器是两个元件，那么c = a + b和d = a / b就可以同时计算，而除法计算较慢，可能除法还在计算过程中，加法就已经算完了，这时CPU就又会让更后面arr[a] = c的寻址功能占用加法器。

通过这种乱序执行，CPU可以最大限度的利用好内部的每个元件，但是代价就是多核场景中，无法保证读写顺序总是和代码逻辑一致。

阻止CPU重排

当然，为了正确实现多核同步，也有些指令会阻止CPU重排代码，例如X86的mfence指令，ARM的dmb、isb指令等。另外有些存取指令本身也会阻止CPU重排它们周围的存取指令。

volatile关键字

volatile关键字在C和Java中都存在，有着类似但不完全一致的功能。在C/C++中，volatile仅仅是指明某块内存是易变的，它可能会被外部DMA设备或其他线程更改。编译器在生成访问volatile修饰的变量的代码时，不会将多次访问变为一次，也不会将后面的访问重排到前面。但是，编译器并不会插入阻止CPU重排的指令。而在Java中，volatile变量也明确保证会阻止CPU的指令重排。

内存序

显然，执行计算的代码只需要保证依赖关系没有错误，就可以随意重排。但是访存代码的重排就可能导致多线程访问与预期不符的问题。例如下面的代码。

int flag = 0;
int data1 = 0;
int data2 = 0;
void *read_thread(void *ctx) {
    while (flag == 0) {
    }
    int d1 = data1;
    int d2 = data2;
    printf("data1 = %d, data2 = %d\n", d1, d2);
    return NULL;
}

void *write_thread(void *ctx) {
    data1 = 100;
    data2 = 200;
    flag = 1;
    return NULL;
}

上面的代码中，读线程理论上应该总是能打印出data1 = 100, data2 = 200，但如果是ARM架构的设备，则可能出现打印不符合预期的情况。这可能是因为读线程中data1和data2先读，而flag后读导致的；也有很小的可能性是写线程中data1和data2后写，flag先写导致的。

而如果是X86的设备，这里则总是能保证正确地打印出来，这也就引出了本文的另一个话题：内存序模型。

不同的CPU实现有着不同的内存序模型，人们按照其对内存读写的重排力度大致分为两类：强内存序模型和弱内存序模型。常见的CPU中，X86是强内存序模型，ARM是弱内存序模型。

弱内存序模型

CPU 指令中存在很多内存读写的操作，但涉及到重排时，我们可以只关注两条指令会如何发生重排，并基于此加以推广。而每条指令又有读和写两种类型，所以一共有四种重排的基础模型：读读（Load-Load）、读写（Load-Store）、写读（Store-Load）、写写（Store-Store）。弱内存序模型下，这四种情况均有可能发生重排。

ldr w1, [x0, 12]       ; #1 如果按顺序执行，它应该先执行
ldr w2, [x0]           ; #2 它应该后执行

上面的ARM64汇编中，读读乱序会导致 #2可能实际发生在 #1 之前。类似的几种重排在下面的代码中展示。

ldr w1, [x0, 12]  ; #3
str w3, [x2, 8]   ; #4
str w4, [x2, 16]  ; #5
ldr w5, [x2, 32]  ; #6

其中，#3和#4发生重排，则是读写重排，也就是#4实际上发生在#1之前。而#4和#5的重排则是写写重排，#5和#6的重排为写读重排。由于数据没有依赖，在ARM架构中，它们的完成顺序完全是任意的。

ARM提供的是弱内存序模型，这能更好地利用CPU的乱序执行与分支预测能力，简化CPU的架构设计，提供更多的性能提升空间。但相对的，程序员就需要更小心地使用内存同步原语。当然，在高级语言内，程序员并不需要担心这种问题，无论是Java的volatile，还是C、C++ 中的 mutex、 atomic_t，都为程序员提供了开箱即用的完整内存序模型，无需担心上层应用出现问题。

重排原因

两条访存指令可能由于多种原因而出现实际执行顺序和代码顺序不一致的情况。例如一条访存指令直接使用寄存器访问，而另一条则有加法和算数左移。又例如一条指令访问的位置处于L1缓存内，而另一条完全没有缓存。很多原因都可能导致重排发生。

强内存序模型

相对的，X86提供的是强内存序模型。在强内存序模型下，读读、读写、写写重排都被禁止了。但是，X86也并未实现顺序一致性，仍然保留了写读重排。

难以避免的写读重排

什么即使X86这种强内存序架构也会出现乱序呢？显然这是出于执行效率优化的原因，处理器愿意放弃一些内存序保证来提高速度。
指令被CPU前端解码后，会进入重排缓冲区 (ROB, ReOrder Buffer)。当指令的所有相关动作都正确完成后，指令会离开重排缓冲区，此时指令执行完成。

X86虽然是强内存序模型的CPU，但是为了提高运行速度，其内部也并不会真的强制让每个读写指令都按顺序执行。CPU会将读写指令放入对应的缓冲区 (load buffer / store buffer) 以同时保证内部的乱序执行和最终一致性。

写入指令处于重排中时，其写入的值会暂存在写缓冲区，而当其retire后，写缓冲区中的内容会尽快提交到缓存。但是显然，执行过程中很有可能发生分支预测失败。当分支预测失败时，该写入的值会随乱序缓冲区一起被丢弃，不会污染缓存。如果读指令必须等待写指令完毕才能执行，就相当于必须要等到这条写指令的分支正确执行完成，在分支预测正确率较高的情况下，这样会损失相当多的性能。
相对的，对于一个可缓存读，其并不会产生缓存刷新（包括CPU缓存和页面缓存）以外的副作用，无论分支预测是否失败，其后的对其没有依赖的指令都可以执行，对性能的影响相对较小。

基于以上原因，CPU没有会使用强一致性模型，至少都会允许后面的读发生在前面的写之前，即Store-Load reordering。

下面是ARM和X86实现中重排机制的表格。

	Load-Load	Load-Store	Store-Load	Store-Store
X86	No	No	Yes	No
ARM	Yes	Yes	Yes	Yes

预测执行与重排

指令重排会和预测执行同时出现，我在公司遇到了相关的问题。

简单复述一下这个bug。CPU与硬件共享内存，通过环形请求/完成队列进行通信。出现任务时，软件会将任务提交至请求队列，并轮询完成队列。硬件读到请求项后进行处理，处理完成后将完成事件送入完成队列。软件轮询到完成队列有新事件后，则会读完成队列内的数据，并返回给调用者。

void submit_request(struct sq_entry *sqe) {   
}

struct cq_event *get_complete_event() {
}
void process_cqe(struct cq_event *cqe) {
    int d = cqe->data;
    // ... 处理完成事件
}
void event_loop() {
    while (g_running) {
        submit_request(submission_queue_entry);
        // ...... 其他处理
        struct cq_event *event = get_complete_event();
        if (event != NULL) {
            process_cqe(event);
        }
    }
}

正常思维来讲，毕竟cqe已经获取到了，那么事件内部的数据结构一定是正确的才对，但是事实上就是会偶现内部数据结构不正确的情况，经过漫长的定位后，最终确定为读取cqe和读取事件的内容时出现了读读乱序，导致先读了事件内容的垃圾数据，后来才读到cqe，而此时判断cqe为真，这些垃圾数据也就留下来了。

通常，多线程只需要合理运用锁和原子操作，就不需要关心汇编层面的问题，但是这个问题并非两个线程间出现数据竞争，而是一个CPU核心与另一个CPU内的其他硬件出现数据竞争。由于另一个是硬件，也无法让它使用锁，最后是通过软件增加dmb ish指令解决。

有个同事不认为读垃圾数据再写会先于判断之前发生，于是我写了下面的POC代码来反驳。

#include <stdio.h>
#include <stdint.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdlib.h>

#define likely(x) __builtin_expect(!!(x), 1)
#define unlikely(x) __builtin_expect(!!(x), 0)

struct data128 {
    // 发生错误的结构体长度就是128bits长，这里复制了结构体的关键字段
    // 但出于保密原因，隐去了不必要的名字，并将关键名称修改为counter和parity
    uint32_t counter;
    uint32_t unused1;
    uint32_t unused2;
    uint32_t rsvd1 : 16;
    uint32_t parity : 1;
    uint32_t rsvd2 : 15;
};

__attribute__((aligned(64))) volatile bool write = false;
typedef __int128_t i128;

void *save_thread(void *arg)
{
    uint32_t cnt = 0;
    uint32_t p = 0;
    while (true) {
        p = 1 - p;
        cnt++;
        struct data128 tmp = {.counter = cnt, .parity = p};
        i128 *tmp2 = (i128 *)&tmp;
        i128 *data = (i128 *)arg;
        // 原子性地将内容存入data，并且保证顺序
        __atomic_store_n(data, *tmp2, __ATOMIC_SEQ_CST);
        // 增加 isb 指令进行强内存序保证，保证写write前，所有指令都已经完成
        // 这里其实没什么必要，因为上面是由内存序保证的原子写操作，不会让后面的写比前面的先执行
        // 又加上这些主要是为了演示使用
        asm volatile("isb" ::: "memory");
        // 同样原子性写write，并保证后面的指令都已经完成
        __atomic_store_n(&write, false, __ATOMIC_SEQ_CST);
        // 标记write所在缓存为失效，这里可能没什么意义，因为写线程的缓存用处不大
        asm volatile("dc civac, %0" ::"r"(&write));
        while (!__atomic_load_n(&write, __ATOMIC_ACQUIRE)) {
            // 等待write为true，当write为true时进行下一次循环
            asm volatile("" ::: "memory");
        }
        // 等待所有指令完成再进行下一个循环
        asm volatile("isb" ::: "memory");
        asm volatile("dc civac, %0" ::"r"(&write));
    }
}

void *read_thread(void *arg)
{
    while (true) {
        if (unlikely(write)) {
            // 如果还没观察到写线程将write写为true，就循环忙等，直到观察到write为true。
            // 编译期标记为较为不可能，让推测执行尽可能执行下方的语句。
            continue;
        }
        struct data128 *data = (struct data128 *)arg;
        // 此时读线程已经观察到write为false了
        // 如果没有 Load-Load reordering，那么下面读到的值一定都是更新后的，且奇偶性一致，
        // 因为写线程使用内存序保证write可见时，counter和parity一定也已经可见。
        // 但实际上会有极小概率cnt与p的奇偶性不一致
        // 此时就是因为先发生了预测执行的读取cnt和p，且在预测期间缓存发生了更新，所以cnt和p不一致
        // 而write则是在缓存更新后被读取，所以write为假，可以走到该流程
        // 一种可能的路径
        /*
        下面文字顺序是代码顺序，而前面的标号则代表reorder之后的实际执行顺序
        2. 读 write
        3. 判断 write 为假
        1. 读counter （读后发生了缓存刷新，counter为旧值）
        4. 读parity
        */
        int cnt = data->counter;
        int p = data->parity;
        if (cnt % 2 != p) {
            // 这里实测很难触发，但是跑个十几分钟也能复现
            printf("Error: counter = %d, parity = %d\n", cnt, p);
            printf("load-load reordering\n");
            printf("write = %u, w = %u\n", write, 0);
            abort();
        }
        // 将write置为true，写线程读到write为true后会再写一次。
        // 这相当于读线程提交了一个事件给写线程，写线程获得该事件后就尝试再写一次。
        __atomic_store_n(&write, true, __ATOMIC_RELEASE);
        // 将缓存标记为失效，尽可能延迟读取到 write 变量的时间
        asm volatile("dc civac, %0" ::"r"(&write));
        // 等待所有指令完成再进行下一个循环
        asm volatile("isb" ::: "memory");
    }
}

void start_threads()
{
    pthread_t tid1;
    pthread_t tid2;
    void *data = calloc(128, 1);
    // 缓存行对齐
    data = (void *)(((uintptr_t)data + 63) & (~63ULL));
    pthread_create(&tid1, NULL, read_thread, data);
    pthread_create(&tid2, NULL, save_thread, data);
    pthread_join(tid1, NULL);
    pthread_join(tid2, NULL);
    return;
}

// gcc ldr_reorder_test.c -O2 -g -o ldr_reorder_test
int main()
{
    printf("sizof(struct data128) = %d\n", sizeof(struct data128));
    start_threads();
    return 0;
}

指令缓存与数据缓存

X86代码的内存一致性模型不仅体现在保守的乱序执行上，对于自修改代码，它也会保证数据缓存与指令缓存的一致性。这意味着在X86 JIT 编译器的开发中，不需要特殊的同步指令来将写入的JIT代码同步至CPU的指令缓存。
例如在下面的C代码中，如果写入该JIT代码后立即执行，则有概率会发生Illegal Instruction异常。这是因为数据缓存还未同步到指令缓存。

void *write_some_aarch64_jit_code() {
    unsigned char code[] = {
        0x00, 0x00, 0x80, 0xd2,  // mov x0, #1
        0xc0, 0x03, 0x5f, 0xd6   // ret
    };
    void *mem = mmap(NULL, sizeof(code), PROT_READ | PROT_WRITE | PROT_EXEC,
                     MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
    if (mem == MAP_FAILED) {
        perror("mmap");
        return NULL;
    }
    for (int i = 0; i < sizeof(code); i++) {
        ((unsigned char *)mem)[i] = code[i];
    }
    return mem;
}

这种情况需要手动同步指令缓存与数据缓存，在ARM64中可以使用如下汇编进行同步。

; 假设要同步的数据/代码的内存地址为 x0
DC CVAC, x0     ; 清除数据缓存
DSB SY          ; 确保缓存清除已完成
IC IVAU, x0     ; 清除指令缓存
DSB SY          ; 确保数据同步完成
ISB             ; 确保指令流水线反映最新的指令

这种方式可以清除x0所指向地址的缓存行（通常为64字节），如果有更长的指令需要清除，则需要在循环中执行该指令。
当然，GCC为我们提供了内置的帮助函数。

void __builtin___clear_cache(char *begin, char *end)

使用该函数，可以自动完成上面的汇编指令的功能，并可以方便的指定长度。