文章

内存模型

内存模型

什么是内存模型

Memory Model,是C++ 11引入的一个重要概念,它定义了:

多线程环境下,不同线程如何观察内存,以及编译器和CPU可以做哪些优化。

1 为什么需要内存模型

  • 硬件层面的乱序执行:为了提高CPU利用率,CPU会进行指令重排,且存在多级缓存(L1,L2,L3)和写缓冲区(Write Buffer)。导致不两只核心看到的内存修改顺序可能不同。
  • 编译器优化:为优化性能,编译器会重排没有依赖关系的指令。
  • C++内存模型的职责:它是一份契约。规定了编写多线程代码时,哪些是编译器和硬件必须保证的,哪引起是程序员必须保证的。

先来看个单线程的例子。

int a = 1;
int b = 2;

程序员看到 a=1 -> b =2 顺序执行。

实际上由于编译器优化、CPU乱序执行、Cache延迟同步等,机器执行可能变成 b = 2 -> a = 1 的顺序。

这对于单线程并无影响,因为结果一样。

多线程问题。

1
2
3
4
5
6
7
8
9
10
int data = 0;
bool ready = false;

# Thread A
data = 100;
ready = true;

# Thread B
while(!ready);
std::cout << data;

我们以为一定输出100,实际可能输出0。 因为 data = 100; ready = true; 可能被CPU重排序为 ready = true; data = 100。 线程B看到 ready 为 true, 而 data 为 0。

这就是经典的可见性问题。

2 数据竞争与未定义行为

  • 定义:当多个线程同时访问一个内存位置,且至少有一个是写操作,同时他们之间没有使用同步机制(互斥锁或原子操作)来建立顺序关系时,就发生了数据竞争。
  • C++标准规定

Data Race = UB (Undefined Behavior)

比如这里的读写x不是原子操作,可能结果是1 1,也可能1 2,甚至出现奇怪的问题。

1
2
3
4
5
6
 int x = 0;
 # Thread 1
 x++;

 # Thread 2
 x++;

可以使用Atomic来实现安全

1
2
 std::atomic<int> cnt{0};
 cnt++;

实际相当于lock, read, modify, write, unlock。 这样一来,多个线程的cnt++就不会发生覆盖。

3 修改顺序

  • Modification Order: 每个原子变量都有自己的修改顺序。所有线程看到该变量所有写操作的顺序必须是一致的。哪怕线程A看到的比线程B慢,但他们看到的数值演变方向必须是一致的。

内存顺序(Memory Order)

C++11引入了std::memory_order,提供了6种不同的内存顺序选项,用于平衡性能和可见性。

1 memory_order_relaxed (松散顺序)

特性: 仅保证当前操作的原子性,不提供任何跨线程的同步或推导顺序。编译器和CPU可以任意重排非原子代码。

应用场景: 纯计数器(如 std::shared_ptr 的引用计数增加,但不包括减少到0时的析构)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
// ====================================================================
// 1. memory_order_relaxed: 只保证"原子性", 不保证任何顺序/可见性关系
// ====================================================================
void demo_relaxed_counter() {
    section("1. memory_order_relaxed —— 只有原子性, 没有同步语义");

    std::atomic<long> counter{0};
    constexpr int kThreads = 4;
    constexpr int kIter = 500000;

    std::vector<std::thread> ts;
    for (int i = 0; i < kThreads; ++i) {
        ts.emplace_back([&] {
            for (int j = 0; j < kIter; ++j) {
                counter.fetch_add(1, std::memory_order_relaxed); // 只要求不丢数据
            }
        });
    }
    for (auto& t : ts) t.join();

    std::cout << kThreads << " 个线程各自 relaxed fetch_add " << kIter << " 次\n";
    std::cout << "counter = " << counter.load()
              << " (期望 " << (long)kThreads * kIter << ")\n";
    std::cout << "结论: relaxed 保证 fetch_add 不会因为并发而“丢加”(原子性),\n"
                 "      但它不建立 happens-before 关系, 不能用来同步其它变量的可见性。\n";
}

