用信號量和讀寫鎖解決讀者寫者問題

讀者寫者問題是非常經典的同步問題，本文首先用信號量來解決這個問題，並結合代碼分析什么是讀者優先、什么是寫者優先，然后給出讀寫鎖的解決方案，並指出在Linux下讀寫鎖的注意事項。

 

讀者寫者問題

　　讀者寫者問題描述的是這么一種情況：對象在多個線程（或者進程）之間共享，其中一些線程只會讀數據，另外一些線程只會寫數據。為了保證寫入和讀取的正確性，我們需要保證，只要有線程在寫，那么其他線程不能讀，否則可能讀到寫了一半的數據；另外，也不能有兩個線程同時寫，否則導致數據錯亂。當然，多個線程是可以同時讀數據。

　　讀者寫者問題在計算機領域非常普遍，大家最容易想到的就是數據庫，數據庫的讀寫分離也是為了減少因為讀者寫者問題加鎖帶來的對並發的影響。

　　讀者寫者問題的解決方案一般都有兩種不同的側重：讀者優先或者寫者優先。簡單來說，讀者優化就是盡量滿足並發的讀操作，當已經有線程在讀數據的時候，其他讀線程無需等待，而寫線程需要等待所有正在進行的讀操作之后才能執行。寫者優先就是盡量滿足寫操作，盡管寫操作不能並發，但是可以排隊，優先於等待的讀線程獲得執行權。

信號量解決讀者寫者問題

　　關於信號量的基礎知識，可以參見上一篇文章《並發與同步、信號量與管程、生產者消費者問題》的講解。這個章節通過信號量來解決讀者寫者問題，為了更好的分析，可以用下面的情況來概括所有的並發的可能。首先r代表讀線程 w代表寫線程，我們給出三元組（X， Y Z）， X代表當前已經獲取資源訪問權的線程， Y和Z代表隨后並發到達的線程（注意，YZ是並發的），那么綜合有以下情況：

　　(r rr)
　　(r rw)
　　(r wr)
　　(r ww)
　　(w rr)
　　(w rw)
　　(w wr)
　　(w ww)

　　另外，當V操作釋放信號量時，如果有多個線程阻塞在當前信號量，那么該喚醒哪一個線程呢？有的教程說有隊列，先來的先服務（FCFS）；也有文章說，it is undefined，所以不要做任何假設。為了方便后文分析，假設喚醒的線程是隨機的。因此上面的八種情況中YZ的值為 rw 和 wr是一樣的。

　　另外YZ為rr 和 ww的情況也很簡單，因此實際上需要分析的是下面兩種情況：

　　（r1 r2w2）： 當已經有寫線程執行的時候，同時來了一個讀線程（w2）和一個寫線程(r2)，那么是讀線程w2先執行還是寫線程r2先執行

　　（w1 r2w2）：當已經有讀線程執行的時候，同時來了一個讀線程和一個寫線程，那么是讀線程w2先執行還是寫線程r2先執行

 

　　后面的分析都會結合這兩種情況。

　　

Linux下信號量API

　　Linux下提供了信號量相關的API，定義在/usr/include/semaphore.h, 我們常用的是以下四個函數

int sem_init (sem_t *__sem, int __pshared, unsigned int __value);  
int sem_destroy (sem_t *__sem);
int sem_wait (sem_t *__sem);
int sem_post (sem_t *__sem);

　　這些API都是需要配對使用，如果配合C++，可以使用RAII來保證。sem_init初始化信號量，注意第二個參數，代表了該信號量是否與其他進程共享，第三個參數是共享資源的數量。sem_wait即P操作，嘗試獲取資源；sem_wait即V操作，釋放資源。個人感覺sem_wait這個名字不是很友好，讓人以為會wait，但事實上不一定會wait，所以感覺叫acquire好一些。

　　

讀者優先：

　　由於讀者寫者問題運行並發的讀操作，寫操作必須是互斥的，所以寫出來的程序最直觀的就是讀者優先。源代碼（rw_semaphore_rp.cpp）

#include <stdio.h>
#include <pthread.h>
#include <time.h>
#include <stdlib.h>
#include <unistd.h>
#include <semaphore.h>
#include <time.h>
struct data{
    int read_counter;
    int write_counter;
    int value;
};

