OSTEP - 并发部分笔记

本文最后更新于 2026年6月27日凌晨

第26章并发：介绍

线程

def: 进程内部的一条执行流。

在多线程进程中，一个进程内有多个执行流：

每个线程有自己的PC等寄存器和自己的栈，用于保存上下文
所有线程共享一套地址空间
调度器在不同线程之间切换
OS使用TCB (Thread Control Block) 来管理线程状态

所谓的共享的数据，指的是内存里面的数据；对于一个线程，寄存器的数据是不共享的，每个线程有自己的一套寄存器

使用线程的理由

并行
- CPU是多核的，一个大任务可以拆成好几份，让多个线程同时来做
避免慢操作阻塞程序
- 比如磁盘I/O，网络，页错误，这个时候其他线程可以继续运行
线程的优势在于共享数据方便。

并发的问题

线程一旦创建，什么时候运行由调度器决定，而不是由代码书写顺序完全决定。

举个例子：
```
static volatile int counter = 0;

void *mythread(void *arg) {
    for (int i = 0; i < 1e7; i++) {
        counter = counter + 1;
    }
    return NULL;
}
```
运行之后结果可能是这样的：
```
counter = 19345221
counter = 19221041
```
这是因为，counter = counter + 1;，对应汇编是这样的：
```
mov 0x8049a1c, %eax
add $0x1, %eax
mov %eax, 0x8049a1c
```
线程的切换，是可能发生在这三条汇编指令之间的。
线程的等待：有时候一个线程必须等待另一个线程完成才能继续运行，即等待/唤醒问题。

核心术语总结

critical section 临界区：访问共享m资源并且不应被多个线程同时执行的代码段，比如上面的代码，临界区就是counter = counter + 1;
indeterminate program 不确定程序：字面意思
race condition 竞态条件：指程序结果依赖线程的执行时机（调度顺序决定）
mutual exclusion 互斥：是我们想要的性质：当一个线程正在临界区里时，其他线程不能进入同一段临界区

原子性

def：一个操作是一个整体，不可分割。

举个例子，我们希望上面的三条汇编指令，像是有一条不可分割的memory-add 0x8049a1c, $0x1指令一样执行。

但是现实中我们不可能为所有复杂操作提供原子操作，因此需要OS和线程库在其上构建通用数据原语，比如锁。实现共享数据同步，临界区互斥的目标。

第27章线程API

介绍 POSIX 线程库 pthread 提供的几个基本 API：

创建线程`pthread_create()`

int pthread_create(pthread_t *thread,
                   const pthread_attr_t *attr,
                   void *(*start_routine)(void *),
                   void *arg);

thread：用来保存新线程的标识，后面可以用它 join 这个线程。
attr：线程属性，比如栈大小、调度属性；通常传 NULL，表示默认属性。
start_routine：新线程从哪个函数开始执行。
arg：传给这个线程函数的参数。

线程启动函数必须长这样：

void *mythread(void *arg) {

}

如果要传多个参数可以传一个结构体打包进去。

线程一旦创建出来，它就成为一个新的执行流，有自己的调用栈，并和原来的线程共享同一个地址空间。

等待线程结束`pthread_join()`

int pthread_join(pthread_t thread, void **value_ptr);

thread是等待结束的进程
value_ptr是期望得到的返回值

为什么这里是参数是void**？因为线程函数的返回值是void*。

举个例子：

void *mythread(void *arg) {
    myret_t *rvals = malloc(sizeof(myret_t));
    rvals->x = 1;
    rvals->y = 2;
    return rvals;
}
int main() {
    pthread_t p;
    pthread_create(&p, NULL, mythread, &args);
    myret_t *rvals;
    pthread_join(p, (void **) &rvals);
}

线程的返回值务必要放在堆上，如果返回栈上的局部变量地址，函数结束之后就被销毁了。

互斥锁

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

典型用法：

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
// 或者 
// pthread_mutex_t lock;
// pthread_mutex_init(&lock, NULL);
pthread_mutex_lock(&lock);
x = x + 1;
pthread_mutex_unlock(&lock);