输出:

1
2
3
4
5
========== 1. memory_order_relaxed —— 只有原子性, 没有同步语义 ==========
4 个线程各自 relaxed fetch_add 500000 次
counter = 2000000 (期望 2000000)
结论: relaxed 保证 fetch_add 不会因为并发而“丢加”(原子性),
      但它不建立 happens-before 关系, 不能用来同步其它变量的可见性。

2 memory_order_seq_cst (顺序一致性)

特性: 默认选项。最严格的模型。所有线程看到的全局操作顺序完全一致,如同所有操作都在一个单核CPU上串行执行一样。它还在所有线程间建立了一个全局唯一的总线顺序。

代价: 性能损耗最大(会插入大量的内存屏障/Memory Barrier指令)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class PetersonLock {
public:
    // me 只能是 0 或 1, 是"我"的编号, 另一个线程编号是 1-me
    void lock(int me) {
        int other = 1 - me;
        flag_[me].store(true, std::memory_order_seq_cst);   // "我要进临界区了"
        turn_.store(other, std::memory_order_seq_cst);       // "如果有争用, 优先让对方进"
        // 只要对方也想进(flag_[other]==true) 且当前轮到对方(turn_==other), 就一直等
        while (flag_[other].load(std::memory_order_seq_cst) &&
               turn_.load(std::memory_order_seq_cst) == other) {
            // busy wait
        }
    }
    void unlock(int me) {
        flag_[me].store(false, std::memory_order_seq_cst);  // "我出临界区了"
    }
private:
    std::atomic<bool> flag_[2]{false, false};
    std::atomic<int> turn_{0};
};

void demo_pure_seq_cst() {
    section("2. memory_order_seq_cst 示例 —— Peterson 互斥锁(全程只用 seq_cst)");

    PetersonLock lock;
    long counter = 0;                     // 故意用非原子的 long, 完全依赖锁来保护
    constexpr long kIter = 2000000;

    std::thread t0([&] {
        for (long i = 0; i < kIter; ++i) {
            lock.lock(0);
            ++counter;                    // 临界区: 如果互斥失效, 这里就会丢计数
            lock.unlock(0);
        }
    });
    std::thread t1([&] {
        for (long i = 0; i < kIter; ++i) {
            lock.lock(1);
            ++counter;
            lock.unlock(1);
        }
    });
    t0.join();
    t1.join();

    std::cout << "两个线程各自加锁/解锁 " << kIter << " 次, 临界区里对同一个非原子变量 ++counter\n";
    std::cout << "counter = " << counter << " (期望 " << 2 * kIter << ")\n";
    std::cout << "结论: 全程只用 memory_order_seq_cst, 不需要像 acquire/release 那样操心\n"
                 "      “谁 release、谁 acquire、跟哪个变量配对”, 只要每个原子操作都是\n"
                 "      seq_cst, 直接按“好像所有操作真的按某个全局顺序依次执行”去推理即可——\n"
                 "      这正是 seq_cst 被定为默认值(不写 memory_order 参数时就是它)的原因:\n"
                 "      心智负担最小、最不容易用错, 代价是每个操作都带一次全内存屏障, 更慢。\n";
}

输出

1
2
3
4
5
6
7
8
========== 2. 纯 memory_order_seq_cst 示例 —— Peterson 互斥锁(全程只用 seq_cst) ==========
两个线程各自加锁/解锁 2000000 次, 临界区里对同一个非原子变量 ++counter
counter = 4000000 (期望 4000000)
结论: 全程只用 memory_order_seq_cst, 不需要像 acquire/release 那样操心
      “谁 release、谁 acquire、跟哪个变量配对”, 只要每个原子操作都是
      seq_cst, 直接按“好像所有操作真的按某个全局顺序依次执行”去推理即可——
      这正是 seq_cst 被定为默认值(不写 memory_order 参数时就是它)的原因:
      心智负担最小、最不容易用错, 代价是每个操作都带一次全内存屏障, 更慢。

3 Acquire-Release(获取-释放语义)

