一句话 + 关键问题

上一章我们亲手造出了锁；这一章换个角度：拿着锁去用。给数据结构加上锁、让多个线程可以安全地使用它，这叫让它变成线程安全（thread safe）的。锁加在哪里、加几把，同时决定了正确性和性能。

本章依次看四种常见结构：计数器、链表、队列、哈希表。每一种都先用"一把大锁"得到正确版本，再看怎样提升并发度——以及什么时候不值得提升。下一章（条件变量）会在本章的队列基础上，造出能"队列空就等待"的有界队列。

核心概念速览

并发计数器：从一把锁开始

计数器是最简单的数据结构。先看没有任何同步的版本（Figure 29.1）：

typedef struct __counter_t {
    int value;
} counter_t;

void init(counter_t *c)      { c->value = 0; }
void increment(counter_t *c) { c->value++; }   // 读-改-写，三条指令，会竞争！
void decrement(counter_t *c) { c->value--; }
int  get(counter_t *c)       { return c->value; }

c->value++ 正是第 26 章那个 counter = counter + 1：一次读、一次加、一次写，两个线程交错执行就会丢更新。修法同样直接——加一把锁（Figure 29.2）：

typedef struct __counter_t {
    int             value;
    pthread_mutex_t lock;         // ① 锁和数据放在一起
} counter_t;

void init(counter_t *c) {
    c->value = 0;
    Pthread_mutex_init(&c->lock, NULL);
}
void increment(counter_t *c) {
    Pthread_mutex_lock(&c->lock);    // ② 进入操作时加锁
    c->value++;
    Pthread_mutex_unlock(&c->lock);  // ③ 离开时解锁
}
int get(counter_t *c) {
    Pthread_mutex_lock(&c->lock);    // ④ 连读也要加锁，保证读到一致的值
    int rc = c->value;
    Pthread_mutex_unlock(&c->lock);
    return rc;
}

这是最基本的并发数据结构设计模式：调用操作时获取一把锁，返回时释放。它和管程（monitor）很像——管程在调用和返回对象方法时自动加锁和解锁。

它是正确的。问题在于性能：原书在一台 4 核 iMac 上让每个线程把计数器加 100 万次，1 个线程大约 0.03 秒，2 个线程并发却要 5 秒以上，线程越多越糟。原因见下图——所有线程都挤在同一把锁上，临界区被完全串行化了；更糟的是，锁和 value 所在的缓存行在各个 CPU 之间来回搬运。

可扩展的计数：近似计数器

研究者为"可扩展计数"下了多年功夫，而且它真的重要：对 Linux 在多核上的性能分析发现，没有可扩展的计数器，一些负载会遭遇严重的扩展性问题。本章介绍其中一种方法：近似计数器（approximate counter），Linux 社区也叫它 sloppy counter。

工作方式：

1. 运行在某个核上的线程要加计数时，只加本核的局部计数器，用对应的局部锁同步。各 CPU 的局部计数器互不相干，更新没有争用，因此可扩展。
2. 为了让全局计数器保持"大致最新"（万一有线程要读），局部值会周期性地转移到全局：拿全局锁，把局部值加进全局，然后把局部计数器清零。
3. 多久转移一次由阈值 S 决定。S 越小，越像前面那个不可扩展的计数器；S 越大，越可扩展，但全局值可能偏离真实计数越远。

为什么还需要局部锁？原书脚注解释：因为一个核上可能运行不止一个线程。如果保证每核只有一个线程，局部锁就可以省掉。想得到精确值也可以：按固定顺序（避免死锁）拿齐所有局部锁和全局锁再求和——但这就不可扩展了。

原书追踪表（Figure 29.3）

S = 5，4 个 CPU 上的线程各自更新 L1…L4，时间向下增长。每一步某些局部计数器 +1；某个局部值达到 S 时，就转移到全局 G 并清零。

注意第 7 步结束时：真实计数是 0+2+4+0+10 = 16，而 get() 只会返回 G = 10——滞后了 6。下面的模拟器可以逐步复现这张表，并让你改 CPU 数和 S，观察滞后怎样随 S 变化。

把模拟器里 S 调大调小，你会看到原书说的这个准确度/性能权衡：

• S 很小（极端 S = 1）：每次更新都转移，全局锁获取次数 = 更新次数，和一把锁的精确计数器一样争用激烈，但 G 永远准确。
• S 很大：全局锁几乎不被碰到，性能极好；但 G 落后得多。原书说滞后"至多为 CPU 数乘以 S"。更精确地说，每个局部计数器在两次转移之间最多攒到 S−1（攒到 S 立即转移），所以滞后上界是 CPU 数 × (S − 1)。

下面是原书的实现（Figure 29.4），关键行已标注：