data share_data;

int READ_THREAD_NUM = 5;
int WRITE_THREAD_NUM = 3;
int TOTAL_TRY = 10000; // 嘗試這么多次程序就結束吧

bool stop = false;

int reader_num = 0; 
sem_t lock;
sem_t lock_writer;
sem_t w_or_r;


void* func_read(void *ptr){
    while(!stop){
        sem_wait(&lock);
        reader_num += 1;
        share_data.read_counter += 1;
        if(reader_num == 1)
            sem_wait(&w_or_r);
        sem_post(&lock);

        // do sth with share_data
        int tmp = share_data.value;        

        sem_wait(&lock);
        reader_num -= 1;
        if(reader_num == 0)
            sem_post(&w_or_r);
        sem_post(&lock);
        sleep(0.1);
    }
    return NULL;
}

void* func_write(void *ptr){
    int idx = *(int*)ptr;
    while(!stop){
#ifdef LOCK_WRITER
        sem_wait(&lock_writer);
#endif
        sem_wait(&w_or_r);

        share_data.write_counter += 1;
        share_data.value = idx;

        if(share_data.write_counter >= TOTAL_TRY)
            stop = true;

        sem_post(&w_or_r);
 #ifdef LOCK_WRITER
        sem_post(&lock_writer);
#endif


        sleep(0.1);
    }
    return NULL;
}

int main(int argc, char * argv[]){
    share_data.read_counter = 0;
    share_data.write_counter = 0;

    sem_init(&lock, 0, 1);
#ifdef LOCK_WRITER
    sem_init(&lock_writer, 0, 1);
#endif
    sem_init(&w_or_r, 0, 1);

    pthread_t readers[READ_THREAD_NUM];
    pthread_t writers[WRITE_THREAD_NUM];

    for(int i = 0; i < READ_THREAD_NUM; i++){
        pthread_create(&readers[i], NULL, func_read, NULL);
    }

    int thread_args[WRITE_THREAD_NUM];
    for(int i = 0; i < WRITE_THREAD_NUM; i++){
        thread_args[i] = i + 1;
        pthread_create(&writers[i], NULL, func_write, (thread_args + i));
    }

    for(int i = 0; i < READ_THREAD_NUM; i++)
        pthread_join(readers[i],0);

    //for(int i = 0; i < WRITE_THREAD_NUM; i++)
    //   pthread_join(writers[i],0);
    
    sem_destroy(&lock);
#ifdef LOCK_WRITER
    sem_destroy(&lock_writer);
#endif
    sem_destroy(&w_or_r);
    printf("Finally read count %d, write count %d\n", share_data.read_counter, share_data.write_counter);
    return 0;
}