memory_order_release(释放): 当前线程中,所有在该操作之前的写操作,都不能被重排到该操作之后。

当该原子写操作完成后,所有之前的修改对其他进行了 acquire 读的线程可见。

memory_order_acquire(获取): 当前线程中,所有在该操作之后的读/写操作,都不能被重排到该操作之前。

确保能看到同步的那个 release 线程之前的所有写入。

release 限制了它前面的代码不能往下掉。

acquire 限制了它后面的代码不能往上翻。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
// ====================================================================
// 2. memory_order_release / memory_order_acquire: 经典的"消息传递"模式
// ====================================================================
void demo_release_acquire() {
    section("2. memory_order_release + memory_order_acquire —— 建立跨线程 happens-before");

    constexpr int kRounds = 3000;
    int fail = 0;

    for (int i = 0; i < kRounds; ++i) {
        std::atomic<int> data{0};
        std::atomic<bool> ready{false};

        std::thread producer([&] {
            data.store(42, std::memory_order_relaxed);      // ① 先准备好数据
            ready.store(true, std::memory_order_release);   // ② release: 之前的写对 acquire 端一定可见
        });
        std::thread consumer([&] {
            while (!ready.load(std::memory_order_acquire)) {} // ③ acquire: 与②的 release 同步
            if (data.load(std::memory_order_relaxed) != 42) { // ④ 因为②③配对, 这里必然能看到 42
                ++fail;
            }
        });
        producer.join();
        consumer.join();
    }

    std::cout << kRounds << " 轮 release/acquire 消息传递, 观察到的失败次数 = " << fail
              << " (标准保证恒为 0)\n";
    std::cout << "结论: release-store 与 acquire-load 配对后,\n"
                 "      producer 在 release 之前的所有写入, 对 consumer 在 acquire 之后一定可见。\n"
                 "      这是无锁编程里最常用、最重要的同步模式。\n";
}

输出

1
2
3
4
5
========== 2. memory_order_release + memory_order_acquire —— 建立跨线程 happens-before ==========
3000 轮 release/acquire 消息传递, 观察到的失败次数 = 0 (标准保证恒为 0)
结论: release-store 与 acquire-load 配对后,
      producer 在 release 之前的所有写入, 对 consumer 在 acquire 之后一定可见。
      这是无锁编程里最常用、最重要的同步模式。

4 memory_order_consume(消费语义)

特性: 弱化版的 acquire。它只限制对当前原子变量有数据依赖(Data Dependency)的后续操作不能重排。

现状: 由于过于复杂且编译器难以完美优化,目前标准委员会不建议显式使用它(很多编译器直接将其提升为 acquire)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
// ====================================================================
// 4. memory_order_consume: 基于"数据依赖链"的同步, 比 acquire 更弱、更省
//    (但实践中几乎所有主流编译器都把它当 acquire 实现, C++ 委员会也在讨论修订/弃用)
// ====================================================================
void demo_consume() {
    section("4. memory_order_consume —— 依赖链同步(弱于 acquire, 现代编译器基本等同 acquire 实现)");

    struct Data { int value; };
    Data payload{123};
    std::atomic<Data*> ptr{nullptr};

    std::thread producer([&] {
        payload.value = 123;
        ptr.store(&payload, std::memory_order_release);  // 发布指针
    });
    std::thread consumer([&] {
        Data* p;
        while ((p = ptr.load(std::memory_order_consume)) == nullptr) {}
        // consume 只承诺: 通过 p 这条“数据依赖链”能访问到的内存(如 p->value)是可见的;
        // 它不像 acquire 那样保证“之后所有的读写都不会被重排到前面”,
        // 只保证依赖于 p 的那些访问。理论上比 acquire 更省一次内存屏障(在弱顺序架构上)。
        std::cout << "consumer 通过 consume 读取到 p->value = " << p->value << "\n";
    });
    producer.join();
    consumer.join();

    std::cout << "注意: GCC / Clang 目前都直接把 memory_order_consume 当作\n"
                 "      memory_order_acquire 来实现, 因为精确追踪“依赖链”极其复杂、容易出错。\n"
                 "      日常代码里几乎不需要手写 consume。\n";
}