typedef struct __counter_t {
    int             global;           // 全局计数
    pthread_mutex_t glock;            // 全局锁
    int             local[NUMCPUS];   // 每 CPU 局部计数
    pthread_mutex_t llock[NUMCPUS];   // ……和各自的锁
    int             threshold;        // 转移频率 S
} counter_t;

void update(counter_t *c, int threadID, int amt) {
    int cpu = threadID % NUMCPUS;              // ① 线程 → 哪个局部计数器
    pthread_mutex_lock(&c->llock[cpu]);       // ② 只拿局部锁：通常无人争用
    c->local[cpu] += amt;
    if (c->local[cpu] >= c->threshold) {       // ③ 攒够了：
        pthread_mutex_lock(&c->glock);         //    才去拿全局锁
        c->global += c->local[cpu];
        pthread_mutex_unlock(&c->glock);
        c->local[cpu] = 0;                     //    局部清零（假设 amt > 0）
    }
    pthread_mutex_unlock(&c->llock[cpu]);
}

int get(counter_t *c) {
    pthread_mutex_lock(&c->glock);
    int val = c->global;                       // ④ 只读全局值
    pthread_mutex_unlock(&c->glock);
    return val;                                // 只是近似值！
}

原书的测量（Figure 29.5）：S = 1024 时，4 个线程在 4 个处理器上共更新 400 万次，耗时和 1 个线程在 1 个处理器上更新 100 万次相差无几；而精确计数器随线程数急剧变慢。Figure 29.6 则固定 4 线程、改变 S：S 小时性能差（但全局值总是很准），S 大时性能极好（但全局值有滞后）。本页"动手实验"一节会在本机重做这两个实验。

并发链表

先只关注插入和查找（删除留给读者思考）。最基本的做法仍然是一把大锁：插入入口加锁、出口解锁（Figure 29.7）：

// Figure 29.7：最初版本
int List_Insert(list_t *L, int key) {
    pthread_mutex_lock(&L->lock);
    node_t *new = malloc(sizeof(node_t));
    if (new == NULL) {
        perror("malloc");
        pthread_mutex_unlock(&L->lock);  // 失败路径也得记得解锁
        return -1;
    }
    new->key  = key;
    new->next = L->head;
    L->head   = new;
    pthread_mutex_unlock(&L->lock);
    return 0;
}
int List_Lookup(list_t *L, int key) {
    pthread_mutex_lock(&L->lock);
    node_t *curr = L->head;
    while (curr) {
        if (curr->key == key) {
            pthread_mutex_unlock(&L->lock); // 第二个出口
            return 0;
        }
        curr = curr->next;
    }
    pthread_mutex_unlock(&L->lock);
    return -1;
}

// Figure 29.8：重写后
int List_Insert(list_t *L, int key) {
    // 不需要同步：malloc 本身线程安全
    node_t *new = malloc(sizeof(node_t));
    if (new == NULL) {
        perror("malloc");
        return -1;          // 此时还没拿锁，直接返回
    }
    new->key = key;
    pthread_mutex_lock(&L->lock);    // 只锁真正的临界区
    new->next = L->head;
    L->head   = new;
    pthread_mutex_unlock(&L->lock);
    return 0;
}
int List_Lookup(list_t *L, int key) {
    int rv = -1;
    pthread_mutex_lock(&L->lock);
    node_t *curr = L->head;
    while (curr) {
        if (curr->key == key) {
            rv = 0; break;   // 跳出循环，走唯一出口
        }
        curr = curr->next;
    }
    pthread_mutex_unlock(&L->lock);
    return rv;
}

第一版有个小麻烦：malloc() 失败时，必须在返回前记得解锁。这种罕见的异常控制流非常容易出错——一项对 Linux 内核补丁的研究发现，近 40% 的 bug 出现在这类很少走到的代码路径上。

重写的思路有两点：

• 插入：只锁真正的临界区。假设 malloc() 本身是线程安全的，分配和填充新节点就不需要锁；只有修改共享链表（改 head）时才需要持锁。于是失败路径上根本没有锁要放。
• 查找：单一出口。找到后用 break 跳出循环，统一走到函数末尾解锁并返回。加锁/解锁点越少，越不容易"忘了解锁就 return"。

让链表扩展：手递手锁

一把锁的链表同样扩展性差。研究者探索过一种提高并发度的技术：手递手锁（hand-over-hand locking），也叫锁耦合（lock coupling）。思路很简单：不再给整个链表一把锁，而是每个节点一把锁；遍历时，先拿下一个节点的锁，再放当前节点的锁——像双手交替抓绳子往前爬。

