第10章：信号量与管程
信号量
信号量使用
互斥访问
条件同步
管程
经典同步问题
哲学家就餐问题
读者-写者问题
10.1 信号量（Semaphore）
回顾
- 并发问题
多线程并发导致资源竞争 - 同步概念
协调多线程对共享数据的访问
任何时刻只能有一个线程执行临界区代码 - 确保同步正确的方法
底层硬件支持
高层次的编程抽象
信号量
- 信号量是操作系统提供的一种协调共享资源访问的方法
软件同步是平等线程间的一种同步协商机制
操作系统是管理者，地位高于进程
用信号量表示系统资源的数量 - 由Dijkstra在20世纪60年代提出
早期的操作系统的主要同步机制
现在很少用（但在计算机科学研究还是非常重要） - 信号量是一种抽象的数据类型
由一个整型（sem）变量和两个原子操作组成
- P()（Prolaag（荷兰语尝试减少））
sem减1
如sem<0，进入等待，否则继续 - V()（Verhoog（荷兰语增加））
sem加1
如sem<=0，唤醒一个等待进程
信号量的特性
- 信号量是被保护的整数变量
初始化完成后，只能通过P()和V()操作修改
由操作系统保证，PV操作是原子操作 - P()可能阻塞，V()不会阻塞
- 通常假定信号量是“公平的”
线程不会被无限期阻塞在P()操作
假定信号量等待按先进先出排队
自旋锁是做不到先进先出的，因为自旋锁需要占用CPU随时地去查，有可能临界区使用者退出的时候它刚查完，下一个进入者是谁去查它就能进去
信号量的实现
ClassSemaphore {
int sem;
WaitQueue q;
}
Semaphore :: P() {
sem --;
if (sem<0) {
Add this thread t to q;
block(P);
}
}
Semaphore :: V() {
sem ++;
if (sem<=0) {
Remove a thread t from q;
wakeup(t);
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
10.2 信号量的使用
- 信号量分类
可分为两种信号量
二进制信号量：资源数目为0或1
资源信号量：资源数目为任何非负值 - 两者等价
基于一个可以实现另一个 - 信号量使用
互斥访问
临界区的互斥访问控制
条件同步
线程间的事件等待 - 用信号量实现临界区的互斥访问
//每类资源设置一个信号量，其初值为1
mutex = new Semaphore(1); //初始化
mutex -> P(); //第一个线程进来之后，由1变成0，能进去；若第二个线程也想进入临界区，由0变成-1，在P操作进入等待状态
critical section;
mutex -> V(); //第一次线程执行结束，进行V操作，使信号量计数由-1变成0， 唤醒第二个进程，则第二个进程就会在第一个进程结束时进入临界区
1
2
3
4
- 必须成对使用P()和V()操作
P()操作保证互斥访问临界资源
V()操作在使用后释放临界资源
PV操作不能次序错误、重复或遗漏
用信号量实现条件同步
//条件同步设置一个信号量，其初值为0
condition = new Semaphore(0);
//线程A 线程B
...M...
condition -> P();
...N... //使用数据 ...X... //准备数据
condition -> v();
...Y...
1
2
3
4
5
6
7
8
9
- 条件等待：线程A要等待线程B执行完X模块才能执行N模块
假设B先执行完X，此时信号量变成1，则N能直接往下执行
假设A先执行完M，此时信号量变成-1，阻塞，进入等待状态，然后进程B进行V操作，释放，此时信号量为0。若信号量小于或等于0，则说明有线程在等待，此时唤醒等待中的线程A，执行N模块
生产者-消费者问题
生产者 -> 缓冲区 -> 消费者
- 有界缓冲区的生产者-消费者问题描述
一个或多个生产者在生成数据后放在一个缓冲区里
单个消费者从缓冲区取出数据处理
任何时刻只能有一个生产者或消费者可访问缓冲区 - 问题分析
任何时刻只能有一个线程操作缓冲区（互斥访问）
缓冲区空时，消费者必须等待生产者（条件同步）
缓冲区满时，生产者必须等待消费者（条件同步） - 用信号量描述每个约束
二进制信号量 mutex
资源信号量 fullBuffers
资源信号量 emptyBuffers
//信号量实现
CLass BoundedBuffer {
mutex = new Semaphore(1);
fullBuffers = new Semaphore(0);
emptyBuffers = new Semaphore(n);
}
BoundedBuffer :: Deposit(c) {
emptyBuffer -> P();
mutex -> P();
Add c to the buffer;
mutex -> V();
fullBuffers -> V();
}
BoundedBuffer :: Remove(c) {
fullBuffer -> P();
mutex -> P();
Remove c from buffer;
mutex -> V();
emptyBuffers -> V();
}
- 使用信号量的困难
读/开发代码比较困难
程序员需要能运用信号量机制
容易出错
使用的信号量已经被另一个线程占用
忘记释放信号量
不能够处理死锁问题
10.3 管程（Monitor）
- 改进信号量在处理临界区的时候的一些麻烦
- 管程是一种用于多线程互斥访问共享资源的程序结构
采用面向对象方法，简化了线程的同步机制
任一时刻最多只有一个线程执行管理代码
正在管程中的线程可临时放弃管理的互斥访问，等待事件出现时恢复
注 - 一个线程在临界区执行，必须执行到它退出临界区，它才能放弃临界区的互斥访问，而管程允许在执行的过程当中，临时放弃。放弃之后其他线程就可进入管程区域
管程的使用
- 在对象/模块中，收集相关共享数据
- 管程组成
一个锁
控制管程代码的互斥访问 - 0个或多个条件变量
0个等同于临界区
管理共享数据的并发访问
- 条件变量（Condition Variable）
条件变量是管程内的等待机制
进入管程的线程因资源被占用而进入等待状态
每个条件变量表示一种等待原因，对应一个等待队列 - Wait()操作
将自己阻塞在等待队列中
唤醒一个等待着或释放管程的互斥访问 - Signal()操作
将等待队列中的一个线程唤醒
如果等待队列为空，则等同空操作
//条件变量实现
Class Condition {
int numWaiting = 0;
WaitQueue = q;
}
Condition :: wait(lock) {
numWaiting ++;
Add this thread t to q;
release(lock);
schedule();//need mutex
require(lock);
}
Condition :: Signal() {
if (numWaiting>0) {
Remove a thread from q;
wakeup(t);//need mutex
numWaiting --;
}
}
用管程解决生产者-消费者问题
ClassBoundedBuffer {
...
Lock lock;
int count = 0;
Condition notFull,notEmpty;
}
BoundedBuffer :: Deposit(c) {
lock -> Acquire();
while (count == n)
notFull.Wait(&lock);
Add c to the buffer;
count ++;
notEmpty.Signal();
lock -> Release();
}
BoundedBuffer :: Remove(c) {
lock -> Acquire(c);
while (count == 0)
notEmpty.Wait(&lock);
Remove c from buffer;
count --;
notFull.Signal();
lock -> Release();
}
管理条件变量的释放处理方式
Hansen 管程
l.acquire()
...
x.wait() //T1进入等待
//T2进入管程 l.acquire()
...
x.Signal()
...
//T1退出管程 l.release()
... //T1恢复管程执行
l.release()
Hoare 管程
l.acquire()
...
x.wait() //T1进入等待
//T2进入管程 l.acquire()
...
//T2进入等待 x.Signal()
... //T1恢复管程执行
l.release() //T1结束
... //T2恢复管程执行
l.release()
Hansen 管程和Hoare 管程
- Hansen 管程
主要用于真实操作系统和Java中
当前正在执行的线程优先级更高
效率更高，少了一次切换 - Hoare 管程
内部的线程优先执行
优先角度考虑更合理，更容易证明正确性 - Hansen 管程和Hoare 管程区别
Hansen-style：Deposit() {
lock -> acquire();
while (count == n) {
notFull.Wait(&lock);
}
Add thing;
count ++;
notEmpty.Signal();
lock -> Release();
}
Hoare-style：Deposit() {
lock -> acquire();
if (count == n) {
notFull.Wait(&lock);
}
Add thing;
count ++;
notEmpty.Signal();
lock -> Release();
}
10.4 经典同步问题
（一）哲学家就餐问题
- 问题描述
5个哲学家围绕着一张圆桌而坐
桌上放着5支叉子
每两个哲学家之间放1支
哲学家的动作包括思考和进餐
进餐时需要同时拿到左右两边的叉子
思考时将两支叉子放回原处
如何保证哲学家们的动作有序进行？
如：不出现永远都难不到叉子的情况
- 方案1
define N 5 //哲学家个数
semaphore fork[5]; //信号量初值为1
void philosopher(int i) //哲学家编号：0-4
while (TRUE)
{
think(); //哲学家在思考
P(fork[i]); //去拿左边的叉子
P(fork[(i+1)%N]); //去拿右边的叉子
eat(); //吃面条中
V(fork[i]); //放下左边的叉子
V(fork[(i+1)%N]); //放下右边的叉子
}
不正确，可能导致死锁
5个人同时拿到左边刀叉时，就会进入死锁状态
- 方案2
一旦无法充分利用系统资源，就把所有资源打成一个整包，只有一个进程可占用这一整包的资源，不管它是不是都需要，这种情况系统也能正常运行，只是效率低一些。
#define N 5 //哲学家个数
semaphore fork[5]; //信号量初值为1
semaphore mutex; //互斥信号量，初值为1
void philosopher(int i) //哲学家编号：0-4
while (TRUE)
{
think(); //哲学家在思考
P(mutex); //进入临界区，任一时刻只能有一个进程申请到二进制信号量
P(fork[i]); //去拿左边的叉子
P(fork[(i+1)%N]); //去拿右边的叉子
eat(); //吃面条中
V(fork[i]); //放下左边的叉子
V(fork[(i+1)%N]); //放下右边的叉子
V(mutex); //退出临界区
}
互斥访问正确，但每次只允许一个人进餐，性能不好
- 方案3
#define N 5 //哲学家个数
semaphore fork[5]; //信号量初值为1
void philosopher(int i) //哲学家编号：0-4
while (TRUE)
{
think(); //哲学家在思考
if (i%2 == 0) {
P(fork[i]); //去拿左边的叉子
P(fork[(i+1)%N]); //去拿右边的叉子
} else {
P(fork[(i+1)%N]); //去拿右边的叉子
P(fork[i]); //去拿左边的叉子
}
eat(); //吃面条中
V(fork[i]); //放下左边的叉子
V(fork[(i+1)%N]); //放下右边的叉子
}
没有死锁，可有多人同时访问
（二）读者-写者问题
- 问题描述
共享数据的两类使用者
读者：只读取数据，不修改
写者：读取和修改数据
读者-写者问题描述：对共享数据的读写
- “读-读”允许
同一时刻，允许有多个读者同时读 - “读-写”互斥
没有写者时读者才能读
没有读者时写者才能写 - “写-写”互斥
没有其他写者时写者才能写
用信号量解决读者-写者问题
用信号量描述每个约束
- 信号量WriteMutex
控制读写操作互斥
初始化为1 - 读者计数Rcount
正在进行读操作的读者数目
初始化为0 - 信号量CountMutex
控制对读者计数的互斥修改
初始化为1
//Writer
P(WriteMutex);
write;
V(WriteMutex);
//Reader
P(CountMutex);
if(Rcount == 0) //第一个读者才需要申请读写互斥信号量，后来读者只需读者计数器加1
P(WriteMutex);
++ Rcount;
V(CountMutex);
read;
P(CountMutex);
-- Rcount;
if(Rcount == 0) //只有最后一个读者才需要释放读写互斥信号量、
V(WriteMutex);
V(CountMutex);
- 特征：读者优先
即后来的读者一定会先于等待的写者进行写操作
读者-写者问题：优先策略
- 读者优先策略
只要有读者正在读状态，后来的读者都能直接进入
如读者持续不断进入，则写者就处于饥饿 - 写者优先策略
只要有写者就绪，写者应尽快执行写操作
如写者持续不断就绪，则读者就处于饥饿
用管程解决读者-写者问题
两个基本方法
Database :: Read() {
Wait until no writers;
read database;
check out - wake up waiting writers;
}
Database :: Write() {
Wait until no readers/writers;
write database;
check out - wake up waiting readers/writers;
}
管程的状态变量
AR=0; //# of active readers
AW=0; //# of active writers
// 只有一个>0
WR=0; //# of waiting readers
WW=0; //# of waiting writers
//可能都>0
Lock lock;
Condition okToRead;
Condition okToWrite;
解决方案详情：读者
AR=0; //# of active readers
AW=0; //# of active writers
WR=0; //# of waiting readers
WW=0; //# of waiting writers
Lock lock;
Condition okToRead;
Condition okToWrite;
Public Database :: Read() {
//Wait until no writers;
StartRead();
read database;
//check out - wake up waiting writers;
DoneRead();
}
Private Database :: StartRead() {
lock.Acquire();
while ((AW+WW)>0) { //条件决定了优先策略
WR++;
okToRead.wait(&lock);
WR--;
}
AR++;
lock.Release();
}
Private Database :: DoneRead() {
lock.Acquire();
AR--;
if ((AR==0 && WW)>0) { //条件决定了优先策略
okToWrite.signal();
}
lock.Release();
}
解决方案详情：写者
AR=0; //# of active readers
AW=0; //# of active writers
WR=0; //# of waiting readers
WW=0; //# of waiting writers
Lock lock;
Condition okToRead;
Condition okToWrite;
Public Database :: Write() {
//Wait until no readers/writers;
StartWrite();
write database;
//check out - wake up waiting readers/writers;
DoneWrite();
}
Private Database :: StartWrite() {
lock.Acquire();
while ((AW+AR)>0) {
WW++;
okToWrite.wait(&lock);
WW--;
}
AW++;
lock.Release();
}
Private Database :: DoneWrite() {
lock.Acquire();
AW--;
if (WW>0) {
okToWrite.signal();
} else if (WR>0) {
okToRead.broadcast();
}
lock.Release();
}
