841 lines
24 KiB
Markdown
841 lines
24 KiB
Markdown
# 第07讲：多线程编程
|
||
|
||
> **本节目标**：理解线程的概念与优势，掌握 POSIX 线程 API（pthread），理解竞态条件的成因与解决方案（互斥锁、信号量），能够实现并行计算和生产者-消费者模型。
|
||
|
||
## 前置知识
|
||
|
||
- [[06_进程控制]] -- 进程的概念与创建
|
||
- [[08_进程间通信]] -- 进程间通信方式（与线程通信对比）
|
||
|
||
---
|
||
|
||
## 一、线程概念
|
||
|
||
### 1.1 什么是线程
|
||
|
||
**线程（Thread）** 是进程内的一个执行单元。一个进程可以包含多个线程，这些线程共享进程的地址空间，但每个线程拥有自己独立的栈和寄存器状态。
|
||
|
||
**生活比喻**：
|
||
- **进程** = 一家公司（独立的办公室、财务系统、人员编制）
|
||
- **线程** = 公司里的员工（共享办公室和财务系统，但各有各的工作任务）
|
||
|
||
### 1.2 进程 vs 线程
|
||
|
||
```mermaid
|
||
graph TB
|
||
subgraph "多进程模型"
|
||
P1["进程1"] --> M1["独立的地址空间"]
|
||
P2["进程2"] --> M2["独立的地址空间"]
|
||
M1 -.->|"IPC 通信<br/>(管道/共享内存)"| M2
|
||
end
|
||
|
||
subgraph "多线程模型"
|
||
T1["线程1"] --> SM["共享的地址空间"]
|
||
T2["线程2"] --> SM
|
||
T3["线程3"] --> SM
|
||
SM --> S1["代码段 (共享)"]
|
||
SM --> S2["数据段 (共享)"]
|
||
SM --> S3["堆 (共享)"]
|
||
SM --> S4["栈 (各自独立)"]
|
||
end
|
||
|
||
style P1 fill:#e1f5fe
|
||
style P2 fill:#e1f5fe
|
||
style T1 fill:#e8f5e9
|
||
style T2 fill:#e8f5e9
|
||
style T3 fill:#e8f5e9
|
||
style SM fill:#fff3e0
|
||
```
|
||
|
||
| 对比维度 | 进程 | 线程 |
|
||
|----------|------|------|
|
||
| 地址空间 | 独立（互不可见） | 共享（可见彼此的变量） |
|
||
| 创建开销 | 大（复制页表、地址空间） | 小（只创建栈和寄存器） |
|
||
| 切换开销 | 大（切换页表、刷新 TLB） | 小（只切换栈指针和寄存器） |
|
||
| 通信方式 | 需要 IPC 机制（管道、共享内存等） | 直接读写共享变量 |
|
||
| 安全性 | 高（地址空间隔离） | 低（需要同步机制保护） |
|
||
| 典型应用 | 独立的程序（如浏览器和音乐播放器） | 同一程序内的并发任务 |
|
||
|
||
### 1.3 线程的共享与私有
|
||
|
||
| 资源 | 共享/私有 | 说明 |
|
||
|------|-----------|------|
|
||
| 代码段 | 共享 | 所有线程执行相同的代码 |
|
||
| 数据段（全局变量） | 共享 | 所有线程可见同一份全局变量 |
|
||
| 堆（malloc 分配的内存） | 共享 | 任一线程分配的内存，其他线程可访问 |
|
||
| 栈 | **私有** | 每个线程有自己的栈，存放局部变量 |
|
||
| 寄存器状态 | **私有** | 每个线程有自己的程序计数器、栈指针等 |
|
||
| 信号处理 | 共享 | 信号发送给进程，任意线程可处理 |
|
||
|
||
---
|
||
|
||
## 二、POSIX 线程 API
|
||
|
||
### 2.1 核心函数
|
||
|
||
| 函数 | 作用 | 说明 |
|
||
|------|------|------|
|
||
| `pthread_create()` | 创建新线程 | 类似 `fork()`，但开销小得多 |
|
||
| `pthread_join()` | 等待线程结束 | 类似 `wait()`，阻塞调用线程 |
|
||
| `pthread_exit()` | 终止当前线程 | 类似 `exit()`，但只终止当前线程 |
|
||
| `pthread_self()` | 获取当前线程 ID | 类似 `getpid()` |
|
||
| `pthread_detach()` | 分离线程 | 线程结束后自动释放资源 |
|
||
| `pthread_cancel()` | 取消线程 | 请求终止另一个线程 |
|
||
|
||
### 2.2 pthread_create 详解
|
||
|
||
```c
|
||
int pthread_create(
|
||
pthread_t *tid, // 输出参数：新线程的 ID
|
||
const pthread_attr_t *attr, // 线程属性（通常传 NULL）
|
||
void *(*start_routine)(void*), // 线程函数（新线程从这里开始执行）
|
||
void *arg // 传递给线程函数的参数
|
||
);
|
||
// 返回值：成功返回 0，失败返回错误码
|
||
```
|
||
|
||
```mermaid
|
||
graph LR
|
||
A["主线程调用<br/>pthread_create()"] --> B["操作系统创建新线程"]
|
||
B --> C["新线程执行<br/>start_routine(arg)"]
|
||
C --> D["线程函数返回"]
|
||
D --> E["主线程调用<br/>pthread_join() 回收"]
|
||
|
||
style A fill:#e1f5fe
|
||
style C fill:#e8f5e9
|
||
style E fill:#fff3e0
|
||
```
|
||
|
||
### 2.3 线程创建示例
|
||
|
||
参考 `实例源代码/chap6/pthread1.c`：
|
||
|
||
```c
|
||
#include "wrapper.h"
|
||
void *peertask(void *vargp);
|
||
|
||
int main()
|
||
{
|
||
pthread_t tid;
|
||
int i;
|
||
|
||
pthread_create(&tid, NULL, peertask, NULL); // 创建新线程
|
||
|
||
for (i = 0; i < 2; i++) {
|
||
printf("main thread looped %d times\n", i);
|
||
usleep(10);
|
||
}
|
||
|
||
pthread_join(tid, NULL); // 等待新线程结束
|
||
exit(0);
|
||
}
|
||
|
||
void *peertask(void *vargp) // 线程函数
|
||
{
|
||
int i;
|
||
for (i = 0; i < 4; i++) {
|
||
printf("peer thread looped %d times\n", i);
|
||
usleep(10);
|
||
}
|
||
return NULL;
|
||
}
|
||
```
|
||
|
||
**输出示例**（顺序不确定，因为主线程和新线程并发执行）：
|
||
```
|
||
main thread looped 0 times
|
||
peer thread looped 0 times
|
||
main thread looped 1 times
|
||
peer thread looped 1 times
|
||
peer thread looped 2 times
|
||
peer thread looped 3 times
|
||
```
|
||
|
||
**关键点**：
|
||
- `pthread_create` 创建新线程后，两个线程**并发执行**
|
||
- `pthread_join` 阻塞主线程，直到新线程结束
|
||
- 输出顺序取决于调度器，每次运行可能不同
|
||
- 编译时需要加 `-lpthread` 链接 pthread 库
|
||
|
||
---
|
||
|
||
## 三、线程安全问题 -- 竞态条件
|
||
|
||
### 3.1 什么是竞态条件
|
||
|
||
**竞态条件（Race Condition）** 是指多个线程同时访问和修改共享变量，最终结果取决于线程执行的精确时序，导致程序行为不可预测。
|
||
|
||
### 3.2 竞态条件演示
|
||
|
||
参考 `实例源代码/chap6/threadrace.c`：
|
||
|
||
```c
|
||
#include "wrapper.h"
|
||
#define N 4
|
||
|
||
void *thread(void *vargp);
|
||
|
||
int main()
|
||
{
|
||
pthread_t tid[N];
|
||
int i;
|
||
|
||
for (i = 0; i < N; i++)
|
||
pthread_create(&tid[i], NULL, thread, &i); // 传递 i 的地址
|
||
|
||
for (i = 0; i < N; i++)
|
||
pthread_join(tid[i], NULL);
|
||
exit(0);
|
||
}
|
||
|
||
void *thread(void *vargp)
|
||
{
|
||
usleep(1);
|
||
int myid = *((int *)vargp); // 读取线程 ID
|
||
printf("Hello from thread %d\n", myid);
|
||
return NULL;
|
||
}
|
||
```
|
||
|
||
**问题分析**：所有线程共享同一个变量 `i` 的地址。当线程读取 `*vargp` 时，`i` 的值可能已经被主线程的循环修改了。
|
||
|
||
**可能出现的输出**：
|
||
```
|
||
Hello from thread 2
|
||
Hello from thread 2
|
||
Hello from thread 3
|
||
Hello from thread 3
|
||
```
|
||
|
||
### 3.3 竞态条件的时序分析
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant 主线程
|
||
participant 线程0
|
||
participant 线程1
|
||
participant 线程2
|
||
participant 线程3
|
||
|
||
主线程->>线程0: pthread_create(&i)
|
||
主线程->>主线程: i++ (i=1)
|
||
主线程->>线程1: pthread_create(&i)
|
||
主线程->>主线程: i++ (i=2)
|
||
主线程->>线程2: pthread_create(&i)
|
||
主线程->>主线程: i++ (i=3)
|
||
主线程->>线程3: pthread_create(&i)
|
||
|
||
Note over 线程0: 读取 *vargp，但 i 已经是 3
|
||
Note over 线程1: 读取 *vargp，i 也是 3
|
||
Note over 线程2: 读取 *vargp，i 也是 3
|
||
Note over 线程3: 读取 *vargp，i 是 3
|
||
|
||
线程0-->>主线程: Hello from thread 3 (错误!)
|
||
线程1-->>主线程: Hello from thread 3 (错误!)
|
||
线程2-->>主线程: Hello from thread 3 (错误!)
|
||
线程3-->>主线程: Hello from thread 3
|
||
```
|
||
|
||
### 3.4 经典竞态：badcount.c
|
||
|
||
参考 `实例源代码/chap6/badcount.c`：
|
||
|
||
```c
|
||
#include "wrapper.h"
|
||
int count = 0;
|
||
|
||
void *increase(void *arg) {
|
||
unsigned int i, niters = (unsigned int)arg;
|
||
for (i = 0; i < niters; i++)
|
||
count++; // 非原子操作！
|
||
return NULL;
|
||
}
|
||
|
||
void *decrease(void *arg) {
|
||
unsigned int i, niters = (unsigned int)arg;
|
||
for (i = 0; i < niters; i++)
|
||
count--; // 非原子操作！
|
||
return NULL;
|
||
}
|
||
|
||
int main(int argc, char **argv) {
|
||
unsigned int niters = atoll(argv[1]);
|
||
pthread_t tid1, tid2;
|
||
pthread_create(&tid1, NULL, increase, (void*)niters);
|
||
pthread_create(&tid2, NULL, decrease, (void*)niters);
|
||
pthread_join(tid1, NULL);
|
||
pthread_join(tid2, NULL);
|
||
|
||
if (count != 0)
|
||
printf("Error! cnt=%d\n", count); // 期望为0，但实际可能不为0
|
||
else
|
||
printf("Correct cnt=%d\n", count);
|
||
exit(0);
|
||
}
|
||
```
|
||
|
||
**为什么 `count++` 和 `count--` 不是原子操作？**
|
||
|
||
```mermaid
|
||
graph LR
|
||
subgraph "count++ 的三条指令"
|
||
A1["LOAD count -> 寄存器"] --> A2["寄存器 = 寄存器 + 1"]
|
||
A2 --> A3["STORE 寄存器 -> count"]
|
||
end
|
||
|
||
subgraph "count-- 的三条指令"
|
||
B1["LOAD count -> 寄存器"] --> B2["寄存器 = 寄存器 - 1"]
|
||
B2 --> B3["STORE 寄存器 -> count"]
|
||
end
|
||
|
||
style A1 fill:#e1f5fe
|
||
style A2 fill:#fff3e0
|
||
style A3 fill:#ffcdd2
|
||
style B1 fill:#e1f5fe
|
||
style B2 fill:#fff3e0
|
||
style B3 fill:#ffcdd2
|
||
```
|
||
|
||
当两个线程的 LOAD/STORE 指令交错执行时，就会出现丢失更新的问题。
|
||
|
||
---
|
||
|
||
## 四、互斥锁（Mutex）
|
||
|
||
### 4.1 互斥锁的原理
|
||
|
||
互斥锁（Mutex）保证同一时刻只有一个线程可以进入**临界区**（访问共享资源的代码段）。其他试图获取已锁定的互斥锁的线程会被阻塞，直到锁被释放。
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant 线程1
|
||
participant 互斥锁
|
||
participant 线程2
|
||
participant 共享变量
|
||
|
||
线程1->>互斥锁: pthread_mutex_lock()
|
||
Note over 互斥锁: 锁状态：已锁定
|
||
线程1->>共享变量: 读取 count (0)
|
||
|
||
线程2->>互斥锁: pthread_mutex_lock()
|
||
Note over 互斥锁: 锁已被占用，线程2 阻塞
|
||
|
||
线程1->>共享变量: count = count + 1 = 1
|
||
线程1->>互斥锁: pthread_mutex_unlock()
|
||
Note over 互斥锁: 锁状态：已解锁
|
||
互斥锁-->>线程2: 被唤醒，获得锁
|
||
|
||
线程2->>共享变量: 读取 count (1)
|
||
线程2->>共享变量: count = count - 1 = 0
|
||
线程2->>互斥锁: pthread_mutex_unlock()
|
||
|
||
Note over 共享变量: count = 0 (正确!)
|
||
```
|
||
|
||
### 4.2 互斥锁 API
|
||
|
||
| 函数 | 作用 |
|
||
|------|------|
|
||
| `pthread_mutex_init()` | 初始化互斥锁 |
|
||
| `pthread_mutex_lock()` | 加锁（阻塞等待） |
|
||
| `pthread_mutex_trylock()` | 尝试加锁（非阻塞） |
|
||
| `pthread_mutex_unlock()` | 解锁 |
|
||
| `pthread_mutex_destroy()` | 销毁互斥锁 |
|
||
|
||
**初始化方式**：
|
||
```c
|
||
// 方式1：静态初始化
|
||
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
|
||
|
||
// 方式2：动态初始化
|
||
pthread_mutex_t mutex;
|
||
pthread_mutex_init(&mutex, NULL);
|
||
```
|
||
|
||
### 4.3 互斥锁修复竞态
|
||
|
||
参考 `实例源代码/chap6/mutex1.c`：
|
||
|
||
```c
|
||
// mutex1.c - 使用互斥锁保护共享变量
|
||
#include "wrapper.h"
|
||
int count = 0;
|
||
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
|
||
|
||
void *thread_fun(void *no) {
|
||
pthread_mutex_lock(&mutex); // 加锁 -- 进入临界区
|
||
count++;
|
||
printf("thread %d, count = %d\n", (int)no, count);
|
||
pthread_mutex_unlock(&mutex); // 解锁 -- 离开临界区
|
||
}
|
||
|
||
int main() {
|
||
pthread_t thread_id1, thread_id2;
|
||
Pthread_create(&thread_id1, NULL, thread_fun, (void*)1);
|
||
Pthread_create(&thread_id2, NULL, thread_fun, (void*)2);
|
||
Pthread_join(thread_id1, NULL);
|
||
Pthread_join(thread_id2, NULL);
|
||
Pthread_exit(0);
|
||
}
|
||
```
|
||
|
||
**输出**（保证正确）：
|
||
```
|
||
thread 1, count = 1
|
||
thread 2, count = 2
|
||
```
|
||
|
||
### 4.4 另一种修复方式：避免共享
|
||
|
||
参考 `实例源代码/chap6/norace.c`，通过为每个线程分配独立的内存来避免共享：
|
||
|
||
```c
|
||
// norace.c - 使用动态分配避免竞态
|
||
for (i = 0; i < N; i++) {
|
||
ptr = malloc(sizeof(int)); // 为每个线程分配独立的内存
|
||
*ptr = i;
|
||
pthread_create(&tid[i], NULL, thread, ptr);
|
||
}
|
||
```
|
||
|
||
**原理**：每个线程拥有自己独立的 `int` 变量，不再共享 `i`，从根本上消除了竞态条件。
|
||
|
||
---
|
||
|
||
## 五、信号量（Semaphore）
|
||
|
||
### 5.1 信号量的概念
|
||
|
||
信号量是一种比互斥锁更通用的同步机制，由 Dijkstra 提出。信号量是一个非负整数计数器，支持两个原子操作：
|
||
|
||
- **P 操作**（`sem_wait`）：计数器减 1，如果计数器 < 0 则阻塞
|
||
- **V 操作**（`sem_post`）：计数器加 1，如果有等待者则唤醒
|
||
|
||
```mermaid
|
||
graph LR
|
||
subgraph "信号量 S"
|
||
S["S = 计数器"]
|
||
end
|
||
|
||
P["sem_wait (P操作)<br/>S--"] -->|S >= 0| OK["允许通过"]
|
||
P -->|S < 0| BLOCK["阻塞等待"]
|
||
|
||
V["sem_post (V操作)<br/>S++"] --> WAKE["唤醒等待者"]
|
||
|
||
style S fill:#fff3e0
|
||
style OK fill:#e8f5e9
|
||
style BLOCK fill:#ffcdd2
|
||
```
|
||
|
||
### 5.2 信号量 API
|
||
|
||
| 函数 | 作用 |
|
||
|------|------|
|
||
| `sem_init(&sem, pshared, value)` | 初始化信号量，value 为初始值 |
|
||
| `sem_wait(&sem)` | P 操作：等待信号量（S--，若 S<0 则阻塞） |
|
||
| `sem_post(&sem)` | V 操作：释放信号量（S++，唤醒等待者） |
|
||
| `sem_getvalue(&sem, &val)` | 获取信号量当前值 |
|
||
| `sem_destroy(&sem)` | 销毁信号量 |
|
||
|
||
### 5.3 互斥锁 vs 信号量
|
||
|
||
| 特性 | 互斥锁 (Mutex) | 信号量 (Semaphore) |
|
||
|------|----------------|-------------------|
|
||
| 计数范围 | 0 或 1（二值） | 0 到 N（计数） |
|
||
| 用途 | 互斥（保护临界区） | 同步（协调执行顺序） |
|
||
| 加锁/解锁者 | 谁加锁谁解锁 | 一个线程 sem_wait，另一个 sem_post |
|
||
| 典型场景 | 保护共享变量 | 生产者-消费者、资源池 |
|
||
|
||
### 5.4 生产者-消费者问题
|
||
|
||
生产者-消费者是经典的同步问题：生产者向缓冲区放入数据，消费者从缓冲区取出数据。需要解决：
|
||
- 缓冲区满时生产者必须等待
|
||
- 缓冲区空时消费者必须等待
|
||
- 同一时刻只有一个线程访问缓冲区
|
||
|
||
**使用信号量解决**：
|
||
|
||
```c
|
||
#include <semaphore.h>
|
||
|
||
#define N 10 // 缓冲区大小
|
||
int buf[N];
|
||
int in = 0, out = 0;
|
||
|
||
sem_t mutex; // 互斥信号量，初值 1
|
||
sem_t empty; // 空槽位数，初值 N
|
||
sem_t full; // 满槽位数，初值 0
|
||
|
||
void *producer(void *arg) {
|
||
while (1) {
|
||
int item = produce_item();
|
||
sem_wait(&empty); // 等待空槽位 (empty--)
|
||
sem_wait(&mutex); // 进入临界区
|
||
buf[in] = item;
|
||
in = (in + 1) % N;
|
||
sem_post(&mutex); // 离开临界区
|
||
sem_post(&full); // 增加满槽位 (full++)
|
||
}
|
||
}
|
||
|
||
void *consumer(void *arg) {
|
||
while (1) {
|
||
sem_wait(&full); // 等待满槽位 (full--)
|
||
sem_wait(&mutex); // 进入临界区
|
||
int item = buf[out];
|
||
out = (out + 1) % N;
|
||
sem_post(&mutex); // 离开临界区
|
||
sem_post(&empty); // 增加空槽位 (empty++)
|
||
consume_item(item);
|
||
}
|
||
}
|
||
```
|
||
|
||
**生产者-消费者流程图**：
|
||
|
||
```mermaid
|
||
graph TD
|
||
subgraph "生产者"
|
||
P1["生产数据"] --> P2["sem_wait(empty)<br/>等待空槽位"]
|
||
P2 --> P3["sem_wait(mutex)<br/>进入临界区"]
|
||
P3 --> P4["放入缓冲区"]
|
||
P4 --> P5["sem_post(mutex)<br/>离开临界区"]
|
||
P5 --> P6["sem_post(full)<br/>通知消费者"]
|
||
P6 --> P1
|
||
end
|
||
|
||
subgraph "消费者"
|
||
C1["sem_wait(full)<br/>等待满槽位"] --> C2["sem_wait(mutex)<br/>进入临界区"]
|
||
C2 --> C3["取出数据"]
|
||
C3 --> C4["sem_post(mutex)<br/>离开临界区"]
|
||
C4 --> C5["sem_post(empty)<br/>通知生产者"]
|
||
C5 --> C6["处理数据"]
|
||
C6 --> C1
|
||
end
|
||
|
||
P6 -.->|"full++ 唤醒"| C1
|
||
C5 -.->|"empty++ 唤醒"| P2
|
||
|
||
style P1 fill:#e1f5fe
|
||
style C6 fill:#e8f5e9
|
||
```
|
||
|
||
### 5.5 实验任务：修复 badcount.c
|
||
|
||
实验中的 `task62.c` 要求使用信号量修复 `badcount.c` 的竞态条件：
|
||
|
||
```c
|
||
#include <semaphore.h>
|
||
sem_t mutex;
|
||
sem_init(&mutex, 0, 1); // 互斥信号量，初值 1
|
||
|
||
void *increase(void *arg) {
|
||
for (i = 0; i < niters; i++) {
|
||
sem_wait(&mutex); // P 操作
|
||
count++;
|
||
sem_post(&mutex); // V 操作
|
||
}
|
||
}
|
||
|
||
void *decrease(void *arg) {
|
||
for (i = 0; i < niters; i++) {
|
||
sem_wait(&mutex); // P 操作
|
||
count--;
|
||
sem_post(&mutex); // V 操作
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## 六、并行计算
|
||
|
||
### 6.1 并行求和问题
|
||
|
||
参考 `实例源代码/chap6/psum64.c`，将一个大数组的求和任务分配给多个线程并行计算：
|
||
|
||
```c
|
||
#include "wrapper.h"
|
||
#define MAXTHREADS 32
|
||
|
||
unsigned long long psum[MAXTHREADS]; // 每个线程的部分和
|
||
unsigned long long nelems_per_thread; // 每个线程处理的元素数
|
||
|
||
void *sum(void *vargp) {
|
||
int myid = *((int *)vargp);
|
||
unsigned long long start = myid * nelems_per_thread;
|
||
unsigned long long end = start + nelems_per_thread;
|
||
unsigned long long i, lsum = 0;
|
||
|
||
for (i = start; i < end; i++)
|
||
lsum += i;
|
||
|
||
psum[myid] = lsum; // 将部分和存入全局数组
|
||
return NULL;
|
||
}
|
||
|
||
int main(int argc, char **argv) {
|
||
unsigned long long i, nelems, nthreads, result = 0;
|
||
pthread_t tid[MAXTHREADS];
|
||
int myid[MAXTHREADS];
|
||
|
||
nthreads = atoi(argv[1]);
|
||
nelems = (1L << atoi(argv[2])); // 2^log_nelems
|
||
nelems_per_thread = nelems / nthreads;
|
||
|
||
// 创建线程
|
||
for (i = 0; i < nthreads; i++) {
|
||
myid[i] = i;
|
||
pthread_create(&tid[i], NULL, sum, &myid[i]);
|
||
}
|
||
for (i = 0; i < nthreads; i++)
|
||
pthread_join(tid[i], NULL);
|
||
|
||
// 汇总部分和
|
||
for (i = 0; i < nthreads; i++)
|
||
result += psum[i];
|
||
|
||
// 验证结果
|
||
if (result == (nelems * (nelems - 1) / 2))
|
||
printf("Correct Result=%lld\n", result);
|
||
else
|
||
printf("Error: result=%lld\n", result);
|
||
|
||
exit(0);
|
||
}
|
||
```
|
||
|
||
**并行求和示意图**：
|
||
|
||
```mermaid
|
||
graph TD
|
||
subgraph "数据划分"
|
||
DATA["0, 1, 2, 3, 4, 5, 6, 7, ..."]
|
||
DATA --> T0["线程0: 0 ~ N/p-1"]
|
||
DATA --> T1["线程1: N/p ~ 2N/p-1"]
|
||
DATA --> T2["线程2: 2N/p ~ 3N/p-1"]
|
||
DATA --> T3["线程3: 3N/p ~ N-1"]
|
||
end
|
||
|
||
T0 --> S0["psum[0]"]
|
||
T1 --> S1["psum[1]"]
|
||
T2 --> S2["psum[2]"]
|
||
T3 --> S3["psum[3]"]
|
||
|
||
S0 --> SUM["result = psum[0]+psum[1]+psum[2]+psum[3]"]
|
||
S1 --> SUM
|
||
S2 --> SUM
|
||
S3 --> SUM
|
||
|
||
style DATA fill:#e1f5fe
|
||
style SUM fill:#e8f5e9
|
||
```
|
||
|
||
### 6.2 加速比与效率
|
||
|
||
**加速比（Speedup）**：S_p = T_1 / T_p
|
||
- T_1：单线程执行时间
|
||
- T_p：p 个线程并行执行时间
|
||
- 理想加速比 S_p = p（线性加速）
|
||
|
||
**效率（Efficiency）**：E_p = S_p / p = T_1 / (p * T_p)
|
||
- 理想效率 E_p = 1（100%）
|
||
- 实际效率 < 1，因为存在线程创建、同步、通信等开销
|
||
|
||
```mermaid
|
||
graph LR
|
||
subgraph "加速比曲线"
|
||
X["线程数 p"] --> Y["加速比 Sp"]
|
||
IDEAL["理想: Sp=p"] -.-> Y
|
||
ACTUAL["实际: Sp<p"] --> Y
|
||
end
|
||
```
|
||
|
||
### 6.3 实验任务：测量加速比
|
||
|
||
实验中的 `task64.c` 要求：
|
||
1. 修改 `psum64.c`，添加计时功能
|
||
2. 分别用 1、2、4、8 个线程运行
|
||
3. 计算每个配置的加速比和效率
|
||
4. 分析为什么加速比不是线性的
|
||
|
||
**预期结果**：
|
||
|
||
| 线程数 | 执行时间 | 加速比 | 效率 |
|
||
|:------:|:--------:|:------:|:----:|
|
||
| 1 | T1 | 1.0 | 100% |
|
||
| 2 | T2 | ~1.8 | ~90% |
|
||
| 4 | T4 | ~3.2 | ~80% |
|
||
| 8 | T8 | ~5.5 | ~69% |
|
||
|
||
---
|
||
|
||
## 七、fork vs pthread_create 性能对比
|
||
|
||
### 7.1 为什么线程更快
|
||
|
||
`fork()` 创建进程时需要复制父进程的整个地址空间（虽然有写时复制优化），而 `pthread_create()` 只需要分配一个新的栈和少量元数据。
|
||
|
||
```mermaid
|
||
graph LR
|
||
subgraph "fork() 创建进程"
|
||
F1["复制页表"] --> F2["复制地址空间<br/>(写时复制)"]
|
||
F2 --> F3["创建新的 PCB"]
|
||
F3 --> F4["开销大: ~ms 级"]
|
||
end
|
||
|
||
subgraph "pthread_create() 创建线程"
|
||
T1["分配栈空间"] --> T2["设置线程属性"]
|
||
T2 --> T3["创建线程元数据"]
|
||
T3 --> T4["开销小: ~us 级"]
|
||
end
|
||
|
||
style F4 fill:#ffcdd2
|
||
style T4 fill:#e8f5e9
|
||
```
|
||
|
||
### 7.2 实验任务：测量创建时间
|
||
|
||
实验中的 `task66.c` 要求分别测量 `fork()` 和 `pthread_create()` 的创建时间：
|
||
|
||
```c
|
||
// 测量 fork 创建时间
|
||
clock_gettime(CLOCK_MONOTONIC, &start);
|
||
for (i = 0; i < N; i++) {
|
||
if (fork() == 0) exit(0); // 子进程立即退出
|
||
wait(NULL);
|
||
}
|
||
clock_gettime(CLOCK_MONOTONIC, &end);
|
||
|
||
// 测量 pthread_create 创建时间
|
||
clock_gettime(CLOCK_MONOTONIC, &start);
|
||
for (i = 0; i < N; i++) {
|
||
pthread_create(&tid, NULL, dummy, NULL);
|
||
pthread_join(tid, NULL);
|
||
}
|
||
clock_gettime(CLOCK_MONOTONIC, &end);
|
||
```
|
||
|
||
**预期结果**：`pthread_create` 通常比 `fork` 快 10-100 倍。
|
||
|
||
---
|
||
|
||
## 八、动态线程池
|
||
|
||
### 8.1 为什么需要线程池
|
||
|
||
频繁创建和销毁线程会带来较大开销。线程池预先创建一组线程，当有任务到来时从池中取出一个线程执行，执行完毕后线程归还池中，避免反复创建销毁。
|
||
|
||
### 8.2 sbuf_t 带缓冲区的线程池
|
||
|
||
实验中的 `task67.c` 实现了一个动态线程池，核心数据结构 `sbuf_t`：
|
||
|
||
```c
|
||
typedef struct {
|
||
int *buf; // 缓冲区数组
|
||
int n; // 缓冲区容量
|
||
int front; // 队首索引
|
||
int rear; // 队尾索引
|
||
sem_t mutex; // 互斥访问
|
||
sem_t slots; // 空槽位信号量
|
||
sem_t items; // 满槽位信号量
|
||
} sbuf_t;
|
||
```
|
||
|
||
**动态调整策略**：
|
||
- **缓冲区满时翻倍**：`buf = realloc(buf, 2 * n)`
|
||
- **缓冲区空时减半**：`buf = realloc(buf, n / 2)`
|
||
|
||
```mermaid
|
||
graph TD
|
||
subgraph "线程池架构"
|
||
MAIN["主线程<br/>接收任务"] -->|"放入缓冲区"| BUF["sbuf_t 缓冲区"]
|
||
BUF -->|"取出任务"| W1["工作线程1"]
|
||
BUF -->|"取出任务"| W2["工作线程2"]
|
||
BUF -->|"取出任务"| W3["工作线程3"]
|
||
end
|
||
|
||
subgraph "动态调整"
|
||
FULL["缓冲区满"] -->|"realloc"| DOUBLE["容量翻倍"]
|
||
EMPTY["缓冲区空"] -->|"realloc"| HALF["容量减半"]
|
||
end
|
||
|
||
style MAIN fill:#e1f5fe
|
||
style BUF fill:#fff3e0
|
||
style W1 fill:#e8f5e9
|
||
style W2 fill:#e8f5e9
|
||
style W3 fill:#e8f5e9
|
||
```
|
||
|
||
---
|
||
|
||
## 九、线程安全的数据结构
|
||
|
||
### 9.1 tickets 售票问题
|
||
|
||
参考 `实例源代码/chap6/tickets1.c`（有竞态条件）和 `tickets2.c`（使用互斥锁修复）：
|
||
|
||
```c
|
||
// tickets2.c - 使用互斥锁保护售票
|
||
int tickets = 10;
|
||
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
|
||
|
||
void *counter(void *no) {
|
||
while (tickets > 0) {
|
||
pthread_mutex_lock(&mutex); // 加锁
|
||
if (tickets > 0) { // 双重检查
|
||
printf("柜台%d 卖出一张票，票号为%d\n", (int)no, tickets);
|
||
usleep(1);
|
||
tickets--;
|
||
}
|
||
pthread_mutex_unlock(&mutex); // 解锁
|
||
usleep(1);
|
||
}
|
||
}
|
||
```
|
||
|
||
**关键点**：`while` 条件检查和 `if` 条件检查都需要在锁内进行（双重检查），否则可能出现超卖。
|
||
|
||
---
|
||
|
||
## 十、实验任务概览
|
||
|
||
本讲对应 [[实验03_多线程编程]]，包含以下任务：
|
||
|
||
| 任务 | 文件 | 内容 | 核心知识点 |
|
||
|------|------|------|------------|
|
||
| 任务一 | task61.c | 3 个对等线程打印姓名/学号/时间 | pthread_create/join |
|
||
| 任务二 | task62.c | 用信号量修复 badcount.c 的竞态条件 | sem_wait/sem_post |
|
||
| 任务三 | task63.c | k 个生产者 + m 个消费者，信号量同步 | 生产者-消费者模型 |
|
||
| 任务四 | task64.c | psum64.c 并行求和，测量不同线程数的加速比 | 并行计算 + 性能分析 |
|
||
| 任务五 | task66.c | 测量 fork vs pthread_create 时间 | 进程/线程创建开销对比 |
|
||
| 任务六 | task67.c | 动态线程池（sbuf_t 缓冲区满翻倍/空减半） | 线程池 + 动态数据结构 |
|
||
|
||
---
|
||
|
||
## 知识关联
|
||
|
||
- 线程是轻量级进程，理解 [[06_进程控制]] 中的 fork 有助于理解线程的优势
|
||
- 线程间的同步问题在死锁章节中会进一步讨论
|
||
- 生产者-消费者模型在 [[08_进程间通信]] 中也有管道和共享内存的实现方式
|
||
- 线程池在并发网络服务器中是核心组件
|
||
|
||
---
|
||
|
||
## 思考题
|
||
|
||
1. **互斥锁 vs 信号量**：互斥锁和二值信号量（初值为 1 的信号量）看起来功能相同，它们有什么本质区别？什么时候必须用信号量而不能用互斥锁？
|
||
2. **死锁的产生**：如果一个线程对已经加锁的互斥锁再次调用 `pthread_mutex_lock()`，会发生什么？如果两个线程分别持有对方需要的锁呢？
|
||
3. **线程安全的函数**：为什么 `printf()` 是线程安全的，而 `count++` 不是？如何判断一个函数是否线程安全？
|
||
4. **并行计算的瓶颈**：为什么 p 个线程的加速比通常达不到理想的 p 倍？哪些因素限制了并行加速？
|
||
5. **fork vs pthread**：在什么场景下应该用多进程而不是多线程？多进程模型有什么优势？
|
||
|
||
---
|
||
|
||
## 扩展阅读
|
||
|
||
- 《UNIX环境高级编程》第11章：线程
|
||
- 《深入理解计算机系统》第12章：并发编程
|
||
- 《POSIX 多线程编程指南》
|
||
- [Pthreads Tutorial (LLNL)](https://computing.llnl.gov/tutorials/pthreads/)
|