参考：《操作系统导论》《操作系统之哲学原理》《Linux高性能服务器编程》

由于没有具体的应用场景，笔者之前一直对锁、条件变量和信号量感觉迷迷糊糊，总觉得它们很相似但又有所区别。这两天在写线程池时需要用到任务队列，主线程生产任务，工作线程则竞争地从队列中取出任务——也就是我们常说的“生产者/消费者问题”，接触到这个具体的场景后，笔者突然就明白了它们的区别。

互斥锁用来保证多线程/进程之间对共享资源的互斥访问，也就是保证同一时刻只有一个执行流在临界区中。

POSIX 的互斥锁操作主要有如下几个函数：

1
2
3
4
5
6
7
8
9
10
//初始化锁,将锁的各个字段都初始化为0
int pthread_mutex_init(pthread_mutex_t* mutex, const pthread_mutexattr* attr);
//销毁锁,释放系统资源
int pthread_destroy(pthread_mutex_t* mutex);
//上锁
int pthread_lock(pthread_mutex_t* mutex);
//解锁
int pthread_unlock(pthread_mutex_t* mutex);
//非阻塞上锁,如果锁已经被持有,则返回错误EBUSY
int pthread_trylock(pthread_mutex_t* mutex);

以上函数成功时都返回 0，否则返回 errno 错误码。

  • 除了使用 pthread_mutex_init 初始化锁，也可以采用 PTHREAD_MUTEX_INITIALIZER 如下：

    1
    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

    注意，这种方式必须在初始化锁使用，不能在声明后赋值。

  • attr 是锁的属性，常用的有两种——pshared 和 type，前者指定锁是否跨进程共享，后者指定锁的类型。互斥锁的类型有普通所、检错锁、递归锁，这里笔者没有深入研究，读者可自行了解。正常情况下，attr 设置为 NULL 即可。

  • 需要注意，同一个执行流重复持有一个锁会导致死锁，此时需要使用递归锁。两个线程按照不同的顺序来申请两个互斥锁也可能导致死锁，具体情境参见《Linux高性能服务器编程》P278 。

条件变量则提供了一种线程之间的通知机制，当某个条件满足时再唤醒沉睡在这个条件上的线程。 就笔者遇到的场景来说，只有当任务队列中存在任务时，线程才能获取任务并继续它的工作，否则只能睡眠。相关函数如下：

1
2
3
4
5
int pthread_cond_init(pthread_cond_t* cond, const pthread_condattr_t* attr); //初始化条件变量
int pthread_cond_destroy(pthread_cond_t* cond); // 销毁条件变量,释放系统资源
int pthread_cond_wait(pthread_cond_t* cond, pthread_mutex_t* mutex); //在条件变量上睡眠
int pthread_cond_signal(pthread_cond_t* cond);//唤醒一个线程,具体唤醒哪个则取决于优先级和调度策略
int pthread_cond_broadcast(pthread_cond_t* cond);//唤醒所有线程

以上函数成功时都返回 0，否则返回 errno 错误码。

  • 同样可以使用如下方式初始化条件变量：

    1
    pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

  • 我们看到 pthread_cond_wait() 函数的第二个参数是互斥锁，那么为什么在条件变量上等待时还需要上锁呢？ 假设等待条件变量时不需要加锁，考虑这样的情形：线程 A 中某个时刻，条件变量 cond 还未成立，于是调用 pthread_cond_wait() 准备睡眠，而在**睡眠前一刻** ，执行流切换到线程 B，B 使条件变量 cond 成立，然后调用 pthread_cond_signal() 唤醒在 cond 上睡眠的线程，但线程 A 还没有完全睡眠，所以等待队列中没有可以唤醒的线程（这也就是所谓的 虚假唤醒 ）。问题来了，现在执行流切换到 A，接着 A 完全睡眠。于是，线程 A 错过了 B 发送的唤醒信号，继而引发死锁。上面的问题在于，从 pthread_cond_signal() 函数被执行到调用线程被放入等待队列的这段时间内条件变量发生了改变。 所以我们必须保证这段空窗期内条件变量不会被修改，这也就是 pthread_cond_wait() 中锁参数的作用。因此 调用 pthread_cond_wait() 前必须保证 mutex 已经上锁 。

    调用 pthread_cond_wait() 时加锁是强制要求（该函数的第二个参数），但 pthread_cond_signal() 则不一定需要在加锁时调用。但是作为一般化的规则，在 wait 和 signal 时都持有锁总是正确的 。

  • pthread_cond_wait() 在睡眠前一刻会释放锁，以使其他线程能够进入临界区；被唤醒后执行流从该函数返回，锁又会被该线程持有，以保证互斥访问临界区。

  • pthread_cond_broadcast() 函数有什么应用场景？参见《操作系统导论》 P261

