【CO003】作業系統筆記4 —— IPC 問題

筆者：YY同學

生命不息，代碼不止，好玩的專案盡在GitHub

什么是 IPC 問題？

IPC（Inter Process Communication）問題是行程或執行緒之間相互通信時可能會發生的一系列問題，一般是共享資料的讀寫沖突問題，通常情況下，對于共享資料，我們希望在某個時刻只有一個行程（執行緒）在操作（增、刪、查、改）它，這就意味著此時其他行程（執行緒）無法進入該資料域，這種性質就是互斥性（Mutual Exclusion），而該資料域又稱沖突域（Critical Section），IPC 問題大多數是解決沖突域的問題，解決方案需要具備良好的互斥性，

沖突域（Critical Section）

沖突域是指代碼中可能存在讀寫沖突且含有共享變數的區域，上圖表示的是 IPC 問題的一般解決思路，由進入沖突、沖突部分和退出沖突三個部分的代碼塊組成，其中，在沖突部分中實作單行程（執行緒）的讀寫會比較安全，這意味著這一程序將沒有其他行程（執行緒）參與，

IPC 問題解決方案必須滿足的四大準則

五種 IPC 問題解決方案

解決方案一：禁用中斷

• 描述：在沖突域內部禁止使用背景關系切換，
• 優點：可以杜絕其他行程（執行緒）對進入沖突域的執行行程（執行緒）的干擾，
• 缺點：當有行程（執行緒）進入沖突域之后，會對所有行程進行強制中斷，
• 評價：該解決方案雖然能達到目的，但是缺乏限制，太過隨意，

解決方案二：添加 🔒 變數

• 描述：通過 lock 變數來判斷當前沖突域是否有行程（執行緒）進入，如有則制止其他行程（執行緒）進入，
• 優點：想法很不錯，看似可行，
• 缺點：但是 lock 本身也是共享變數，變數的公有性賦予了所有行程（執行緒）對其更改值的權力，
• 評價：該解決方案沒有辦法達到目的，因為每個行程（執行緒）在進入沖突域前都可以更改 lock 的值，相當于門鎖不起任何作用，任何人都可以從外面開鎖，

解決方案三：加入 turn 變數嚴格輪換

• 描述：通過 turn 的值來判斷當前輪到哪個行程（執行緒）進入沖突域并且制止其他行程（執行緒）進入，
• 優點：與 lock 變數不同，turn 變數更改的位置是在沖突域之后，并且是多行程（執行緒）輪換，相當于只有等當前行程（執行緒）離開沖突域之后，turn 的值才會變化，別的行程（執行緒）此時才能進入沖突域，這符合我們的設計需要，
• 缺點：CPU 執行效率較低，而且違反了第三準則，當 Process 0 離開沖突域時，turn 的值變為 1，此時只有 Process 1 能夠進入沖突域，但是此時 Process 1 可能才剛剛開始執行沖突域之前部分的代碼，此時 Process 1 并未進入沖突域，但是它把其他行程（執行緒）鎖在外面了，
• 評價：變數設定上可行，但是 CPU 執行效率較低且違背第三準則，所以該解決方案沒有辦法達到目的，

解決方案四：彼得遜解決方案

• 描述：使用 True 和 False 作為信號，每個行程（執行緒）信號獨立，
• 優點：采用分塊理念，設定控制信號，使用函式封裝塊增加使用的便捷性，提高效率，
• 缺點：浪費 CPU 執行時間，優先級顛倒問題（優先級高的行程可能會陷入忙等），
• 評價：結合前三種解決方案的優點，彌補其缺點，雖然仍有不足，但不失為一個可行的解決方案，

解決方案五：信號量（Semaphore）

• 描述：在彼得遜解決方案的基礎上推廣到一般情況，設定信號量 *s，初始值為 1，當 *s = 0 的時候行程（執行緒）鎖定，
• 優點：使用原子操作函式 up() 和 down()，杜絕了沖突域內的背景關系切換，
• 缺點：無，
• 評價：對于每個行程（執行緒），在進入沖突域前先鎖定，離開沖突域之后再把鎖打開，原子性保證互斥性的同時，也將時間效率最大化，是一個較完美的解決方案，

原子操作函式偽代碼

void down(semaphore *s) { 
    if( *s > 0 )
        *s = *s - 1;
    else
        sleep();
}

void up(semaphore *s) { 
    if( *s == 0 )
        wakeup_proc();
    *s = *s + 1;
}

三類 IPC 問題模型

1. 生產者-消費者模型（Producer-Consumer Model）

