信號(hào)量

背景

信號(hào)量（semaphore）

• 抽象數(shù)據(jù)類(lèi)型

  • 一個(gè)整形（sem），兩個(gè)原子操作

  • P()：sem減1，如果sem<0，等待，否則繼續(xù)

  • V()：sem加1，如果sem<=0，喚醒一個(gè)等待的P

信號(hào)量使用

特性

• 信號(hào)量是被保護(hù)的整數(shù)變量

  • 初始化完成后，只能通過(guò)P()和V()操作修改

  • 由操作系統(tǒng)保證，PV操作是原子操作

• P()可能阻塞，V()不會(huì)阻塞

• 通常假定信號(hào)量是”公平的“

  • 線(xiàn)程不會(huì)被無(wú)限期阻塞在P()操作

  • 假定信號(hào)量等待按先進(jìn)先出排隊(duì)

分類(lèi)

• 二進(jìn)制信號(hào)量：可以是0或1

• 一般/計(jì)數(shù)（資源）信號(hào)量：可取任何非負(fù)值

• 兩者等價(jià)：基于一個(gè)可以實(shí)現(xiàn)另一個(gè)

使用

• 互斥訪(fǎng)問(wèn)：臨界區(qū)的互斥訪(fǎng)問(wèn)控制

• 條件同步：線(xiàn)程間的事件等待

生產(chǎn)者-消費(fèi)者問(wè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)者可訪(fǎng)問(wèn)緩沖區(qū)

• 問(wèn)題分析

  • 任何時(shí)刻只能有一個(gè)線(xiàn)程操作緩沖區(qū)（互斥訪(fǎng)問(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

class BoundedBuffer{
    mutex =new Semaphore(1);
    fullBuffers =new Semaphore(0);
    emptyBuffers =new Semaphore(n);
}
BoundedBuffer::Producter(c){
    emptyBuffers ->P();
    mutex ->P();
    ADD C TO THE BUFFER;
    mutex ->V();
    fullBuffers ->V();
}
BoundedBuffer::Consumer(c){
    fullBuffers ->P();
    mutex ->P();
    REMOVE C FROM BUFFER;
    mutex ->V();
    emptyBuffers ->V();
}

信號(hào)量實(shí)現(xiàn)

• 實(shí)現(xiàn)

class Semaphore{
    int sem;
    WaitQueue q;
}
Semaphore: P(){
    sem--;
    if(sem<0){
        ADD THIS THREAD t TO q;
        block(q);
    }
}
Semaphore: V(){
    sem++;
    if(sem<=0){
        REMOVE a THREAD t FROM q;
        wakeup(t);
    }
}

• 困難

  • 讀/開(kāi)發(fā)代碼比較困難

  • 容易出錯(cuò)。使用的信號(hào)量已經(jīng)被另一個(gè)線(xiàn)程占用；忘記釋放信號(hào)量

  • 不能夠處理死鎖問(wèn)題

管程（Moniter）

• 管程是一種用于多線(xiàn)程互斥訪(fǎng)問(wèn)共享資源的程序結(jié)構(gòu)

  • 采用面向?qū)ο蠓椒?，?jiǎn)化了線(xiàn)程間的同步控制

  • 任一時(shí)刻最多只有一個(gè)線(xiàn)程執(zhí)行管程代碼

  • 正在管程中的線(xiàn)程可臨時(shí)放棄管程的互斥訪(fǎng)問(wèn)，等待事件出現(xiàn)時(shí)恢復(fù)

• 使用

  • 在對(duì)象/模塊中，收集相關(guān)共享數(shù)據(jù)

  • 定義訪(fǎng)問(wèn)共享數(shù)據(jù)的方法

• 組成

  • 一個(gè)鎖：控制管程代碼的互斥訪(fǎng)問(wèn)

  • 0或者多個(gè)條件變量：管理共享數(shù)據(jù)的并發(fā)訪(fǎng)問(wèn)

• 條件變量

  • 條件變量是管程內(nèi)的等待機(jī)制

    • 進(jìn)入管程的線(xiàn)程因資源被搶占而進(jìn)入等待狀態(tài)

    • 每個(gè)條件變量表示一種等待原因，對(duì)應(yīng)一個(gè)等待隊(duì)列

  • wait()操作

    • 將自己阻塞在等待隊(duì)列中

    • 喚醒一個(gè)等待者或釋放管程的互斥訪(fǎng)問(wèn)

  • signal()操作

    • 將等待隊(duì)列中的一個(gè)線(xiàn)程喚醒

    • 如果等待隊(duì)列為空，則等同空操作

• 條件變量實(shí)現(xiàn)

class Condition{
    int numWaiting =0;
    WaitQueue q;
}
Condition::Wait(lock){
    numWaiting++;
    ADD this thread t to q;
    release(lock);
    schedule();
    require(lock);
}
Condition::Signal(){
    if(numWaiting >0){
        REMOVE a thread t from q;
        wakeup(t);
        numWaiting--;
    }
}

• 管程解決生產(chǎn)者-消費(fèi)者問(wèn)題

class BoundedBuffer {
    …
    Lock lock;
    int count = 0;
    Condition notFull, notEmpty;
}
BoundedBuffer::Producter(c) {
    lock->Acquire();
    while (count == n)
        notFull.Wait(&lock);
    Add c to the buffer;
    count++;
    notEmpty.Signal();
    lock->Release();
}
BoundedBuffer::Consumer(c) {
    lock->Acquire();
    while (count == 0)    
      notEmpty.Wait(&lock);
    Remove c from buffer;
    count--;
    notFull.Signal();
    lock->Release();
}

經(jīng)典同步問(wèn)題

讀者-寫(xiě)者問(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ě)

• 信號(hào)量方法

class Writer(){
    P(WriteMutex);
    write;
    V(WriteMutex);
}
class Reader(){
    P(CountMutex);
    if(Rcount ==0)
        P(WriteMutex);
    ++Rcount;
    V(CountMutex);
    
    read;
    
    P(CountMutex);
    --Rcount;
    if(Rcount ==0)
        V(WriteMutex);
    V(CountMutex);
}

• 管程方法

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;

//讀者
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();
}
Public Database::Read() {
  //Wait until no writers;
  StartRead(); 
  read database;
  //check out – wake up waiting writers; 
  DoneRead(); 
}

//寫(xiě)者
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();
}
Public Database::Read() {
  //Wait until no writers;
  StartRead(); 
  read database;
  //check out – wake up waiting writers; 
  DoneRead(); 
}

哲學(xué)家就餐問(wèn)題

• 問(wèn)題描述

  • 5個(gè)哲學(xué)家圍繞一張圓桌而坐

    • 桌子上放著5支叉子

    • 每?jī)蓚€(gè)哲學(xué)家之間放一支

  • 哲學(xué)家的動(dòng)作包括思考和進(jìn)餐

    • 進(jìn)餐時(shí)需同時(shí)拿到左右兩邊的叉子

    • 思考時(shí)將兩只叉子放回原處

更多精彩，關(guān)注“咋家”