23.2 原子对象和原子操作

原子对象和原子操作是 C11 标准引入的重要特性，主要用于多线程编程中的同步操作，可以保证指定的共享对象操作具有原子性，避免数据竞争。

1. 原子类型的作用

多个线程同时读写同一个普通对象而不加同步，会构成数据竞争，行为未定义（见第 23 章）。对原子对象的访问是原子操作，不构成数据竞争：其他线程只能看到访问之前或之后的值，而不会看到写到一半的值。C11 在 <stdatomic.h> 头文件中定义了原子类型，并提供了一组配套的原子操作函数。

原子类型适合计数器、状态标志这类只涉及单个对象的同步。原子类型只保证对单个对象的每次访问是原子的；若需要同时更新多个对象并保持它们之间的一致关系，应使用互斥量（见 23.3）。

2. 原子类型

最常用的原子类型包括：

atomic_bool              // bool 的原子版本
atomic_char              // char 的原子版本
atomic_int               // int 的原子版本
atomic_uint              // unsigned int 的原子版本
atomic_long              // long 的原子版本
atomic_ulong             // unsigned long 的原子版本
atomic_size_t            // size_t 的原子版本

也可以使用 _Atomic 关键字得到其他类型的原子版本：

_Atomic int counter;     // 声明原子整型对象

atomic_int 这类名字是标准库提供的类型别名，例如 atomic_int 就是 _Atomic int。两种写法得到相同的类型，选择哪种取决于代码风格。

3. 原子操作

3.1 基本原子操作

注：以下函数是为所有的原子类型 A 定义的泛型函数：类型 C 为 A 对应的非原子类型。

原子存储和加载：

void atomic_store(volatile A* obj, C desired);    // 原子存储（原子写）
C atomic_load(const volatile A* obj);             // 原子加载（原子读）

上面两个函数默认以 memory_order_seq_cst 内存序执行读写操作。

这两个函数有对应的 _explicit 版本，可以指定内存序：

void atomic_store_explicit(volatile A* obj, C desired, memory_order order);
C atomic_load_explicit(const volatile A* obj, memory_order order);

atomic_load_explicit 指定的内存序只能是 memory_order_acquire、memory_order_relaxed、memory_order_consume 或 memory_order_seq_cst；
atomic_store_explicit 指定的内存序只能是 memory_order_release、memory_order_relaxed 或 memory_order_seq_cst。

示例：

atomic_int counter = 0;
atomic_store_explicit(&counter, 1, memory_order_release);
int value = atomic_load_explicit(&counter, memory_order_acquire);

原子交换：

C atomic_exchange(volatile A* obj, C desired);  // 原子交换

原子比较与交换：

bool atomic_compare_exchange_strong(volatile A* obj,
                                    C* expected,
                                    C desired);
bool atomic_compare_exchange_weak(volatile A* obj,
                                 C* expected,
                                 C desired);
bool atomic_compare_exchange_weak_explicit(volatile A* obj,
                                          C* expected,
                                          C desired,
                                          memory_order success,
                                          memory_order failure);
bool atomic_compare_exchange_strong_explicit(volatile A* obj,
                                            C* expected,
                                            C desired,
                                            memory_order success,
                                            memory_order failure);

3.2 原子算术操作

注：以下函数是为所有的原子类型 A 定义的泛型函数：类型 C 为 A 对应的非原子类型；如果 A 为原子整数类型，则类型 M 为 A 对应的非原子类型；如果 A 为原子指针类型，则类型 M 为 ptrdiff_t。

// 原子加法
C atomic_fetch_add(volatile A* obj, M arg);

// 原子减法
C atomic_fetch_sub(volatile A* obj, M arg);

// 原子与运算
C atomic_fetch_and(volatile A* obj, M arg);

// 原子或运算
C atomic_fetch_or(volatile A* obj, M arg);

// 原子异或运算
C atomic_fetch_xor(volatile A* obj, M arg);