输出

1
2
3
4
5
========== 4. memory_order_consume —— 依赖链同步(弱于 acquire, 现代编译器基本等同 acquire 实现) ==========
consumer 通过 consume 读取到 p->value = 123
注意: GCC / Clang 目前都直接把 memory_order_consume 当作
      memory_order_acquire 来实现, 因为精确追踪“依赖链”极其复杂、容易出错。
      日常代码里几乎不需要手写 consume。

5 memory_order_acq_rel

特性: 同时具备 acquire 和 release 语义。通常用于 Read-Modify-Write(RMW)操作,如 fetch_add。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
// ====================================================================
// 5. memory_order_acq_rel: 既要 acquire 又要 release 的原子 RMW
//    典型场景: 引用计数递减(最后一个线程需要看到所有其它线程之前的写入)
// ====================================================================
void demo_acq_rel() {
    section("5. memory_order_acq_rel —— RMW 操作同时需要 acquire + release 语义(引用计数)");

    constexpr int kThreads = 8;
    std::atomic<int> refcount{kThreads};
    std::vector<int> payload(kThreads, 0);   // 每个线程只写自己的下标, 无数据竞争
    std::atomic<int> last_one_count{0};

    std::vector<std::thread> ts;
    for (int i = 0; i < kThreads; ++i) {
        ts.emplace_back([&, i] {
            payload[i] = i * 100 + 1;        // ① 各自写入自己的数据(普通写, 不是原子操作)
            // fetch_sub 需要同时具备:
            //   release: 让①对"接下来看到归零的那个线程"可见
            //   acquire: 让本线程能看到"之前所有已经减过引用的线程"写的数据
            if (refcount.fetch_sub(1, std::memory_order_acq_rel) == 1) {
                int sum = 0;
                for (int v : payload) sum += v;   // ② 必须能看到全部 8 个线程写的 payload
                std::cout << "最后归零的线程看到 payload 总和 = " << sum
                          << " (期望 " << [&]{ int s=0; for(int k=0;k<kThreads;++k) s+=k*100+1; return s; }()
                          << ")\n";
                last_one_count.fetch_add(1, std::memory_order_relaxed);
            }
        });
    }
    for (auto& t : ts) t.join();

    std::cout << "判定“自己是最后一个”的线程数 = " << last_one_count.load()
              << " (必须恰好是 1, 否则会重复释放/重复销毁)\n";
    std::cout << "结论: 如果这里只用 memory_order_release(没有 acquire),\n"
                 "      最后归零的线程就不保证能看到其它线程写的 payload —— 这正是\n"
                 "      shared_ptr / boost::intrusive_ptr 等引用计数实现选择 acq_rel 的原因。\n";
}

输出

1
2
3
4
5
6
========== 5. memory_order_acq_rel —— RMW 操作同时需要 acquire + release 语义(引用计数) ==========
最后归零的线程看到 payload 总和 = 2808 (期望 2808)
判定“自己是最后一个”的线程数 = 1 (必须恰好是 1, 否则会重复释放/重复销毁)
结论: 如果这里只用 memory_order_release(没有 acquire),
      最后归零的线程就不保证能看到其它线程写的 payload —— 这正是
      shared_ptr / boost::intrusive_ptr 等引用计数实现选择 acq_rel 的原因。

安全的单例模式

先来看一个DCLP(Double-Checked Locking Pattern),即双重检查锁定标准写法。

class Singleton {
public:
    static Singleton* getInstance() {
        if (instance == nullptr) {      // 第一次检查
            std::lock_guard<std::mutex> lock(mtx);

            if (instance == nullptr) {  // 第二次检查
                instance = new Singleton();
            }
        }
        return instance;
    }

private:
    static Singleton* instance;
    static std::mutex mtx;
};

第一次检查,避免每次都加锁。 第二次检查,防止多个线程同时创建对象。 上面这个代码的问题在于,instance = new Singleton(); 这句话不是一个原子操作。 理想顺序是:allocate -> construct -> publish pointer , 但实际可能发生重排序, allocate -> publish pointer -> construct 。

