第6章 进程同步

操作系统OperatingSystem第6章进程同步Chapter6:ProcessesSynchronization进程同步：间接相互制约关系（竞争）想一想打印工作过程待打印文件队列out=4m.docprog.c…w.pdf…4567in=7进程1进程2打印进程应用程序提交打印任务应用程序去做其它工作打印任务被放进打印队列打印进程从队列中取出任务打印进程控制打印机打印进程同步：直接相互制约关系（合作）输入进程A输出进程B缓冲区out=4…4567in=7输入进程输出进程1如果缓冲区满，进程A被阻塞否则将数据放入缓冲区，唤醒进程B如果缓冲区空，进程B被阻塞否则从缓冲区中取出数据，唤醒进程A进程同步（合理&有序）对多个相关进程在执行次序上进行协调，使并发执行的诸进程直接按一定的规则或时序共享系统资源，能很好地相互合作，使程序执行具有可再现性进程走走停停输入进程：缓冲区满，停；加入数据了，发信号让输出进程走输出进程：缓冲区空，停；输入数据了，发信号让输入进程走例：司机与售票员默契生产者

消费者#defineBUFFER_SIZE10typedefstruct{...}item;itembuffer[BUFFER_SIZE];intin=out=counter=0;共享数据while(true){while(counter==BUFFER_SIZE) ;buffer[in]=item;in=(in+1)%BUFFER_SIZE;counter++;}生产者进程while(true){while(counter==0) ;item=buffer[out];out=(out+1)%BUFFER_SIZE;counter--;}消费者进程生产者

消费者引出的一个问题共享变量counter会出现语义错误如生产者进程和消费者进程各执行一次register=counter;register=register+1;counter=register;生产者P消费者Cregister=counter;register=register-1;counter=register;P.register=counter;P.register=P.register+1;C.register=counter;C.register=C.register-1;counter=P.register;counter=C.register;一个可能的执行序列初始情况counter=5;术语1:竞争条件(RaceCondition)竞争条件:和时间有关的共享数据语义错误含义1:多个进程并发访问和操作共享数据含义2:和时间有关就是和调度顺序有关含义3:这样的错误非常难于调试P.register=counter;P.register=P.register+1;C.register=counter;C.register=C.register-1;counter=P.register;counter=C.register;第i次执行P.register=counter;P.register=P.register+1;counter=P.register;C.register=counter;C.register=C.register-1;counter=C.register;第j次执行避免竞争条件的一个尝试在写counter时阻断其他进程访问counterP.register=counter;P.register=P.register+1;C.register=counter;C.register=C.register-1;counter=P.register;counter=C.register;一个可能的执行序列生产者PP.register=counter;P.register=P.register+1;给counter上锁消费者Ccounter=P.register;检查counter锁生产者P给counter开锁消费者CC.register=counter;C.register=C.register-1;counter=C.register;给counter上锁给counter开锁一段代码一次只允许一个进程进入如何实现:一段代码一次只允许一个进程进入?术语2:临界区(CriticalSection)临界区:一次只允许一个进程进入的一段代码counter--;消费者counter++;生产者这就是临界区临界区定义了进程间的协作协议，因此一个非常重要的工作:找出进程中的临界区代码剩余区进入区临界区退出区剩余区进程代码结构这是本部分的中心内容如何进入临界区?

进入原则1.互斥进入:如果一个进程在临界区中执行，则其他进程不允许进入这些进程间的约束关系称为互斥(mutualexclusion)一个好的临界区进入方法还需考虑…这是临界区进入的基本原则(正确性保证)2.有空让进:若干进程要求进入空闲临界区时，应尽快使一进程进入临界区3.有限等待:从进程发出进入请求到允许进入，不能无限等待进入临界区的一个尝试轮换法

