计算机系统基础-08：并发

并发编程

并发

并发指多个逻辑控制流在时间上重叠执行。它出现在硬件异常处理、进程、Unix signal handler、线程和网络服务器等场景中。

应用层并发的常见用途：

• 响应异步事件。
• 在多处理器上并行计算。
• 访问慢速 I/O 设备时隐藏等待时间。
• 与用户交互。
• 同时服务多个网络 client。

并发程序常见实现方式：

方式	特点
Process	地址空间隔离，通信需要 IPC
Thread	共享同一进程地址空间，通信方便但容易产生 race
I/O multiplexing	单线程管理多个 I/O 事件，本部分不展开

并发程序的主要难点是需要考虑多种可能的执行交错。常见错误包括 data race、deadlock、livelock 和 starvation。

Iterative Server

Iterative echo server 一次只服务一个 client：

int main(int argc, char **argv)
{
    int listenfd, connfd;
    struct sockaddr_in clientaddr;
    socklen_t clientlen = sizeof(clientaddr);

    /* 创建监听描述符，等待客户端连接。 */
    listenfd = Open_listenfd(argv[1]);
    while (1) {
        /* 迭代式服务器：一次只接受并处理一个连接。 */
        connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);
        echo(connfd);
        /* 当前连接结束后关闭 connfd，再接受下一个连接。 */
        Close(connfd);
    }
}

流程：

1. Server 在 accept 中等待连接。
2. Client 连接成功后，server 调用 echo(connfd)。
3. echo 持续处理该 client，直到 client 关闭连接。
4. Server 才能回到 accept 接收下一个 client。

问题：若 client 1 已连接但长时间不发送数据，server 会阻塞在读取 client 1 的数据上，client 2 即使发起连接也无法被及时服务。

解决思路：使用 concurrent server，为不同 client 创建不同控制流，使多个连接可以重叠处理。

基于 Process 的并发服务器

基于 process 的并发服务器为每个 client fork 一个 child process：

void sigchld_handler(int sig)
{
    /* 循环回收所有已终止子进程，WNOHANG 防止阻塞在 handler 中。 */
    while (waitpid(-1, 0, WNOHANG) > 0)
        ;
}

int main(int argc, char **argv)
{
    int listenfd, connfd;
    struct sockaddr_in clientaddr;
    socklen_t clientlen = sizeof(clientaddr);

    Signal(SIGCHLD, sigchld_handler);
    listenfd = Open_listenfd(argv[1]);

    while (1) {
        connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);
        /* 每个连接 fork 一个子进程处理。 */
        if (Fork() == 0) {
            /* 子进程不再接受新连接，关闭继承来的监听描述符。 */
            Close(listenfd);
            echo(connfd);
            Close(connfd);
            exit(0);
        }
        /* 父进程不处理该连接，关闭自己的 connfd 副本。 */
        Close(connfd);
    }
}

执行流程：

1. Parent process 在 accept 中等待连接。
2. 某个 client 连接后，accept 返回 connfd。
3. Parent 调用 fork 创建 child。
4. Child 关闭自己的 listenfd，使用 connfd 服务该 client。
5. Parent 关闭自己的 connfd，继续回到 accept。

为什么 parent 必须关闭 connfd？

fork 后，parent 和 child 都持有同一个 connected socket 的引用。若 parent 不关闭自己的 connfd，即使 child 结束并关闭 connfd，内核中的 socket 引用计数仍不为 0，连接无法按预期释放。

After fork:
refcnt(connfd) = 2

child close(connfd)  -> refcnt = 1
parent close(connfd) -> refcnt = 0，连接可关闭

三个 Close 是否都必须保留？

Process-based concurrent server 中通常有三个 Close：

if (Fork() == 0) {
    /* 子进程不接受新连接，关闭继承的监听描述符。 */
    Close(listenfd);
    echo(connfd);
    /* 子进程完成服务后关闭自己的连接描述符副本。 */
    Close(connfd);
    exit(0);
}
/* 父进程不处理该连接，关闭自己的 connfd 副本。 */
Close(connfd);

解析：

1. Child 中的 Close(listenfd)：不是功能正确性的必要条件。Child 即使保留 listenfd，只要不调用 accept，就不会接收新连接；并且 child exit 时 OS 会关闭其描述符。但关闭它更清晰，可避免无关 descriptor 留在 child 的 descriptor table 中。
2. Child 中的 Close(connfd)：若后面立即 exit(0)，从功能上也可由 OS 在进程退出时关闭。但显式关闭能表达连接服务结束，也避免未来修改代码时形成泄漏。
3. Parent 中的 Close(connfd)：必须保留。Parent 是 long-running server，不会很快退出；若不关闭自己的 connfd，每次连接都会在 parent descriptor table 中留下一个无用引用，使 descriptor 和内核 socket 资源长期无法释放。

因此，前两个 Close 主要是清理 child 不需要的资源；parent 的 Close(connfd) 是长期运行 server 中避免 resource leak 的关键。

第一次 accept 为什么常返回 connfd = 4？

Unix descriptor 按进程维护，通常从最小空闲编号分配。普通进程启动后，0/1/2 分别是 stdin/stdout/stderr；open_listenfd 创建 listening socket 后通常占用 3。第一次 accept 会创建新的 connected socket，因此常返回最低空闲 descriptor 4。

fork 会复制 parent 的 descriptor table。复制后 parent 和 child 的 listenfd、connfd 指向相同的内核 file/socket object，对应引用计数增加。Close 只删除当前进程 descriptor table 中的一项，并使对应内核对象引用计数减一；引用计数降到 0 时，内核对象才会释放。

为什么需要回收 child？

Child 退出后会成为 zombie。Parent 需要处理 SIGCHLD 并调用 waitpid 回收，否则 zombie 会保留内核资源。

waitpid(-1, 0, WNOHANG) 的含义：

• -1：回收任意 child。
• WNOHANG：没有可回收 child 时立即返回。
• while：一次 signal 可能对应多个 child 退出，需要循环回收。

Process 方案的优缺点：

优点：

• 可同时处理多个连接。
• 地址空间隔离，默认不共享全局变量，错误隔离较清晰。
• 编程模型直接。

缺点：

• Process 创建和回收开销较高。
• 进程间共享数据不方便，需要 IPC，例如 FIFO、System V shared memory 等。
• 若要统计所有 client 的总接收字节数，不能直接用普通全局变量完成。

Thread 模型

传统视角下，process 包含 process context、code、data 和 stack。另一种视角下，process 可看作 thread 加上 code、data 和 kernel context。

一个 process 可以包含多个 thread。每个 thread 有：

• 独立逻辑控制流。
• 独立 thread context，包括 TID、寄存器、栈指针、程序计数器、条件码。
• 独立 thread stack。

同一 process 内的 thread 共享：

• code segment。
• data segment。
• heap。
• shared libraries。
• open files。
• installed signal handlers。
• kernel context 中的地址空间与描述符表。