下面我们来看条件变量的局限性。先引入具体代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
int food;
pthread_cond_t cond;
pthread_mutex_t mutex;
void* producer() //如果还有food,就不生产,等待没有food时再生产
{
    pthread_mutex_lock(&mutex);
    if(food == true)
        pthread_cond_wait(&cond, &mutex);
    add_food(&food);
    pthread_cond_signal(&cond);
    pthread_mutex_unlock(&mutex);
}
void* consumer() //如果没有food,就等待,直到producer发送信号
{
    pthread_mutex_lock(&mutex);
    if(food == false)
        pthread_cond_wait(&cond, &mutex);
    eat_food(&food);
    pthread_cond_signal(&cond);
    pthread_mutex_unlock(&mutex);
}

上面的代码在只有两个线程时能够准确无误的运行，但当线程数多于两个时，就会发生问题：消费者 A 发现没有 food，于是沉睡在条件变量 food 上，然后执行流转移到生产者，生产者 add_food 后唤醒消费者 A，于是消费者 A 进入就绪队列（只是就绪，但没有运行）。问题来了，此时消费者 B 抢先执行 ，发现 food 不为空，于是 eat_food 。紧接着切换到消费者 A ，从 17 行返回，由于 food 已经被消费者 B 吃掉，所以执行第 18 行 eat_food() 时将引发错误（只有 food 不为空时才能 eat）。

我们很容易知道解决上面问题的办法：不让消费者 B 抢先执行。也就是说，唤醒消费者 A 后立刻调度。 实际上，这种方式被称为 Hoare 语义，而前面的只唤醒，不保证立刻调度的方式称为 Mesa 语义。然而，由于 Hoare 语义的实现难度较大，几乎所有的操作系统都采用 Mesa 语义。

那么该如何解决这个问题呢？也很简单，将第 7、16 行的 if 改成 while 即可。消费者被唤醒后，总是再次检查共享变量 food，如果不满足，则再次睡眠。因此，谨记使用条件变量的通用规则：总是使用 while() 判断条件是否成立 。

上面的方案依旧存在问题：当消费者 A 发出 signal 时，会唤醒哪个线程呢？按道理来说应该是唤醒生产者，但线程调度不保证这一点。设想，当消费者 A 发出 signal 后唤醒的是消费者 B，而 B 发现 food 仍然为空，于是陷入睡眠；而生产者根本没有被唤醒，于是一直相互等待。这个问题的原因在于信号没有指向性 ，显然消费者不应该唤醒消费者，而应该唤醒生产者。解决办法也很简单——使用两个条件变量：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
int food;
pthread_cond_t empty;
pthread_cond_t filled;
pthread_mutex_t mutex;
void* producer() //如果还有food,就不生产,等待没有food时再生产
{
    pthread_mutex_lock(&mutex);
    if(food == true)
        pthread_cond_wait(&empty, &mutex); //如果还有food,就等待被消耗完再添加
    add_food(&food); 
    pthread_cond_signal(&full);
    pthread_mutex_unlock(&mutex);
}
void* consumer() //如果没有food,就等待,直到producer发送信号
{
    pthread_mutex_lock(&mutex);
    if(food == false)
        pthread_cond_wait(&filled, &mutex);
    eat_food(&food);
    pthread_cond_signal(&empty);
    pthread_mutex_unlock(&mutex);
}

