313 lines
12 KiB
ReStructuredText
313 lines
12 KiB
ReStructuredText
条件变量机制
|
||
=========================================
|
||
|
||
本节导读
|
||
-----------------------------------------
|
||
|
||
到目前为止，我们已经了解了操作系统提供的互斥锁和信号量。但应用程序在使用这两者时需要非常小心，
|
||
如果使用不当，就会产生效率低下、竞态条件、死锁或者其他一些不可预测的情况。为了简化编程、避免错误，
|
||
计算机科学家针对某些情况设计了一种更高层的同步互斥原语。具体而言，在有些情况下，
|
||
线程需要检查某一条件（condition）满足之后，才会继续执行。
|
||
|
||
我们来看一个例子，有两个线程 first 和 second 在运行，线程 first 会把全局变量 A 设置为
|
||
1，而线程 second 在 ``A != 0`` 的条件满足后，才能继续执行，如下面的伪代码所示：
|
||
|
||
.. code-block:: C
|
||
:linenos:
|
||
|
||
static int A = 0;
|
||
void first() {
|
||
A=1;
|
||
...
|
||
}
|
||
|
||
void second() {
|
||
while (A==0) {
|
||
// 忙等或睡眠等待 A==1
|
||
};
|
||
//继续执行相关事务
|
||
}
|
||
|
||
在上面的例子中，如果线程 second 先执行，会忙等在 while 循环中，在操作系统的调度下，线程
|
||
first 会执行并把 A 赋值为 1 后，然后线程 second 再次执行时，就会跳出 while 循环，进行接下来的工作。
|
||
配合互斥锁，可以正确完成上述带条件的同步流程，如下面的伪代码所示：
|
||
|
||
.. code-block:: C
|
||
:linenos:
|
||
|
||
static int A = 0;
|
||
void first() {
|
||
mutex.lock();
|
||
A=1;
|
||
mutex.unlock();
|
||
...
|
||
}
|
||
|
||
void second() {
|
||
mutex.lock();
|
||
while A == 0 {
|
||
mutex.unlock();
|
||
// give other thread chance to lock
|
||
mutex.lock();
|
||
}
|
||
mutex.unlock();
|
||
//继续执行相关事务
|
||
}
|
||
|
||
这种实现能执行，但效率低下，因为线程 second 会忙等检查，浪费处理器时间。我们希望有某种方式让线程
|
||
second 休眠，直到等待的条件满足，再继续执行。于是，我们可以写出如下的代码：
|
||
|
||
.. code-block:: C
|
||
:linenos:
|
||
|
||
static int A = 0;
|
||
void first() {
|
||
mutex.lock();
|
||
A=1;
|
||
wakup(second);
|
||
mutex.unlock();
|
||
...
|
||
}
|
||
|
||
void second() {
|
||
mutex.lock();
|
||
while (A==0) {
|
||
wait();
|
||
};
|
||
mutex.unlock();
|
||
//继续执行相关事务
|
||
}
|
||
|
||
粗略地看，这样就可以实现睡眠等待了。但请同学仔细想想，当线程 second 在睡眠的时候， ``mutex``
|
||
是否已经上锁了？ 确实，线程 second 是带着上锁的 ``mutex`` 进入等待睡眠状态的。
|
||
如果这两个线程的调度顺序是先执行线程 second，再执行线程first，那么线程 second 会先睡眠且拥有
|
||
``mutex`` 的锁；当线程 first 执行时，会由于没有 ``mutex`` 的锁而进入等待锁的睡眠状态。
|
||
结果就是两个线程都睡了，都执行不下去，这就出现了 **死锁** 。
|
||
|
||
这里需要解决的两个关键问题： **如何等待一个条件？** 和 **在条件为真时如何向等待线程发出信号** 。
|
||
我们的计算机科学家给出了 **管程（Monitor）** 和 **条件变量（Condition Variables）**
|
||
这种巧妙的方法。接下来，我们就会深入讲解条件变量的设计与实现。
|
||
|
||
条件变量的基本思路
|
||
-------------------------------------------
|
||
|
||
管程有一个很重要的特性，即任一时刻只能有一个活跃线程调用管程中的过程，
|
||
这一特性使线程在调用执行管程中过程时能保证互斥，这样线程就可以放心地访问共享变量。
|
||
管程是编程语言的组成部分，编译器知道其特殊性，因此可以采用与其他过程调用不同的方法来处理对管程的调用.
|
||
因为是由编译器而非程序员来生成互斥相关的代码，所以出错的可能性要小。
|
||
|
||
管程虽然借助编译器提供了一种实现互斥的简便途径，但这还不够，还需要一种线程间的沟通机制。
|
||
首先是等待机制：由于线程在调用管程中某个过程时，发现某个条件不满足，那就在无法继续运行而被阻塞。
|
||
其次是唤醒机制：另外一个线程可以在调用管程的过程中，把某个条件设置为真，并且还需要有一种机制，
|
||
及时唤醒等待条件为真的阻塞线程。为了避免管程中同时有两个活跃线程，
|
||
我们需要一定的规则来约定线程发出唤醒操作的行为。目前有三种典型的规则方案：
|
||
|
||
- Hoare 语义：线程发出唤醒操作后，马上阻塞自己，让新被唤醒的线程运行。注：此时唤醒线程的执行位置还在管程中。
|
||
- Hansen 语义：是执行唤醒操作的线程必须立即退出管程，即唤醒操作只可能作为一个管程过程的最后一条语句。
|
||
注：此时唤醒线程的执行位置离开了管程。
|
||
- Mesa 语义：唤醒线程在发出行唤醒操作后继续运行，并且只有它退出管程之后，才允许等待的线程开始运行。
|
||
注：此时唤醒线程的执行位置还在管程中。
|
||
|
||
一般开发者会采纳 Brinch Hansen 的建议，因为它在概念上更简单，并且更容易实现。这种沟通机制的具体实现就是
|
||
**条件变量** 和对应的操作：wait 和 signal。线程使用条件变量来等待一个条件变成真。
|
||
条件变量其实是一个线程等待队列，当条件不满足时，线程通过执行条件变量的 wait
|
||
操作就可以把自己加入到等待队列中，睡眠等待（waiting）该条件。另外某个线程，当它改变条件为真后，
|
||
就可以通过条件变量的 signal 操作来唤醒一个或者多个等待的线程（通过在该条件上发信号），让它们继续执行。
|
||
|
||
早期提出的管程是基于 Concurrent Pascal 来设计的，其他语言如 C 和 Rust 等，并没有在语言上支持这种机制。
|
||
我们还是可以用手动加入互斥锁的方式来代替编译器，就可以在 C 和 Rust 的基础上实现原始的管程机制了。
|
||
在目前的 C 语言应用开发中，实际上也是这么做的。这样，我们就可以用互斥锁和条件变量，
|
||
来重现上述的同步互斥例子：
|
||
|
||
.. code-block:: C
|
||
:linenos:
|
||
|
||
static int A = 0;
|
||
void first() {
|
||
mutex.lock();
|
||
A=1;
|
||
condvar.wakup();
|
||
mutex.unlock();
|
||
...
|
||
}
|
||
|
||
void second() {
|
||
mutex.lock();
|
||
while (A==0) {
|
||
condvar.wait(mutex); //在睡眠等待之前，需要释放mutex
|
||
};
|
||
mutex.unlock();
|
||
//继续执行相关事务
|
||
}
|
||
|
||
有了上面的介绍，我们就可以实现条件变量的基本逻辑了。下面是条件变量的 wait 和 signal 操作的伪代码：
|
||
|
||
.. code-block:: C
|
||
:linenos:
|
||
|
||
void wait(mutex) {
|
||
mutex.unlock();
|
||
<block and enqueue the thread>;
|
||
mutex.lock();
|
||
}
|
||
|
||
void signal() {
|
||
<unblock a thread>;
|
||
}
|
||
|
||
条件变量的wait操作包含三步，1. 释放锁；2. 把自己挂起；3. 被唤醒后，再获取锁。条件变量的 signal
|
||
操作只包含一步：找到挂在条件变量上睡眠的线程，把它唤醒。
|
||
|
||
注意，条件变量不像信号量那样有一个整型计数值的成员变量，所以条件变量也不能像信号量那样有读写计数值的能力。
|
||
如果一个线程向一个条件变量发送唤醒操作，但是在该条件变量上并没有等待的线程，则唤醒操作实际上什么也没做。
|
||
|
||
实现条件变量
|
||
-------------------------------------------
|
||
|
||
使用 condvar 系统调用
|
||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||
|
||
我们通过例子来看看如何实际使用条件变量。下面是面向应用程序对条件变量系统调用的简单使用，
|
||
可以看到对它的使用与上一节介绍的信号量系统调用类似。 在这个例子中，主线程先创建了初值为 1
|
||
的互斥锁和一个条件变量，然后再创建两个线程 First 和 Second。线程 First 会先睡眠 10ms，而当线程
|
||
Second 执行时，会由于条件不满足执行条件变量的 wait 操作而等待睡眠；当线程 First 醒来后，通过设置
|
||
A 为 1，让线程 second 等待的条件满足，然后会执行条件变量的 signal 操作，从而能够唤醒线程 Second。
|
||
这样线程 First 和线程 Second 就形成了一种稳定的同步与互斥关系。
|
||
|
||
.. code-block:: C
|
||
:linenos:
|
||
:emphasize-lines: 34,44,39
|
||
|
||
static int A = 0; //全局变量
|
||
|
||
const int CONDVAR_ID = 0;
|
||
const int MUTEX_ID = 0;
|
||
|
||
void first() {
|
||
sleep(10);
|
||
puts("First work, Change A --> 1 and wakeup Second");
|
||
mutex_lock(MUTEX_ID);
|
||
A = 1;
|
||
condvar_signal(CONDVAR_ID);
|
||
mutex_unlock(MUTEX_ID);
|
||
...
|
||
}
|
||
void second() {
|
||
puts("Second want to continue,but need to wait A=1");
|
||
mutex_lock(MUTEX_ID);
|
||
while (A == 0) {
|
||
condvar_wait(CONDVAR_ID, MUTEX_ID);
|
||
}
|
||
mutex_unlock(MUTEX_ID);
|
||
...
|
||
}
|
||
int main() {
|
||
// create condvar & mutex
|
||
assert_eq(condvar_create(), CONDVAR_ID);
|
||
assert_eq(mutex_blocking_create(), MUTEX_ID);
|
||
// create first, second threads
|
||
...
|
||
}
|
||
|
||
int condvar_create()
|
||
{
|
||
return syscall(SYS_condvar_create);
|
||
}
|
||
|
||
int condvar_signal(int cid)
|
||
{
|
||
return syscall(SYS_condvar_signal, cid);
|
||
}
|
||
|
||
int condvar_wait(int cid, int mid)
|
||
{
|
||
return syscall(SYS_condvar_wait, cid, mid);
|
||
}
|
||
|
||
- 第 26 行，创建了一个 ID 为 ``CONDVAR_ID`` 的条件量，对应第 34 行 ``SYSCALL_CONDVAR_CREATE`` 系统调用；
|
||
- 第 19 行，线程 Second 执行条件变量 ``wait`` 操作（对应第 44 行 ``SYSCALL_CONDVAR_WAIT`` 系统调用），
|
||
该线程将释放 ``mutex`` 锁并阻塞；
|
||
- 第 5 行，线程 First 执行条件变量 ``signal`` 操作（对应第 39 行 ``SYSCALL_CONDVAR_SIGNAL`` 系统调用），
|
||
会唤醒等待该条件变量的线程 Second。
|
||
|
||
|
||
实现 condvar 系统调用
|
||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||
|
||
操作系统如何实现条件变量系统调用呢？在线程的眼里，条件变量是一种每个线程能看到的共享资源，
|
||
且在一个进程中，可以存在多个不同条件变量资源，所以我们可以把所有的条件变量资源放在一起让进程来管理，
|
||
如下面代码第9行所示。这里需要注意的是： ``condvar_list: Vec<Option<Arc<Condvar>>>``
|
||
表示的是条件变量资源的列表。而 ``Condvar`` 是条件变量的内核数据结构，由等待队列组成。
|
||
操作系统需要显式地施加某种控制，来确定当一个线程执行 ``wait`` 操作和 ``signal`` 操作时，
|
||
如何让线程睡眠或唤醒线程。在这里， ``wait`` 操作是由 ``Condvar`` 的 ``wait`` 方法实现，而 ``signal``
|
||
操作是由 ``Condvar`` 的 ``signal`` 方法实现。
|
||
|
||
.. code-block:: C
|
||
:linenos:
|
||
:emphasize-lines: 5,13,17,29,38
|
||
|
||
// os/proc.h
|
||
struct proc {
|
||
int pid; // Process ID
|
||
uint next_condvar_id;
|
||
struct condvar condvar_pool[LOCK_POOL_SIZE];
|
||
...
|
||
};
|
||
|
||
// os/sync.h
|
||
struct condvar {
|
||
struct queue wait_queue;
|
||
// "alloc" data for wait queue
|
||
int _wait_queue_data[WAIT_QUEUE_MAX_LENGTH];
|
||
};
|
||
|
||
// os/sync.c
|
||
struct condvar *condvar_create()
|
||
{
|
||
struct proc *p = curr_proc();
|
||
if (p->next_condvar_id >= LOCK_POOL_SIZE) {
|
||
return NULL;
|
||
}
|
||
struct condvar *c = &p->condvar_pool[p->next_condvar_id];
|
||
p->next_condvar_id++;
|
||
init_queue(&c->wait_queue, WAIT_QUEUE_MAX_LENGTH, c->_wait_queue_data);
|
||
return c;
|
||
}
|
||
|
||
void cond_signal(struct condvar *cond)
|
||
{
|
||
struct thread *t = id_to_task(pop_queue(&cond->wait_queue));
|
||
if (t) {
|
||
t->state = RUNNABLE;
|
||
add_task(t);
|
||
}
|
||
}
|
||
|
||
void cond_wait(struct condvar *cond, struct mutex *m)
|
||
{
|
||
// conditional variable will unlock the mutex first and lock it again on return
|
||
mutex_unlock(m);
|
||
struct thread *t = curr_thread();
|
||
// now just wait for cond
|
||
push_queue(&cond->wait_queue, task_to_id(t));
|
||
t->state = SLEEPING;
|
||
sched();
|
||
mutex_lock(m);
|
||
}
|
||
|
||
首先是核心数据结构：
|
||
|
||
- 第 5 行，进程控制块中管理的条件变量列表。
|
||
- 第 13 行，条件变量的核心数据成员：等待队列。
|
||
|
||
然后是重要的三个成员函数：
|
||
|
||
- 第 17 行，创建条件变量，即创建了一个空的等待队列。
|
||
- 第 29 行，实现 ``signal`` 操作，将从条件变量的等待队列中弹出一个线程放入线程就绪队列。
|
||
- 第 38 行，实现 ``wait`` 操作，释放 ``m`` 互斥锁，将把当前线程放入条件变量的等待队列，
|
||
设置当前线程为挂起状态并选择新线程执行。在恢复执行后，再加上 ``m`` 互斥锁。
|
||
|
||
Hansen, Per Brinch (1993). "Monitors and concurrent Pascal: a personal history". HOPL-II:
|
||
The second ACM SIGPLAN conference on History of programming languages. History of Programming
|
||
Languages. New York, NY, USA: ACM. pp. 1–35. doi:10.1145/155360.155361. ISBN 0-89791-570-4.
|