同一个进程中的线程像是一个“线程池”里的同级成员；processes 通常形成父子层级。

Pthreads 基本接口

Pthreads 是 POSIX 线程接口，常用函数包括：

函数	作用
pthread_create	创建 thread
pthread_join	等待 joinable thread 结束并回收
pthread_self	获取当前 thread ID
pthread_cancel	请求取消某个 thread
pthread_exit	结束当前 thread
exit	结束整个 process，包括所有 threads

Hello world 示例：

void *thread(void *vargp)
{
    /* thread 入口函数必须返回 void *。 */
    printf("Hello, world!\n");
    return NULL;
}

int main()
{
    pthread_t tid;

    /* 创建一个新线程执行 thread(NULL)。 */
    Pthread_create(&tid, NULL, thread, NULL);
    /* 主线程等待 tid 对应线程结束。 */
    Pthread_join(tid, NULL);
    exit(0);
}

流程：

1. Main thread 调用 Pthread_create 创建 peer thread。
2. Peer thread 执行 thread。
3. Main thread 调用 Pthread_join 等待 peer thread 结束。
4. Peer thread return NULL 等价于隐式 pthread_exit(NULL)。
5. Pthread_join 返回后，main thread 调用 exit 结束进程。

基于 Thread 的并发服务器

Thread-per-connection server 为每个连接创建一个 thread：

int main(int argc, char **argv)
{
    int listenfd, *connfdp;
    socklen_t clientlen;
    struct sockaddr_in clientaddr;
    pthread_t tid;

    listenfd = Open_listenfd(argv[1]);
    while (1) {
        clientlen = sizeof(clientaddr);
        /* 为每个连接动态分配描述符存储，避免把栈变量地址传给多个线程。 */
        connfdp = Malloc(sizeof(int));
        *connfdp = Accept(listenfd, (SA *)&clientaddr, &clientlen);
        /* 新线程负责释放 connfdp 并处理连接。 */
        Pthread_create(&tid, NULL, thread, connfdp);
    }
}

void *thread(void *vargp)
{
    /* 复制连接描述符值，随后即可释放参数块。 */
    int connfd = *((int *)vargp);

    /* 分离线程，结束后资源自动回收，无需主线程 join。 */
    Pthread_detach(pthread_self());
    Free(vargp);
    echo(connfd);
    Close(connfd);
    return NULL;
}

关键点：

• connfdp 需要动态分配，确保每个 thread 得到独立参数。
• Thread routine 复制 connfd 后立即 Free(vargp)。
• Thread 调用 pthread_detach(pthread_self())，结束时自动释放资源。

为什么 server thread 要 detached？

Thread 有两种状态：

• joinable：其他 thread 可通过 pthread_join 回收；若不 join，会造成资源未释放。
• detached：结束后资源自动回收，不能被 join。

Server 通常不逐个 join worker thread，因此 worker 应设为 detached。

错误示例：

while (1) {
    connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);
    /* 错误：每个线程都拿到同一个栈变量 connfd 的地址。 */
    Pthread_create(&tid, NULL, thread, &connfd);
}

connfd 是 main thread 栈上的一个变量，循环下一次 accept 可能覆盖它。多个 worker thread 可能读到同一个地址中的新值，导致连接描述符混乱。

正确做法：为每个连接分配独立 int，或用能安全承载整数的 intptr_t 做值传递。直接把 int 强转为 void * 在 64-bit 平台上不可移植。

为什么传 &connfd 会产生 data race？

connfd 是 main thread 栈上的同一个局部变量。Main thread 第一次 accept 后可能得到 4，创建 worker thread 后立刻进入下一轮 accept，并把同一内存位置覆盖为 5。若第一个 worker 还没来得及读取 *connfdp，它也可能读到 5。

该错误不是因为 descriptor 本身不能共享，而是因为多个 thread 共享了同一个参数存放位置。动态分配 int 的作用是让每个 worker 拥有不同地址；worker 复制出 connfd 后立即 Free(vargp)，即可避免参数位置被下一轮循环覆盖。

如何比较 pthread_create 和 fork 的开销？

可以分别循环创建并回收大量 thread/process，记录总时间并除以次数。测量时应注意：

• 每轮都要包含回收成本，例如 pthread_join 或 waitpid。
• 子任务内容应尽量为空，避免业务逻辑掩盖创建成本。
• 需要预热和重复测量，观察均值和波动。
• 创建开销只是维度之一；process 还提供地址空间隔离，thread 则提供更低成本的数据共享。

Threads & Processes

相同点：

• 都有独立逻辑控制流。
• 都可并发执行，也可能运行在不同 core 上。
• 都会发生 context switch。

不同点：

对比项	Process	Thread
地址空间	通常独立	同一 process 内共享
数据共享	需要 IPC	可直接共享全局数据和 heap
创建与回收开销	较高	较低
错误隔离	较强	较弱

Thread 方案的优点是共享数据方便、开销较低；缺点是 unintended sharing 可能引入难以复现的错误。

Thread Memory Model

概念模型：

• 每个 thread 运行在同一个 process context 中。
• 每个 thread 有自己的 thread context。
• 所有 threads 共享 process virtual address space 中除 thread stack 以外的区域。

实际模型中，寄存器值是隔离的，但一个 thread 可以读写另一个 thread 的 stack，只要它获得相应地址。因此“每个 thread 的 stack 私有”是编程约定，不是硬件强制隔离。

Shared Variable 判定

变量实例与内存位置有关：

变量类型	实例位置	实例数量
Global variable	data segment	每个 process 一个实例
Local automatic variable	thread stack	每次调用、每个 thread 各自实例
Local static variable	data segment	每个 process 一个实例

定义：变量 x 是 shared variable，当且仅当多个 thread 引用了 x 的至少同一个实例。

示例：

#define N 2
char **ptr;

int main()
{
    int i;
    pthread_t tid;
    char *msgs[N] = {
        "Hello from foo",
        "Hello from bar"
    };

    /* ptr 指向主线程栈上的 msgs 数组；多个线程共享该全局指针。 */
    ptr = msgs;
    for (i = 0; i < N; i++)
        /* 将整数 id 强制转换为 void* 传递，仅用于示例。 */
        Pthread_create(&tid, NULL, thread, (void *)i);
    /* 主线程退出但不终止进程，等待其他线程继续运行。 */
    Pthread_exit(NULL);
}

void *thread(void *vargp)
{
    int myid = (int)vargp;
    /* static 局部变量被所有线程共享，++cnt 存在数据竞争。 */
    static int cnt = 0;

    printf("[%d]: %s (cnt=%d)\n", myid, ptr[myid], ++cnt);
    return NULL;
}

共享性分析：

变量实例	main thread	peer thread 0	peer thread 1	是否共享
ptr	是	是	是	是
cnt	否	是	是	是
i in main	是	否	否	否
msgs in main	是	是	是	是
myid in thread 0	否	是	否	否
myid in thread 1	否	否	是	否

