条件变量机制#
本节导读#
到目前为止,我们已经了解了操作系统提供的互斥锁和信号量。但在某些情况下,应用程序在使用这两者时需要非常小心,如果使用不当,就会产生效率低下、竞态条件、死锁或者一些不可预测的情况。为了简化编程,避免错误,计算机科学家针对某些情况设计了一种高层的同步互斥原语。具体而言,在有些情况下,线程需要检查某一条件(condition)满足之后,才会继续执行。
我们来看一个例子,有两个线程first和second在运行,线程first会把全局变量 A设置为1,而线程second在 ! A == 0
的条件满足后,才能继续执行,如下面的伪代码所示:
1static mut A: usize = 0;
2unsafe fn first() -> ! {
3 A=1;
4 ...
5}
6
7unsafe fn second() -> ! {
8 while A==0 {
9 // 忙等或睡眠等待 A==1
10 };
11 //继续执行相关事务
12}
在上面的例子中,如果线程second先执行,会忙等在while循环中,在操作系统的调度下,线程first会执行并把A赋值为1后,然后线程second再次执行时,就会跳出while循环,进行接下来的工作。配合互斥锁,可以正确完成上述带条件的同步流程,如下面的伪代码所示:
1static mut A: usize = 0;
2unsafe fn first() -> ! {
3 mutex.lock();
4 A=1;
5 mutex.unlock();
6 ...
7}
8
9unsafe fn second() -> ! {
10 mutex.lock();
11 while A==0 { };
12 mutex.unlock();
13 //继续执行相关事务
14}
这种实现能执行,但效率低下,因为线程second会忙等检查,浪费处理器时间。我们希望有某种方式让线程second休眠,直到等待的条件满足,再继续执行。,于是,我们可以写出如下的代码:
1static mut A: usize = 0;
2unsafe fn first() -> ! {
3 mutex.lock();
4 A=1;
5 wakup(second);
6 mutex.unlock();
7 ...
8}
9
10unsafe fn second() -> ! {
11 mutex.lock();
12 while A==0 {
13 wait();
14 };
15 mutex.unlock();
16 //继续执行相关事务
17}
粗略地看,这样就可以实现睡眠等待了。但请同学仔细想想,当线程second在睡眠的时候,mutex是否已经上锁了? 确实,线程second是带着上锁的mutex进入等待睡眠状态的。如果这两个线程的调度顺序是先执行线程second,再执行线程first,那么线程second会先睡眠且拥有mutex的锁;当线程first执行是,会由于没有mutex的锁而进入等待锁的睡眠状态。结果就是两个线程都睡了,都执行不下去,这就出现了 死锁 。
这里需要解决的两个关键问题: 如何等待一个条件? 和 ** 在条件为真时如何向等待线程发出信号 ** 。我们的计算机科学家给出了 ** 管程(Monitor)** 和 ** 条件变量(Condition Variables)** 这种巧妙的方法。接下来,我们就会深入讲解条件变量的设计与实现。
条件变量的基本思路#
注解
Brinch Hansen(1973)和Hoare(1974)结合操作系统和Concurrent Pascal编程语言,提出了一种高级同步原语,称为管程(monitor)。一个管程是一个由过程(procedures,Pascal语言的术语,即函数)、共享变量及数据结构等组成的一个集合。线程可以调用管程中的过程,但线程不能在管程之外声明的过程中直接访问管程内的数据结构。
1monitor m1
2 integer i; //共享变量
3 condition c; //条件变量
4
5 procedure f1();
6 ... //对共享变量的访问,以及通过条件变量进行线程间的通知
7 end;
8
9 procedure f2();
10 ... //对共享变量的访问,以及通过条件变量进行线程间的通知
11 end;
12end monitor
管程有一个很重要的特性,即任一时刻只能有一个活跃线程调用管程中过程,这一特性使线程在调用执行管程中过程时能保证互斥,这样线程就可以放心地访问共享变量。管程是编程语言的组成部分,编译器知道其特殊性,因此可以采用与其他过程调用不同的方法来处理对管程的调用,比如编译器可以在管程中的每个过程的入口/出口处加上互斥锁的加锁/释放锁的操作。因为是由编译器而非程序员来生成互斥锁相关的代码,所以出错的可能性要小。
管程虽然借助编译器提供了一种实现互斥的简便途径,但这还不够,还需要一种线程间的沟通机制。首先是等待机制:由于线程在调用管程中某个过程时,发现某个条件不满足,那就在无法继续运行而被阻塞。这里需要注意的是:在阻塞之前,操作系统需要把进入管程的过程入口出的互斥锁给释放掉,这样才能让其他线程有机会调用管程的过程。
其次是唤醒机制:另外一个线程可以在调用管程的过程中,把某个条件设置为真,并且还需要有一种机制及时唤醒等待条件为真的阻塞线程。这里需要注意的是:唤醒线程(本身执行位置在管程的过程中)如果把阻塞线程(其执行位置还在管程的过程中)唤醒了,那么需要避免两个活跃的线程都在管程中导致互斥被破坏的情况。为了避免管程中同时有两个活跃线程,我们需要一定的规则来约定线程发出唤醒操作的行为。目前有三种典型的规则方案:
Hoare语义:线程发出唤醒操作后,马上阻塞自己,让新被唤醒的线程运行。注:此时唤醒线程的执行位置还在管程中。
Hansen语义:是执行唤醒操作的线程必须立即退出管程,即唤醒操作只可能作为一个管程过程的最后一条语句。注:此时唤醒线程的执行位置离开了管程。
Mesa语义:唤醒线程在发出行唤醒操作后继续运行,并且只有它退出管程之后,才允许等待的线程开始运行。注:此时唤醒线程的执行位置还在管程中。
一般开发者会采纳Brinch Hansen的建议,因为它在概念上更简单,并且更容易实现。这种沟通机制的具体实现就是 ** 条件变量 ** 和对应的操作:wait和signal。线程使用条件变量来等待一个条件变成真。条件变量其实是一个线程等待队列,当条件不满足时,线程通过执行条件变量的wait操作就可以把自己加入到等待队列中,睡眠等待(waiting)该条件。另外某个线程,当它改变条件为真后,就可以通过条件变量的signal操作来唤醒一个或者多个等待的线程(通过在该条件上发信号),让它们继续执行。
早期提出的管程是基于Concurrent Pascal来设计的,其他语言,如C和Rust等,并没有在语言上支持这种机制。我们还是可以用手动加入互斥锁的方式来代替编译器,就可以在C和Rust的基础上实现原始的管程机制了。在目前的C语言应用开发中,实际上也是这么做的。这样,我们就可以用互斥锁和条件变量来重现实现上述的同步互斥例子:
1static mut A: usize = 0;
2unsafe fn first() -> ! {
3 mutex.lock();
4 A=1;
5 condvar.wakup();
6 mutex.unlock();
7 ...
8}
9
10unsafe fn second() -> ! {
11 mutex.lock();
12 while A==0 {
13 condvar.wait(mutex); //在睡眠等待之前,需要释放mutex
14 };
15 mutex.unlock();
16 //继续执行相关事务
17}
有了上面的介绍,我们就可以实现条件变量的基本逻辑了。下面是条件变量的wait和signal操作的伪代码:
1fn wait(mutex) {
2 mutex.unlock();
3 <block and enqueue the thread>;
4 mutex.lock();
5}
6
7fn signal() {
8 <unblock a thread>;
9}
条件变量的wait操作包含三步,1. 释放锁;2. 把自己挂起;3. 被唤醒后,再获取锁。条件变量的signal操作只包含一步:找到挂在条件变量上睡眠的线程,把它唤醒。
注意,条件变量不像信号量那样有一个整型计数值的成员变量,所以条件变量也不能像信号量那样有读写计数值的能力。如果一个线程向一个条件变量发送唤醒操作,但是在该条件变量上并没有等待的线程,则唤醒操作实际上啥也没做。
实现条件变量#
使用condvar系统调用#
我们通过例子来看看如何实际使用条件变量。下面是面向应用程序对条件变量系统调用的简单使用,可以看到对它的使用与上一节介绍的信号量系统调用类似。 在这个例子中,主线程先创建了初值为1的互斥锁和一个条件变量,然后再创建两个线程 First和Second。线程First会先睡眠10ms,而当线程Second执行时,会由于条件不满足执行条件变量的wait操作而等待睡眠;当线程First醒来后,通过设置A为1,让线程second等待的条件满足,然后会执行条件变量的signal操作, 从而能够唤醒线程Second。这样线程First和线程Second就形成了一种稳定的同步与互斥关系。
1static mut A: usize = 0; //全局变量
2
3const CONDVAR_ID: usize = 0;
4const MUTEX_ID: usize = 0;
5
6unsafe fn first() -> ! {
7 sleep(10);
8 println!("First work, Change A --> 1 and wakeup Second");
9 mutex_lock(MUTEX_ID);
10 A=1;
11 condvar_signal(CONDVAR_ID);
12 mutex_unlock(MUTEX_ID);
13 ...
14}
15unsafe fn second() -> ! {
16 println!("Second want to continue,but need to wait A=1");
17 mutex_lock(MUTEX_ID);
18 while A==0 {
19 condvar_wait(CONDVAR_ID, MUTEX_ID);
20 }
21 mutex_unlock(MUTEX_ID);
22 ...
23}
24pub fn main() -> i32 {
25 // create condvar & mutex
26 assert_eq!(condvar_create() as usize, CONDVAR_ID);
27 assert_eq!(mutex_blocking_create() as usize, MUTEX_ID);
28 // create first, second threads
29 ...
30}
31
32pub fn condvar_create() -> isize {
33 sys_condvar_create(0)
34}
35pub fn condvar_signal(condvar_id: usize) {
36 sys_condvar_signal(condvar_id);
37}
38pub fn condvar_wait(condvar_id: usize, mutex_id: usize) {
39 sys_condvar_wait(condvar_id, mutex_id);
40}
第26行,创建了一个ID为 CONDVAR_ID 的条件量,对应的是第33行 SYSCALL_CONDVAR_CREATE 系统调用;
第19行,线程Second执行条件变量wait操作(对应的是第39行 SYSCALL_CONDVAR_WAIT 系统调用),该线程将释放mutex锁并阻塞;
第5行,线程First执行条件变量signal操作(对应的是第36行 SYSCALL_CONDVAR_SIGNAL 系统调用),会唤醒等待该条件变量的线程Second。
实现condvar系统调用#
操作系统如何实现条件变量系统调用呢?我们还是采用通常的分析做法:数据结构+方法,即首先考虑一下与此相关的核心数据结构,然后考虑与数据结构相关的相关函数/方法的实现。
在线程的眼里,条件变量 是一种每个线程能看到的共享资源,且在一个进程中,可以存在多个不同条件变量资源,所以我们可以把所有的条件变量资源放在一起让进程来管理,如下面代码第9行所示。这里需要注意的是: condvar_list: Vec<Option<Arc<Condvar>>> 表示的是条件变量资源的列表。而 Condvar 是条件变量的内核数据结构,由等待队列组成。操作系统需要显式地施加某种控制,来确定当一个线程执行wait操作和signal操作时,如何让线程睡眠或唤醒线程。在这里,wait操作是由Condvar的wait方法实现,而signal操作是由Condvar的signal方法实现。
1pub struct ProcessControlBlock {
2 // immutable
3 pub pid: PidHandle,
4 // mutable
5 inner: UPSafeCell<ProcessControlBlockInner>,
6}
7pub struct ProcessControlBlockInner {
8 ...
9 pub condvar_list: Vec<Option<Arc<Condvar>>>,
10}
11pub struct Condvar {
12 pub inner: UPSafeCell<CondvarInner>,
13}
14pub struct CondvarInner {
15 pub wait_queue: VecDeque<Arc<TaskControlBlock>>,
16}
17impl Condvar {
18 pub fn new() -> Self {
19 Self {
20 inner: unsafe { UPSafeCell::new(
21 CondvarInner {
22 wait_queue: VecDeque::new(),
23 }
24 )},
25 }
26 }
27 pub fn signal(&self) {
28 let mut inner = self.inner.exclusive_access();
29 if let Some(task) = inner.wait_queue.pop_front() {
30 add_task(task);
31 }
32 }
33 pub fn wait(&self, mutex:Arc<dyn Mutex>) {
34 mutex.unlock();
35 let mut inner = self.inner.exclusive_access();
36 inner.wait_queue.push_back(current_task().unwrap());
37 drop(inner);
38 block_current_and_run_next();
39 mutex.lock();
40 }
41}
首先是核心数据结构:
第9行,进程控制块中管理的条件变量列表。
第15行,条件变量的核心数据成员:等待队列。
然后是重要的三个成员函数:
第18行,创建条件变量,即创建了一个空的等待队列。
第27行,实现signal操作,将从条件变量的等待队列中弹出一个线程放入线程就绪队列。
第33行,实现wait操作,释放mutex互斥锁,将把当前线程放入条件变量的等待队列,设置当前线程为挂起状态并选择新线程执行。在恢复执行后,再加上mutex互斥锁。
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.