第10章：信號(hào)量與管程
信號(hào)量
信號(hào)量使用
互斥訪問(wèn)
條件同步
管程
經(jīng)典同步問(wèn)題
哲學(xué)家就餐問(wèn)題
讀者-寫(xiě)者問(wèn)題
10.1 信號(hào)量（Semaphore）
回顧
- 并發(fā)問(wèn)題
多線程并發(fā)導(dǎo)致資源競(jìng)爭(zhēng) - 同步概念
協(xié)調(diào)多線程對(duì)共享數(shù)據(jù)的訪問(wèn)
任何時(shí)刻只能有一個(gè)線程執(zhí)行臨界區(qū)代碼 - 確保同步正確的方法
底層硬件支持
高層次的編程抽象
信號(hào)量
- 信號(hào)量是操作系統(tǒng)提供的一種協(xié)調(diào)共享資源訪問(wèn)的方法
軟件同步是平等線程間的一種同步協(xié)商機(jī)制
操作系統(tǒng)是管理者星岗，地位高于進(jìn)程
用信號(hào)量表示系統(tǒng)資源的數(shù)量 - 由Dijkstra在20世紀(jì)60年代提出
早期的操作系統(tǒng)的主要同步機(jī)制
現(xiàn)在很少用（但在計(jì)算機(jī)科學(xué)研究還是非常重要） - 信號(hào)量是一種抽象的數(shù)據(jù)類(lèi)型
由一個(gè)整型（sem）變量和兩個(gè)原子操作組成
- P()（Prolaag（荷蘭語(yǔ)嘗試減少））
sem減1
如sem<0瞒爬，進(jìn)入等待恍风，否則繼續(xù) - V()（Verhoog（荷蘭語(yǔ)增加））
sem加1
如sem<=0棒旗，喚醒一個(gè)等待進(jìn)程
信號(hào)量的特性
- 信號(hào)量是被保護(hù)的整數(shù)變量
初始化完成后，只能通過(guò)P()和V()操作修改
由操作系統(tǒng)保證皮壁，PV操作是原子操作 - P()可能阻塞椭更，V()不會(huì)阻塞
- 通常假定信號(hào)量是“公平的”
線程不會(huì)被無(wú)限期阻塞在P()操作
假定信號(hào)量等待按先進(jìn)先出排隊(duì)
自旋鎖是做不到先進(jìn)先出的，因?yàn)樽孕i需要占用CPU隨時(shí)地去查蛾魄，有可能臨界區(qū)使用者退出的時(shí)候它剛查完虑瀑，下一個(gè)進(jìn)入者是誰(shuí)去查它就能進(jìn)去
信號(hào)量的實(shí)現(xiàn)
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 信號(hào)量的使用
- 信號(hào)量分類(lèi)
可分為兩種信號(hào)量
二進(jìn)制信號(hào)量：資源數(shù)目為0或1
資源信號(hào)量：資源數(shù)目為任何非負(fù)值 - 兩者等價(jià)
基于一個(gè)可以實(shí)現(xiàn)另一個(gè) - 信號(hào)量使用
互斥訪問(wèn)
臨界區(qū)的互斥訪問(wèn)控制
條件同步
線程間的事件等待 - 用信號(hào)量實(shí)現(xiàn)臨界區(qū)的互斥訪問(wèn)
//每類(lèi)資源設(shè)置一個(gè)信號(hào)量湿滓，其初值為1
mutex = new Semaphore(1); //初始化
mutex -> P(); //第一個(gè)線程進(jìn)來(lái)之后，由1變成0舌狗，能進(jìn)去叽奥；若第二個(gè)線程也想進(jìn)入臨界區(qū)，由0變成-1痛侍，在P操作進(jìn)入等待狀態(tài)
critical section;
mutex -> V(); //第一次線程執(zhí)行結(jié)束朝氓，進(jìn)行V操作，使信號(hào)量計(jì)數(shù)由-1變成0主届， 喚醒第二個(gè)進(jìn)程膀篮，則第二個(gè)進(jìn)程就會(huì)在第一個(gè)進(jìn)程結(jié)束時(shí)進(jìn)入臨界區(qū)
1
2
3
4
- 必須成對(duì)使用P()和V()操作
P()操作保證互斥訪問(wèn)臨界資源
V()操作在使用后釋放臨界資源
PV操作不能次序錯(cuò)誤、重復(fù)或遺漏
用信號(hào)量實(shí)現(xiàn)條件同步
//條件同步設(shè)置一個(gè)信號(hào)量岂膳，其初值為0
condition = new Semaphore(0);
//線程A 線程B
...M...
condition -> P();
...N... //使用數(shù)據(jù) ...X... //準(zhǔn)備數(shù)據(jù)
condition -> v();
...Y...
1
2
3
4
5
6
7
8
9
- 條件等待：線程A要等待線程B執(zhí)行完X模塊才能執(zhí)行N模塊
假設(shè)B先執(zhí)行完X，此時(shí)信號(hào)量變成1磅网，則N能直接往下執(zhí)行
假設(shè)A先執(zhí)行完M谈截，此時(shí)信號(hào)量變成-1，阻塞涧偷，進(jìn)入等待狀態(tài)簸喂，然后進(jìn)程B進(jìn)行V操作，釋放燎潮，此時(shí)信號(hào)量為0喻鳄。若信號(hào)量小于或等于0，則說(shuō)明有線程在等待确封，此時(shí)喚醒等待中的線程A除呵，執(zhí)行N模塊
生產(chǎn)者-消費(fèi)者問(wèn)題
生產(chǎn)者 -> 緩沖區(qū) -> 消費(fèi)者
- 有界緩沖區(qū)的生產(chǎn)者-消費(fèi)者問(wèn)題描述
一個(gè)或多個(gè)生產(chǎn)者在生成數(shù)據(jù)后放在一個(gè)緩沖區(qū)里
單個(gè)消費(fèi)者從緩沖區(qū)取出數(shù)據(jù)處理
任何時(shí)刻只能有一個(gè)生產(chǎn)者或消費(fèi)者可訪問(wèn)緩沖區(qū) - 問(wèn)題分析
任何時(shí)刻只能有一個(gè)線程操作緩沖區(qū)（互斥訪問(wèn)）
緩沖區(qū)空時(shí)，消費(fèi)者必須等待生產(chǎn)者（條件同步）
緩沖區(qū)滿(mǎn)時(shí)爪喘，生產(chǎn)者必須等待消費(fèi)者（條件同步） - 用信號(hào)量描述每個(gè)約束
二進(jìn)制信號(hào)量 mutex
資源信號(hào)量 fullBuffers
資源信號(hào)量 emptyBuffers
//信號(hào)量實(shí)現(xiàn)
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();
}
- 使用信號(hào)量的困難
讀/開(kāi)發(fā)代碼比較困難
程序員需要能運(yùn)用信號(hào)量機(jī)制
容易出錯(cuò)
使用的信號(hào)量已經(jīng)被另一個(gè)線程占用
忘記釋放信號(hào)量
不能夠處理死鎖問(wèn)題
10.3 管程（Monitor）
- 改進(jìn)信號(hào)量在處理臨界區(qū)的時(shí)候的一些麻煩
- 管程是一種用于多線程互斥訪問(wèn)共享資源的程序結(jié)構(gòu)
采用面向?qū)ο蠓椒ㄑ赵?jiǎn)化了線程的同步機(jī)制
任一時(shí)刻最多只有一個(gè)線程執(zhí)行管理代碼
正在管程中的線程可臨時(shí)放棄管理的互斥訪問(wèn)，等待事件出現(xiàn)時(shí)恢復(fù)
注 - 一個(gè)線程在臨界區(qū)執(zhí)行秉剑，必須執(zhí)行到它退出臨界區(qū)泛豪，它才能放棄臨界區(qū)的互斥訪問(wèn)，而管程允許在執(zhí)行的過(guò)程當(dāng)中侦鹏，臨時(shí)放棄诡曙。放棄之后其他線程就可進(jìn)入管程區(qū)域
管程的使用
- 在對(duì)象/模塊中，收集相關(guān)共享數(shù)據(jù)
- 管程組成
一個(gè)鎖
控制管程代碼的互斥訪問(wèn) - 0個(gè)或多個(gè)條件變量
0個(gè)等同于臨界區(qū)
管理共享數(shù)據(jù)的并發(fā)訪問(wèn)
- 條件變量（Condition Variable）
條件變量是管程內(nèi)的等待機(jī)制
進(jìn)入管程的線程因資源被占用而進(jìn)入等待狀態(tài)
每個(gè)條件變量表示一種等待原因略水，對(duì)應(yīng)一個(gè)等待隊(duì)列 - Wait()操作
將自己阻塞在等待隊(duì)列中
喚醒一個(gè)等待著或釋放管程的互斥訪問(wèn) - Signal()操作
將等待隊(duì)列中的一個(gè)線程喚醒
如果等待隊(duì)列為空价卤，則等同空操作
//條件變量實(shí)現(xiàn)
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 --;
}
}
用管程解決生產(chǎn)者-消費(fèi)者問(wèn)題
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();
}
管理?xiàng)l件變量的釋放處理方式
Hansen 管程
l.acquire()
...
x.wait() //T1進(jìn)入等待
//T2進(jìn)入管程 l.acquire()
...
x.Signal()
...
//T1退出管程 l.release()
... //T1恢復(fù)管程執(zhí)行
l.release()
Hoare 管程
l.acquire()
...
x.wait() //T1進(jìn)入等待
//T2進(jìn)入管程 l.acquire()
...
//T2進(jìn)入等待 x.Signal()
... //T1恢復(fù)管程執(zhí)行
l.release() //T1結(jié)束
... //T2恢復(fù)管程執(zhí)行
l.release()
Hansen 管程和Hoare 管程
- Hansen 管程
主要用于真實(shí)操作系統(tǒng)和Java中
當(dāng)前正在執(zhí)行的線程優(yōu)先級(jí)更高
效率更高，少了一次切換 - Hoare 管程
內(nèi)部的線程優(yōu)先執(zhí)行
優(yōu)先角度考慮更合理聚请，更容易證明正確性 - Hansen 管程和Hoare 管程區(qū)別
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 經(jīng)典同步問(wèn)題
（一）哲學(xué)家就餐問(wèn)題
- 問(wèn)題描述
5個(gè)哲學(xué)家圍繞著一張圓桌而坐
桌上放著5支叉子
每?jī)蓚€(gè)哲學(xué)家之間放1支
哲學(xué)家的動(dòng)作包括思考和進(jìn)餐
進(jìn)餐時(shí)需要同時(shí)拿到左右兩邊的叉子
思考時(shí)將兩支叉子放回原處
如何保證哲學(xué)家們的動(dòng)作有序進(jìn)行荠雕？
如：不出現(xiàn)永遠(yuǎn)都難不到叉子的情況
- 方案1
define N 5 //哲學(xué)家個(gè)數(shù)
semaphore fork[5]; //信號(hào)量初值為1
void philosopher(int i) //哲學(xué)家編號(hào)：0-4
while (TRUE)
{
think(); //哲學(xué)家在思考
P(fork[i]); //去拿左邊的叉子
P(fork[(i+1)%N]); //去拿右邊的叉子
eat(); //吃面條中
V(fork[i]); //放下左邊的叉子
V(fork[(i+1)%N]); //放下右邊的叉子
}
不正確稳其，可能導(dǎo)致死鎖
5個(gè)人同時(shí)拿到左邊刀叉時(shí)，就會(huì)進(jìn)入死鎖狀態(tài)
- 方案2
一旦無(wú)法充分利用系統(tǒng)資源炸卑，就把所有資源打成一個(gè)整包既鞠，只有一個(gè)進(jìn)程可占用這一整包的資源，不管它是不是都需要盖文，這種情況系統(tǒng)也能正常運(yùn)行嘱蛋，只是效率低一些。
#define N 5 //哲學(xué)家個(gè)數(shù)
semaphore fork[5]; //信號(hào)量初值為1
semaphore mutex; //互斥信號(hào)量五续，初值為1
void philosopher(int i) //哲學(xué)家編號(hào)：0-4
while (TRUE)
{
think(); //哲學(xué)家在思考
P(mutex); //進(jìn)入臨界區(qū)洒敏，任一時(shí)刻只能有一個(gè)進(jìn)程申請(qǐng)到二進(jìn)制信號(hào)量
P(fork[i]); //去拿左邊的叉子
P(fork[(i+1)%N]); //去拿右邊的叉子
eat(); //吃面條中
V(fork[i]); //放下左邊的叉子
V(fork[(i+1)%N]); //放下右邊的叉子
V(mutex); //退出臨界區(qū)
}
互斥訪問(wèn)正確，但每次只允許一個(gè)人進(jìn)餐疙驾，性能不好
- 方案3
#define N 5 //哲學(xué)家個(gè)數(shù)
semaphore fork[5]; //信號(hào)量初值為1
void philosopher(int i) //哲學(xué)家編號(hào)：0-4
while (TRUE)
{
think(); //哲學(xué)家在思考
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]); //放下右邊的叉子
}
沒(méi)有死鎖凶伙，可有多人同時(shí)訪問(wèn)
（二）讀者-寫(xiě)者問(wèn)題
- 問(wèn)題描述
共享數(shù)據(jù)的兩類(lèi)使用者
讀者：只讀取數(shù)據(jù)，不修改
寫(xiě)者：讀取和修改數(shù)據(jù)
讀者-寫(xiě)者問(wèn)題描述：對(duì)共享數(shù)據(jù)的讀寫(xiě)
- “讀-讀”允許
同一時(shí)刻它碎，允許有多個(gè)讀者同時(shí)讀 - “讀-寫(xiě)”互斥
沒(méi)有寫(xiě)者時(shí)讀者才能讀
沒(méi)有讀者時(shí)寫(xiě)者才能寫(xiě) - “寫(xiě)-寫(xiě)”互斥
沒(méi)有其他寫(xiě)者時(shí)寫(xiě)者才能寫(xiě)
用信號(hào)量解決讀者-寫(xiě)者問(wèn)題
用信號(hào)量描述每個(gè)約束
- 信號(hào)量WriteMutex
控制讀寫(xiě)操作互斥
初始化為1 - 讀者計(jì)數(shù)Rcount
正在進(jìn)行讀操作的讀者數(shù)目
初始化為0 - 信號(hào)量CountMutex
控制對(duì)讀者計(jì)數(shù)的互斥修改
初始化為1
//Writer
P(WriteMutex);
write;
V(WriteMutex);
//Reader
P(CountMutex);
if(Rcount == 0) //第一個(gè)讀者才需要申請(qǐng)讀寫(xiě)互斥信號(hào)量函荣，后來(lái)讀者只需讀者計(jì)數(shù)器加1
P(WriteMutex);
++ Rcount;
V(CountMutex);
read;
P(CountMutex);
-- Rcount;
if(Rcount == 0) //只有最后一個(gè)讀者才需要釋放讀寫(xiě)互斥信號(hào)量、
V(WriteMutex);
V(CountMutex);
- 特征：讀者優(yōu)先
即后來(lái)的讀者一定會(huì)先于等待的寫(xiě)者進(jìn)行寫(xiě)操作
讀者-寫(xiě)者問(wèn)題：優(yōu)先策略
- 讀者優(yōu)先策略
只要有讀者正在讀狀態(tài)扳肛，后來(lái)的讀者都能直接進(jìn)入
如讀者持續(xù)不斷進(jìn)入傻挂，則寫(xiě)者就處于饑餓 - 寫(xiě)者優(yōu)先策略
只要有寫(xiě)者就緒，寫(xiě)者應(yīng)盡快執(zhí)行寫(xiě)操作
如寫(xiě)者持續(xù)不斷就緒挖息，則讀者就處于饑餓
用管程解決讀者-寫(xiě)者問(wèn)題
兩個(gè)基本方法
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;
}
管程的狀態(tài)變量
AR=0; //# of active readers
AW=0; //# of active writers
// 只有一個(gè)>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) { //條件決定了優(yōu)先策略
WR++;
okToRead.wait(&lock);
WR--;
}
AR++;
lock.Release();
}
Private Database :: DoneRead() {
lock.Acquire();
AR--;
if ((AR==0 && WW)>0) { //條件決定了優(yōu)先策略
okToWrite.signal();
}
lock.Release();
}
解決方案詳情：寫(xiě)者
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();
}
