23. 原子操作与并发

C11 在标准库中加入了线程和原子操作支持。<threads.h> 提供线程及同步接口，<stdatomic.h> 提供原子类型和原子操作。

1. 本章内容

线程：创建、等待和结束线程。
原子操作：对共享对象进行原子访问。
互斥：用互斥量保护临界区。
条件等待：等待共享条件发生变化。
线程专属存储：让每个线程拥有各自的对象。

2. 先记住

多个线程并发访问同一对象，并且至少有一次访问会修改它时，必须使用标准允许的同步方式；否则可能形成数据竞争，而数据竞争会导致未定义行为。不同实现对 <threads.h> 的支持程度可能不同，使用前还应确认目标实现是否提供这些接口。

3. 习题

本章的五节分别讲线程、原子操作、互斥、条件等待和线程专属存储。下面的四道题从最简单的计数器一路做到条件变量与线程专属存储，每道都要求先跑出错误现象，再动手改，最后用重复运行确认修好了。全部题目用 gcc -std=c17 -Wall -Wextra 编译。MinGW 不带 <threads.h>，请在 Linux 或装了 C11 线程库的工具链上做。

3.1 巩固

#12301

⚡2⏳3

四个线程各把三个计数器加 25000 次，写法各不相同。下面是计数器与工作线程，main 自己补全（初始化互斥量、创建工作线程、等待结束、打印结果）：

static atomic_int a = 0;
static atomic_int b = 0;
static mtx_t lock;
static int c = 0;

static int worker(void *arg) {
    (void)arg;
    for (int i = 0; i < 25000; ++i) {
        a = a + 1;
        b += 1;
        mtx_lock(&lock);
        ++c;
        mtx_unlock(&lock);
    }
    return 0;
}

编译运行，跑五遍，记录三个计数器的值；再回答：

哪个计数器每次都是 100000，哪个不是，为什么。
a = a + 1 与 b += 1 作用在同一个原子对象上，为什么结果不同。
把 c 换成普通 int 去掉互斥量，结果有什么不同，能不能据此说这段代码没竞争。
上面的代码没有检查 thrd_create 与 thrd_join 的返回值。补上检查：创建失败时打印返回的错误码并退出，thrd_join 失败时也要报告。说明为什么这两个函数的返回值不能像 printf 那样忽略。

验收：a 的取值随运行变化且小于 100000，b、c 恒为 100000；补上检查后程序行为不变。

#12302

⚡3⏳3

下面的程序用一个普通 int 作标志，让主线程等待两个工作线程结束：

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

static int counter = 0;
static int ready = 0;

int worker(void *arg) {
    (void)arg;
    for (int i = 0; i < 100000; ++i) {
        counter++;
    }
    ready = 1;
    return 0;
}

int main(void) {
    thrd_t a, b;
    long spins = 0;
    thrd_create(&a, worker, NULL);
    thrd_create(&b, worker, NULL);
    while (!ready) {
        ++spins;
    }
    thrd_join(a, NULL);
    thrd_join(b, NULL);
    printf("counter=%d spins=%ld\n", counter, spins);
    return 0;
}

各跑三遍，记录 counter 与 spins，然后回答：

counter 是否等于 200000，spins 是什么量级。
ready 上的读写构成数据竞争，按标准程序行为未定义。这里能不能指望循环一定反复去读 ready？用 objdump -d 或 gcc -S 看生成的代码，找出循环被改成了什么。
用 _Atomic int ready 加 atomic_store 与 atomic_load 改写，spins 和 counter 应当怎样；实际验证，并说明为什么这时 counter++ 仍然可能丢更新。

验收：能指出 spins 的量级可能远小于预期（本机实测为 0），而 counter 在原子改写版下仍可能小于 200000。偶尔凑巧得到 200000 也不能证明代码正确：counter++ 仍是「读—改—写」三步，只有把 counter 也换成原子类型或用互斥量保护，才谈得上正确。

3.2 编程

#12303

⚡5⏳4

下面的有界缓冲区让一个生产者、两个消费者共用一个条件变量，消费者用 if 判断缓冲是否为空。题面只给出两个线程函数和共享对象，main 与头文件自己补全，TOTAL 为 120，两个消费者各取 60 个：

#define CAP 4
#define TOTAL 120

static mtx_t lock;
static cnd_t cond;
static int buf[CAP];
static size_t head, tail, count;
static long sum;
static int got;

static int producer(void *arg) {
    (void)arg;
    for (int i = 1; i <= TOTAL; ++i) {
        mtx_lock(&lock);
        while (count == CAP) {
            cnd_wait(&cond, &lock);
        }
        buf[tail] = i;
        tail = (tail + 1) % CAP;
        ++count;
        cnd_signal(&cond);
        mtx_unlock(&lock);
    }
    return 0;
}