4. 内存序

原子操作可以指定内存序（memory order）来控制操作的顺序：

enum memory_order {
    memory_order_relaxed,     // 最宽松的内存序
    memory_order_consume,     // 消费序
    memory_order_acquire,     // 获取序
    memory_order_release,     // 释放序
    memory_order_acq_rel,     // 获取释放序
    memory_order_seq_cst      // 顺序一致性（默认）
};

示例：

atomic_int counter = 0;

// 使用指定的内存序进行原子操作
atomic_store_explicit(&counter, 1, memory_order_release);
int value = atomic_load_explicit(&counter, memory_order_acquire);

5. 使用示例

一个使用原子操作的简单计数器示例：

#include <stdatomic.h>
#include <stdio.h>
#include <threads.h>

atomic_int counter = 0;

int thread_func(void* arg) {
    for (int i = 0; i < 1000000; i++) {
        atomic_fetch_add(&counter, 1);
    }
    return 0;
}

int main() {
    thrd_t th1, th2;

    // 创建两个线程
    thrd_create(&th1, thread_func, NULL);
    thrd_create(&th2, thread_func, NULL);

    // 等待线程结束
    thrd_join(th1, NULL);
    thrd_join(th2, NULL);

    printf("Final counter value: %d\n", atomic_load(&counter));
    return 0;
}

两个线程均创建成功并完成循环时，输出为：

bash

Final counter value: 2000000

6. 习题

本页的题目从原子操作的返回值一直做到自旋锁和内存序的代码生成。全部题目在 Linux 上用 gcc -std=c17 -Wall -Wextra 编译，用 timeout 限制运行时间。

6.1 巩固

#12309

⚡2⏳2

下面的程序依次调用几个原子操作，并把返回值保存下来：

#include <stdatomic.h>
#include <stdio.h>

int main(void) {
    atomic_int n = 0;

    int a = atomic_fetch_add(&n, 5);
    int b = atomic_fetch_add(&n, 3);
    int c = atomic_exchange(&n, 100);
    int d = atomic_fetch_sub(&n, 1);

    atomic_uint flags = 0;
    unsigned old1 = atomic_fetch_or(&flags, 0x6u);
    unsigned old2 = atomic_fetch_and(&flags, 0x4u);

    printf("%d %d %d %d | %u %u %u | %d\n", a, b, c, d, old1, old2,
           atomic_load(&flags), atomic_load(&n));
    return 0;
}

要求：

先逐行写出每个对象在执行完该行后的值，再写出 printf 的完整输出，然后编译运行核对。
atomic_fetch_add 返回的是加之前的值还是加之后的值？atomic_exchange 呢？把 atomic_exchange(&n, 100) 换成 atomic_store(&n, 100)，c 还能取到值吗？
用 atomic_fetch_xor 写一个 static void toggle(atomic_uint *p, unsigned mask)，把 p 中 mask 为 1 的位翻转。main 里从 0xF0u 出发连续调用两次 toggle(&v, 0x3Cu)，打印每次的结果和返回值。
对 atomic_int counter，比较 counter++、++counter 与 atomic_fetch_add(&counter, 1) 三者。哪些返回修改前的值，哪些返回修改后的值？若只关心最终计数、不使用返回值，三者可以互换吗？

验收：第 1 步输出为 0 5 8 100 | 0 6 4 | 99；第 2 步 c 处无法从 atomic_store 取得返回值，因为它的返回类型是 void；第 3 步两次调用后 v 回到 0xF0u，返回值依次为 0xF0u 与 0xCCu。

6.2 编程

#12310

⚡3⏳3

8 个线程各把一个二进制位合并进同一个 atomic_uint，主线程用 go 统一发令，让它们同时开始：

#include <stdatomic.h>
#include <stdio.h>
#include <threads.h>

#define N 8

static atomic_uint flags;
static atomic_int go;