其中 cnt 是 local static variable，两个 peer threads 共享同一个实例。++cnt 若不加同步，会发生 data race。

Data Race 示例：cnt++

两个 thread 各执行 NITERS 次 cnt++：

#define NITERS 100000000
unsigned int cnt = 0;

void *count(void *arg)
{
    int i;
    for (i = 0; i < NITERS; i++)
        /* cnt++ 不是原子操作，会被拆成读、加、写。 */
        cnt++;
    return NULL;
}

期望结果是 2 * NITERS。实际可能小于期望值：

BOOM! cnt=198841183
BOOM! cnt=198261801

原因：cnt++ 不是单条原子操作。它通常被编译为 load、update、store：

L: load   cnt -> register
U: update register + 1
S: store  register -> cnt

错误交错示例：

初始 cnt = 0

Thread 1: L1  读 cnt = 0
Thread 1: U1  得到 1
Thread 2: L2  读 cnt = 0
Thread 1: S1  写 cnt = 1
Thread 2: U2  得到 1
Thread 2: S2  写 cnt = 1

两个 thread 都完成一次 increment，但最终 cnt 只增加 1。

Progress Graph

Progress graph 用于描述两个并发 thread 的离散执行状态空间：

• 每个轴对应一个 thread 的指令顺序。

• 每个点表示一个执行状态，例如 (L1, S2) 表示 thread 1 已完成 L1，thread 2 已完成 S2。
  

• 一条 trajectory 表示一种合法执行交错。
  

• 对共享变量 cnt，L、U、S 构成 critical section。两个 thread 的 critical section 不应交错。会导致交错的状态集合称为 unsafe region。

结论：

• trajectory 不接触 unsafe region，则称为 safe。
• 对 cnt++ 来说，trajectory 正确当且仅当它是 safe trajectory。

Semaphore

Semaphore 是非负整数同步变量。Dijkstra 的 P / V 操作：

P(s): while (s == 0) wait(); s--;
V(s): s++;

操作系统保证 P 和 V 中修改 semaphore 的部分不可分割：

• 同一时刻只有一个 P 或 V 能修改 s。
• 当 P 的等待结束时，只有该 P 能执行 s--。
• V 可唤醒被阻塞的 P。
• Semaphore 不保证唤醒顺序。

POSIX semaphore 接口：

#include <semaphore.h>

int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_wait(sem_t *s);  /* P(s)：若计数为 0 则阻塞，否则减 1 */
int sem_post(sem_t *s);  /* V(s)：计数加 1，并唤醒可能等待的线程 */

CSAPP wrapper：

/* CSAPP 对 sem_wait 的包装，失败时统一报错。 */
void P(sem_t *s);
/* CSAPP 对 sem_post 的包装，失败时统一报错。 */
void V(sem_t *s);

将 semaphore 初始化为 1，可作为 mutex 使用：

unsigned int cnt;
sem_t sem;

int main()
{
    /* 初始化二元信号量为 1，用作互斥锁。 */
    Sem_init(&sem, 0, 1);
    /* create two threads */
}

void *count(void *arg)
{
    int i;
    for (i = 0; i < NITERS; i++) {
        /* 进入临界区前获取信号量。 */
        P(&sem);
        cnt++;
        /* 离开临界区后释放信号量。 */
        V(&sem);
    }
    return NULL;
}

P(&sem) 与 V(&sem) 把 critical section 包围起来。Progress graph 中，这相当于增加 forbidden region，使任何合法 trajectory 都不能进入 unsafe region。

Producer-Consumer

单元素 buffer 使用两个 semaphore。一个进程负责“生产数据”，另一个进程负责“消费数据”。两个进程通过一个共享缓冲区交接数据，其中 Producer 不能在缓冲区还没被消费时继续覆盖写入；Consumer 不能在缓冲区还没有数据时读取。所以这里用两个 semaphore 来协调顺序：

• empty：空槽数量，初始为 1。
• full：已有 item 数量，初始为 0。

#define NITERS 5

struct {
    int buf;      /* 单槽缓冲区 */
    sem_t full;   /* full=1 表示缓冲区中有可消费数据 */
    sem_t empty;  /* empty=1 表示缓冲区可写入 */
} shared;

int main()
{
    /* 初始时缓冲区为空：empty=1，full=0。 */
    Sem_init(&shared.empty, 0, 1);
    Sem_init(&shared.full, 0, 0);
    /* create producer and consumer */
}

Producer：

void *producer(void *arg)
{
    int i, item;

    for (i = 0; i < NITERS; i++) {
        item = i;
        /* 等待空槽。 */
        P(&shared.empty);
        shared.buf = item;
        /* 通知消费者缓冲区已有数据。 */
        V(&shared.full);
    }
    return NULL;
}

Consumer：

void *consumer(void *arg)
{
    int i, item;

    for (i = 0; i < NITERS; i++) {
        /* 等待生产者放入数据。 */
        P(&shared.full);
        item = shared.buf;
        /* 取走数据后释放空槽。 */
        V(&shared.empty);
        printf("consumed %d\n", item);
    }
    return NULL;
}

流程解析：

1. 初始 empty = 1, full = 0。
2. Consumer 若先运行，会阻塞在 P(full)。
3. Producer 执行 P(empty) 后写入 buf，再 V(full) 唤醒 consumer。
4. Consumer 执行 P(full) 后读取 buf，再 V(empty) 允许 producer 写入下一项。

Readers-Writers Problem

定义

• Reader 只读共享对象。
• Writer 修改共享对象。
• Writer 必须独占访问对象。
• 多个 reader 可同时访问对象。

First Readers-Writers Problem

First readers-writers problem 偏向 reader：除非 writer 已经获得对象，否则 reader 不应等待。若 writer 在等待期间仍有新 reader 到达，新 reader 可继续优先进入，因此 writer 可能 starvation。

实现：

int readcnt;       /* initially 0 */
sem_t mutex, w;    /* both initially 1 */

void reader(void)
{
    while (1) {
        /* mutex 保护 readcnt 的更新。 */
        P(&mutex);
        readcnt++;
        /* 第一个读者锁住写者，允许后续读者并发进入。 */
        if (readcnt == 1)
            P(&w);
        V(&mutex);

        /* reading */

        P(&mutex);
        readcnt--;
        /* 最后一个读者离开时释放写者锁。 */
        if (readcnt == 0)
            V(&w);
        V(&mutex);
    }
}

void writer(void)
{
    while (1) {
        /* 写者独占 w，保证写期间没有读者或其他写者。 */
        P(&w);
        /* writing */
        V(&w);
    }
}

冲突情况：