while(turn!=0);临界区turn=1;剩余区进程P0while(turn!=1);临界区turn=0;剩余区进程P1满足互斥进入要求问题:P0完成后不能接着再次进入，尽管进程P1不在临界区…(不满足有空让进)许多复杂的道理往往就埋藏在日常生活中!进入临界区的又一个尝试似乎没有任何头绪…可借鉴生活中的道理回到家里，牛奶放进冰箱3:30买牛奶3:25到达商店回到家里，牛奶放进冰箱3:20离开家去商店买牛奶3:15离开家去商店3:05打开冰箱，没有牛奶了3:00打开冰箱，没有牛奶了到达商店3:10妻子丈夫时间上面的轮换法类似于值日更好的方法应该是立即去买，留一个便条进入临界区的又一个尝试标记法计算机考虑问题的方式flag[0]=true;while(flag[1]);临界区flag[0]=false;剩余区进程P0flag[1]=true;while(flag[0]);临界区flag[1]=false;剩余区进程P1if(noMilk){if(noNote){leaveNote;

buymilk;

removenote;

}}leaveNote;if(noMilk){if(noNote){buymilk;

}}removenote;标记法能否解决问题?flag[0]=true;while(flag[1]);临界区flag[0]=false;剩余区进程P0flag[1]=true;while(flag[0]);临界区flag[1]=false;剩余区进程P1考虑下面的执行顺序(1)(2)(3)(4)flag[0]=trueflag[1]=true此时P0和P1的进入请求会无限等待进入临界区的再一次尝试非对称标记A带名字的便条+让一个人更加勤劳丈夫是那个更勤劳的人各种情况都正确关键:选择一个进程进入，另一个进程循环等待leavenoteA;while(noteB){//X

donothing;} if(noMilk){buymilk;}

removenoteA;丈夫(A)leavenoteB;

if(noNoteA){//Y

if(noMilk){

buymilk;}}removenoteB;妻子(B)进入临界区Peterson算法结合了标记和轮转两种思想

flag[0]=true;turn=1;while(flag[1]&&turn==1);临界区flag[0]=false;剩余区进程P0

flag[1]=true;turn=0;while(flag[0]&&turn==0);临界区flag[1]=false;剩余区进程P1Peterson算法的正确性

flag[i]=true;turn=j;while(flag[j]&&turn==j);临界区flag[i]=false;剩余区进程Pi满足互斥进入:如果两个进程都进入，则flag[0]=flag[1]=true，turn==0==1，矛盾!满足有空让进:如果进程P1不在临界区，则flag[1]=false，或者turn=0，都P0能进入!满足有限等待:P0要求进入，flag[0]=true；P1在临界区有turn=0，P1不再改变turn值，所有后面的P1会循环，P0进入。多个进程怎么办?

面包店算法仍然是标记和轮转的结合如何轮转:每个进程都获得一个序号，序号最小的进入如何标记:进程离开时序号为0，不为0的序号即标记面包店:每个进入商店的客户都获得一个号码，号码最小的先得到服务；号码相同时，名字靠前的先服务。

choosing[i]=true;num[i]=max(num[0],…,num[n-1])+1;choosing[i]=false;

for(j=0;j<n;j++){while(choosing[j]);while((num[j]!=0)&&(num[j],j)<(num[i],i]));}临界区num[i]=0;剩余区进程Pi面包店算法的正确性

choosing[i]=true;num[i]=max(num[0],…,num[n-1])+1;choosing[i]=false;

for(j=0;j<n;j++){while(choosing[j]);while((num[j]!=0)&&(num[j],j)<(num[i],i]));}临界区num[i]=0;进程Pi满足互斥进入:Pi在临界区内，Pk试图进入，一定有(num[i],i)<(num[k],k)，Pk循环等待。满足有空让进:如果没有进程在临界区中，最小序号的进程一定能够进入。满足有限等待:离开临界区的进程再次进入一定排在最后(FIFO)，所以任一个想进入的进程至多等待n个进程一个显然的感觉是:太复杂了!让用户考虑这样复杂的事显然不合适怎么办?给出简单的互斥办法回顾一下临界区的概念：只允许一个进程进入

再想一想：另一个进程怎么样能执行?

被调度:另一个进程只有被调度才能执行，才可能进入临界区自然就能想到一个互斥办法:阻止调度怎么阻止调度?子曰：“温故而知新，可以为师矣。”进入临界区的另一类解法:禁止中断