如果没有其他线程持有锁，当前线程拿到锁，进入临界区。
如果其他线程已经持有锁，当前线程会阻塞在 pthread_mutex_lock()，直到锁被释放。
只有拿到锁的线程，才应该调用 pthread_mutex_unlock()。
如果程序结束后不再使用这个锁，还可以调用 pthread_mutex_destroy()清理。

这些API都是有返回值的，当返回值为0的时候才正常，上面应该检查一下，只是略了。

条件变量

int pthread_cond_wait(pthread_cond_t *cond,
                      pthread_mutex_t *mutex);
int pthread_cond_signal(pthread_cond_t *cond);

前者是睡眠，后者是唤醒
条件变量总是和锁一起用，线程睡眠时释放锁，唤醒时拿回锁

Thread API 使用原则

线程之间的交互越简单越好
锁与条件变量必须初始化
所有调用必须检查返回值
返回值别返回栈上的
每个线程有自己的栈；要共享数据，应放在堆或全局可访问的位置

第28章锁

锁的基本思想

锁的状态：locked and unlocked

相当于给程序员调度进程的手段。

锁的评价标准

正确性：真的能够提供互斥的功能，只允许一个线程进入临界区
公平性：防止某个线程一直拿不到锁，导致饥饿
性能：锁本身的开销：
- 没有竞争时拿/放锁的效率
- 单CPU多线程竞争、多CPU多线程竞争的效率

接下来会列出一系列锁的实现。

控制中断

void lock() {
    DisableInterrupts();
}

void unlock() {
    EnableInterrupts();
}

缺点：

要求用户程序能执行特权操作，危险
不支持多CPU，因为别的CPU的线程没被关掉中断
关闭中断会导致无法进行I/O（回忆，I/O等操作是要trap之后中断陷入内核态的）

test-and-set

void lock(lock_t *mutex) {
    while (mutex->flag == 1)
        ;              // spin
    mutex->flag = 1;
}

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

这段代码直觉来看是会失败的，因为检查flag和设置flag不是一个原子操作。

实现锁时，不能把「检查锁是否空闲」和「把锁设为已占用」分成两个可被打断的步骤。

所以我们需要依靠硬件的支持，假设有这样一个原子操作（这里的C语言代码只是为了方便理解，实际上是原子的）：

int TestAndSet(int *old_ptr, int new) {
    int old = *old_ptr;
    *old_ptr = new;
    return old;
}

则可以实现一个简单自旋锁：

void lock(lock_t *lock) {
    while (TestAndSet(&lock->flag, 1) == 1)
        ; // spin
}

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

自旋锁的问题：

不保证公平性，可能导致饿死；
性能不好，其他没占到锁的线程会不停自旋，浪费时间片。

其他硬件原语

Compare and Swap

int CompareAndSwap(int *ptr, int expected, int new) {
    int original = *ptr;
    if (original == expected)
        *ptr = new;
    return original;
}

void lock(lock_t *lock) {
    while (CompareAndSwap(&lock->flag, 0, 1) == 1)
    ; // spin
}

Load-Linked / Store-Conditional

先 LoadLinked() 读一个值，然后 StoreConditional() 尝试写入；如果中间没人改过这个地址，写入成功，否则失败。

Fetch-and-Add

代码C · 19 行

int FetchAndAdd(int *ptr) {
    int old = *ptr;
    *ptr = old + 1;
    return old;
}

typedef struct lock_t {
    int ticket = 0;
    int turn = 0;
} lock_t;

void lock(lock_t *lock) {
    int myTurn = FetchAndAdd(&lock->ticket);
    while(lock->turn != myTurn)
        ; //spin
}
void unlock(lock_t *lock) {
    lock->turn = lock->turn + 1;
}

ticket：下一个来排队的人应该拿到几号；
turn：现在轮到几号进入临界区。

只有当turn == myTurn时，线程才被允许进入临界区，释放之后，轮到下一个线程。