• 如何避免 W-W conflict？ 所有 writer 都必须持有 w，因此同一时刻只有一个 writer 能写。
• 如何避免 R-W conflict？ 第一个 reader 执行 P(&w)，最后一个 reader 执行 V(&w)。只要有 reader 在读，writer 无法获得 w。
• 如何支持 R-R concurrency？ Reader 只在修改 readcnt 时短暂持有 mutex，实际 reading 阶段不持有 mutex，多个 reader 可并发读。
• 如何实现 reader first？ 当已有 reader 在读时，readcnt > 0，新 reader 只需更新 readcnt，不会等待 writer。

Second Readers-Writers Problem

Second readers-writers problem 偏向 writer：一旦 writer 准备写，后续到达的 reader 必须等待，直到 writer 完成。该策略减少 writer starvation，但 reader 可能 starvation。

实现思路通常需要额外状态，例如：

• readcnt：当前 reader 数量。
• writecnt：正在等待或写入的 writer 数量。
• rmutex / wmutex：保护计数器。
• readTry：writer 等待时阻止新 reader 进入。
• resource：保护共享对象。

核心约束是：writer 到达后先关闭 reader 入口；已有 reader 退出后，writer 尽快获得资源。

一种等价的实现方式是使用 lock 和 condition variable 维护状态：

• readers：正在读的 reader 数量。
• writers：正在写的 writer 数量，取值通常为 0 或 1。
• waiting_writers：正在等待写入的 writer 数量。
• ok_to_read / ok_to_write：分别用于唤醒 reader 和 writer。

Reader 获取读锁时，需要在 writers > 0 或 waiting_writers > 0 时等待；这表示只要有 writer 正在写或已经排队，新的 reader 就不能插队。Writer 获取写锁时，先增加 waiting_writers，然后等待 readers == 0 && writers == 0，满足后减少 waiting_writers 并设置 writers = 1。

释放规则：

• Reader 释放读锁后，若 readers == 0 且存在等待 writer，应 signal ok_to_write。
• Writer 释放写锁后，若仍有等待 writer，优先 signal 下一个 writer；否则 broadcast 给等待的 readers。

为什么该实现偏向 writer？

关键在于 reader 获取读锁时检查 waiting_writers。若某个 writer 已经到达并等待，即使当前还有 reader 在读，后续新 reader 也必须等待。这样已有 reader 退出后，writer 能尽快获得资源，避免 reader 连续到达导致 writer 长期 starvation。

Prethreaded Concurrent Server

Prethreaded server 在启动时创建 worker thread pool。Master thread 只负责 accept 连接并把 connfd 插入共享 buffer；worker threads 从 buffer 中取出 connfd 并服务 client。

优势：

• 避免每次连接都创建新 thread。
• 可限制并发 worker 数量。
• 适合高频短连接或 IPC 服务。

worker thread 数量如何设置？

取决于 workload：

• CPU-bound：通常接近 core 数，避免过多 context switch。
• I/O-bound：可高于 core 数，用更多 worker 覆盖 I/O 等待。
• 受内存、栈空间、调度开销和 tail latency 约束。

需要通过压测确定。固定经验值不能替代实际负载下的测量。

shared buffer 大小如何设置？

Buffer 太小会使 master thread 频繁阻塞在插入 connfd 上，无法及时 accept 后续连接；buffer 太大会占用更多内存，并可能增加排队时间，使 client 连接已经建立但迟迟得不到 worker 处理。

该大小应结合连接到达速率、平均服务时间、worker 数量和延迟目标确定。可以用压测观察队列长度、拒绝率和 tail latency，再调整 buffer 容量。

Android / HarmonyOS 中 prethreading 可用于什么？

可用于 Binder IPC 等服务线程池。系统服务提前准备线程处理 IPC 请求，避免每次请求临时创建线程，并通过线程池上限控制资源使用。

Threads for Parallelism

并发程序不一定并行。Parallel program 是 concurrent program 的子集，要求多个任务同时在不同硬件执行单元上运行。

相对加速比定义为 Speedup = T1 / Tp，其中 T1 是单线程时间，Tp 是 p 个线程时间。

并行求和：mutex 版本

long gsum;
long nelems_per_thread;
sem_t mutex;

void *sum_mutex(void *vargp)
{
    int myid = *((int *)vargp);
    long start = myid * nelems_per_thread;
    long end = start + nelems_per_thread;
    long i;

    for (i = start; i < end; i++) {
        /* 每次加法都进入临界区，正确但同步开销很高。 */
        P(&mutex);
        gsum += i;
        V(&mutex);
    }
    return NULL;
}

问题：每次累加都进入 critical section，同步开销占比高，线程越多竞争越强。

优化原则：

• 若能避免共享写，就避免共享写。
• 若不能避免同步，应增加每次同步之间的有效计算量，摊薄同步成本。

为什么 mutex 版本可能越并行越慢？

每次循环只做一次简单加法，但都要执行 P/V。同步操作需要原子修改、阻塞和唤醒等机制，开销远高于一次加法。

单线程时 P 通常立即成功，V 也不需要唤醒其他 thread；多线程时会出现真实竞争，等待者可能阻塞，释放者可能唤醒其他 thread。若机器有 4 个 core，从 1 个 thread 增加到 4 个 thread 时，更多 thread 同时竞争同一把锁，contention 增强，运行时间可能上升。超过 core 数后，更多 thread 处于 run queue 中，性能变化还会受到调度、唤醒和缓存状态影响，不一定呈单调趋势。

并行求和：数组部分和

void *sum_array(void *vargp)
{
    int myid = *((int *)vargp);
    long start = myid * nelems_per_thread;
    long end = start + nelems_per_thread;
    long i;

    for (i = start; i < end; i++)
        /* 不同线程写 psum 的不同元素，避免共享全局和互斥锁。 */
        psum[myid] += i;
    return NULL;
}

每个 thread 写自己的 psum[myid]，main thread 在 join 后合并。该方法减少了锁竞争，但循环中仍持续写共享数组位置，可能有 cache 影响。

并行求和：local sum

void *sum_local(void *vargp)
{
    int myid = *((int *)vargp);
    long start = myid * nelems_per_thread;
    long end = start + nelems_per_thread;
    long i, sum = 0;

    for (i = start; i < end; i++)
        /* 使用线程私有局部变量累积，避免频繁写共享数组。 */
        sum += i;

    /* 最后只写一次共享结果数组。 */
    psum[myid] = sum;
    return NULL;
}

该版本在 local variable 中累加，最后写一次 psum[myid]，减少共享写和 cache coherence 流量。

并行求和的性能应如何核对？

若每个 thread 处理独立区间，且循环主体几乎没有同步，理想情况下 p 个 thread 的时间接近 T1 / p。实际性能会在 core 数附近到达平台瓶颈；例如 4-core 机器上，4 个 thread 后继续增加到 8 或 16 个 thread 通常不会继续线性加速。

若测量结果不符合预期，应检查：是否仍有共享写、是否存在 false sharing、任务划分是否均匀、计时是否包含 thread 创建和 join、是否受内存带宽或其他后台任务影响。