cli();临界区sti();剩余区进程Pi结论:让用户开关中断是不明智的!缺点:会死机while(true){

cli();while(counter==BUFFER_SIZE);buffer[in]=item;in=(in+1)%BUFFER_SIZE;counter++;

sti();}优点:简单进入临界区其他解法的再尝试如果右边的三条机器指令原子执行就解决问题了!P.register=counter;P.register=P.register+1;counter=P.register;需修改一下硬件!显然针对右边的代码没法设计这样的原子指令!怎么办?

while(true){while(counter==BUFFER_SIZE);buffer[in]=item;in=(in+1)%BUFFER_SIZE;counter++;}进入临界区的硬件原子指令解法booleanTestAndSet(boolean&target){booleanrv=target;target=true;returnrv;}提供硬件原子指令TestAndSet一次执行完毕

while(TestAndSet(&lock));临界区lock=false;剩余区进程Pi用户需要查看硬件手册…显然也不方便将这些进入临界区(互斥)方法整理一下…给出了三种临界区进入方法(即互斥方法)软件方法:面包店算法关中断方法硬件原子指令(TestAndSet)方法都不适合由普通用户实现当然由系统承担这个复杂而易错的工作…引入“锁”

mylock=CreateLock();Acquire(&mylock);临界区Release(&mylock);剩余区进程Pi都是系统调用CreateLock(){intlock=0;returnlock;}Acquire(&lock){while(TestAndSet(&lock));}Release(&lock){

lock=0;}交给系统许多事情就不一样了!用“锁”来解生产者消费者问题while(true){Acquire(&lock);while(counter==BUFFER_SIZE);buffer[in]=item;in=(in+1)%BUFFER_SIZE;counter++;

Release(&lock);

}想一想会出什么问题?临界区可能会执行很长时间…Acquire(&lock){while(TestAndSet(&lock));

}忙等待!怎么办?不忙等，让出CPU成熟的“锁”intguard=0;Acquire(&lock){while(TestAndSet(guard));

if(lock==BUSY){

放入等待队列;

睡眠&guard=0;}else{lock=BUSY;guard=0;}}?保证这段代码的互斥进入!也要忙等，但这段代码的执行时间是确定的，且不长…用来保证用户临界区代码的互斥进入!成熟的“锁”

ReleaseRelease(&lock){ while(TestAndSet(guard)); if(等待队列上有等待进程){

从该队列上取出一个进程;

将其放入就绪队列;}else{

lock=FREE; } guard=0;}完全解决了吗?仍然存在忙等…while(true){

Acquire(&lock);//成熟版

while(counter==BUFFER_SIZE);…...

Release(&lock);

}仍然存在忙等…解决的方法仍然是sleep()while(true){

Acquire(&lock);//成熟版

if(counter==BUFFER_SIZE)sleep();

buffer[in]=item;in=(in+1)%BUFFER_SIZE;counter++;

if(counter==1)wakeup(消费者);Release(&lock);}生产者while(true){

Acquire(&lock);//成熟版

if(counter==0)sleep();item=buffer[out];out=(out+1)%BUFFER_SIZE;counter--;

if(counter==BUFFER_SIZE-1)wakeup(生产者);

Release(&lock);

}消费者术语3:同步(Synchronization)同步:多个进程按确定的协作顺序执行同步的例子随处可见…司机

while(true){

启动车辆；

正常运行；

到站停车；}售票员

while(true){

关门；

售票；

开门；}实例1:读者写者问题两组并发进程：读者和写者，共享一个数据区。(1)允许多个读者同时执行读操作；(2)不允许读者、写者同时操作；(3)不允许多个写者同时操作。实例2:可惜的是，上面的方案仍然存在问题…while(true){

Acquire(&lock);//成熟版

if(counter==BUFFER_SIZE)sleep();

…counter=counter+1;

if(counter==1)wakeup(消费者);Release(&lock);}生产者while(true){

Acquire(&lock);//成熟版

if(counter==0)sleep();…counter=counter–1;

if(counter==BUFFER_SIZE-1)wakeup(生产者);

Release(&lock);

}消费者场景:(1)

