过桥问题
一条河上架设了由N个桥墩组成的一座桥。若一个桥墩只能站一个人,过河的人只能沿着桥向前走而不能向后退。过河时,只要对岸无人过,就可以过。但不允许河对岸的两个人同时过,以防止出现死锁。请给出两个方向的人顺利过河的同步算法。
不妨将两个方向称为正方和反方。为了考虑这个问题:
- 首先要想第一个过桥人的情况,如果正反双方都要过桥,那么第一个上桥的那个人就获得对桥的控制权,从而使另一方必须等待。
- 其次,假定正方人在过桥,因为只有N个桥墩,每个桥墩只能站一个人,所以在正方过桥时只能控制在N个人之内。
- 最后,最后一个离桥的人要释放对桥的控制权。
值得注意的是,第一个上桥的人(占有桥)不一定是最后一个离桥的人(释放桥),这个问题暗含的一个思想是,等待过桥的任何一个人都可能是第一个上桥的人,正在过桥的这些人中任何一个都可能是最后一个离桥的人。
为此,必须解决以下4个问题:
- 如何确定是第一个上桥的人和最后一个离桥的人?
- 如何获得和释放对桥的控制权?
- 谁来获得、释放对桥的控制权,为什么?
- 如何控制在N个人之内?
关于这四个问题的解答是:
为了确定第一个上桥的和最后一个离桥的人,可以设置一个计数器counter=0,只要上桥一个人就令其加一、只要下桥一个人就令其减一,这样,如果counter等于1就说明这是第一个上桥的人,如果counter等于0说明这是最后一个离桥的人。因为过桥的人都要操作这个counter,从而counter也就变成了一个临界资源,要给它设置一个信号量semCounter=1(同时只能有一个人来操作变量counter),否则就有可能出现counter的值错乱情况。
因为正反双方都要过同一个桥,所以桥是一个临界资源,从而要设一个信号量semBridge=1,获得对桥的控制权就是要P(semBridge),而释放对桥的控制权就是V(semBridge)。
-
那么谁来获得、谁来释放对桥的控制权呢?也许你会考虑,每一个上桥的人都执行P(semBridge)来获得上桥资格,每一个离桥的人都执行V(semBridge)来释放资源。事实上这是错误的也是不必要的。这是因为:
- 每一方过桥的人数事先是不固定的(尽管不能超过N,但是并不强制规定一定要是N个人),所以如果按照这种PV设计思路,semBridge的初始值是无法设置的
- 如果semBridge初始值设置为N,就有可能出现2种情况:
- 正方上桥的人不足N人,他们执行P原语后semBridge仍然非负,这样反方的人执行P原语后也上了桥,两方人员就会相对而发生死锁,这是题目不允许的;
- 又或者正方上桥的人恰好为N个人,执行N次P原语后semBridge为0,这样反方的人执行P原语后进入等待状态,似乎是安全的。但是存在一个致命的失误:每一个离桥的人都要执行V原语来释放。这样,一旦正方的人至少有一人下桥,反方的人就有可能获得上桥资格,但时此时正方的人尚未全部下桥,因此也会出现死锁情况。
正确的策略应该是:第一个上桥的人执行P(semBridge),最后一个离桥的人执行V(semBridge)。这样就能保证一旦正方的人上桥,除非最后一个人也下桥,否则反方人员必须等待。这也是为什么semBridge的初始值是1的原因。
换句话说,正方向第一个人执行P原语后,第二个人、第三个人……都不必执行P原语而直接上桥(if语句实现)。
而反方向要上桥时,必然有第一个上桥人,这个上桥人必须先对反方的counter的信号量semCounter执行P原语,之后因为检测到counter为1后才会对semBridge执行P原语,此后进入等待,注意因为进入等待状态,信号量semCounter根本没有被释放!,从而反方向的第二个人、第三个人……依次对semCounter(而不是semBridge)执行P原语进入等待,根本没有上桥机会。 过桥的人数如何控制在N个人以内呢?设置一个控制人数的信号量semMax=N再适合不过。这个信号量可以作为正方双方的共用信号量,因为对它的操作是在获取到桥的控制权(P(semBridge))之后,所以不会发生冲突。
注意到在之前已经定义了一个记录人数的变量counter,为什么不使用counter来统计人数呢?因为如果已经有N个人在过桥,那么正方等待过桥的人就必须等待。信号量的PV操作有令其等待的功能,而单纯的变量则没有,尽管可以对其进行数值比较来查看是否超过人数上限。
执行代码如下:
//设置计数变量
counterA=0,counterB=0; //正反双方的计数变量
//设置信号量
semaphore
semCounterA=1, semCounterB=1,
semBridge=1,
semMax=N
//正方向上桥的过程
void directA(int i)
{
//判断是否是第一个上桥人,若是则锁住桥
P(semCounterA);
counterA++;
if(counterA==1)
P(semBridge);
V(semCounterA);
//控制人数不超过N
P(semMax);
上桥,过桥,下桥;
V(semMax);
//离桥时减去人数,若是最后一个离桥则释放桥
P(semCounterA);
counterA--;
if(counterA==0)
V(semBridge);
V(counterA);
}
//反方向与正方向同理
void directB(int i){
/*....*/
}
注意这个设计的一个极限情况是,假设正方向的人开始过桥,并且过桥的人有无穷个人,那么反方向的人永远也无法获得过桥机会。因为不存在最后一个人来释放桥。
过桥问题的一个变种——第二类读写者问题将会对此加以限制。
读写者问题(第一类)
某数据库有一个写进程、多个读进程,它们之间读、写操作的互斥要求是:写进程运行时,其他读、写进程不能对数据库进行操作。读进程之间不互斥,可以同时读数据库。请用信号量及PV操作描述这一组进程的工作过程。
分析此题的约束条件:
- 读者之间不发生互斥
- 写者与读者(甚至是其他写者)之间发生互斥
- 数据库是一个临界资源区
多个读者之间可以共存,所以第一个读者会与写者竞争资源,一旦读者进入临界资源区,它就会“锁”住这个资源,写者进入等待,而其他读者可以进入。最后一个读者退出临界资源区时释放资源。
为了识别第一个读者和最后一个读者,设置计数变量readerCounter=0,计数变量保护信号量semReaderCounter=1。数据库资源区公有信号量semDB=1。
由于读者的个数不受限制,所以没有必要设置semMax。
由此写出的伪代码是:
//读者技术变量
readerCounter=0;
//信号量
semaphore
semReaderCounter=1,
semDB=1;
//写者进程
writer(){
P(semDB)
写操作
V(semDB)
}
//读者进程
reader(){
//判定是否是第一个读者,如果是则竞争
P(semReaderCounter)
semReaderCounter++;
if(semReaderCounter==1)
P(semDB)
V(semReaderCounter)
执行读操作
//判定是否是最后一个离开数据库的读者,若是,则释放资源
P(semReaderCounter)
semReaderCounter--;
if(semReaderCounter==0)
V(semDB)
V(semReaderCounter)
}
注意到,此时也存在一个隐患:一旦读者进程争取到数据库,并且此后有源源不断的读者进程也进入数据库,且读取时间很长,数据库资源得不到释放,那么写者进程就有可能长时间被阻塞而发生“饥饿现象”。为此,考虑第二类的读写者问题。
读写者问题(第二类)
将只读数据的进程称为“读者”进程,而写或修改数据的进程称为“写者”进程。允许多个“读者”同时读数据,但不允许“写者”与其他“读者”或“写者”同时访问数据。另外,要保证:一旦有“写者”等待时,新到达的“读者”必须等待,直到该“写者”完成数据访问为止。试用P、V 操作正确实现“读者”与“写者”的同步。
这个问题的约束条件在第一类读写者问题的基础上增加了一条:
- 一旦有“写者”等待时,新到达的“读者”必须等待,直到该“写者”完成数据访问为止。
这是基于“写者优先”的策略。写者优先是因为,当共享数据区被读者占用时,紧邻的读者可能会立刻进入,并且长时间读取,而写者相当长的时间被阻塞无法进入数据区,从而造成“写者饥饿”。
这个问题是过桥问题的一个变种。若将读者、写者分别看作是两方过桥的人,而将共享数据区看作是一座桥(可以通过任意数目的人,即过桥问题中的N=+∞)。由此可以转述为如下描述:
读者、写者分别在桥两边准备过桥。过桥的人只能沿着桥向前走而不能向后退。对于读者而言,只要对岸没有写者,就可以通过任意数目的人,但是一旦对岸有写者准备过桥,准备过桥的读者将等待写者过桥后才能过桥。对于读者而言,如果桥上有任何人(读者或写者),都必须等他们过了桥后,自己才能过桥。
大体的代码框架没有变化。变量和信号量需要:
- 桥(共享数据区)的互斥信号量:semDataMutex=1
- 读者、写者的计数变量:readerCounter=0, writerCounter=0
- 计数变量保护信号量:semReaderCounter=1, semWriterCounter=1
由于对于读者而言不限制过桥人数,因此没有必要设置semMax信号量。
在过桥问题中,读者的步骤分为三个部分:
//第一部分:准备上桥
if(是第一个上桥的读者){
试图获取过桥的权力
if(没有成功) 进入等待队列
}
//第二部分:过桥
过桥
//第三部分:准备下桥
if(是最后一个下桥的读者){
释放桥资源
}
可以看出,只要有一个读者上了桥,并且桥上还有读者,后续读者都可以上桥。为了保证“一旦对岸有写者准备过桥,准备过桥的读者将等待写者过桥后才能过桥”,必须对这种行为加以限制:
考虑存在某个信号量,它指示是否有写者准备过河,每个读者上桥之间都要检测这个信号量,如果发生写者准备过桥的情况,它就要进入等待队列。显然,对这个信号量的检测操作必须在准备上桥进行:
if(有写者准备过河){
进入等待序列
}else{
//第一部分:准备上桥
}
//第二部分:过桥
//第三部分:准备下桥
对于写者:
- 如果他是第一个准备上桥的人,那么他要发出某个信号告知对岸那些准备过桥的读者他要准备过河,请读者等待。
- 根据写者优先,如果有很多个写者准备过桥,这个信号只有等最后一个写者过完桥后才能撤销,在此期间,等待的读者还是在等待。
- 因为写者必须等桥上正在过河的任何人(无论是写者还是读者)都过了他才能过桥(互斥),所以它要竞争过桥
因此它的步骤分为5个部分
//第一部分:发出信号
if(若是第一个上桥的写者){
发出准备过河信号
}
//第二部分:准备过桥
试图过桥(竞争桥资源)
if(没有成功) //桥上还有人
进入等待队列 //等待他们都通过桥
//第三部分:过桥
过桥
//第四部分:准备下桥
释放桥资源 //保证桥上最多只有一个写者,并且只有写者全部通过读者才能继续过
if(是最后一个下桥的写者){
撤销写者过河信号,等待中的读者可以准备过桥
}
综上所述,这个“准备过河的信号”可以设为semWRMutex=1
//变量
readerCounter=0, writerCounter=0 //计数变量
//信号量
semaphore
semDataMutex=1 //共享资源互斥
semReaderCounter=1, semWriterCounter=1 //计数变量保护
semWRMutex=1 //有写者准备过河
//读者:
reader(){
//探测是否有写者准备进入,这里不要把P理解为wait,
//在这里需要有一个挂入等待队列的功能,P是合适的
P(semWRMutex)
//如果没有写者准备进入,则semWRMutex=1,
//P(semWRMutex) 后semWRMutex为0,然后这个读者进入
P(semReaderCounter)
readerCounter++;
if(readerCounter==1)
P(semsemDataMutex)
V(semReaderCounter)
//在下面的语句中又立刻将semWRMutex设为1
V(semWRMutex)
//执行读操作
/*................*/
//准备退出
P(semReaderCounter)
readerCounter--;
if(readerCounter==0)
V(semsemDataMutex)
V(semReaderCounter)
}
//写者
writer(){
P(semWriterCounter)
WriterCounter++;
if(WriterCounter==1)
//第一个准备进入的写者发出准备进入信号
//这个信号直到最后一个写者结束后才撤销
//注意这里的“发出”并不是signal(V原语)
P(semWRMutex)
V(semWriterCounter)
//由于互斥,必须竞争
P(semsemDataMutex)
//执行写操作
//退出
V(semsemDataMutex)
//探测是否是最后一个写者
P(semWriterCounter)
WriterCounter--;
if(WriterCounter==0)
//最后一个写者结束后撤销信号,通知相关等待读者进入
V(semWRMutex)
V(semWriterCounter)
}
仓库问题
有一个仓库,可以存放A 和B 两种产品,但要求:
(1)每次只能存入一种产品(A 或B);
(2)-N<A 产品数量-B 产品数量<M。其中,N 和M 是正整数。
试用同步算法描述产品A 与产品B 的入库过程。
对于不等式-N<A 产品数量-B 产品数量<M可以拆分为两个不等式:B-A<N且A-B<M。也就是说,B的数量至多只能比A多N-1个,且A的数量至多只能比B多M-1个。
显然,仓库是一个临界资源,所以需要设置一个互斥信号量semRepository=1。
然而如何保证A、B的数量满足约束条件呢?考虑到,如果A、B之间的数量差一旦超过约束范围,比如某时刻A-B=M-1了,那么接下来A就不能再放入了,这个时候A必须进入等待队列等待。所以很容易想到使用信号量来保证约束条件。
分别为A、B设置约束信号量semEnableA,semEnableB,分别表示A、B还可以存入的数量,那么它们的初值是多少呢?或者说,这两个信号量该如何使用?
考虑在任意时刻,仓库中A、B的数量都在约束范围内,假定现在要放入一个A,必须检测这个A能否放入,即执行P(semEnableA)操作,如果判定不能放入,则A进入等待队列;另外,如果判定能放入,那么A放入后,B还可存入数量也要加一个(semEnableB++),这样双方数量差才能不超过界限,这就要求要执行V(semEnableB)。对于B而言也是同理。也就是:semEnableA在A进程中被P,但是却是在B进程中被V。
设semEnableA初值为K。在初始时刻,仓库是空的。考虑一个极端情况,就是只放入A而不放B,从而不断执行P(semEnableA)操作,semEnable不断自减,由于B的数量始终是0,所以这个过程直到A的数量为M-1后停止,此时semEnableA=K-(M-1)。之后再放入A时执行P(semEnableA)将导致A被挂起等待,这蕴含了semEnableA=0,也就是K=M-1,所以semEnableA的初始值为M-1。对于B而言,同理可得semEnableB的初始值为N-1。
由此写出如下为伪代码:
semaphore
semRepository=1,
semEnableA=M-1, semEnableB=N-1
//放入A
putA(){
取A产品
P(semEnableA) //检测能否放入A
P(semRepository) //若能,则锁住仓库
放入A
V(semRepository)
V(semEnableB) //将B可放入的数量加一个
}
//放入B
putB(){
取B产品
P(semEnableB) //检测能否放入B
P(semRepository) //若能,则锁住仓库
放入B
V(semRepository)
V(semEnableA) //将A可放入的数量加一个
}
多任务同步问题(状态合并和拓扑排序)
设有如下5个任务,每个节点代表一个任务。对每一个任务而言,只有其所有前驱任务都被执行完毕,该任务才能开始执行,请使用P、V原语实现。
这是一个典型的进程同步问题。一种普适的做法是为每一个箭头标明一个使用信号量,由于此处有7个箭头,于是存在7个私用信号量,它们的初始值都是0。
对于某个任务Ti,它的框架是:
Ti {
对Ti的入度信号量执行P原语
执行Ti
对Ti的出度信号量执行V原语
}
例如对于T1有:
T1{
do T1 //由于T1没有前驱任务,所以不必执行P原语
V(S12)
V(S13)
}
对于T2有:
T1{
P(S12)
do T2
V(S24)
}
对于T6
T6{
P(S46)
P(S56)
do T6 //由于T6没有后继任务,所以不必执行V原语
}
但是实际上,可以通过合并箭头的方式来减少信号量的数目。考虑下面两种情况:
-
一对多情况
即存在n个进程Pik(k=1,2,...,n),它们的前驱进程是P1。事实上可以将这n个箭头合并为一个箭头,从而只用设置一个信号量S1i=0:
对于P1而言,它的程序框架是:
而对Pik而言,它的代码框架是:P1{ /*...前驱事务代码*/ V(S1i) V(S1i) ...... //执行n次V(S1i) V(S1i) }
对于其解释是这样的:假定在极短的时间内进程P1和Pik(k=1,2,...,n)进程同时执行了。那么由于信号量S1i初始值为0,且P(S1i)被执行n次,从而S1i=-n并且Pik(k=1,2,...,n)均被挂入等待对队列进行等待。然后,等到P1执行完它的前驱事务代码后,依次执行n次V(S1i),每次执行一次就会使得等待队列中的一个进程Pik进入就绪队列准备执行。由于这n个后继进程不在同一条链上,所以它们可以并发执行,同时也满足了同步关系。Pik{ P(S1i) /*...后继事务代码*/ }
- 多对一情况
对于这种情况,我们仍然可以将箭头进行合并,并专设一个信号量Si1=0来表示同步关系
对于P1而言,它的程序框架是:
而对Pik而言,它的代码框架是:P1{ P(S1i) P(S1i) ...... //执行n次P(Si1) P(S1i) /*...后继事务代码*/ }
对于这种情况的解释是:当进程P1执行第一句P(Si1)时,由于信号量Si1初值为0,从而P1进入等待队列,此时Si1=-1。然后,任意一个前驱进程Pik执行完毕后执行V(Si1),使得Si1=0,并且P1又进入就绪队列准备执行。P1执行完第二个P(Si1)后,Si1又等于-1且其自身又进入等待队列被阻塞,直到下一个前驱进程Pik'执行完毕后执行V(Si1)使其继续运行……如此循环往复,直到所有的这n个前驱进程Pik都执行完毕,进程P1才能执行它的后继事务代码,由此完成同步功能。Pik{ /*...前驱事务代码*/ V(Si1) }
那么如何进行合并呢?第k节点的所有入边都标上semIn(k)的标记,所有出边都标上semOut(k)标记。设第k个结点有numIn个入边,有numOut个出边,则其算法为:
T(k){
P(semIn(k)) × numIn //表示依次执行P(semIn(k))共numIn次
do work
V(semOut(k)) × numOut //表示依次执行V(semIn(k))共numOut次
}
对于本题的图:
它的信号量可以缩减为4个:
所以代码如下:
semaphore
S1,S2,S3,S4;
S1=S2=S3=S4=0;
T1{
do work
V(S1) × 2
}
T2{
P(S1)
do work
V(S3)
}
T3{
P(S1)
do work
V(S2)
}
T4{
P(S3) × 2
do work
V(S4)
}
T5{
P(S2)
do work
V(S3)
V(S4)
}
T6{
P(S4) × 2
do work
}
其执行过程表如下,可见是保持了同步关系,并且信号量只需要四个而不是七个:
吃水果问题(简单)
桌子上有一只盘子,盘子中只能放一只水果。爸爸专向盘子中放苹果,妈妈专向盘子中放橘子,一个儿子专等吃盘子中的橘子,一个女儿专等吃盘子中的苹果。用PV 操作实现他们之间的同步机制。
由于不能同时放苹果和橘子,所以盘子是一个临界资源,而父亲母亲就是互斥关系。如果盘子中没有水果时,儿子女儿是不能取水果的,所以父亲母亲和女儿儿子之间是同步关系。所以,思路如下:
- 父亲母亲对盘子信号量操作,获得放水果的许可
- 父亲(母亲)放了水果后,发出私用信号量来通知女儿(儿子)取水果
- 女儿(儿子)取完水果后通知父亲(母亲)放水果,并释放盘子资源
由于这个盘子中只能放一个水果,所以父亲与女儿、母亲与儿子之间各只需一个私用信号量。
这个题的核心思想在于其同步过程实现:父亲与女儿是同步的,父亲锁住的盘子只有女儿才能释放。母亲与儿子是同步的,母亲锁住的盘子只有儿子才能释放。而盘子的互斥信号量又是公用的。
这就保证了:父亲放了水果后,除非女儿进程拿了水果并释放盘子,否则其他进程都要等待。即父亲母亲获取不到盘子而等待,儿子因为盘中没有橘子的信号(母亲没有更改semOrange仍然是0)而陷入等待。
semaphore
semPlate=1 //盘子的互斥信号量
semApple=0, semOrange=0 //注意同步进程的信号量初始值是0
//父亲放苹果
putApple(){
//获得盘子,它的释放是在取苹果进程中进行的
//P(semPlate) 后,除非女儿进程去了苹果释放盘子
//否则父亲。母亲、儿子三个进程都取不到盘子
P(semPlate)
放苹果
V(semApple) //通知女儿吃苹果,由于semApple初始值0,V原语后变为1
//此后,当女儿还没有取到苹果并释放盘子时,下一轮放苹果
//操作会在P(semPlate)这一步被挂起等待
}
//女儿取苹果
getApple(){
//接收到取苹果信号。该操作使得semApple从1又变为0
//从而如果父亲未放苹果,semApple始终是0,则在下一轮取苹果操作时
//在这一步被挂起等待
P(semApple)
取苹果
V(semPlate) //释放盘子资源
}
//取橘子和拿橘子进程如法炮制
putOrange(){
/*.......*/
}
getOrange(){
/*.......*/
}
吃水果问题(生产者-消费者)
桌子上有一只盘子,盘子中最多放N个水果。爸爸专向盘子中放苹果,妈妈专向盘子中放橘子,且每次只能放一个。儿子女儿每次可以拿一个水果来吃。用PV 操作实现他们之间的同步机制。
这是上个吃水果问题的一个变种。它的变化是:
- 盘子(缓冲区)的容量是N
- 生产者生产的东西消费者都可以使用,女儿可以吃苹果橘子,儿子也可以。
在上个简单的吃水果问题中,父亲锁住盘子后放了苹果,当且仅当它的同步进程——女儿进程取完苹果后释放盘子资源,否则其他所有进程都会进入等待。这就保证了父亲女儿同步进行。
而在这个问题中,盘子可放的水果不止一个,生产者进程锁住盘子后放了水果,但是不应该在消费者进程释放盘子,否则生产者放一个水果,当且仅当消费者拿走后才能放下一个水果,这样盘子任意时刻至多只能有一个水果,这显然是不合理的。无论是生产者还是消费者,在对盘子操作之前一个锁住盘子,而在操作完成后要自己释放盘子。
盘子设置一个互斥信号量semPlate=1,由于盘子最多可以放N个水果,所以信号量要增设两个:目前可以取多少个水果semEnableGet,以及目前有多少个空位可以放水果semEnablePut。初始时,有N个位置可以放水果,而有0个水果可以取,所以初始值semEnableGet=0,semEnablePut=N。
对这两个信号量的操作是这样的:
- 生产者检测是否可以放水果(P(semEnablePut))。如果可以就放(semEnablePut也就减少一个空位),否则要等待空位。
- 得到放置许可后,生产者要锁住盘子(P(semPlate))
- 然后放水果
- 生产者释放盘子(V(semPlate)),并且通知可以取水果了(V(semEnableGet)后semEnableGet自增1,表示多一个水果可取)
- 消费者检测是否有水果可以取(P(semEnableGet))。如果可以就取(semEnableGet也就减少一个水果),否则要等待水果。
- 得到放置许可后,消费者要锁住盘子(P(semPlate))
- 然后取水果
- 消费者释放盘子(V(semPlate)),并且通知可以放水果了(V(semEnablePut)后semEnablePut自增1,表示多一个空位可以放水果)
注意,V原语次序可以任意,但是P原语顺序不能乱,否则可能会引发死锁。
哲学家问题:
假设有五位哲学家围坐在一张圆形餐桌旁,做以下两件事情之一:吃饭,或者思考。吃东西的时候,他们就停止思考,思考的时候也停止吃东西。餐桌中间有一大碗意大利面,每两个哲学家之间有一只筷子,哲学家必须用两只筷子吃东西。他们只能使用自己左右手边的那两只筷子。条件是:
1. 只有拿到左右两只筷子才吃饭
2. 如果筷子在别人手上,必须等他人吃完后才能拿筷子
3. 在未拿到两只筷子吃饭前,绝对不会放下自己手中的筷子
描述一个保证不会有两个相邻的人同时要求吃饭的通信算法是简单的:
semaphore
chopstick[0]~chopstick[4]=1 //5根筷子的互斥信号量
philosopher(i){ //第i个哲学家进程
P(chopstick[i]); //拿左边的筷子
P(chopstick[i+4 mod 5]) //拿右边的筷子,注意mod的使用
吃饭
V(chopstick[i+4 mod 5]) //放下右边的筷子
V(chopstick[i]); //放下左边的筷子
}
尽管这个算法是简单的,但是它有一个潜在的隐患:可能会引发死锁。假设5个哲学家同时拿起左手边的筷子,但是都没有右边的筷子,且根据题意,他们都不会放下自己的筷子,从而没有人能吃饭。
为了避免死锁,改进上述算法。因为所有人都要先拿自己左手边的筷子,这样可能每个人手中都恰好有一根筷子,这是引发死锁的关键。
自然想到,如果两个相邻的哲学家都首先取第一根筷子既可避免。让偶数编号的哲学家先拿左手边的筷子(第i个哲学家拿第i个筷子)而让奇数编号的哲学家拿右手边的筷子(第i个哲学家拿第 (i+4)mod5 个筷子):
这样,任何一个哲学家拿到一根筷子时,就会阻止他相邻的哲学家拿到该筷子,这名哲学家从而只能等待:
semaphore
chopstick[0]~chopstick[4]=1 //5根筷子的互斥信号量
philosopher(i){ //第i个哲学家进程
if(i是偶数)
P(chopstick[i]); //拿左边的筷子
P(chopstick[i+4 mod 5]) //拿右边的筷子,注意mod的使用
吃饭
V(chopstick[i+4 mod 5]) //放下右边的筷子
V(chopstick[i]); //放下左边的筷子
else
P(chopstick[i+4 mod 5]); //拿右边的筷子
P(chopstick[i]) //拿左边的筷子
吃饭
V(chopstick[i+4 mod 5]) //放下右边的筷子
V(chopstick[i]); //放下左边的筷子
}
读卡机问题
读卡机上读卡片。这一项工作由三个进程get,copy和put以及两个缓冲区buffer1和 buffer2 完成。进程get的功能是把一张卡片上的信息从读卡机上读进buffer1;进程copy的功能是把buffer1中的信息复制到buffer2;进程put的功能是取出buffer2中的信息并从行式打印机上打印输出。
这是一个同步问题。有一个错误的解法是这样的:
因为get和copy、copy和put之间都是同步关系。如果buffer1中get没有放入数据,copy是取不到数据的,只能等待。对于buffer2也同理。所以:
//错误做法: ERROR SOLUTION
semaphore
enable1=0,enable2=0; //两个缓冲区是否可以放入数据
get{
放入数据;
V(enable1)
}
copy{
P(enable1)
取数据
放入buffer2
V(enable2)
}
put{
P(enable2)
取数据
}
乍一看似乎是合理有效的:一开始都没有数据,enable1和enable2都是0,copy在运行 P(enable1)时会进入等待,put在运行P(enable2)也会进入等待。直到get放入数据并执行 V(enable1)时,copy才被激活,然后copy执行 V(enable2)时,put才被激活取数据。此后get不再放入数据,copy和put继续等待,如此往复。
但是这里有一个重大隐患,那就是没有考虑到两个缓冲区是临界资源没有被保护。你可能会想,get和copy是同步关系而 不是互斥关系 ,如果get不通知copy来去数据,那么copy就会在P(enable1)这一步时陷入等待,根本不会访问buffer1,似乎临界资源不产生冲突。
但是恰恰忽略的一点是,copy在取数据时,因为buffer1没有被保护而使得copy也在往里面填充数据。这就会引发脏数据。所以尽管时同步,也要对临界资源进行互斥:
semaphore
enable1=0,enable2=0, //两个缓冲区是否可以放入数据
bufMutex1=1,bufMutex2=1; //两个缓冲区的互斥信号量
get{
P(bufMutex1)
放入数据;
V(bufMutex1)
V(enable1)
}
copy{
P(enable1)
P(bufMutex1)
取数据
V(bufMutex1)
P(bufMutex2)
放入buffer2
V(bufMutex2)
V(enable2)
}
put{
P(enable2)
P(bufMutex2)
取数据
V(bufMutex2)
}
这样在copy取buffer1中数据时,get需要等待而不能放入数据。
注意到,enable1这个信号量在这里仅表示能否装入,所以只适合表示缓冲区仅能放置一份数据的情况。
但是假如是一个缓冲队列,里面可以放置N份数据,那么只使用enable1就明显不够了,为此需要设置两个信号量:
- 当前还能放多少数据:enablePutData=N(相当于empty)
- 当前有多少份数据可以取:enableGetData=0(相当于full)
enablePutData初始值是N,表示刚开始可以放N份数据;enableGetData初始值是0,表示刚开始没有数据可以取:
- 每放一份数据后,执行P(enablePutData)来使enablePutData自减1,表示少一个单元可放数据,如果enablePutData等于0,表示没有单元可以放数据,则生产者需要等待。
放完数据后,需要执行V(enableGetData)来使enableGetData自增1,表示缓冲区多一个数据可取,就是刚刚放入的那一份。 - 每取一份数据后,执行P(enableGetData)来使enableGetData自减1,表示少一个数据可取,就是刚刚取了的那一份,如果enableGetData等于0,表示没有数据可取,则消费者需要等待。
取完数据后,需要执行V(enablePutData)来使enablePutData自增1,表示缓冲区多一个空位可以放数据。
所以拓展后可以写为:
semaphore
enablePutData1=N,enablePutData2=N,
enableGetData1=0,enableGetData2=0,
bufMutex1=1,bufMutex2=1; //两个缓冲区的互斥信号量
get{
//先测试能否放数据再互斥临界区,P原语顺序错乱可能会引发死锁
P(enablePutData1)
P(bufMutex1)
放入数据;
V(bufMutex1)
V(enableGetData1)
}
copy{
P(enableGetData1) //测试buffer1能不能取数据
P(bufMutex1)
取数据
V(bufMutex1)
V(enablePutData1)
//测试buffer2能不能放数据
P(enablePutData2)
P(bufMutex2)
放入buffer2
V(bufMutex2)
V(enableGetData2)
}
put{
//测试buffer2能不能取数据
P(enableGetData2)
P(bufMutex2)
取数据
V(bufMutex2)
V(enablePutData2)
}
如果N=1就是本题的情况,测试enable信号量可以减少为1个。
游览车问题
已知风景区内的轿车总量为M辆,游客总数为N,约定:
(1)每辆轿车限乘一位游客;
(2)如果有空闲的轿车,应当允许想游览的游客乘坐;
(3)无空闲轿车时,游客只能排队等待;
(4)若没有想游览的游客,空闲的轿车也要等待。
显然存在一个轿车进程Car()和游客进程Passager()。题目中的M、N在这里并不是想要告诉你车数和人数,而是想告诉你车辆数目和游客数目是固定的。这里的“固定”指游客游览完可以继续游览,轿车游览完可以继续拉客。所以,本题的相关信号量的值和M、N都没有关系!。
这道题巧用了信号量和等待机制。首先用自然语言表述出这两个进程要干什么。
对于轿车进程Car(),每当它运行时:
- 可用轿车数目加1,通知(唤醒)等待的游客。
- 在游客进程没有发出发信号表明上车可以出发之前,轿车是在等待状态。
- 带上乘客游览,在这一段时间内,游客进程什么也不做,从而处于等待状态
- 发出游览完成信号,通知(唤醒)乘客下车
- 在接收到游客下车完成信号之前,一直等待游客下车。然后返回第一步。
对于游客进程Passager(),每当它运行时:
- 检查是否有可用轿车(轿车有可用信号),如果有则选定一个轿车,否则等待
- 开始上车,在这个过程中,轿车进程等待
- 上车后,发出开车信号,唤醒轿车进程出发
- 游览过程中游客不需要做什么,这就是处于等待状态,需要轿车进程发出游览完成信号后,游客进程才从等待状态唤醒,执行下一步
- 游客下车,在此过程中,轿车进程等待
- 下车完成后,发出下车完成信号,本次游览结束,游客继续返回第一步准备游览。
上述描述中可以学到如下技巧:
- A进程与B进程交互运行时,当某个操作要在A进程完成,此间B进程陷入等待,直到A进程完成后发出信号激活B进程。为此,设置一个信号量semaphore=0,B中执行P(semaphore)使得B等待,而当A完成操作后,执行V(semaphore)来唤醒B进程,因为B进程挂在semaphore的等待队列中。
- 游客进程通过对可用车数semEnableCar进行P操作后,如果没有可用的轿车,就挂入该信号量等待队列中。而后无论来多少辆车,因为车要声明可用车数加1,也就是执行V操作,这就会唤醒semEnableCar等待队列中的游客进程,让游客上车。
反过来说,如果轿车进程执行V操作后,如果没有游客进程,那么就不可能收到开车信号,也就会陷入等待中。
所以,semEnableCar的初始值就是0,而不是M或者N什么的。当没有一个车时,游客的P(semEnableCar)才能导致游客等待。
semaphore
semEnableCar=0, //可用车数目信号量
semCanStart=0, //可以开车信号量
semFinish=0, //游览完成信号量
semHasTakenOff=0 //游客已经下车信号量
Car(){
V(semEnableCar) //可用车+1,通知等待的游客
P(semCanStart) //等待游客的“可以出发”信号
开始游览,在这个过程中,游客正在等待,也就是阻塞状态
V(semFinish) //向游客进程发出游览完成信号
P(semHasTakenOff) //等待游客的“已经下车”信号
Car() //循环运行
}
Passager(){
P(semEnableCar) //等待有可用车
上车,此时轿车等待
V(semCanStart) //发出“可以出发”信号
P(Finish) //在收到轿车“游览完成”信号之前,进入阻塞状态,表示“正在游览”
游客开始下车,此间轿车正在等待
V(semHasTakenOff) //游客“已经下车”信号给轿车,激活轿车
Passager() //循环运行
}
理发问题
有一个理发师,一把理发椅和N把供等候理发的顾客坐的椅子。如果没有顾客,则理发师等待;当一个顾客到来时,唤醒理发师进行理发。如果理发师正在理发时又有新顾客到来,有空椅子可坐,他就坐下来等,如果没有空椅子,就立即离开。为理发师和顾客各编一段程序描述他们的行为,要求不能带有竞争条件。
这是一个比较复杂的进程同步问题。需要设计两个进程:
- 顾客进程Customer()
- 理发师进程Barber()
需要注意到,在顾客与理发师发生同步交互之前,顾客需要先判断是否需要等待(这个过程,理发师进程不需要介入)。其逻辑如下:
首先要解决的一个问题是,如何判断是否有等待位置。很容易想到设置一个信号量来指示是否有等待位置,但是这么做是不合理的,理由如下:
假设存在了这么一个信号量,那么第一个顾客来到就要进行P操作而陷入等待(V操作显然是不合理的,因为第一个顾客来之前不可能有顾客在等待)。同时还必须通知理发,否则所有顾客将无限期等待下去。
一种情况是理发师通知等待的顾客理发,为此理发师进程必须对这个信号量进行V操作,考虑到在第一个顾客尚未来到之前,不可能有顾客在等待,所以V操作将变得没有意义。
另一种情况是等待区顾客通知理发师理发,为此必须设置另外一个信号量作为通知使用。但是由于顾客已经陷入等待,所以附设的通知信号量将无法被顾客发出,理发师也将永远收不到通知信号。
此外,单一的通知信号量将起不到约束顾客等待数量在N之内的作用。
综上所述,设置一个信号量只是是否有等待位置是不合理的。
那么如何做呢?可以设置一个变量waitingNumbers,当一个顾客来到时先判断等待人数waitingNumbers是否超过可等待人数N,如果是则离开(即顾客进程结束),否则waitingNumbers++,顾客准备等待。显然,所有顾客都要对waitingNumbers操作,从而waitingNumbers也就成为一个临界资源,为此要进行保护,故而设置互斥量semWaitingNumbersMutex=1,因为同一时间至多只能由一个顾客进行操作。
现在如果顾客进程还没有退出,那么他就要与理发师进程发生交互。
注意,在这一阶段有两个关键的等待过程。当顾客尚未到来或者还没有发出准备理发的信号之前,理发师是在等待的。另一方面,如果客户没有离开,而理发师没有对其发出可以理发信号时,客户就需要等待,这个等待就是题目中“坐在等待区椅子上”的等待情况。
也就是说,顾客发信号唤醒等待的理发师,理发师发信号唤醒等待的顾客(★★★核心思想)
针对理发师的等待,可以设置一个信号量semCustomers=0。理发师对其进行P操作(wait),使得semCustomers为-1,从而理发师进入等待。直到顾客对其进行V(signal)操作,使得semCustomers为0,重新唤醒理发师进程。
这就又引发一个问题:如果第二个顾客、第三个……第K个顾客依次到来(且没有离开,K≤N),那么他们每个人都要进行V(semCustomers)操作,因为第一个顾客操作后semCustomers等于0,那么后续这K人的V操作将会使得semCustomers等于K,并且因为没有等待进程(针对信号量semCustomers的唯一可以等待的只有理发师进程),所以这些顾客进程将继续后续操作。【V原语妙用】
在后续操作中必须使这些人陷入等待,设置一个信号量semBarber=0。所有人在V(semCustomers)后必须进行P(semBarber),即等待理发师发出可以理发通知,即等待理发师的V(semBarber)操作。第一个顾客在收到理发师通知进行理发时,semBarber又重新归零,这样后来的K个人对其P操作后使得semBarber等于-1、-2、……、-K,这K个人也就进入了等待。
现在,论证上述过程的可持续性:当第一位顾客离开后,理发师进程又重新进行P(semCustomers)操作,因为此时semCustomers为K,所以理发师不会等待,而是继续执行。理发师接下来执行V(semBarber)唤醒顾客,由于semBarber为-K,K个人都挂在semBarber信号量的等待队列中,所以任选一个唤醒,进行理发。由此可以循环执行。
int waitingNumbers=0; //等待的顾客
semaphore
semWaitingNumbersMutex=1,
semBarber=0,
semCustomers=0,
semCut=0, //可以开始理发
semFinish=0, //理发完成
Customer(){
P(semWaitingNumbersMutex)
if(waitingNumbers>N)
V(semWaitingNumbersMutex) //此后,进程结束,表示离开
else{
waitingNumbers++;
V(semWaitingNumbersMutex);
V(semCustomers); //通知理发师有顾客
P(semBarber)//等待理发师通知
//接到了通知,就坐理发,注意等待座位要空出来
P(semWaitingNumbersMutex)
waitingNumbers--;
V(semWaitingNumbersMutex)
//告诉理发师可以开始理发
V(semCut)
//等待理发完成信号
P(semFinish)
}
}
Barber(){
P(semCustomers) //等待顾客到来
V(semBarbers) //唤醒等待的顾客
P(semCut) //等待顾客可以开始理发信号
//理发
V(semFinish) //理发完成
}
实际上可以推广为M个理发师理发,而执行代码并不发生变化。这是因为在系统中运行着M个Barber()进程,它们刚开始都执行第一个P(semCustomers)语句而陷入等待,直到有顾客到来执行V(semCustomers)而唤醒任意一个理发师理发。任意一个理完头发的理发师发信号来唤醒下一个等待的顾客。
上机问题
某高校计算机系开设有网络课并安排了上机实习,假设机房共有2m台机器,有2n 名学生选该课,规定:
① 每两个学生组成一组,各占一台机器,协同完成上机实习;
② 只有一组的两个学生到齐,并且此时机房有空闲机器时,该组学生才能进入机房;
③ 上机实习由一名教师检查,检查完毕,一组学生同时离开机房。
试用P、V操作模拟上机实习过程。
在这里,除了学生进程Student()和教师进程Teacher()之外,还应该有一个进程,用以实现第一个约束条件——即两个学生到齐后才允许进入,不妨将该进程命名为Monitor()。显然教师进程不具有此功能,它只是用来完成第三约束条件,即“检查”功能。
可以设置一个信号量semStudent=0,当来一个学生时执行V操作,通知Monitor,而Monitor执行P操作来接收通知信号量。为了实现“两个学生到齐才能进去”,Monitor需要执行两次P(semStudent),然后才发出两次允许进入信号——分别给两个申请进入的学生使用。其核心的步骤是:
- 初始,semStudent=0,
- Monitor执行第一个P(semStudent) => semStudent=-1,陷入等待
- 第一个学生执行V(semStudent) => semStudent=0,唤醒Monitor,该学生等待接收可进入信号。
- Monitor执行第二个P(semStudent) => semStudent=-1,陷入等待
- 第二个学生执行V(semStudent) => semStudent=0,唤醒Monitor,该学生等待接收可进入信号。
- Monitor继续执行,分别执行两次V(semEnter) => 两个等待的学生接收到信号,进入。
后面的步骤就很简单了,因为必须有空闲的电脑,否则学生等待,因为电脑总数固定,所以可以设置一个电脑的信号量semComputer=2M。教师和学生进程之间是互相同步的关系。
BEGIN
Semaphore:
semStudent=0, semComputer=2M ,semEnter=0, semFinish=0, semCheck=0;
COBEGIN
Process Procedure semStudent()
begin
V(semStudent); // 表示有学生到达
P(semComputer); // 等待获取一台计算机
P(semEnter); // 等待进入许可
DO it with partner();
V(semFinish); // 实习完成
P(semCheck); // 等待老师检查
V(semComputer): // 释放计算机资源
end
Process Procedure Teacher()
begin
P(semFinished); // 等待学生实习完成
P(semFinished); // 等待另一学生实习完成
check the work();
V(semCheck); // 表示检查完成
V(semCheck); // 表示检查完成
end
Process Procedure Monitor()
begin
P(semStudent); // 等待学生到达
P(semStudent); // 等待另一学生到达
V(semEnter); // 允许学生进入
V(semEnter); // 允许学生进入
end
Coend
END