Thread-Safe 与 Reentrant

Thread-safe function：被多个 concurrent threads 反复调用时，总能产生正确结果。

常见 thread-unsafe 函数类别：

类别	原因	修复方向
Class 1	未保护共享变量	使用锁或 semaphore
Class 2	依赖跨调用持久状态	把状态作为参数传入
Class 3	返回 static variable 指针	调用者提供 buffer，或 lock-and-copy
Class 4	调用 thread-unsafe 函数	改用 thread-safe 版本

Class 2 示例：rand

unsigned int next = 1;

int rand(void)
{
    /* next 是全局状态，多个线程同时更新会产生数据竞争。 */
    next = next * 110351524 + 12345;
    return (unsigned int)((next / 65536) % 32768);
}

void srand(unsigned int seed)
{
    /* 修改同一个全局 next，会影响所有线程后续 rand 结果。 */
    next = seed;
}

next 是共享持久状态，多个 thread 会互相干扰。改为传入状态指针：

int rand_r(int *nextp)
{
    /* 调用者传入私有状态指针，从而避免共享全局 next。 */
    *nextp = *nextp * 110351524 + 12345;
    return (unsigned int)((*nextp / 65536) % 32768);
}

Class 3 示例：ctime

char *ctime(const time_t *timep)
{
    static char *p;
    /* convert time to string */
    /* 返回 static buffer 指针，多个线程会共享同一存储。 */
    return p;
}

多个 thread 调用时会共享同一个 static buffer。可改为调用者传入 private buffer，或使用 lock-and-copy：

char *ctime_ts(const time_t *timep, char *privatep)
{
    char *sharedp;

    /* 用互斥锁保护对 thread-unsafe ctime 的调用。 */
    P(&mutex);
    sharedp = ctime(timep);
    /* 把共享缓冲区内容复制到调用者提供的私有缓冲区。 */
    strcpy(privatep, sharedp);
    V(&mutex);
    return privatep;
}

Reentrant function 是 thread-safe function 的重要子集。若函数在多线程调用时不访问任何共享变量，则它是 reentrant。Reentrant function 不需要同步。

为什么 signal handler 中不能使用 printf？

printf 不是 async-signal-safe。它可能使用内部锁和共享缓冲区。若 signal 在主程序持有 printf 内部锁时打断执行，handler 再调用 printf，可能发生 deadlock 或破坏内部状态。

Thread Local Storage

Thread local storage（TLS）为每个 thread 提供独立变量实例，适合“逻辑上全局、实际应按 thread 隔离”的数据。

示例：

thread_local int i = 0;

void f(int newval) {
    // 每个线程都有独立的 i，因此这里不会改到其他线程的 i。
    i = newval;
}

void g() {
    // 输出当前线程自己的 i。
    std::cout << i;
}

void threadfunc(int id) {
    f(id);
    // 自增也只作用于当前线程的 thread_local 变量。
    ++i;
    g();
}

若 main 中创建 id 为 1、2、3 的三个 thread，每个 thread 的 i 独立：

• thread 1 输出 2。
• thread 2 输出 3。
• thread 3 输出 4。
• main thread 的 i 仍为 0。

可能输出为 2340、3240、4230、4320、2430、3420 等排列，但最后一位总是 main thread 输出的 0。

TLS 应用：

• Random number generator：每个 thread 维护独立 seed。
• errno：每个 thread 维护独立错误码，避免互相覆盖。

同一条指令如何访问不同 thread 的 TLS 实例？

多个 thread 共享同一份 code，不能依赖 PC-relative addressing 区分 TLS 实例，因为执行到同一条指令时 PC 相同。常见实现是为每个 thread 维护 TLS base address，并把它放在专用寄存器或线程控制块中。

编译器把 TLS 变量访问转换为“TLS base + 固定 offset”。Offset 在编译或链接阶段确定；TLS base 在 thread 创建时由运行时/内核设置。于是同一条机器指令在不同 thread 中使用不同 TLS base，最终访问不同内存位置。

例如 errno 逻辑上像全局变量，但每个 thread 都有独立实例。一个 thread 的系统调用失败不应覆盖另一个 thread 看到的 errno。

Race 与 Deadlock 示例：

Race 的本质：程序正确性依赖“某个 thread 先到达控制点 x，另一个 thread 后到达控制点 y”。

错误示例：

#define N 4

int main()
{
    pthread_t tid[N];
    int i;

    for (i = 0; i < N; i++)
        /* 错误：传入的是循环变量 i 的同一个地址。 */
        pthread_create(&tid[i], NULL, thread, &i);
    for (i = 0; i < N; i++)
        pthread_join(tid[i], NULL);
}

void *thread(void *vargp)
{
    /* 读取时 i 可能已经被主线程改成其他值。 */
    int myid = *((int *)vargp);
    printf("Hello from th. %d\n", myid);
    return NULL;
}

问题：所有 thread 都获得同一个 &i。Thread 读取 i 的时刻不确定，可能输出重复 id 或越界 id。应为每个 thread 准备独立参数，例如 myid[i] = i 后传 &myid[i]。

Deadlock 示例：

sem_t mutex[2];

void *count(void *vargp)
{
    int id = (int)vargp;

    /* 两个线程可能分别持有 mutex[0] 和 mutex[1]。 */
    P(&mutex[id]);
    /* 再请求对方持有的锁时形成循环等待。 */
    P(&mutex[1 - id]);
    cnt++;
    V(&mutex[id]);
    V(&mutex[1 - id]);
    return NULL;
}

若 thread 0 获得 mutex[0]，thread 1 获得 mutex[1]，随后二者都等待对方持有的 semaphore，就进入 deadlock。

避免方法：所有 thread 按同一全局顺序获取锁，例如总是先获取编号较小的锁，再获取编号较大的锁。

Pthread Mutex

POSIX mutex 提供 mutual exclusion：

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

/* 也可以用 pthread_mutex_init 显式初始化互斥锁。 */
pthread_mutex_init(&lock, NULL);
/* 进入临界区前加锁。 */
pthread_mutex_lock(&lock);
balance = balance + 1;
/* 更新共享变量后解锁。 */
pthread_mutex_unlock(&lock);

常见扩展接口：

/* 非阻塞加锁；若锁已被持有，立即返回错误码。 */
int pthread_mutex_trylock(pthread_mutex_t *mutex);
/* 带绝对超时时间的加锁；超时前未获得锁则返回。 */
int pthread_mutex_timedlock(pthread_mutex_t *mutex,
                            struct timespec *abs_timeout);

• trylock：若锁已被持有，立即失败返回。
• timedlock：获得锁或超时后返回。

评价一个 lock 主要看：

• mutual exclusion：能否保证互斥。
• fairness：等待者是否可能长期得不到锁。
• performance：低竞争和高竞争下开销是否可接受。

为什么锁的性能也重要？