解决饥饿的问题，但是没解决自旋锁的性能问题。

解决自旋过多的问题

一个简单改进是拿不到锁就yield()：当前线程主动放弃 CPU，回到 ready 队列，让别人运行。

void lock(lock_t *lock) {
    while(TestAndSet(&flag, 1) == 1) {
        yield();
    }
}

问题：多线程反复竞争一把锁时，会有大量的上下文切换的成本。

另一个改进：等待锁的线程睡眠。

这里以Solaris的两个API为例：

park()：让当前线程睡眠；
unpark(threadID)：唤醒指定线程。

typedef struct lock_t {
    int flag;
    int guard;
    queue_t *q;
} lock_t;

flag：真正的锁是否被持有；
guard：保护锁内部数据结构的小自旋锁；
q：等待这个锁的线程队列。

则：

线程拿不到锁时，把自己加入队列，park()睡眠；

void lock(lock_t *lock) {
    while(TestAndSet(&lock->guard, 1) == 1)
        ;
    if (lock->flag == 0) {
        lock->flag = 1;
        lock->guard = 0;
    }
    else {
        queue_add(lock->q);
        lock->guard = 0;
        park();
    }
}

释放锁时，如果队列非空，unpark()唤醒下一个睡眠的进程。

可能的问题：

线程A刚把自己加入队列
上下文切换到线程B，线程B释放锁，unpark(A)
结果切换回A的时候又park了，继续睡

这里可以通过设计别的OS原语解决。

两阶段锁

第一阶段：先自旋一小会儿（设置固定次数），希望锁马上释放；
第二阶段：如果还拿不到，就睡眠，等待唤醒。

这也是一种混合方案，Linux Kernel采用。

第29章基于锁的并发数据结构

如标题所示，讲并发数据结构的设计与其中的trade-offs。

最简单的方法：给整个数据结构的CRUD环节直接加一把锁，但是可能会导致性能的缺陷（所有线程都竞争同一把锁）。

所以可能需要把锁拆细，但也不是越细越好。

并发计数器

给计数器上一把锁：

typedef struct counter_t {
    int value;
    pthread_mutex_t lock;
} counter_t;

void increment(counter_t *c) {
    pthread_mutex_lock(&c->lock);
    c->value++;
    pthread_mutex_unlock(&c->lock);
}

正确，但缺点：所有线程竞争同一把锁，更新越频繁竞争越严重。

Approximate Counter

只是一种idea。

typedef struct __counter_t {
    int global;
    pthread_mutex_t glock;
    int local[NUMCPUS];
    pthread_mutex_t llock[NUMCPUS];
    int threshold;
} counter_t;

有一个全局计数器global，全局锁glock，只在更新全局计数器时使用；
有一系列的局部计数器local[i]和局部锁llock[i]，在更新局部计数器时上锁

更新逻辑：

void increment(counter_t *c, int cpu, int amt) {
    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;
    }
    pthread_mutex_unlock(&c->llock[cpu]);
}

即：

先给局部计数器上锁，amt是每次的递增数，cpu可以理解为线程的编号
如果局部计数器的大小大于了阈值threshold，则更新给全局计数器，这里也需要上锁
最后得到的全局计数器是approximate的

trade-offs：

通过这样的设计，防止了所有线程一直竞争同一把锁
但是得到的结果是不精确的
阈值的设置也有权衡：
- 阈值大，性能好，但是全局计数器可能滞后
- 阈值小则反之

并发链表

最简单的并发链表：上一把大锁：

typedef struct __list_t {
    node_t *head;
    pthread_mutex_t lock;
} list_t;

插入：

void insert(list_t *l, int key) {
	node_t *new = malloc(sizeof(node_t));
	if (new == NULL) {
        return -1;
    }
    new->key = key;
    pthread_mutex_lock(&l->lock);
    new->next = l->head;
    l->head = new;
    pthread_mutex_unlock(&l->lock);
}