1
2
3
4
5
6
7
memory = operator new(...)
Singleton(memory)
instance = memory
// 变成
memory = operator new(...)
instance = memory      // 提前发布
Singleton(memory)      // 还在构造

假设两个线程

1
2
3
4
5
6
7
8
9
# Thread A
instance = new Singleton(); // allocate , publich pointer, 
// 构造还没完成

# Thread B
if(instance != nullptr)
{
    return instance; // 此时对象可能构造函数还没执行完,成员变量还没初始化
}

本质的问题,对象地址已可见 ≠ 对象状态已可见。 这就是一个典型的内存模型问题,因为这里缺少happens before关系。 线程A的写到线程B的读之间如果没有同步机制,则不存在happens-before。行为是未定义。

以下的修复代码,利用原子变量,通过

1
2
3
4
5
release
    ↓
synchronizes-with
    ↓
acquire

来保证

1
2
3
构造完成
    happens-before
看到指针
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
#include <atomic>
#include <mutex>

class Singleton {
private:
    static std::atomic<Singleton*> instance;
    static std::mutex mtx;
  
    Singleton() {} // 构造函数

public:
    static Singleton* getInstance() {
        // 1. 第一次读:使用 acquire 确保看到初始化完成后的内存
        Singleton* tmp = instance.load(std::memory_order_acquire);
        if (tmp == nullptr) {
            std::lock_guard<std::mutex> lock(mtx);
            tmp = instance.load(std::memory_order_relaxed);
            if (tmp == nullptr) {
                tmp = new Singleton();
                // 2. 写入:使用 release 确保 new 内部的构造函数执行完毕后,指针才被发布
                instance.store(tmp, std::memory_order_release);
            }
        }
        return tmp;
    }
};

// 初始化静态成员
std::atomic<Singleton*> Singleton::instance{nullptr};
std::mutex Singleton::mtx;

c++ 11之后有更简介的写法

1
2
3
4
5
6
7
8
9
class Singleton {
public:
    static Singleton& getInstance() {
        static Singleton instance;   // C++11 保证: 初始化线程安全, 且只执行一次
        return instance;
    }
private:
    Singleton() {}
};

C++11 标准明确写了这句话(大意):

如果控制流并发地进入一个正在初始化的局部 static 变量声明,并发的执行必须等待该初始化完成。

也就是说,函数局部 static 变量的”线程安全初始化 + 只初始化一次”是语言标准直接保证的,不需要你手写任何原子操作或锁——编译器会替你生成。

总结

分清原子性与可见性: 原子性保证操作“不被打断”,内存模型保证修改“能被别人看见”。

牢记硬件常识: x86 架构是强内存模型(Strong Memory Model),硬件层面上天然保证大部分 Acquire-Release 语义(只存在 Store-Load 乱序)。但 ARM 架构是弱内存模型(Weak Memory Model),乱序非常激进。因此,在 x86 上测试没问题的无锁代码,到了 ARM 上极易因为内存顺序选错而崩掉。

什么是 Happens-Before 和 Synchronizes-With Happens-Before(先行发生): 描述操作之间的可见性关系。如果 A happens-before B,那么 A 的结果对 B 必然可见。注意,它不代表时间上的先后,而是逻辑上的可见性。 Synchronizes-With(同步): 描述跨线程的 Happens-Before 关系。一个线程的 release 操作,与另一个线程对同一原子变量的 acquire 操作,构成 Synchronizes-With 关系。

无锁队列(Lock-free Queue)中,内存顺序怎么选

控制结构(如链表的 head 或 tail 指针)在发布新节点时,必须用 memory_order_release。 消费者线程在读取 head 或 tail 以获取新节点时,必须用 memory_order_acquire,以确保新节点内部的数据已被完全写入。 如果是单纯的并发计数(如队列长度计),使用 memory_order_relaxed 即可。

本文由作者按照 CC BY 4.0 进行授权