static int setter(void *arg) {
    (void)arg;
    while (!atomic_load(&go)) {
        thrd_yield();
    }
    unsigned v = atomic_load(&flags);
    thrd_yield();
    v |= 1u << *(int *)arg;
    atomic_store(&flags, v);
    return 0;
}

int main(void) {
    thrd_t t[N];
    int ids[N];

    for (int i = 0; i < N; ++i) {
        ids[i] = i;
        thrd_create(&t[i], setter, &ids[i]);
    }
    atomic_store(&go, 1);
    for (int i = 0; i < N; ++i) {
        thrd_join(t[i], NULL);
    }
    printf("flags = 0x%02X\n", atomic_load(&flags));
    return 0;
}

要求：

连跑 5 次，记录 flags。8 个线程各置一位，正确结果应当是 0xFF，实际为什么小于它？这个程序里每一次读和每一次写都是原子的，为什么整体仍然出错？
把 setter 里的读—改—写换成一次 atomic_fetch_or(&flags, 1u << id)，再跑 5 次。验收：5 次都是 0xFF。
主线程取走全部标志位：用 atomic_exchange(&flags, 0) 取得旧值并清零。在 thrd_join 之后调用一次，打印取到的值和清零后的 flags。验收：取到 0xFF，之后 flags = 0。为什么这里用 atomic_exchange 而不是先 atomic_load 再 atomic_store？
用同样的问题检验 atomic_fetch_and 与 atomic_fetch_xor：分别写一个「清除若干位」和「翻转若干位」的函数，各用 8 个线程并发调用，验证结果与单线程一致。

验收：第 1 步多次运行得到不同的值（本机 5 次在 0x81 到 0xC0 之间）；第 2 步恒为 0xFF；第 3 步取到 0xFF，flags 变为 0。

#12311

⚡4⏳4

下面的四个线程各把 counter 加 100000 次，bump 想用比较交换完成加法，但结果只有十万量级：

#include <stdatomic.h>
#include <stdio.h>
#include <threads.h>

static atomic_int counter;

static void bump(atomic_int *p) {
    int expected = atomic_load(p);
    int desired = expected + 1;
    while (!atomic_compare_exchange_weak(p, &expected, desired)) {
    }
}

static int worker(void *arg) {
    (void)arg;
    for (int i = 0; i < 100000; ++i) {
        bump(&counter);
    }
    return 0;
}

int main(void) {
    thrd_t t[4];

    for (int i = 0; i < 4; ++i) {
        thrd_create(&t[i], worker, NULL);
    }
    for (int i = 0; i < 4; ++i) {
        thrd_join(t[i], NULL);
    }
    printf("counter = %d (want 400000)\n", atomic_load(&counter));
    return 0;
}

(1) [4.4] 连跑 5 次，记录 counter。本机测得 11.7 万到 16.4 万之间。找出 bump 的缺陷：交换失败时 atomic_compare_exchange_weak 会把内存中的当前值写进 expected，但这里没有据此重新计算 desired。说明失败一次之后，这个循环会做什么。只改一处，使 4 个线程的结果恒为 400000。

(2) [4.3] atomic_compare_exchange_strong 与 _weak 的区别在于后者可能「伪失败」：值相等时也会返回假。写出两种写法的循环，说明为什么 _weak 必须放在循环里，而 _strong 可以只调用一次。用 4 个线程各加 100000 次，比较两者的运行时间。

(3) [4.5] 用比较交换实现 static int atomic_fetch_max(atomic_int *p, int value)：若 *p < value 就把它改成 value，返回修改前的值。用 {3, 9, 4, 12, 7, 12, 1} 依次调用，验收输出为

text

put  3 -> prev  0, now  3
put  9 -> prev  3, now  9
put  4 -> prev  9, now  9
put 12 -> prev  9, now 12
put  7 -> prev 12, now 12
put 12 -> prev 12, now 12
put  1 -> prev 12, now 12