概念上这很合理，它允许链表操作有很高的并发度。但原书指出：实践中很难让它比一把锁更快——遍历时对每个节点都要加锁、解锁，这些开销大得惊人。即使链表很长、线程很多，允许多个遍历同时进行带来的并发收益，也不太可能超过"拿一把锁、做完、放掉"的简单做法。也许某种混合方案（每隔若干个节点才换一把锁）值得研究。下面的模拟器让你调"加锁代价"和"每个节点的工作量"，亲眼看看手递手什么时候赢、什么时候输。

并发队列：Michael & Scott 双锁队列

给队列加一把大锁的做法留给你自己想。这里看一个并发度稍高的设计，来自 Michael 和 Scott（Figure 29.9）。它有两把锁：一把管队头（head_lock），一把管队尾（tail_lock）。目的是让入队和出队可以并发：通常情况下，入队只碰尾锁，出队只碰头锁。

typedef struct __node_t { int value; struct __node_t *next; } node_t;
typedef struct __queue_t {
    node_t         *head, *tail;
    pthread_mutex_t head_lock, tail_lock;     // ① 头尾各一把锁
} queue_t;

void Queue_Init(queue_t *q) {
    node_t *tmp = malloc(sizeof(node_t));    // ② 哑节点（dummy）
    tmp->next = NULL;
    q->head = q->tail = tmp;
    pthread_mutex_init(&q->head_lock, NULL);
    pthread_mutex_init(&q->tail_lock, NULL);
}

void Queue_Enqueue(queue_t *q, int value) {
    node_t *tmp = malloc(sizeof(node_t));    // ③ 锁外准备新节点
    assert(tmp != NULL);
    tmp->value = value;
    tmp->next  = NULL;
    pthread_mutex_lock(&q->tail_lock);       // ④ 只拿尾锁
    q->tail->next = tmp;
    q->tail       = tmp;
    pthread_mutex_unlock(&q->tail_lock);
}

int Queue_Dequeue(queue_t *q, int *value) {
    pthread_mutex_lock(&q->head_lock);       // ⑤ 只拿头锁
    node_t *tmp      = q->head;              //    head 永远指向哑节点
    node_t *new_head = tmp->next;
    if (new_head == NULL) {
        pthread_mutex_unlock(&q->head_lock);
        return -1;                           //    只有哑节点 = 队列空
    }
    *value  = new_head->value;               // ⑥ 真正的队首是哑节点的下一个
    q->head = new_head;                      //    它变成新的哑节点
    pthread_mutex_unlock(&q->head_lock);
    free(tmp);                               // ⑦ 旧哑节点在锁外释放
    return 0;
}

哑节点（dummy node）是 Michael 和 Scott 的关键技巧：它在初始化时分配，使头部操作和尾部操作得以分离。如果没有哑节点，空队列时 head = tail = NULL，第一次入队必须同时改 head 和 tail、最后一次出队也必须同时改两者——两个操作就又得碰两把锁了。有了哑节点，队列里至少永远有一个节点，入队只改 tail 和 tail->next，出队只改 head。

原书也提醒：只用锁的队列往往不能完全满足多线程程序的需要。更完整的有界队列——队列空或满时能让线程等待——是下一章条件变量的重点。

并发哈希表

最后是一个简单又非常实用的结构：不需要扩容的哈希表（Figure 29.10）。它直接用前面的并发链表搭成：

#define BUCKETS (101)
typedef struct __hash_t { list_t lists[BUCKETS]; } hash_t;   // 每个桶是一条带锁的链表

void Hash_Init(hash_t *H) {
    for (int i = 0; i < BUCKETS; i++) List_Init(&H->lists[i]);
}
int Hash_Insert(hash_t *H, int key) { return List_Insert(&H->lists[key % BUCKETS], key); }
int Hash_Lookup(hash_t *H, int key) { return List_Lookup(&H->lists[key % BUCKETS], key); }

它性能好的原因就是每个桶一把锁，而不是整个结构一把锁，于是大量并发操作可以同时进行。原书 Figure 29.11 在同一台 4 核 iMac 上测试：4 个线程各做 1 万到 5 万次并发插入，并发哈希表的耗时几乎不随插入量增长（扩展性"极好"），而一把锁的链表耗时则随插入量迅速上升。

先预测再验证

动手实验：精确计数器 vs 近似计数器

cd labs/29-lock-data-structures && make
./counters            # 完整实验，约 2~3 秒
./counters-nopad      # 原书写法：局部计数器不做缓存行填充
./counters approx 4 1000000 1024   # 单独跑一个配置

示例输出（本机：AMD Ryzen 7 8745H，8 核 16 线程，WSL2，gcc 14 -O2；你的机器上数字会不同）：

$ ./counters
NUMCPUS (online) = 16, 每线程 1000000 次递增, 每项取 3 次运行的中位数