这里注意并发的写法，我们只把修改整个链表的临界区上锁了，而把前面的部分抽离出来，真正修改共享链表时加锁。

Hand-over-hand Locking

这个例子主要是展示，锁拆太细也不好。

假设我们不是给链表的list_t结构体加锁，而是给node_t结构加锁，即链表的每个结点都有一个锁：

直觉上可以增加并发，但是在实践中会导致lock / unlock的开销远大于并发带来的收益

锁更细 ≠ 性能更好

并发队列

展示「合理拆锁」

typedef struct __queue_t {
    node_t *head;
    node_t *tail;
    pthread_mutex_t head_lock, tail_lock;
} queue_t;

入队只需要管头指针，出队只需要管尾指针
所以这样拆分锁，效率很高

并发哈希表

复用并发链表：

#define BUCKETS (101)

typedef struct __hash_t {
    list_t lists[BUCKETS];
} hash_t;

CRUD操作只操作某个bucket：

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);
}

性能好，因为不同的key会落到不同的bucket，bucket之间相互独立，不容易产生对锁的竞争。

总结

more concurrency isn't necessarily faster；

avoid premature optimization

第30章条件变量

回忆之前说的，并发的两个问题：一个是需要互斥，一个是需要等待/唤醒机制。条件变量就是解决等待/唤醒机制的。

条件变量解决的问题

假设我们有一个场景需求是父线程需要等子线程结束：

parent: begin;
child: operates;
parent: end;

我们可以使用一个共享变量解决此问题：

volatile int done = 0;

void *child(void *arg) {
    printf("child\n");
    done = 1;
    return NULL;
}

int main() {
    pthread_t c;
    printf("parent: begin\n");
    Pthread_create(&c, NULL, child, NULL);
    while (done == 0)
        ;   // spin
    printf("parent: end\n");
}

问题：父线程一直在空转，占着CPU。

因此引入睡眠/唤醒机制：当线程需要某个条件成立时，先睡眠而非一直spin，直至被唤醒。

条件变量

def: 条件变量是一个等待的集合。

线程发现自己所需要的条件还不成立，则：
- 把自己挂在这个条件变量的等待集合上
- 睡眠
之后另一个线程改变了状态，使得条件满足，则：
- 在这个条件变量上发 signal，唤醒一个等待的线程

这里以POSIX为例，睡眠和唤醒的API是：

int pthread_cond_wait(pthread_cond_t *cond,
                      pthread_mutex_t *mutex);
int pthread_cond_signal(pthread_cond_t *cond);

其实有返回值，但是我们下文不予检查。

wait：条件不成立，释放锁并睡眠（这个操作是原子的）
signal：状态变了，唤醒起来看看（不意味着继续，后面解释）

为什么条件变量一定要与锁连用

首先需要明晰一个事情，条件变量本身不表示条件，共享变量 / 状态变量表示条件。

给出一段正确的代码：

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

void thr_exit() {
    Pthread_mutex_lock(&m);
    done = 1;
    Pthread_cond_signal(&c);
    Pthread_mutex_unlock(&m);
}

void thr_join() {
    Pthread_mutex_lock(&m);
    while (done == 0)
        Pthread_cond_wait(&c, &m);
    Pthread_mutex_unlock(&m);
}

如果在thr_join()处不加锁，则：

done == 0成立，准备wait；
但是此时被切换（注意这里没加锁导致的）；
子线程运行得到done == 1，发送signal；
但是此时由于cond_t c上没有挂载线程，相当于没有发送；
然后切回父线程，直接wait，进入睡眠；
之后再也无法苏醒，因为无法被signal。

而且，wait里面也有一个参数是锁（回忆wait的作用），所以使用锁是强制的。

生产者-消费者问题

问题背景

假设有一个共享缓冲区，此处里面一次只能放一个整数，这里不妨直接写成int count，值取0或1：

生产者线程负责往里面put一个值，且只有在缓冲区不满的时候，才能put；
消费者线程负责从里面get一个值，且只有在缓冲区非空的时候才能get。