在缓冲区装满后生产者又生产了两个item，此时有两个生产者进程在等待队列中。(2)消费者进程执行1次，counter==BUFFER_SIZE-1，一个生产者进程被唤醒。(3)消费者再执行1次，counter=BUFFER_SIZE-2，另一个生产者进程没有被唤醒。需要记录“积累下来”的唤醒操作个数需互斥修改:生产者要减少它，消费者要增加它怎么记录?

当然是一个整型变量!对这个整型变量的分析

会引起睡眠:生产者将它减到0以后，生产者要睡眠会引起唤醒:消费者要增加它，如果有睡眠的生产者，需要唤醒。这是一个什么样的整型变量?

信号量由伟大人物提出的伟大概念信号量:1965年，由荷兰学者Dijkstra提出的一种特殊整型变量。信号量定义：1个数据结构+2个基本操作(ADT)

structsemaphore{ intvalue;//记录唤醒操作个数

PCB*queue;//等待在该信号量上的进程}P(semaphores);//消费唤醒操作S(semaphores);//产生唤醒操作信号量的P操作Acquire(&lock){while(TestAndSet(guard));

if(lock==BUSY){

放入等待队列;

睡眠&guard=0;}else{lock=BUSY;guard=0;}}P(semaphores){while(TestAndSet(s.guard));

s.value--;

if(s.value<0){

放入等待队列s.queue;

睡眠&s.guard=0;}else{s.guard=0;}}P的名称来源于荷兰语的proberen，即test信号量的V操作V(semaphores){while(TestAndSet(s.guard));

s.value++;

if(s.value<=0){

从s.queue上取出一个进程;

将其放入就绪队列;}}s.guard=0;}V的名称也来源于荷兰语verhogen(increment)Release(&lock){while(TestAndSet(guard));if(等待队列上有进程){

从该队列上取出一个进程;

将其放入就绪队列;}else{

lock=FREE;} guard=0;}信号量

vs.锁信号量是一个一般的锁…锁:{0,1}信号量:{-,…,-2,-1,0,1,2,…,}信号量即信号的数量，唤醒实际上就是发一个信号开锁动作也是发一个信号信号量：一个伟大的概念可以用来实现互斥可以记录睡眠操作的次数(s.value

s.value

>0)可以记录资源个数(s.value

s.value

>0)可以记录等待进程个数(|s.value|

s.value

<0)可以用来实现复杂的进程间同步关系生产者消费者的信号量解法 semaphorefullBuffers=0; semaphoreemptyBuffers=BUFFER_SIZE;

semaphoremutex=1;

Consumer(){

fullBuffers.P();

mutex.P();

item=Dequeue();

mutex.V();

emptyBuffers.V();

} Producer(item){

emptyBuffers.P();

mutex.P();

Enqueue(item);

mutex.V();

fullBuffers.V();}终于给出了一个正确的解……信号量可以控制进程的执行顺序

P1

P2

P4

P5

P3 semaphoreS1,S2,S3;S1=0;S2=0;S3=0; P1:...S1.V();S1 P2:...S2.V();S2 P3:S1.P();S2.P();...S3.V();S3S3 P4:S3.P();... P5:S3.P();...信号量解法的分析用户写起来不算难，但需要经过一定的训练。仍然可能存在一些难于发现的错误… Consumer(){

mutex.P();fullBuffers.P();

item=Dequeue();

mutex.V();

emptyBuffers.V();

} Producer(item){

mutex.P();emptyBuffers.P();

Enqueue(item);

mutex.V();

fullBuffers.V();}这样的顺序会出现死锁引入更高级的工具管程问题关键:在程序的许多地方都有信号量的操作想法:将共享数据、互斥(同步)操作集中隐藏起来，由辅助工具负责展开 monitorexample{intcounter;voidproducer(x);voidconsumer(x);} while(true){ducer(x);} while(true){example.consumer(x);}编译器将其展开 while(true){mutex.P();producer函数体;if(??)xx.V();mutex.V();}管程描述管程将共享数据、同步互斥等封装起来管程里的数据结构只能通过管程中的操作访问初始化共享数据操作一次只允许一个进程调用这些操作(互斥进入)条件变量(condition)关联一个队列，用来实现同步隐式实现互斥条件变量上的操作只有wait(即P)和signal(即V)隐式实现同步条件x用管程解决生产者