(4) [5.4*] 用 4 个线程各插入 50000 个随机数调用 atomic_fetch_max，最后打印最大值并与单线程结果核对。再把同一个函数改成求最小值 atomic_fetch_min，初值取 0，运行后发现最小值恒为 0。说明初值在「比较并替换」里扮演什么角色，求最小值时应当怎样选初值。

验收：第 1 步每次结果都明显小于 400000 且各不相同，改正后恒为 400000；第 3 步输出与上面七行一致；第 4 步能说明 0 比所有随机数都小，替换条件永远不成立。

6.3 综合与挑战

#12312

⚡5⏳4

内存序决定编译器能生成什么样的指令，这一点可以直接从汇编里看出来。下面两个线程通过 ready 传递数据：

#include <stdatomic.h>
#include <stdio.h>
#include <threads.h>

typedef struct {
    int value;
    int pad[15];
} payload_t;

static payload_t payload;
static atomic_int ready;

static int writer(void *arg) {
    (void)arg;
    payload.value = 42;
    atomic_store_explicit(&ready, 1, memory_order_release);
    return 0;
}

static int reader(void *arg) {
    (void)arg;
    while (atomic_load_explicit(&ready, memory_order_acquire) == 0) {
    }
    printf("payload.value = %d\n", payload.value);
    return 0;
}

int main(void) {
    thrd_t w, r;

    thrd_create(&w, writer, NULL);
    thrd_create(&r, reader, NULL);
    thrd_join(w, NULL);
    thrd_join(r, NULL);
    return 0;
}

要求：

先试着写错：把 writer 的存储写成 memory_order_acquire，把 reader 的加载写成 memory_order_release。用 -Wall -Wextra 编译，把编译器给出的警告抄下来，并写出两类操作各自允许的内存序。
把程序改回正确版本，再用 -S 生成汇编，找出 writer 里对 ready 的写。然后把 atomic_store_explicit(&ready, 1, memory_order_release) 换成 atomic_store(&ready, 1)（默认 memory_order_seq_cst），再生成一次汇编，比较两处指令的差别。本机 GCC 上前者是一条普通 movl，后者是 xchgl。
为什么 memory_order_seq_cst 在 x86-64 上要写出一条带锁效果的指令，而 release 只是普通存储？结合这个结果说明：x86 上把内存序全写成 relaxed 的程序往往也能跑对，但不能据此认为程序正确。
把 payload.value 与 ready 之间的存储顺序倒过来（先置 ready，再写 payload.value），运行 20 次。本机能观察到错误值吗？说明这类缺陷为什么难以靠运行发现。

验收：第 1 步警告中出现 invalid memory model 'memory_order_acquire' for '__atomic_store_4' 与 valid models are 'memory_order_relaxed', 'memory_order_seq_cst', 'memory_order_release'（加载端同理，合法值为 relaxed、seq_cst、acquire、consume）；第 2 步两段汇编只在 ready 的写指令上不同；第 4 步本机 20 次运行都打印 42。

#12313

⚡6⏳5

<stdatomic.h> 还提供一个专门用于标志位的类型 atomic_flag：它保证无锁，只有 atomic_flag_test_and_set、atomic_flag_clear 两个操作，初始值必须写成 ATOMIC_FLAG_INIT。

(1) [4.3] 用 atomic_flag 实现一个自旋锁：

static atomic_flag lock = ATOMIC_FLAG_INIT;

static void lock_acquire(void) {
    while (atomic_flag_test_and_set_explicit(&lock, memory_order_acquire)) {
        thrd_yield();
    }
}

static void lock_release(void) {
    atomic_flag_clear_explicit(&lock, memory_order_release);
}

4 个线程各用这把锁保护 ++counter 共 200000 次。验收：counter = 800000。把 lock_release 里的 memory_order_release 改成 memory_order_relaxed，再跑，结果仍可能是 800000。说明为什么在 x86 上观察不到差别，以及这样的代码是否仍然正确。