即，以这里的简化模型为例：

生产者需要等待count == 0这个条件；
消费者需要等待count == 1这个条件。

需要使用条件变量。

为什么使用`while`而非`if`

简而言之就是，被唤醒，不意味着条件仍然成立，使用while可以使得每次醒来之后重新检查条件。

一个条件变量可能不够

while (count == 1)
    wait(cond, mutex);   // producer

while (count == 0)
    wait(cond, mutex);   // consumer

这样的代码可能有bug，因为生产者和消费者挂载在同一个cond上等待，唤醒的次序会有问题。

所以，不同的等待队列，应该使用不同的条件变量：

代码C · 25 行

cond_t empty, fill;
mutex_t mutex;

void *producer(void *arg) {
    for (...) {
        Pthread_mutex_lock(&mutex);
        while (count == 1)
            Pthread_cond_wait(&empty, &mutex);
        put(i);
        Pthread_cond_signal(&fill);
        Pthread_mutex_unlock(&mutex);
    }
}

void *consumer(void *arg) {
    for (...) {
        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);
    }
}

生产者等待空位，睡眠挂在在empty上，反之。

第31章信号量

def: 一个带整数值的同步对象，表示「当前还剩多少可用资源」。

当 S.value > 0 时，表示可立即分配的资源数；
当 S.value = 0 时，表示没有可用资源；
当 S.value < 0 时，其绝对值表示正在等待该信号量的线程/进程数。

两个核心操作：

sem_wait(&s)：信号量减一，如果减完之后信号量为负数，则睡眠等待；
sem_post(&s)：信号量加一，如果之后信号量仍为非正数，则唤醒一个等待的线程。

信号量的初始值的设置很重要。

信号量用作互斥锁 - 二值信号量

sem_t m;
sem_init(&m, 0, 1);

sem_wait(&m);
// critical section
sem_post(&m);

为什么初始值是 1？因为锁一开始是空闲的，允许第一个线程直接拿走这一个许可。

如果线程 A 先执行 sem_wait()：

信号量从 1 变成 0
A 不阻塞，进入临界区

如果线程 B 此时也执行 sem_wait()：

信号量从 0 变成 -1
B 发现值为负，睡眠等待

等 A 执行 sem_post()：

信号量从 -1 变成 0
唤醒 B

信号量用作条件变量 - 顺序控制

信号量初值为0。

sem_t s;

void *child(void *arg) {
    printf("child\n");
    sem_post(&s);
    return NULL;
}

int main() {
    sem_init(&s, 0, 1);
    printf("parent: begin\n");
    Pthread_create(&c, NULL, child, NULL);
    sem_wait(&s);
    printf("parent: end\n");
}

这里初始值为什么是 0？因为一开始子线程还没完成，父线程没有任何“完成信号”可以拿。

有两种执行顺序：

父线程先 sem_wait()：信号量变成负数，父线程睡眠；子线程完成后 sem_post() 唤醒父线程。
子线程先 sem_post()：信号量变成 1；父线程后来 sem_wait() 时直接通过。

信号量用作条件变量 - 生产者消费者问题

int buffer[MAX]; // 共享缓冲区有MAX个槽位
int fill = 0; // use = MAX - 1 - fill，理解为可用的

生产者需要等use不为0
消费者需要等fill不为0

我们的最终方案：

代码C · 21 行

sem_t empty; // 空槽数量
sem_t full;  // 已填充槽数量
sem_t mutex;

sem_init(&empty, 0, MAX);
sem_init(&full, 0, 0);
sem_init(&mutex, 0, 1);

// 生产者
sem_wait(&empty);
sem_wait(&mutex);
put(i);
sem_post(&mutex);
sem_post(&full);

// 消费者
sem_wait(&full);
sem_wait(&mutex);
tmp = get();
sem_post(&mutex);
sem_post(&empty);