消费者问题

monitorProducerConsumer{intcounter;conditionfull,empty;voidenter(x){if(counter==N)full.wait();

将x放入缓冲区;counter++;if(counter==1)empty.signal();}itemremove(){if(counter==0)empty.wait();

从缓存区取出x;counter--;if(counter==N-1)full.signal();}}

voidproducer(){while(true){

生产x;ProducerConsumer.enter(x);}}

voidconsumer(){while(true){x=ProducerConsumer.remove();}}管程是如何展开的?一次只允许一个进程调用这些操作(互斥进入)用互斥信号量(mutex)包裹整个操作条件变量用来实现同步(wait和signal)每个条件变量展成同步信号量，并展开wait和signal复杂之处在于管程内的进程会唤醒其他进程，两个进程在管程内活动，如enter中的empty.signal，怎么办?解决办法:在管程内部再引入一个信号量(next)，唤醒(empty.signal)其他进程后自己睡在next上。管程如何展开?用互斥信号量(mutex)包住整个操作(F)每个条件变量(x)展成同步信号量，并展开wait和signal x展开成:semaphorex_sem=0;intx_count=0; mutex.P();F()函数体;if(next_count>0)next.V();elsemutex.V();x_count++;if(next_count>0)next.V();elsemutex.V();x_sem.P();x_count--;x.wait() if(x_count>0){next_count++;signal(x_sem);next.P();next_count--;}x.signal()生产者展开后的样子