Lock 本身不产生业务结果，它只是保护 critical section。若 critical section 中的有效计算很短，而加锁、等待、唤醒或 cache coherence 的开销很高，整体程序可能比单线程更慢。评价锁时要同时看正确性、公平性和开销：正确性保证互斥，公平性避免 starvation，性能决定并发是否真正带来收益。

构造 Lock

关闭中断

单处理器系统可在 critical section 中关闭中断：

void lock() {
    /* 单处理器内核中关闭中断，可防止当前线程在临界区被抢占。 */
    DisableInterrupts();
}

void unlock() {
    /* 离开临界区后重新允许中断。 */
    EnableInterrupts();
}

问题：

• 允许普通 thread 执行特权操作，依赖调用者不滥用。
• 多处理器上无效，因为其他 CPU 仍可执行。
• 长时间关闭中断会影响系统响应。
• 只适合内核在特定场景保护自己的共享数据结构。

简单 Flag 的错误

尝试用普通变量 flag 表示锁状态：

typedef struct {
    int flag;  /* 0 表示空闲，1 表示已被占用 */
} lock_t;

void init(lock_t *mutex) {
    mutex->flag = 0;
}

void lock(lock_t *mutex) {
    /* 检查和设置不是原子操作，两个线程可能同时通过 while。 */
    while (mutex->flag == 1)
        ;
    mutex->flag = 1;
}

void unlock(lock_t *mutex) {
    /* 释放锁。 */
    mutex->flag = 0;
}

正确性问题：

Thread 1: while(flag == 1)  // sees 0
context switch
Thread 2: while(flag == 1)  // sees 0
Thread 2: flag = 1
context switch
Thread 1: flag = 1

两个 thread 都认为自己获得锁，互斥失效。

性能问题：等待者一直 spin-wait，占用 CPU。单处理器上若持锁 thread 被抢占，其他 thread 自旋不会推进锁释放。

Test-and-Set Spin Lock

Test-and-Set 原子地读取旧值并写入新值。

int TestAndSet(int *old_ptr, int new)
{
    /* 真实硬件会保证读取旧值和写入新值原子完成。 */
    int old = *old_ptr;
    *old_ptr = new;
    return old;
}

基于 Test-and-Set 的 spin lock：

typedef struct {
    int flag;  /* 0 表示锁空闲，1 表示锁被持有 */
} lock_t;

void init(lock_t *lock) {
    lock->flag = 0;
}

void lock(lock_t *lock) {
    /* TestAndSet 返回旧值；旧值为 1 表示锁已被其他线程持有。 */
    while (TestAndSet(&lock->flag, 1) == 1)
        ;
}

void unlock(lock_t *lock) {
    /* 写回 0，使后续 TestAndSet 能观察到锁空闲。 */
    lock->flag = 0;
}

特点：

• 正确性：Test-and-Set 原子化了“检查并设置”。
• 公平性：不保证先等待者先获得锁，可能 starvation。
• 性能：低竞争时开销小；高竞争或单 CPU 上浪费 CPU。

CAS & LL/SC & Ticket Lock

Compare-And-Swap（CAS）原子地比较旧值并按条件写入：

int CompareAndSwap(int *ptr, int expected, int new)
{
    /* 原子地比较 *ptr 与 expected；相等时写入 new。 */
    int actual = *ptr;
    if (actual == expected)
        *ptr = new;
    return actual;
}

void lock(lock_t *lock) {
    /* 只有观察到 flag 从 0 变为 1 的线程成功获得锁。 */
    while (CompareAndSwap(&lock->flag, 0, 1) == 1)
        ;
}

Load-Linked / Store-Conditional（LL/SC）常见于 ARM：

int LoadLinked(int *ptr) {
    /* 读取值，并在硬件中记录该地址的保留状态。 */
    return *ptr;
}

int StoreConditional(int *ptr, int value) {
    /* 若 LoadLinked 之后没有其他写入，则本次写入成功。 */
    if (no one has updated *ptr since the LoadLinked) {
        *ptr = value;
        return 1;
    } else {
        /* 失败时调用者通常需要重新执行 LL/SC 循环。 */
        return 0;
    }
}

Ticket lock 使用 Fetch-and-Add 保证公平性：

typedef struct {
    int ticket;  /* 下一个可领取的排队号 */
    int turn;    /* 当前允许进入临界区的排队号 */
} lock_t;

void init(lock_t *lock) {
    lock->ticket = 0;
    lock->turn = 0;
}

void lock(lock_t *lock) {
    /* FetchAndAdd 原子领取一个排队号。 */
    int myturn = FetchAndAdd(&lock->ticket);
    /* 按 ticket 顺序等待，保证先来先服务。 */
    while (lock->turn != myturn)
        ;
}

void unlock(lock_t *lock) {
    /* 允许下一个 ticket 持有者进入。 */
    lock->turn = lock->turn + 1;
}

每个 thread 获得递增 ticket，只能在 turn == myturn 时进入 critical section，因此不会 starvation。

Test-and-Set + Yield

void lock()
{
    while (TestAndSet(&flag, 1) == 1)
        /* 获锁失败时主动让出 CPU，减少纯自旋浪费。 */
        yield();
}

若只有两个 thread 且单 CPU，yield 可让出 CPU 给持锁 thread。若有很多等待 thread，一个持锁 thread 被抢占后，其他等待者会轮流运行、发现锁被持有、再 yield，仍有调度开销。

Queue + Park/Unpark

改进思路：显式维护等待队列，把不能获得锁的 thread sleep，释放锁时唤醒队列中的一个 thread。

typedef struct {
    int flag;     /* 真正的锁状态 */
    int guard;    /* 保护锁内部队列和 flag 的短自旋锁 */
    queue_t *q;   /* 等待线程队列 */
} lock_t;

• flag：真实锁状态。
• guard：保护 lock 内部状态的短期 spin lock。
• q：等待队列。

Lock 逻辑：

void lock(lock_t *lock)
{
    /* 先获得 guard，确保检查 flag 和入队操作互斥。 */
    while (TestAndSet(&lock->guard, 1) == 1)
        ;

    if (lock->flag == 0) {
        /* 锁空闲：直接占有锁并释放 guard。 */
        lock->flag = 1;
        lock->guard = 0;
    } else {
        /* 锁忙：加入等待队列，释放 guard 后挂起。 */
        queue_add(lock->q, gettid());
        lock->guard = 0;
        park();
    }
}

Unlock 逻辑：

void unlock(lock_t *lock)
{
    /* 修改等待队列前先获得 guard。 */
    while (TestAndSet(&lock->guard, 1) == 1)
        ;

    if (queue_empty(lock->q)) {
        /* 没有等待者，直接把锁标记为空闲。 */
        lock->flag = 0;
    } else {
        /* 有等待者时唤醒一个线程；flag 保持为 1，锁交给被唤醒线程。 */
        unpark(queue_remove(lock->q));
    }
    lock->guard = 0;
}