如果没有mutex作为互斥锁，由于put操作不是原子的（更新buffer和更新fill的过程中间不原子）。假设有两个生产者，可能会导致前者put的值被后者put的值覆盖。
如果sem_wait(&empty); sem_wait(&mutex);交换位置，会导致先拿到mutex锁，然后睡着的事情，导致死锁。

读者-写者锁

对于一个数据结构，需要：

多个读者可以同时进入
写者必须独占
读者和写者不能同时进入

实现方式：

代码C · 35 行

typedef struct __rwlock_t {
    sem_t lock;
    sem_t writelock;
    int readers;
} rwlock_t;

void rwlock_init(rwlock_t *lock) {
    lock->readers = 0;
    Sem_init(&lock->lock, 1); 
    Sem_init(&lock->writelock, 1); 
}

void rwlock_acquire_readlock(rwlock_t *lock) {
    Sem_wait(&lock->lock);
    lock->readers++;
    if (lock->readers == 1)
		Sem_wait(&lock->writelock);
    Sem_post(&lock->lock);
}

void rwlock_release_readlock(rwlock_t *lock) {
    Sem_wait(&lock->lock);
    lock->readers--;
    if (lock->readers == 0)
		Sem_post(&lock->writelock);
    Sem_post(&lock->lock);
}

void rwlock_acquire_writelock(rwlock_t *lock) {
    Sem_wait(&lock->writelock);
}

void rwlock_release_writelock(rwlock_t *lock) {
    Sem_post(&lock->writelock);
}

简而言之：

第一个读者拿住writelock，阻止写者进入；
后续的读者直接进入即可（因为每个读者读完之后会释放lock，但是第一个读者不会释放writelock）
最后一个读者释放writelock，允许写者进入

但是有公平性问题：如果读者一直源源不断进来，写者会饿死。

解决方案：设定一个readers值的上限。

哲学家就餐问题

五个人围着桌子，每两个人之间有一把叉子。每个人吃饭需要左右两把叉子。目的：如何实现getforks()和putforks()函数，保证没有死锁，没有人饿死，并发度最高。

sem_t forks[5];
int left(p) {return p;}
int right(p) {return (p + 1) % 5;}
void getforks() {
    sem_wait(forks[left(p)]);
    sem_wait(forks[right(p)]);
}
void putforks() {
    sem_post(forks[left(p)]);
    sem_post(forks[right(p)]);
}

这个解决方案有问题：死锁：每个人都拿到了左边的但是都在等右边的，就会这样死下去。

解决方案：对于某个特定的人，让其先拿右边再拿左边。

限制并发数

sem_t throttle;
sem_init(&throttle, 0, 10);

...
sem_wait(&throttle);
... code
sem_post(&throttle);

这样就限制了最多有10个线程同时运行这段代码。

如何实现信号量

typedef struct {
    int value;
    pthread_cond_t cond;
    pthread_mutex_t lock;
} Zem_t;

wait：

Mutex_lock(&s->lock);
while (s->value <= 0)
    Cond_wait(&s->cond, &s->lock);
s->value--;
Mutex_unlock(&s->lock);

post：

Mutex_lock(&s->lock);
s->value++;
Cond_signal(&s->cond);
Mutex_unlock(&s->lock);

第32章常见并发bug

非死锁问题

违反原子性

一个MySQL的例子：

// Thread 1
if (thd->proc_info) {
    fputs(thd->proc_info, ...);
}

// Thread 2
thd->proc_info = NULL;

对于Thread 1，这一刻检查到thd->proc_info非空，但是下一刻被切换到Thread 2，将thd->proc_info置空，后面显然会出严重的bug。

解决方案：加一把锁

// thread 1
pthread_mutex_lock(&proc_info_lock);
if (thd->proc_info) {
    fputs(thd->proc_info, ...);
}
pthread_mutex_unlock(&proc_info_lock);

// thread 2
pthread_mutex_lock(&proc_info_lock);
thd->proc_info = NULL;
pthread_mutex_unlock(&proc_info_lock);

顺序混乱

使用合适的条件变量进行约束即可。

死锁问题

举个例子：