== 实验一：线程数扫描（Figure 29.5 风格，近似计数器 S=1024）==
 threads   precise(s)    approx(s)  precise get()   approx get()
       1        0.005        0.007        1000000         999424
       2        0.033        0.007        2000000        1998848
       4        0.064        0.007        4000000        3997696
       8        0.154        0.010        8000000        7995392
      16        0.273        0.017       16000000       15990784

== 实验二：阈值 S 扫描（Figure 29.6 风格，4 线程）==
       S    approx(s)   approx get()          lag
       1        0.096        4000000            0
       2        0.066        4000000            0
       4        0.038        4000000            0
       8        0.012        4000000            0
      16        0.009        4000000            0
      32        0.008        4000000            0
      64        0.009        4000000            0
     128        0.010        3999744          256
     256        0.010        3999744          256
     512        0.009        3999744          256
    1024        0.008        3997696         2304

$ ./counters-nopad
[NOPAD 未填充缓存行] NUMCPUS (online) = 16, 每线程 1000000 次递增, 每项取 3 次运行的中位数

== 实验一：线程数扫描（Figure 29.5 风格，近似计数器 S=1024）==
 threads   precise(s)    approx(s)  precise get()   approx get()
       1        0.005        0.007        1000000         999424
       2        0.030        0.007        2000000        1998848
       4        0.059        0.048        4000000        3997696
       8        0.138        0.050        8000000        7995392
      16        0.257        0.052       16000000       15990784

== 实验二：阈值 S 扫描（Figure 29.6 风格，4 线程）==
       S    approx(s)   approx get()          lag
       1        0.122        4000000            0
       2        0.087        4000000            0
       4        0.077        4000000            0
       8        0.067        4000000            0
      16        0.060        4000000            0
      32        0.056        4000000            0
      64        0.052        4000000            0
     128        0.048        3999744          256
     256        0.053        3999744          256
     512        0.048        3999744          256
    1024        0.049        3997696         2304

怎么读这些数字：

• 精确计数器不扩展。1 线程 0.005 秒；2 线程已是 0.033 秒（约 6.6 倍），16 线程 0.273 秒。趋势和原书一致，但幅度小得多——原书 2 线程超过 5 秒。十几年后的 glibc 互斥锁基于 futex（第 28 章），无争用时只是一条原子指令，争用时的处理也高效得多；CPU 间缓存一致性也更快。但"线程越多越慢"这个本质没变：临界区是串行的，还要为锁和 value 所在的缓存行在核之间反复搬运付出代价。
• 近似计数器接近完美扩展。1、2、4 线程耗时几乎相同（0.007 秒），而总工作量翻了 4 倍。8 线程、16 线程时略有上升：本机只有 8 个物理核，16 个线程意味着两个超线程共用一个核的执行单元；加上线程没有绑核、WSL2 虚拟化的调度噪声，扩展性打了点折扣。
• 1 线程时近似计数器反而稍慢（0.007 vs 0.005）：它每次多做一次数组下标计算和阈值比较。这就是"更并发 ≠ 更快"的一个小例子——没有争用时，简单方案赢。
• S 的权衡：S = 1 时 0.096 秒（每次都抢全局锁，比精确计数器还慢）；S 增到 8 左右耗时就降到底了——本机的全局锁在每 8 次更新才碰一次时就不再是瓶颈（原书那台机器要到更大的 S 才饱和）。代价是滞后：S = 1024 时 get() 少了 2304。
• 滞后为什么正好是 2304？每个线程更新 1,000,000 次，1,000,000 = 976 × 1024 + 576，所以每个局部计数器最后剩 576 没转移，4 个共 4 × 576 = 2304。同理 S = 128 时 1,000,000 mod 128 = 64，滞后 4 × 64 = 256。而 1,000,000 = 2⁶ × 15625，能被 1、2、4……64 整除，所以 S ≤ 64 时恰好没有余数、滞后为 0。无论如何，滞后都不超过 NUMCPUS × (S − 1)。
• NUMCPUS 应该取本机核数。原书 iMac 是 4 核，NUMCPUS = 4；本机有 16 个逻辑 CPU，所以程序取 16。如果 NUMCPUS 比线程数少，多个线程会共享一个局部计数器和局部锁，又产生争用；取多了只是浪费一点内存。更好的实现会按线程实际所在的 CPU（如 sched_getcpu()）选局部计数器，而不是简单地用线程号取模。
• 伪共享（false sharing）。counters-nopad 按原书写法把 local[] 紧挨着放，几个局部计数器落在同一条 64 字节缓存行里。4 线程时耗时从 0.007 秒涨到 0.048 秒，接近 7 倍——虽然逻辑上互不相干，硬件却仍要让这条缓存行在各 CPU 间来回搬运。这是原书没提、但在真实机器上非常重要的一点。

自测题

常见误区与本章小结