内存模型和原子操作的总结

2024-03-07 2025-04-20 约 2500 字预计阅读 5 分钟

内存模型和原子操作的设计目的

通过底层的设计可以让操作系统或 cpu 给用户提供一种操控颗粒度更小的各个线程之间的内存访问同步的功能。

原子操作和互斥量的应用场景

如果仅需要对少量的简单数据进行原子操作，原子锁可能比互斥量更高效。但在涉及到复杂数据结构或需要独占资源访问的情况下，互斥量提供了更强的同步能力，但通常会涉及线程的切换和上下文切换，因此可能会引入一定的性能开销。

内存模型同步模式

原子变量主要用于同步线程之间的共享内存访问。通常，一个线程创建数据，然后存储到原子中。其他线程从此原子读取数据，当看到预期值时，其他线程正在创建的数据将在该线程中完整且可见 (内存可见性)。不同的内存模型模式用于指示线程之间的数据共享联系有多强。

有 3 种模式或模型允许程序员指定线程间同步的类型:

Sequentially Consistent (序列一致)
Relaxed (宽松)
Release-Acquire (释放-获取)
Release-Consume (释放-消费) (是获取-释放内存模型中的进一步微妙改进)

内存可见性

“内存可见性"是指在多线程或多进程环境中，确保对内存的读写操作能够被其他线程或进程正确地感知和同步。

Memory Order 提供四种方式操控可见性:

memory_order_relaxed: 内存不可见性
memory_order_release-memory_order_require: release 之前的内存访问, 在 require 之后是内存可见的。
memory_order_release-memory_order_consume: release 之前对 release 的原子变量有依赖的内存访问, 在 require 之后内存可见的。
memory_order_seq_cst: 有此内存定序的加载操作进行获得操作，存储操作进行释放操作，而读修改写操作进行获得操作和释放操作，再加上存在一个单独全序，其中所有线程以同一顺序观测到所有修改。

cpu 的工作量从低到高:

relaxed
release-consume
release-acquire
seq_cst

指令重排

内存的顺序描述了计算机CPU获取内存的顺序，内存的排序可能静态也可能动态的发生：

静态内存排序：编译器期间，编译器对内存重排。静态内存排序是为了提高代码的利用率和性能，编译器对代码进行了重新排序。
动态内存排序：运行期间，CPU乱序执行。为了优化性能CPU也会进行对指令进行重新排序、延缓执行、各种缓存等等，以便达到更好的执行效果。

总之，程序不会完全按照你原始代码的顺序来执行。

See ref.

Memory Order

memory_order_relaxed:

不限制指令重排 (单个线程的同一原子操作顺序不变)
没有内存可见性 (即不会同步其他数据)

memory_order_release-memory_order_require:

限制指令重排, 不能重排到 release 之后, require 之前。
release 之前的内存访问, 在 require 之后是内存可见的。

memory_order_release-memory_order_consume:

限制指令重排, 不能重排在 release 之后, 依赖于 consume 的值的访问不能重排到 consume 之前。
对 release 的原子变量有依赖的内存访问, 在 consume 之后是内存可见的。
std::kill_dependency 可以调整依赖。

memory_order_acq_cst:

限制指令重排。
加载操作进行获得操作，存储操作进行释放操作，而读修改写操作进行获得操作和释放操作，再加上存在一个单独全序，其中所有线程以同一顺序观测到所有修改。

宽松同步模式

假设 ‘x’ 和 ‘y’ 最初为 0:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
-Thread 1-
y.store (20, memory_order_relaxed)
x.store (10, memory_order_relaxed)

-Thread 2-
if (x.load (memory_order_relaxed) == 10) {
    assert (y.load(memory_order_relaxed) == 20) /* assert A */
    y.store (10, memory_order_relaxed)
}

-Thread 3-
if (y.load (memory_order_relaxed) == 10)
    assert (x.load(memory_order_relaxed) == 10) /* assert B */

上面的例子中, 任一断言实际上都可能失败。虽然根据原子性, 原子操作不能同时发生, 且 store 发生于 load 之前, 则 load 能获得 store 的值。但是因为 x, y, 是不同的原子变量, 且是 Relaxed 同步模式, 所以 x, y 的存储指令重排后顺序是不确定的。同时, Relaxed 同步模式是内存不可见的, 它不会同步其他共享内存数据。

唯一强加的顺序是，一旦在线程 2 中观察到线程 1 中变量的值，线程 2 就无法看到线程 1 中该变量的“较早”值。即，假设“x”最初为 0:

1
2
3
4
5
6
7
8
-Thread 1-
x.store (1, memory_order_relaxed)
x.store (2, memory_order_relaxed)

-Thread 2-
y = x.load (memory_order_relaxed)
z = x.load (memory_order_relaxed)
assert (y <= z)