// Thread 1
pthread_mutex_lock(L1);
pthread_mutex_lock(L2);

// Thread 2
pthread_mutex_lock(L2);
pthread_mutex_lock(L1);

如果 Thread 1 先拿到 L1，Thread 2 先拿到 L2，就会变成：

Thread 1 拿着 L1，等待 L2
Thread 2 拿着 L2，等待 L1

这样就形成了死锁。

死锁发生的条件

必须同时满足这四个条件：

Mutual exclusion：资源是互斥的，比如锁一次只能被一个线程持有。
Hold-and-wait：线程拿着一个资源，又去等另一个资源。
No preemption：资源不能被强行从持有者手里抢走，比如锁只能由持有者释放。
Circular wait：多个线程形成循环等待链。

只要破坏其中任意一个条件，死锁就不会发生。

如何预防死锁

破坏 circular wait

形成循环等待链的问题主要是拿锁的顺序不固定，解决方案是所有线程拿锁的顺序都固定，这样就不可能形成环。

比如，所有线程都规定好，拿锁的时候先拿L1再拿L2。

破坏 hold-and-wait

用一个全局的锁，让「拿锁」这个过程变得原子化：

pthread_mutex_lock(prevention);
pthread_mutex_lock(L1);
pthread_mutex_lock(L2);
pthread_mutex_unlock(prevention);

缺点：并发性下降。

trylock

拿不到第二把锁，就放弃重来：

top:
pthread_mutex_lock(L1);
if (pthread_mutex_trylock(L2) != 0) {
    pthread_mutex_unlock(L1);
    goto top;
}

但可能出现 livelock：两个线程都一直让步、重试，但谁也没真正推进。它不是卡死，因为线程还在运行；但没有进展。

避免 mutual exclusion

就是说，用硬件原语构建原子指令，不用锁。

第33章基于事件的并发

前面几张说「并发=多线程+锁+信号量+条件变量」，其实并非唯一办法。

其实之前JavaScript笔记 - JavaScript的异步部分详细介绍了这一种实现异步的方法。

基本就是：有一个while(1)的事件循环，有一个消息队列（事件队列），事件循环不停遍历事件队列。

所以这里不再赘述了。

课程笔记 > 操作系统

#操作系统 #ostep #课程笔记

一套速通式学习教材的workflow - 以学习OSTEP为例上一篇

Codex - Docs Note 下一篇

OSTEP - 并发部分笔记

第26章 并发：介绍

线程

使用线程的理由

并发的问题

核心术语总结

原子性

第27章 线程API

创建线程pthread_create()

等待线程结束pthread_join()

互斥锁

条件变量

Thread API 使用原则

第28章 锁

锁的基本思想

锁的评价标准

控制中断

test-and-set

其他硬件原语

Compare and Swap

Load-Linked / Store-Conditional

Fetch-and-Add

解决自旋过多的问题

两阶段锁

第29章 基于锁的并发数据结构

并发计数器

Approximate Counter

并发链表

Hand-over-hand Locking

并发队列

并发哈希表

总结

第30章 条件变量

条件变量解决的问题

条件变量

为什么条件变量一定要与锁连用

生产者-消费者问题

问题背景

为什么使用while而非if

一个条件变量可能不够

第31章 信号量

信号量用作互斥锁 - 二值信号量

信号量用作条件变量 - 顺序控制

信号量用作条件变量 - 生产者消费者问题

读者-写者锁

哲学家就餐问题

限制并发数

如何实现信号量

第32章 常见并发bug

非死锁问题

违反原子性

顺序混乱

死锁问题

死锁发生的条件

如何预防死锁

破坏 circular wait

破坏 hold-and-wait

trylock

避免 mutual exclusion

第33章 基于事件的并发

第26章并发：介绍

第27章线程API

创建线程`pthread_create()`

等待线程结束`pthread_join()`

第28章锁

第29章基于锁的并发数据结构

第30章条件变量

为什么使用`while`而非`if`

第31章信号量

第32章常见并发bug

第33章基于事件的并发