　　首先忽略掉宏定義LOCK_WRITER, 那么使用的就是lock 和 w_or_r這兩個信號量。w_or_r使用來保護對共享資源（讀寫的資源）的訪問，在func_write中對share_data的寫操作之前，sem_wait(w_or_r)即信號量的P操作，操作完成之后sem_post。在讀線程func_read中，用reader_num記錄並發的讀線程數量，對這個變量的操作用lock這個信號量來互斥（在並發的讀線程之間互斥）。從代碼可以看到，只有當第一個讀線程發起讀操作時(reader_num==1）才會去sem_wait(w_or_r)，而且只有當所有的讀線程都結束時(reader_num == 0），才會釋放w_or_r。

　　我們來考慮本章開始提出的問題：

　　（r1 r2w2），r1（讀線程）獲取了管程的執行權，那么w2（寫線程）一定會阻塞，而r2不用對w_or_r執行P操作，因此可以立即執行，那么后面r2w2的執行順序一定是r2 然后是 w2

　　（w1 r2w2），w1獲取了管程的執行權，那么r2和w2都一定會阻塞在w_or_r，前面提到多個線程被阻塞時，V操作喚醒的線程是隨機的，因此r2 w2各有50%的概率先執行

　　用以下命令可編譯並執行代碼 EXE=rw_semaphore && g++  -g -lpthread $EXE.cpp -o $EXE && ./$EXE，下面是執行了5次的結果：

Finally read count 14378, write count 10001
Finally read count 3870, write count 10002
Finally read count 8040, write count 10001
Finally read count 9077, write count 10001
Finally read count 14386, write count 10002

　　

　　接下來考慮宏定義LOCK_WRITER, 在這種情況下有lock_writer對func_write加鎖，所以當寫線程正在寫數據時，后面來的讀線程阻塞在lock_writer，而不是w_or_r。

　　我們來考慮本章開始提出的問題：

　　（r1 r2w2），r1（讀線程）獲取了管程的執行權，那么w2（寫線程）一定會阻塞，而r2不用對w_or_r執行P操作，因此可以立即執行，那么后面r2w2的執行順序一定是r2 然后是 w2

　　（w1 r2w2），w1獲取了管程的執行權，那么r2會阻塞在w_or_r，此時w2阻塞在lock_writer，當w1釋放w_or_r時，一定是r2獲得w_or_r, 那么后面r2w2的執行順序一定是r2 然后是 w2

　　用以下命令可編譯並執行代碼 EXE=rw_semaphore && g++ -DLOCK_WRITER -g -lpthread $EXE.cpp -o $EXE && ./$EXE，下面是執行了5次的結果：

Finally read count 25983, write count 10001
Finally read count 14378, write count 10001
Finally read count 54344, write count 10001
Finally read count 35147, write count 10001
Finally read count 27105, write count 10002

　　可以看到，read操作的數量明顯高於之前的版本。

 

寫者優先：

　　按照第一章節對讀者寫者問題的描述，所謂的寫者優先即如果當前是寫線程獲得了執行權，那么后續並發有讀請求和寫請求到來的時候，我們優先考慮后續的讀請求。為了達到這個目的，我們可以記錄當前正在嘗試寫操作的請求數量，當這個數量大於1時，讀操作需要被阻塞，當所有寫操作都完成，再通知等到的讀操作請求，這個解決思路比較簡單，和讀者優先的解決方法也很類似，下面是具體的代碼(rw_semaphore_wnp.cpp):

#include <stdio.h>
#include <pthread.h>
#include <time.h>
#include <stdlib.h>
#include <unistd.h>
#include <semaphore.h>
#include <time.h>
struct data{
    int read_counter;
    int write_counter;
    int value;
    int value1;
};

data share_data;

int READ_THREAD_NUM = 5;
int WRITE_THREAD_NUM = 3;
int TOTAL_TRY = 10000; // 嘗試這么多次程序就結束吧

bool stop = false;
int reader_num = 0;
int writer_num = 0;
sem_t lock, wlock;
sem_t w_or_r;
sem_t try_read;

void* func_read(void *ptr){
    while(!stop){
        sem_wait(&try_read);

        sem_wait(&lock);

        reader_num += 1;
        share_data.read_counter += 1;
        if(reader_num == 1){
            sem_wait(&w_or_r);
        }

        sem_post(&lock);

        sem_post(&try_read);

        // do sth with share_data
        int tmp = share_data.value;        
        sleep(0);
        if(tmp != share_data.value1){
            printf("Error happens %d %d \n", tmp, share_data.value1);
            stop = true;
        }

        sem_wait(&lock);
        reader_num -= 1;
        if(reader_num == 0){
            sem_post(&w_or_r);
        }
        sem_post(&lock);

        sleep(0.1);
    }
    return NULL;
}

void* func_write(void *ptr){
    int idx = *(int*)ptr;
    while(!stop){
        sem_wait(&wlock);
        writer_num += 1;
        if(writer_num == 1)
            sem_wait(&try_read);
        sem_post(&wlock);        

        sem_wait(&w_or_r);

        share_data.write_counter += 1;
        share_data.value = idx;
        share_data.value1 = idx;

        if(share_data.write_counter >= TOTAL_TRY)
            stop = true;
        sem_post(&w_or_r);

        sem_wait(&wlock);
        writer_num -= 1;
        if(writer_num == 0)
            sem_post(&try_read);
        sem_post(&wlock);
        sleep(0.1);
    }
    return NULL;
}

int main(int argc, char * argv[]){
    share_data.read_counter = 0;
    share_data.write_counter = 0;

    sem_init(&lock, 0, 1);
    sem_init(&wlock, 0, 1);
    sem_init(&w_or_r, 0, 1);
    sem_init(&try_read, 0, 1);

    pthread_t readers[READ_THREAD_NUM];
    pthread_t writers[WRITE_THREAD_NUM];

    for(int i = 0; i < READ_THREAD_NUM; i++){
        pthread_create(&readers[i], NULL, func_read, NULL);
    }

    int thread_args[WRITE_THREAD_NUM];
    for(int i = 0; i < WRITE_THREAD_NUM; i++){
        thread_args[i] = i + 1;
        pthread_create(&writers[i], NULL, func_write, (thread_args + i));
    }

    for(int i = 0; i < READ_THREAD_NUM; i++)
        pthread_join(readers[i],0);

    //for(int i = 0; i < WRITE_THREAD_NUM; i++)
    //   pthread_join(writers[i],0);
    
    sem_destroy(&lock);
    sem_destroy(&wlock);
    sem_destroy(&w_or_r);
    sem_destroy(&try_read);
    printf("Finally read count %d, write count %d\n", share_data.read_counter, share_data.write_counter);
    return 0;
}

 

 　　對比讀者優先的代碼，主要是增加了一個計數器（writer_num）和兩個信號量(wlock, try_read)。writer_num用來記錄排隊的寫線程數目，wlock用來保證writer_num的互斥修改。try_read的作用是保證讀線程每次操作之前都得獲得這個信號量，而在func_write中，只有當所有的讀線程都結束(writer_num == 0)時才會釋放try_write這個信號量，這樣就達到了讀操作優先的目的。

　　同樣我們做一下理論上的分析：

　　（r1 r2w2），r1（讀線程）獲取了管程的執行權，此時W2和r2都會try_lock(分別是第70 和 第30行），當r1釋放try_read時，喚醒r2還是w2是隨機的，因此r2 w2各有50%的概率先執行

　　（w1 r2w2），w1獲取了管程的執行權，那么r2會阻塞在try_read，同事w2阻塞在w_or_r。w1是先釋放w_or_r，后釋放try_read, 所以w2會先獲得執行權， 那么后面r2w2的執行順序一定是w2 然后是r2

　　用以下命令可編譯並執行代碼 EXE=rw_semaphore_wnp && g++ -g -lpthread $EXE.cpp -o $EXE && ./$EXE，下面是執行了5次的結果：

Finally read count 3342, write count 10001
Finally read count 2252, write count 10002
Finally read count 7795, write count 10002
Finally read count 8394, write count 10001
Finally read count 676, write count 10002

 

讀寫鎖解決讀者寫者問題

 　　讀者寫者問題是一個如此普遍的並發問題，因此很多語言為了便於開發者的使用都提供的個高級的抽象，即讀寫鎖。

Linux下讀寫鎖API

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
int pthread_rwlockattr_getkind_np(const pthread_rwlockattr_t *attr, int *pref);

　　前面兩個函數用來初始和銷毀讀寫鎖，注意init函數可以指定屬性參數（attr）。中間有三個函數用於讀寫鎖的lock與unlock，pthread_rwlock_rdlock用來獲取read lock（用於讀數據），pthread_rwlock_wrlock用來獲取write lock（用來寫數據）。最后兩個函數用來設置、判斷是讀者優先，還是寫着優先，在pthread.h源碼枚舉值如下：

enum
{
  PTHREAD_RWLOCK_PREFER_READER_NP,
  PTHREAD_RWLOCK_PREFER_WRITER_NP,
  PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP,
  PTHREAD_RWLOCK_DEFAULT_NP = PTHREAD_RWLOCK_PREFER_READER_NP
};

　　從上面可以看到有三種模式：讀者優先，寫者優先，寫者非遞歸優先。但坑爹的是，如果設置讀者優先（即PTHREAD_RWLOCK_PREFER_WRITER_NP）根本不起作用。如果希望寫者優先，那么要使用PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP，這個問題描述可以參見這篇文章和man page。

實現：

　　基於讀寫鎖rw_lock，解決讀者寫者問題就比較簡單了。先設置好優先選項，然后再寫線程中調用pthread_rwlock_wrlock，在讀線程中調用pthread_rwlock_rdlock。具體的代碼如下(rw_rwlock.cpp)：

#include <stdio.h>
#include <pthread.h>
#include <time.h>
#include <stdlib.h>
#include <unistd.h>
#include <time.h>

struct data{
    int read_counter;
    int write_counter;
    int value;
};

data share_data;

int READ_THREAD_NUM = 5;
int WRITE_THREAD_NUM = 3;
int TOTAL_TRY = 10000; // 嘗試這么多次程序就結束吧

bool stop = false;
pthread_rwlock_t rwlock;
pthread_mutex_t lock;

void* func_read(void *ptr){
    while(!stop){
        pthread_rwlock_rdlock(&rwlock);

        pthread_mutex_lock(&lock);
        share_data.read_counter += 1;
        pthread_mutex_unlock(&lock);
        // do sth about value
        int tmp = share_data.value;

        pthread_rwlock_unlock(&rwlock);
        sleep(0.1);
    }
    return NULL;
}

void* func_write(void *ptr){
    int idx = *(int*)ptr;
    while(!stop){
        pthread_rwlock_wrlock(&rwlock);

        share_data.write_counter += 1;
        share_data.value = idx;
        if(share_data.write_counter >= TOTAL_TRY)
            stop = true;
        pthread_rwlock_unlock(&rwlock);
        sleep(0.1);

    }
    return NULL;
}

int main(int argc, char * argv[]){
    share_data.read_counter = 0;
    share_data.write_counter = 0;

    pthread_rwlockattr_t attr;
    pthread_rwlockattr_init(&attr);
    

#ifdef WRITER_NONRECURSIVE_NP
    pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
#endif

    int np;
    pthread_rwlockattr_getkind_np(&attr, &np);
    printf("readnp: %d, writenp: %d, write_nonrecursivenp: %d, and we are using %d\n", \
            PTHREAD_RWLOCK_PREFER_READER_NP, PTHREAD_RWLOCK_PREFER_WRITER_NP, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP , np);

    pthread_rwlock_init(&rwlock, &attr);

    pthread_mutex_init(&lock, 0);

    pthread_t readers[READ_THREAD_NUM];
    pthread_t writers[WRITE_THREAD_NUM];

    for(int i = 0; i < READ_THREAD_NUM; i++){
        pthread_create(&readers[i], NULL, func_read, NULL);
    }

    int thread_args[WRITE_THREAD_NUM];
    for(int i = 0; i < WRITE_THREAD_NUM; i++){
        thread_args[i] = i + 1;
        pthread_create(&writers[i], NULL, func_write, (thread_args + i));
    }

    for(int i = 0; i < READ_THREAD_NUM; i++)
        pthread_join(readers[i],0);

    //for(int i = 0; i < WRITE_THREAD_NUM; i++)
    //   pthread_join(writers[i],0);
    
    pthread_rwlockattr_destroy(&attr);
    pthread_rwlock_destroy(&rwlock);
    pthread_mutex_destroy(&lock);
     printf("Finally read count %d, write count %d\n", share_data.read_counter, share_data.write_counter);
    return 0;
}

 

　　上面的代碼中，為了統計讀請求執行的次數，使用了互斥鎖(lock)，在編譯的時候通過是否預定於WRITER_NONRECURSIVE_NP這個宏來指定是讀者優先還是寫着優先。

　　首先是讀者優先時得指令 EXE=rw_rwlock && g++ -g -lpthread $EXE.cpp -o $EXE && ./$EXE，輸出結果如下（注意，為了統計 所以在讀線程中加了互斥鎖lock，去掉這個互斥鎖 讀操作的次數會更多）：

Finally read count 9820, write count 10002
Finally read count 58371, write count 10002
Finally read count 2297, write count 10002
Finally read count 23586, write count 10001
Finally read count 4217, write count 10002

　　然后是寫者優先時得指令 EXE=rw_rwlock && g++ -DWRITER_NONRECURSIVE_NP -g -lpthread $EXE.cpp -o $EXE && ./$EXE，輸出結果如下

Finally read count 376, write count 10001
Finally read count 711, write count 10001
Finally read count 928, write count 10001
Finally read count 1468, write count 10001
Finally read count 449, write count 10001

references：

readers writers problem

pthread_rwlockattr_setkind_np.3.html

並發與同步、信號量與管程、生產者消費者問題