(2) [4.4] 把自旋锁与互斥量做一次对比：同样的 4×200000 次加一，一个版本用 mtx_t，一个用上面这把自旋锁但不调用 thrd_yield（纯忙等），一个用带 thrd_yield 的自旋锁。用 clock() 测 CPU 时间、timespec_get 测墙钟时间，各跑 3 次。本机结果：mtx_t 墙钟 84 到 90 ms、CPU 294 到 318 ms；纯忙等自旋墙钟 167 到 193 ms、CPU 539 到 663 ms；带 yield 自旋墙钟 28 到 51 ms、CPU 94 到 199 ms。说明三者的差别分别来自哪里，以及自旋锁适合多长的临界区。

(3) [5.4] 用 atomic_int + atomic_compare_exchange_weak_explicit 实现同样的自旋锁，expected 必须在每次重试前重新置 0。验收：counter = 800000。把循环里重置 expected 的那一行删掉，再跑 3 次，记录 counter（本机约 35 万），说明少了这一行之后循环的行为。

(4) [5.5] 把计数器换成两个有关系的对象：4 个写线程各自执行 atomic_fetch_add(&x, 1); atomic_fetch_add(&y, 1);，一个读线程连续读取 x、y 并统计 x != y 的次数。另写一个用互斥量保护 ++x; ++y; 的版本，读线程在同一把锁下读取。两者各跑一次，记录不等的次数。本机前者约 18 万次，后者恒为 0。说明为什么把每个对象单独声明为原子类型不能代替锁。

(5) [6.5**] atomic_is_lock_free(&obj) 与宏 ATOMIC_INT_LOCK_FREE 用来判断某个原子类型是否真的无锁。下面的程序查询 atomic_int、atomic_llong、atomic_flag 与一个 16 字节结构体：

#include <stdatomic.h>
#include <stdio.h>

struct big {
    char data[16];
};

int main(void) {
    atomic_int ai = 0;
    atomic_llong al = 0;
    atomic_flag af = ATOMIC_FLAG_INIT;
    _Atomic struct big ab;

    printf("atomic_int lock free = %d (macro %d)\n",
           atomic_is_lock_free(&ai), ATOMIC_INT_LOCK_FREE);
    printf("atomic_llong lock free = %d (macro %d)\n",
           atomic_is_lock_free(&al), ATOMIC_LLONG_LOCK_FREE);
    printf("atomic_flag lock free = %d\n", atomic_is_lock_free(&af));
    printf("struct big lock free = %d\n", atomic_is_lock_free(&ab));
    return 0;
}

直接编译会在链接阶段报 undefined reference to '__atomic_is_lock_free'。加上 -latomic 后，本机输出为 1 (macro 2)、1 (macro 2)、1、0。说明为什么需要额外链接一个库，宏的取值 2 与 atomic_is_lock_free 返回的 1 分别表示什么，以及为什么 16 字节的结构体不是无锁的。

验收：第 1 步 800000；第 2 步三组数据能给出各自的解释；第 3 步删行后明显小于 800000 且每次不同；第 4 步原子版本不为 0、互斥量版本恒为 0；第 5 步能说明 2 表示「总是无锁」、0 表示「总是使用锁」、1 表示「运行期判断」，atomic_is_lock_free 对 atomic_int 返回 1、对 16 字节结构体返回 0。

3.10 求值

4.3 选择语句

5.5 跳转语句

10.2 结构体

14.3 程序支持

15.1 声明

16.2 对象表示

18.2 标准库中的宏

23.2 原子对象和原子操作 ​

1. 原子类型的作用 ​

2. 原子类型 ​

3. 原子操作 ​

3.1 基本原子操作 ​

3.2 原子算术操作 ​

4. 内存序 ​

5. 使用示例 ​

6. 习题 ​

6.1 巩固 ​

6.2 编程 ​

6.3 综合与挑战 ​