现在我们提出下一个问题：为什么还要引入信号量？实际上，信号量不是必须的，它是对互斥锁和条件变量的封装 ，看看源码便知：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
int sem_wait(sem_t *sem)
{
    if (sem == NULL) {
        errno = EINVAL;
        return -1;
    }
    if (sem->pshared == PTHREAD_PROCESS_PRIVATE && sem->value <= 0) {
        pthread_mutex_lock(&sem->lock);
        while (sem->value <= 0) {
            pthread_cond_wait(&sem->cond, &sem->lock);
        }
        sem->value--;
        pthread_mutex_unlock(&sem->lock);
        return 0;
    } else {
        errno = ENOSYS;
        return -1;
    }
}

int sem_post(sem_t *sem)
{
    if (sem == NULL) {
        errno = EINVAL;
        return -1;
    }
    if (sem->pshared == PTHREAD_PROCESS_PRIVATE) {
        pthread_mutex_lock(&sem->lock);
        sem->value++;
        pthread_cond_signal(&sem->cond);
        pthread_mutex_unlock(&sem->lock);
        return 0;
    } else {
        errno = ENOSYS;
        return -1;
    }
}
1
2
3
4
5
6
typedef struct {
    int value;
    int pshared;
    pthread_mutex_t lock;
    pthread_cond_t cond;
} sem_t;

以上源码由 chatgpt 生成。

可以很清楚地看到，信号量内部使用了互斥锁和条件变量。信号量和条件变量的区别在于：

  1. 信号量内部使用了 value ，而条件变量是在外部使用 value （也就是上面的 food）来计数。

    之前代码中的 food 是二元变量（true\false），实际上你完全可以直接将它改为多值变量。

  2. 由于条件变量是在外部维护的 value，所以操作 value 时必须由程序员负责先持有锁。而信号量也会持有锁，只不过对程序员屏蔽了细节。

  3. 信号量只能一次唤醒一个特定的线程/进程，而条件变量可以广播。

二元信号量可以充当互斥锁，也能够充当条件变量。当信号量的 value 初始化为 1 时即为互斥锁。

信号量的 POSIX 函数如下：

1
2
3
4
5
int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_destroy(sem_t *sem);
int sem_wait(sem_t *sem);
int sem_post(sem_t *sem);
int sem_getvalue(sem_t *sem, int *sval);

上面函数成功时返回 0，失败时返回 -1，并设置 errno。

  • pshared 表示信号量的类型，可以是 PTHREAD_PROCESS_PRIVATE 或 PTHREAD_PROCESS_SHARED，分别表示进程内私有和进程间共享。value 表示信号量的初始值。
  • sem_wait() 将信号量的值减 1 。如果信号量的值为 0，则 sem_wait 陷入阻塞，直到信号量大于 0 。参见上面的源码。
    sem_post() 将信号量的值加 1，并唤醒一个线程。

为了更好地理解信号量，下面使用信号量改写之前的生产者/消费者代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
queue food; //food为一个队列,没有食物时队列为空
pthread_mutex_t mutex;
sem_t sem; //信号量的值即为队列中的食物量
void* producer() 
{
    pthread_mutex_lock(&mutex);
    food.pushback(); //向队列中添加food
    pthread_mutex_unlock(&mutex);
    sem_post(&sem);  //信号量+1,即食物量+1,并唤醒一个等待线程
}
void* consumer() 
{
    sem_wait(&sem); //将信号量-1,如果为0就等待
    pthread_mutex_lock(&mutex);
    food.pop(); //从队列中取出food
    pthread_mutex_unlock(&mutex);
}

个人认为当条件可累积时，信号量比条件变量更方便。从上面的代码也能够看出，即使使用了信号量，在操作共享资源时仍然必须锁来保证互斥访问 。

注意，为了避免死锁，请将互斥锁的获取和释放紧贴着临界区，务必不要将 sem_wait 和 sem_post 放入锁范围内 ！详细参见《操作系统导论》P271