为什么不能在 park() 后再释放 guard？

若 thread 在持有 guard 时调用 park()，它会睡眠且不释放 guard。其他 thread 在 unlock 中需要获得 guard 才能唤醒等待者，于是系统 deadlock。

waiting thread: holds guard -> park()
unlocking thread: waits for guard

为什么唤醒等待者时不把 flag 置为 0？

若等待队列非空，unlock 直接把锁的所有权交给被唤醒 thread。该 thread 从 park() 返回后不会重新执行 flag = 1。因此 flag 应保持为 1，表示锁仍被持有，只是持有者即将切换。

Wakeup/Waiting Race

存在丢失唤醒风险：

Thread 1:
queue_add(q, mytid)
guard = 0

context switch

Thread 2:
unpark(mytid)

context switch

Thread 1:
park()

unpark 发生在 park 之前，Thread 1 之后仍进入睡眠，唤醒丢失。

Solaris 的 setpark() 解决该问题：

/* setpark 先声明自己即将 park，避免 unpark 早于 park 导致丢失唤醒。 */
queue_add(lock->q, gettid());
setpark();
lock->guard = 0;
park();

若 setpark() 后已经发生 unpark()，后续 park() 会立即返回，不会睡眠。

Futex & Two-Stage Lock

Linux futex 与某个用户态内存地址关联，并在内核中维护每个 futex 的等待队列。大部分锁操作在用户态完成；只有真的需要睡眠或唤醒时，才进入内核。

如果发现锁一直被占用，就通过 futex_wait 进入内核睡眠：

/* 仅当 address 当前值仍为 expected 时，线程才进入睡眠。 */
futex_wait(address, expected)

若 *address == expected，调用 thread 睡眠；否则立即返回。

/* 唤醒等待在 address 上的一个或多个线程。 */
futex_wake(address)

唤醒在该地址队列上等待的一个 thread。

Linux mutex 常使用 two-stage 设计：

1. 先在用户态用原子操作尝试获得锁，必要时短暂 spin。
2. 若竞争持续，再通过 futex 进入内核睡眠。

优点：

• 低竞争时无需进入内核，开销低。
• 高竞争时等待者睡眠，减少 CPU 浪费。

Condition Variable

定义

Lock 只能保证互斥；有时 thread 需要等待某个条件成立。例如 parent 需要等待 child 执行完毕后再继续。

自旋等待：

volatile int done = 0;

void *child(void *arg) {
    printf("child\n");
    /* 通知父线程子线程已经完成。 */
    done = 1;
    return NULL;
}

int main()
{
    pthread_t c;
    Pthread_create(&c, NULL, child, NULL);
    while (done == 0)
        /* 忙等等待 done 改变，会浪费 CPU。 */
        ;
    printf("parent: end\n");
}

该方法会浪费 CPU。Condition variable 允许 thread 在条件不满足时进入等待队列，由改变条件的 thread 唤醒。

POSIX 接口：

pthread_cond_t c;

/* 原子地释放 m 并等待条件变量 c；被唤醒后重新持有 m。 */
pthread_cond_wait(pthread_cond_t *c, pthread_mutex_t *m);
/* 唤醒至少一个等待在 c 上的线程。 */
pthread_cond_signal(pthread_cond_t *c);
/* 唤醒所有等待在 c 上的线程。 */
pthread_cond_broadcast(pthread_cond_t *c);

pthread_cond_wait 的语义：

1. 调用时 mutex 必须已加锁。
2. 原子地释放 mutex 并让当前 thread 睡眠。
3. 被唤醒后，先重新获得 mutex，再从 pthread_cond_wait 返回。

Condition Variable 示例：Join

int done = 0;
pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t c = PTHREAD_COND_INITIALIZER;

void thr_exit()
{
    Pthread_mutex_lock(&m);
    /* 状态变量记录 child 已经结束。 */
    done = 1;
    /* 唤醒等待 done 变为 1 的线程。 */
    Pthread_cond_signal(&c);
    Pthread_mutex_unlock(&m);
}

void *child(void *arg)
{
    printf("child\n");
    thr_exit();
    return NULL;
}

void thr_join()
{
    Pthread_mutex_lock(&m);
    while (done == 0)
        /* wait 会原子释放 m 并睡眠；醒来后重新获得 m。 */
        Pthread_cond_wait(&c, &m);
    Pthread_mutex_unlock(&m);
}

两种执行顺序：

• Parent 先执行 thr_join：看到 done == 0，释放锁并睡眠；child 设置 done = 1 后 signal，parent 醒来。
• Child 先执行 thr_exit：设置 done = 1；parent 后执行 thr_join 时不会睡眠。

错误实现：没有状态变量

void thr_exit()
{
    Pthread_mutex_lock(&m);
    /* 错误：没有修改状态变量，signal 不能被后来的 wait 观察到。 */
    Pthread_cond_signal(&c);
    Pthread_mutex_unlock(&m);
}

void thr_join()
{
    Pthread_mutex_lock(&m);
    /* 错误：如果 signal 已经发生，这里会永久等待。 */
    Pthread_cond_wait(&c, &m);
    Pthread_mutex_unlock(&m);
}

若 child 在 parent wait 前 signal，该 signal 不会被保存。Parent 之后调用 wait 会永久睡眠。Condition variable 必须和状态变量配合使用。

错误实现：没有锁

void thr_exit()
{
    /* 错误：修改 done 时没有持有锁，和 thr_join 的检查存在竞态。 */
    done = 1;
    Pthread_cond_signal(&c);
}

void thr_join()
{
    if (done == 0)
        /* 错误：wait 没有与互斥锁配合，无法保证检查和睡眠的原子性。 */
        Pthread_cond_wait(&c);
}

问题：检查 done 和进入等待不是原子操作。可能出现：

parent: if (done == 0) 成立
child:  done = 1; signal
parent: wait

Signal 丢失，parent 睡眠。Mutex 用于保护“检查条件”和“进入等待”的原子性。

Condition Variable & Producer-Consumer

基础 buffer：

int buffer;
int count = 0;

void put(int value)
{
    /* 单槽缓冲区只有空时才能写入。 */
    assert(count == 0);
    count = 1;
    buffer = value;
}

int get()
{
    /* 单槽缓冲区只有满时才能读取。 */
    assert(count == 1);
    count = 0;
    return buffer;
}

仅使用 lock 不够，因为当 buffer 满或空时，thread 需要等待条件改变，而不是持锁循环。

错误版本：用 if

void *consumer(void *arg)
{
    Pthread_mutex_lock(&mutex);
    /* if 只检查一次；被错误唤醒或多个消费者竞争时可能不安全。 */
    if (count == 0)
        Pthread_cond_wait(&cond, &mutex);
    int tmp = get();
    Pthread_cond_signal(&cond);
    Pthread_mutex_unlock(&mutex);
}

多 consumer 场景可能出错：

