System|Concurrency|条件变量
System|Concurrency|条件变量
摘要
本文介绍了条件变量的使用场景，并介绍了条件变量的简易实现机制。
首先从一个经典的并发问题引入
Bounded-buffer problem
有界缓冲区问题，sender向buffer中添加数据，receiver从buffer中取出数据。以两个索引in，out作为未读取数据的上下边界，buf作为存储未读取数据的缓冲区。
在单sender和单receiver的情况下，无需加锁。但是多sender时，则需要对于send操作进行加锁。
send(bb, message):
acquire(bb.send_lock)
while True:
if bb.in – bb.out < N:
bb.buf[bb.in mod N] <- message
bb.in <- bb.in + 1
release(bb.send_lock)
return但是，在buffer满时，我们并不希望sender继续轮询，而是希望直接调度给receiver。这就导致了Yield系统调用。（receiver同理）
Yield
yield():
acquire(t_lock)
"""Suspend running thread"""
id = cpus[CPU].thread
threads[id].state = RUNNABLE
threads[id].sp = SP
"""Choose new thread"""
do:
id = (id + 1) mod N
while threads[id].state != RUNNABLE
"""Resume new thread"""
SP = threads[id].sp
threads[id].state = RUNNING
cpus[CPU].thread = id
release(t_lock)这个系统调用将暂停当前线程，并选择一个新线程进行调度。修改之后的代码为：
send(bb, msg):
acquire(bb.lock)
while True:
if bb.in - bb.out < N:
bb.buf[bb.in mod N] <- msg
bb.in <- bb.in + 1
release(bb.lock)
return
release(bb.lock)
yield()
acquire(bb.lock)问题在于，在yield之后，被唤醒的线程未必就能够满足条件能够执行。我们实际上期望当sender被唤醒时，buf必然不是满的，而yield并不能提供这样的信息。这样事实上执行了一些没有必要的acquire和条件判断，影响到了性能。
Condition Variables
Naive API
WAIT(cv): yield processor and wait to be notified of cv（封装yield）
NOTIFY(cv): notify waiting threads of cv
send(bb, msg):
acquire(bb.lock)
while True:
if bb.in - bb.out < N:
bb.buf[bb.in mod N] <- msg
bb.in <- bb.in + 1
release(bb.lock)
notify(bb.not_empty)
return
release(bb.lock)
wait(bb.not_full)
acquire(bb.lock)The Lost Notify Problem
这三行代码release(bb.lock) wait(bb.not_full) acquire(bb.lock)
如果buf full，在sender wait前，receiver已经notify，此时buf empty，receiver也wait，然后sender wait（错过了notify），那么sender和receiver互相等待，永久休眠。
最好的方法就是避免中间被插入其他代码。
Updated API
我们需要把release(bb.lock) wait(bb.not_full) acquire(bb.lock)变为一个原子性操作。（这个API其实就是Pthread-cond-wait/Pthred-cond-signal，现在明白为啥要加入一个mutex么）
WAIT(cv, lock): release lock, yield CPU, wait to be notified
NOTIFY(cv): notify waiting threads of cv
yield_wait():
"""Suspend running thread"""
id = cpus[CPU].thread
threads[id].sp = SP
"""Choose new thread"""
do:
id = (id + 1) mod N
while threads[id].state != RUNNABLE
"""Resume new thread"""
SP = threads[id].sp
threads[id].state = RUNNING
cpus[CPU].thread = id
----------------------------------------------------------------------------
wait(cv, lock):
acquire(t_lock)
release(lock)
threads[id].cv = cv
threads[id].state = WAITING
yield_wait()
release(t_lock)
acquire(lock)
---------------------------------------------------------------------------
notify(cv):
acquire(t_lock)
for i = 0 to N-1:
if threads[i].cv == cv && threads[i].state == WAITING:
threads[i].state = RUNNABLE
release(t_lock)这个新实现引入了WAITING状态，以避免yield会唤醒，当notify时将WAITING的线程变为Runnable。
do:
id = (id + 1) mod N
while threads[id].state != RUNNABLE然而在这里，如果所有的线程都在RUNNING或WAITING，那么拿着锁的yield_wait将会陷入死循环，因此我们需要在循环体中插入释放锁的代码，否则t_clock将永远不会被释放。
do:
id = (id + 1) mod N
release(t_lock)
acquire(t_lock)
while threads[id].state != RUNNABLE但这也导致了连锁反应，在放了锁之后，如果当前线程的yield_wait还没有结束，而另一个CPU执行了yield_wait，那么另一个CPU使用了原进程的栈(threads[id].sp = SP)，导致污染。（因为本应发生的SP = threads[id].sp没有执行）
id= cpus[CPU].thread
threads[id].sp = SP
SP = cpus[CPU].stack为了避免这个问题，我们在问题代码的前面加入一个临时栈，充当保护，以避免原线程的栈被污染。在找到新进程后，再转入新进程的栈。
Preemption
线程管理器强制某个线程运行一段时间后yield，我们假设一段简单的中断实现，timer interrupt在线程运行一段时间后，强制线程移交给另一线程，防止starvation。
timer_interrupt():
push PC // done by CPU
push registers // this is not a function call;
// cannot assume compiler saves registers
yield()
pop registers
pop PC由于使用yield需要获取t_lock，一旦在yield_wait时发生中断，将会导致dead lock。
解决方法很简单，在获取锁时顺便disable interrupt，释放时顺便enable interrupt。
do:
id = (id + 1) mod N
release(t_lock)
enable_interrupt()
disable_interrupt()
acquire(t_lock)
while threads[id].state != RUNNABLE如果在CPU0执行线程1的yield_wait时，在上面代码锁释放后，另一个CPU1看到线程1是WAITING，使用NOTIFY唤醒了线程1，并且CPU0中又触发了一次中断，导致线程1进入RUNNABLE，然后CPU0又唤醒一次线程1.
此时两个CPU都在运行线程1，无疑是很危险的。
因此我们增加一个CPU独有的临时栈作为保护，它只存在于这个过程中，目的是防止中断对于原线程进行修改。
id= cpus[CPU].thread
cpus[CPU].thread = null
threads[id].sp = SP
SP = cpus[CPU].stack