static int consumer(void *arg) {
    (void)arg;
    for (int i = 0; i < TOTAL / 2; ++i) {
        mtx_lock(&lock);
        if (count == 0) {
            cnd_wait(&cond, &lock);
        }
        int v = buf[head];
        head = (head + 1) % CAP;
        --count;
        sum += v;
        ++got;
        cnd_signal(&cond);
        mtx_unlock(&lock);
    }
    return 0;
}

要求：

先运行五遍，记录 got、sum 和 count。正确结果是 got=120、sum=7260（1 到 120 之和）、count=0。实测五遍都正常结束，但 sum 每次都不同，且都明显小于 7260。解释原因：cnd_wait 返回后消费者不再检查条件，被唤醒时数据可能已经被另一个消费者取走，它读到的 buf[head] 是上一次留下的旧值。于是有些数被读了两遍，有些数从未被读。count 声明为 size_t，条件不成立时继续递减会绕回一个极大的值，缓冲区满与空的判断随之失效，这一点也一并说明。
把消费者的 if 改成 while，再跑五遍，结果都是正确的。这个改法在本例中掩盖了问题，但不能作为结论：一个条件变量上同时排着等空位和等数据的线程，cnd_signal 只叫醒其中一个，被叫醒的那一类如果条件不成立会重新睡下，通知就此丢失。把缓冲区容量改成 1（CAP 改为 1，其余不动），分别用 signal 和 broadcast 各跑五遍，观察前者会卡死（用 timeout 5 运行，退出码 124），后者正常结束且总和为 7260。
说明上面卡死是怎么形成的，再改成两个条件变量（一个表示非空、一个表示非满）的写法，用 signal 跑 20 遍，确认每次都能结束且总和正确。
用 -fsanitize=thread 编译修复后的版本，确认没有报告。若本机 TSan 与 C11 线程库不兼容，改用 glibc 的 pthread 接口重写同一段逻辑再测，并在报告里说明。

验收：第 1 步的 sum 多次运行互不相同；第 2 步中 signal 版在 CAP=1 时稳定卡死，broadcast 版正常；第 3 步连续运行 20 次，got=120、sum=7260、count=0 恒定不变。

3.3 综合与挑战

#12304

⚡6⏳4

下面的程序有四处缺陷，症状是大多数运行都卡死不退出，偶尔退出时 done 也远小于 200000。逐一定位并修复：

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

#define NTHREADS 4
#define PER 50000

static mtx_t lock;
static cnd_t cond;
static tss_t key;
static int ready = 0;
static long done = 0;
static atomic_int freed = 0;

static void cleanup(void *p) {
    free(p);
    atomic_fetch_add(&freed, 1);
}

static int worker(void *arg) {
    int id = *(int *)arg;
    int *slot = malloc(sizeof *slot);
    if (slot == NULL) {
        return 1;
    }
    *slot = id;
    tss_set(key, slot);

    mtx_lock(&lock);
    if (!ready) {
        cnd_wait(&cond, &lock);
    }
    mtx_unlock(&lock);

    for (int i = 0; i < PER; ++i) {
        done += 1;
    }
    return 0;
}

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

    mtx_init(&lock, mtx_plain);
    cnd_init(&cond);
    tss_create(&key, cleanup);

    for (int i = 0; i < NTHREADS; ++i) {
        ids[i] = i;
        thrd_create(&t[i], worker, &ids[i]);
    }

    mtx_lock(&lock);
    ready = 1;
    cnd_signal(&cond);
    mtx_unlock(&lock);

    tss_delete(key);

    for (int i = 0; i < NTHREADS; ++i) {
        thrd_join(t[i], NULL);
    }

    printf("done = %ld, freed = %d\n", done, atomic_load(&freed));
    return 0;
}

要求：

用 timeout 5 跑五遍，记录退出码和输出。rc=124 表示卡死。
找出四处缺陷，各写一句：哪一行、属于哪一类（条件判断、唤醒方式、计数同步、TSS 生命周期），以及它单独存在时会造成什么现象。
逐个修复并单独验证。只改唤醒方式为 cnd_broadcast 时程序不再卡死，但 done 仍明显小于 200000，且 freed 为 0；只改 if 为 while 时仍然会卡死。用这两个中间结果说明缺陷之间是怎样互相掩盖的。
全部修复后连续运行 20 次，输出恒为 done = 200000, freed = 4。

验收：连续 20 次运行退出码为 0，两个数字都不变；在报告里给出四处缺陷的行号与类型。

3.10 求值

4.3 选择语句

5.5 跳转语句

10.2 结构体

14.3 程序支持

15.1 声明

16.2 对象表示

18.2 标准库中的宏

23. 原子操作与并发 ​

1. 本章内容 ​

2. 先记住 ​

3. 习题 ​

3.1 巩固 ​

3.2 编程 ​

3.3 综合与挑战 ​