1. Tc1 发现 count == 0，进入 wait。
2. Producer 放入 item，signal 唤醒 Tc1。
3. Tc1 还未运行，Tc2 先获得锁并消费 item，使 count = 0。
4. Tc1 醒来后从 wait 返回，若只用 if，不会重新检查 count，直接 get()，断言失败。

结论：pthread_cond_wait 返回后，条件不一定仍成立。必须用 while 重新检查条件。

为什么一个 condition variable 仍然不够？

即使用 while，若 producer 和 consumer 共用同一个 cond，多 producer / consumer 场景仍可能唤醒错误角色。

错误流程：

1. Tc1 和 Tc2 都因 count == 0 等待。
2. Producer 放入 item，signal 唤醒 Tc1。
3. Producer 下一轮发现 count == 1，进入 wait。
4. Tc1 消费 item 后 signal，但可能唤醒 Tc2。
5. Tc2 醒来发现 count == 0，继续 wait。
6. Producer 仍在 wait，所有相关 thread 都睡眠。

问题在于同一个 cond 同时表示“非空”和“非满”，signal 可能唤醒不满足条件的一类 thread。

正确版本：两个 condition variables

使用两个条件变量：

• empty：buffer 非满，producer 等待它。
• fill：buffer 非空，consumer 等待它。

cond_t empty, fill;
mutex_t mutex;

void *producer(void *arg)
{
    int i;
    for (i = 0; i < loops; i++) {
        Pthread_mutex_lock(&mutex);
        /* while 在被唤醒后重新检查条件，避免虚假唤醒或条件被其他线程改变。 */
        while (count == 1)
            Pthread_cond_wait(&empty, &mutex);
        put(i);
        /* 缓冲区由空变满，通知消费者。 */
        Pthread_cond_signal(&fill);
        Pthread_mutex_unlock(&mutex);
    }
}

void *consumer(void *arg)
{
    int i;
    for (i = 0; i < loops; i++) {
        Pthread_mutex_lock(&mutex);
        while (count == 0)
            Pthread_cond_wait(&fill, &mutex);
        int tmp = get();
        /* 缓冲区由满变空，通知生产者。 */
        Pthread_cond_signal(&empty);
        Pthread_mutex_unlock(&mutex);
        printf("%d\n", tmp);
    }
}

要点：

• 等待条件必须写成 while。
• Producer 只 signal fill。
• Consumer 只 signal empty。
• 对 count、buffer 的访问必须在 mutex 保护下进行。

多槽 Buffer

多槽 circular buffer：

int buffer[MAX];
int fill_ptr = 0;
int use_ptr = 0;
int count = 0;

void put(int value)
{
    /* fill_ptr 指向下一次写入位置。 */
    buffer[fill_ptr] = value;
    /* 环形缓冲区写指针取模前进。 */
    fill_ptr = (fill_ptr + 1) % MAX;
    count++;
}

int get()
{
    /* use_ptr 指向下一次读取位置。 */
    int tmp = buffer[use_ptr];
    /* 环形缓冲区读指针取模前进。 */
    use_ptr = (use_ptr + 1) % MAX;
    count--;
    return tmp;
}

条件改为：

• Producer 等待 count == MAX 变为 false。
• Consumer 等待 count == 0 变为 false。

signal 与 broadcast

若多个等待者需要不同大小的资源，只 signal 一个 thread 可能唤醒不合适的等待者。

内存分配示例：

int bytesLeft = MAX_HEAP_SIZE;
cond_t c;
mutex_t m;

void *allocate(int size)
{
    Pthread_mutex_lock(&m);
    while (bytesLeft < size)
        /* 可用字节不足时等待 free 增加空间。 */
        Pthread_cond_wait(&c, &m);
    void *ptr = ...;
    /* 分配成功后扣减剩余空间。 */
    bytesLeft -= size;
    Pthread_mutex_unlock(&m);
    return ptr;
}

void free(void *ptr, int size)
{
    Pthread_mutex_lock(&m);
    /* 释放空间后增加剩余字节数。 */
    bytesLeft += size;
    /* 唤醒所有等待者，让不同 size 的请求都重新检查条件。 */
    Pthread_cond_broadcast(&c);
    Pthread_mutex_unlock(&m);
}

使用 broadcast 唤醒所有等待者，让它们各自重新检查 bytesLeft < size。代价是可能造成更多 context switch，但语义更稳妥。

Dining Philosophers

五个 philosophers 和五把 forks 构成环。若每个 philosopher 都先拿左边 fork，再拿右边 fork：

void get_forks(int p)
{
    /* 先拿左叉再拿右叉；所有哲学家同序拿锁可能导致死锁。 */
    sem_wait(&forks[left(p)]);
    sem_wait(&forks[right(p)]);
}

void put_forks(int p)
{
    /* 释放两把叉子，允许相邻哲学家继续竞争。 */
    sem_post(&forks[left(p)]);
    sem_post(&forks[right(p)]);
}

可能 deadlock：

P0 holds f0, waits f1
P1 holds f1, waits f2
P2 holds f2, waits f3
P3 holds f3, waits f4
P4 holds f4, waits f0

一种修复：破坏环形等待，让一个 philosopher 反向拿 fork：

void get_forks(int p)
{
    if (p == 4) {
        /* 让一个哲学家反向拿叉子，打破循环等待条件。 */
        sem_wait(&forks[right(p)]);
        sem_wait(&forks[left(p)]);
    } else {
        sem_wait(&forks[left(p)]);
        sem_wait(&forks[right(p)]);
    }
}

一般原则：避免 circular wait，可规定全局资源顺序，并要求所有 thread 按相同顺序申请资源。

用 Lock + CV 实现 Semaphore

可用 mutex 和 condition variable 实现 semaphore：

typedef struct {
    int value;              /* 信号量计数 */
    pthread_cond_t cond;    /* 等待 value 变为正数的条件变量 */
    pthread_mutex_t lock;   /* 保护 value 的互斥锁 */
} Zem_t;

void Zem_init(Zem_t *s, int value)
{
    /* 初始化计数值及内部同步原语。 */
    s->value = value;
    Cond_init(&s->cond);
    Mutex_init(&s->lock);
}

void Zem_wait(Zem_t *s)
{
    Mutex_lock(&s->lock);
    /* value <= 0 时等待；被唤醒后必须重新检查条件。 */
    while (s->value <= 0)
        Cond_wait(&s->cond, &s->lock);
    /* 消耗一个资源。 */
    s->value--;
    Mutex_unlock(&s->lock);
}

void Zem_post(Zem_t *s)
{
    Mutex_lock(&s->lock);
    /* 释放一个资源并唤醒一个等待者。 */
    s->value++;
    Cond_signal(&s->cond);
    Mutex_unlock(&s->lock);
}

解析：

• lock 保护 value。
• wait 中必须用 while，因为被唤醒后条件可能已变化。
• post 增加 value 后 signal 一个等待者。