voidproducer(){while(true){

生产x;ProducerConsumer.enter(x);}}while(true){生产x;mutext.P();if(counter==N){full_count++;if(next_count>0)next.V();elsemutex.V();full_sem.P();full_count--;}

将x放入缓中区;counter++;if(counter==1){if(empty_count>0){next_count++;empty_sem.V();next.P();next_count--;}if(next_count>0)next.V();elsemutex.V();}

monitorProducerConsumer{intcounter;conditionfull,empty;voidenter(x){if(counter==N)full.wait();

将x放入缓冲区;counter++;if(counter==1)empty.signal();}

上述概念的总结概念那么多，应该适时整理一下。Load/StoreDisableIntsTestAndSet管程(monitors)信号量(semaphores)锁(locks)应用程序硬件高层工具用户锁、信号量、管程是系统提供给普通用户开发并发程序的高级工具…做一个练习!读者

写者问题RRW如共享一个数据库如果整个数据库设一个锁，则效率太低显然应该:多个读者可以同时读，写者互斥、写者和读者互斥读者

写者问题解的基本结构intAR/AW活动的读者/写者个数intWR/WW等待的读者/写者个数Reader()

如果有writer，则等待;

访问数据库;

离开，唤醒等待的writer;Writer()

如果有reader或writer，则等待;

访问数据库;

离开，唤醒等待的reader和writer;需要定义读者和写者计数、互斥(同步)信号量互斥信号量mutex同步信号量ToRead和ToWrite读者

写者问题的信号量解法(1)Reader(){ToRead.P();AR++;

if(AR==1)ToWrite.P();ToRead.V();

开始读数据库;ToRead.P();AR--;if(AR==0)ToWrite.V();ToRead.V();}Writer(){

ToWrite.P();

开始写数据库;

ToWrite.V();

}第一个读者阻塞在ToWrite上以后的读者阻塞在ToRead上写者阻塞在ToWrite上读者优先，会造成写者等待很长时间读者

写者问题的信号量解法(2)Reader(){mutex.P();

while((AW+WW)>0){WR++;mutex.V();ToRead.P();mutex.P();WR—-}AR++;mutex.V();

开始读数据库;mutex.P();AR--;if(AR==0&&WW>0)ToWrite.V();mutex.V();}mutex用来互斥访问AR、AW、WR、WW等写者优先:没有等待的写者时读者才进入所有读者等在ToRead上，所有写者等在ToWrite上要保持警惕!读者

写者问题的信号量解法(2)Writer(){mutex.P();

while((AW+AR)>0){WW++;mutex.V();ToWrite.P();mutex.P();WW—-}AW++;mutex.V();

开始写数据库;mutex.P();AW--;

if(WW>0)ToWrite.V();elseif(WR>0)ToRead.V();mutex.V();}一个显然的感觉:太多的信号量、太多的操作…写者优先是否合适?ToRead.broadcast读者

写者问题的管程解法

monitorReaderWriter{intAR,WR,AW,WW;conditionToRead,ToWrite;itemread(){while((AW+WW)>0){WR++;ToRead.wait();WR--}AR++;读出数据库元素;AR--;if(AR==0&&WW>0)ToWrite.signal();}write(x){while((AW+AR)>0){WW++;ToWrite.wait();WW—-}AW++;将x写入数据库;AW--;if(WW>0)ToWrite.signal();elseif(WR>0)ToWrite.signal();}}(或broadcast())实际操作系统提供哪些同步工具!Linux同步工具原子操作atomic_t自旋锁spinlockLinux包含了几乎所有现代的操作系统具有的同步机制

信号量semaphoreBKL(BigKernelLock)、rwlock、brlock、seqlock、rw_semaphore等不同的场合应用不同的工具会有不同的效果!Linux同步工具原子操作原子操作:一旦开始执行不被任何事件打断

通常用来对资源引用进行计数(atomic_inc(refcnt))gcc不对其进行优化在include/asm/atomic.h中typedefstruct{volatileintcounter;}atomic_t;static__inline__voidatomic_inc(atomic_t*v){__asm____volatile__(LOCK;"incl%0":"=m"(v->counter):"m"(v->counter));}#ifdefCONFIG_SMP#defineLOCK"lock;"#else#defineLOCK""#endif不是SMP，LOCK为空。CPU在单条指令执行完毕前不会中断。一条指令Linux同步工具原子操作什么是SMP?

SMP:SymmetricalMulti-Processing，对称多处理技术CPUCPUCPU内存外设此时LOCK被扩展成指令lock;lock;是一个指令前缀，可使修饰指令在执行期间的LOCK引脚变成逻辑0，禁止其它处理器存取带有lock;前缀的指令的存储器操作数(暂时封锁总线)

Linux同步工具自旋锁自旋锁，一直旋转的锁，大家是否记得…

为什么自旋?Acquire(&lock){while(TestAndSet(&lock));

}一直忙等，所以称为自旋不自旋就要切换上下文，如果切换上下文时间比自旋时间长的多…spinlock_tlock=SPIN_LOCK_UNLOCKED;spin_lock_irq(&lock);

//临界区

spin_unlock_irq(&lock);Linux自旋锁spinlock在include/linux/spinlock.h中#definespin_lock_irq(lock)do{local_irq_disable();spin_lock(lock);}while(0)#definespin_unlock_irq(lock)do{spin_uplock(lock);local_irq_enable();}while(0)在include/asm/system.h中#definelocal_irq_disable()__asm____volatile__("cli":::"memory")#definelocal_irq_enable()__asm____volatile__("sti":::"memory")在单处理器环境中spin_lock(lock)和spin_uplock(lock)及spinlock_t都展开为空，所以单CPU环境中自旋锁就是开关中断…Linux自旋锁spinlock的SMP实现staticinlinevoidspin_lock(spinlock_t*lock)

{__asm____volatile__(spin_lock_string):”=m”(lock->lock)::“memory”);}#definespin_lock_string“\n1:\t”“lock;decb%0\n\t”“js2f\n”“.section.text.lock,\”ax\”\n”“2:\t”“cmpb$0,%0\n\t”“rep;nop\n\t”“jle2b\n\t”“jmp1b\n”“.previo