• 行程（執行緒）池中有若干生產者和消費者，以及一個產品倉庫，
• 生產者負責生產產品存入倉庫，消費者負責從倉庫消化生產者生產的產品，
• 同時只能有一個行程（執行緒）在運行且產品倉庫的容量一定，會實時顯示當前倉庫中有多少產品，
• 當倉庫滿的時候生產者將會停止生產產品（進入 sleep 狀態），直到倉庫有空位出現；同理，當倉庫空的時候，消費者會停止消費產品（進入 sleep 狀態），

偽代碼

#define N 100
typedef int semaphore;
semaphore mutex = 1;  // mutual exclusive，互斥信號量，等于 0 時鎖定
semaphore empty = N;  // 表示一開始有 N 個位置是空的
semaphore full = 0;   // 表示一開始有 0 個位置是滿的

/*  Producer  */
void producer(void) {
    int item;
    while(TRUE) {
        item = produce_item();
        down(&empty);  // 如果滿了就讓自己睡眠，由消費者喚醒
        down(&mutex);  // mutex 必須放在后面，避免發生死鎖（dead lock）
        insert_item(item);
        up(&mutex);
        up(&full);
    }
}

/*  Consumer  */
void consumer(void) {
    int item; 
    while(TRUE) {
        down(&full);   // 如果空了就讓自己睡眠，由生產者喚醒
        down(&mutex);  // mutex 必須放在后面，避免發生死鎖（dead lock）
        item = remove_item();
        up(&mutex);
        up(&empty);
        consume_item(item);
    }
}

2. 資料庫讀寫模型（Database R&W Model）

• 行程（執行緒）池中有若干行程（執行緒），分別代表不同的 Reader 和 Writer，
• 在 Writer 沒有寫入資料的情況下，多個 Reader 可以同時讀取資料，并且在 Reader 讀資料的時候 Writer 無法寫入資料，
• 在 Writer 寫入資料時，所有其他 Reader 和 Writer 無法讀寫，

偽代碼

semaphore db = 1;     // 確保讀寫行程無沖突
semaphore mutex = 1;  // 確保單個行程無沖突
int read_count = 0;

void writer(void) {
    while(TRUE) {
        prepare_write();
        down(&db);  // 寫的時候不能有其他讀寫
        
        write_database();
        up(&db);
    }
}

void reader(void) {
    while(TRUE) {
        down(&mutex);
        read_count++;
        if(read_count == 1)
            down(&db);  // 讀的時候不能寫可以讀，但是讀需要一個個讀，所以要暫時鎖住
        up(&mutex);

        read_database();

        down(&mutex);
        read_count--;
        if(read_count == 0)
            up(&db);  // 一個讀完，打開門鎖，讀下一個
        up(&mutex);
        process_data();
    }
}

3. 哲學家進餐模型（Dining Philosopher Model）

• 5 個哲學家圍著圓桌吃飯，桌上有 5 根筷子，
• 每個哲學家需要拿到一雙筷子（ 2 根）才能吃飯，
• 每個哲學家吃完飯后會把手中的筷子放回桌上，供別的哲學家吃飯，

偽代碼

#define N 5
#define LEFT  ((i+N-1) % N)  // 我們假定每根筷子擺在兩個哲學家中間，誰都可以拿起筷子
#define RIGHT  ((i+1) % N)  // LEFT 和 RIGHT 分別代表當前哲學家左邊和右邊的人

int state[N];  // 三種狀態 THINKING，HUNGRY 和 EATING
semaphore mutex = 1;
semaphore s[N];  // 控制哲學家狀態的信號量

void philosopher(int i) {
    think();
    take_chopsticks(i);  // Entry of Critical Section
    eat();  // Critical Section
    put_chopsticks(i);  // Exit of Critical Section
}

void take_chopsticks(int i) {
    down(&mutex);
    state[i] = HUNGRY;
    test(i);  // 測驗是否能拿起一雙筷子
    up(&mutex);
    down(&s[i]);  // 可以拿起筷子的話就吃飯，不可以就 sleep
}

void test(int i) {
    if(state[i] == HUNGRY && state[LEFT] != EATING && state[RIGHT] != EATING){
        state[i] = EATING;
        up(&s[i]);  // 喚醒還沒吃飯的哲學家
    }
}

void put_chopsticks(int i) {
    down(&mutex);
    state[i] = THINKING;
    test(LEFT);   // 放下筷子后紳士地問左邊的哲學家吃不吃
    test(RIGHT);  // 放下筷子后紳士地問右邊的哲學家吃不吃
    up(&mutex);
}