上面的例子中, 断言不可能失败。一旦线程 2 看到 2 的存储，它就无法再看到值 1。因为线程 1 的指令是操作同一原子变量, 所以指令重排后顺序是不变的。且宽松同步模式, 一旦看到某个原子变量的内存的新值, 就不会再看到它的旧值。单个线程的同一原子操作, 指令重排后顺序不变, 也可以证明这一点。

当程序员只是希望变量本质上是原子的而不是使用它来同步其他共享内存数据的线程时，最常使用宽松模式。

释放-获取同步模式

-Thread 1- y = 20; x.store (10, memory_order_release);

-Thread 2- if (x.load(memory_order_acquire) == 10) assert (y == 20);

1
2
3
4

上面的例子中, 断言不会失败。线程 1 中, Release 会限制 `y = 20` 重排在 Release 之前, 且根据释放-获取的同步模式的内存可见性, Release 之前的内存操作, 对 Acquire 之后是可见的。

### 释放-消费同步模式

-Thread 1- n = 1 m = 1 p.store (&n, memory_order_release)

-Thread 2- t = p.load (memory_order_acquire); assert( *t == 1 && m == 1 );

-Thread 3- t = p.load (memory_order_consume); assert( *t == 1 && m == 1 );

1
2
3
4

线程 2 中的断言将通过, 但是线程 3 中的断言可能会失败。因为到 p 的存储和到 m 的存储之间不再存在依赖关系，因此不需要同步这些值。

### 序列一致同步模式

1
2
3
-Thread 1-       -Thread 2-
y = 1            if (x.load() == 2)
x.store (2);        assert (y == 1)

上面的例子中, 断言不会失败。因为限制指令重排, 对“y”的存储发生在线程 1 中对 x 的存储之前。如果线程 2 中“x”的加载获取了线程 1 中发生的存储结果，则它必须全部看到在存储之前发生的所有操作在线程 1 中，即使是不相关的线程。这意味着优化器无法自由地对线程 1 中的两个存储重新排序，因为线程 2 也必须看到 Y 的存储。

此模式还提供所有线程之间的一致性。此示例中的断言都不会失败（x 和 y 最初均为 0）:

1
2
3
4
5
-Thread 1-       -Thread 2-                   -Thread 3-
y.store (20);    if (x.load() == 10) {        if (y.load() == 10)
x.store (10);      assert (y.load() == 20)      assert (x.load() == 10)
                   y.store (10)
                 }

内存屏障

1
2
3
4
5
6
7
8
9
atom v

thread A {
  F (rel Fence)     // release memory fence
  X (store v)       // store atomic variable `v`
}
thread B {
  Y (acq v)         // acquire atomic variable `v`
}

F 之前的内存访问对 Y 之后是内存可见的。

1
2
3
4
5
6
7
8
9
atom v

thread A {
  X (rel v)
}
thread B {
  Y (load v)
  F (acq Fence)
}

X 之前的内存访问对 F 之后是内存可见的。

1
2
3
4
5
6
thread A {
  FA (rel Fence)
}
thread B {
  FB (acq Fence)
}

FA 之前的内存访问对 FB 之后的内存是可见的。

应用

Relaxed 同步模式常用于计数:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
#include <vector>
#include <iostream>
#include <thread>
#include <atomic>
 
std::atomic<int> cnt = {0};
 
void f() {
    for (int n = 0; n < 1000; ++n)
        cnt.fetch_add(1, std::memory_order_relaxed);        // 运算是读-修改-写的原子操作。只要求原子性，不要求定序或同步。
}
 
int main() {
    std::vector<std::thread> v;
    for (int n = 0; n < 10; ++n)
        v.emplace_back(f);
    for (auto& t : v)
        t.join();
    std::cout << "最终计数器值为 " << cnt << '\n';
}

SpinLock:

 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
struct spinlock {
  std::atomic<bool> lock_ = {0};

  void lock() noexcept {
    for (;;) {
      // Optimistically assume the lock is free on the first try
      if (!lock_.exchange(true, std::memory_order_acquire)) {
        return;
      }
      // Wait for lock to be released without generating cache misses
      while (lock_.load(std::memory_order_relaxed)) {
        // Issue X86 PAUSE or ARM YIELD instruction to reduce contention between
        // hyper-threads
        __builtin_ia32_pause();
      }
    }
  }

  bool try_lock() noexcept {
    // First do a relaxed load to check if lock is free in order to prevent
    // unnecessary cache misses if someone does while(!try_lock())
    return !lock_.load(std::memory_order_relaxed) &&
           !lock_.exchange(true, std::memory_order_acquire);
  }

  void unlock() noexcept {
    lock_.store(false, std::memory_order_release);
  }
};

有尝试加锁次数的 SpinLock:

没有 while acquire, 减少了 cpu 同步的流量。

拓展

mutex/sem 之类的同步原语会使用内存屏障同步的修改。See ref。