圖解執行緒同步 - 條件變數
執行緒同步問題廣泛應用於多種場景下，特別是與網路資料收發等耗時操作有關的場景。執行緒的操作往往比較抽象，執行緒大多執行在程式的後臺，無法直觀的檢視其執行狀態，因此，本文以圖解的形式，為讀者講述執行緒同步的原理，並附以相關例程方便大家除錯。
本文講述一個經典的生產者-消費者執行緒同步模型，用於描述與後臺快取佇列相關執行緒的同步過程。執行緒模型以互斥量(pthread_mutex_t)、條件變數為基礎(pthread_cond_t)，講述互斥量如何與條件變數相互配合，以生產者-消費者的方式共同維護臨界區資源
1、生產者-消費者執行緒模型
本文主要討論posix標準下的生產者-消費者執行緒模型，posix標準多用於類linux相關環境
POSIX: The Portable Operating System Interface (POSIX) is a family of standards specified by the IEEE Computer Society for maintaining compatibility between operating systems. POSIX defines the application programming interface (API), along with command line shells and utility interfaces, for software compatibility with variants of Unix and other operating systems
1.1 執行緒模型工作原理
生產者-消費者(producer-consumer)問題是一個經典的執行緒同步問題，它可以描述為兩個或者多個執行緒共同維護同一個臨界區資源(critical resource)，其中，生成者執行緒負責(例如從網路介面中抽取資料等)向臨界區注入資料，消費者執行緒負責從臨界區抽取資料，並對資料進行處理，。下圖為生產者-消費者執行緒同步模型示意圖。
圖中上面的process_msg為消費者執行緒，下面的enqueue_msg為生產者執行緒，從左到右代表了執行緒執行的時間線。
我們先來看消費者執行緒，消費者執行緒率先獲取互斥鎖物件(圖中的紅點表示互斥鎖物件)，獲得了對臨界區資源的獨佔處理權及cpu資源的優先使用權，然後開始執行自己的執行緒函式。
在消費者的執行緒函式中，首先檢查臨界區資源是否滿足執行條件，若滿足執行條件，則從佇列中取出資料執行自己的邏輯。
如果臨界區資源不滿足執行條件，如佇列為空，此時，消費者執行緒通過在pthread_cond_wait中臨時釋放互斥鎖，並將自己投入休眠狀態，等待被生產者執行緒向臨界區注入資料，將自己喚醒並重新獲得互斥鎖，這時消費者執行緒會阻塞在pthread_cond_wait呼叫中。
消費者執行緒通過在pthread_cond_wait中臨時釋放互斥鎖後，將自己投入休眠狀態，此時生成者執行緒將獲得互斥鎖，並獲得了對臨界區資源的獨佔處理權及cpu資源的優先使用權，然後開始執行自己的執行緒函式。生成者執行緒向臨界區資源注入資料，如向佇列中注入待解碼的資料包，然後，通過pthread_cond_signal喚醒消費者執行緒(圖中虛線所示)，隨即通過unlock_mutex釋放互斥鎖。
在生成者執行緒釋放互斥鎖後，消費者執行緒已被喚醒，並重新獲取互斥鎖，再次檢查臨界區資源，如果滿足條件，則執行自己的執行緒函式，然後釋放互斥鎖，等待下一次執行。
這裡需要說明一下，在實際情況下，並不總是消費者執行緒優先獲得互斥鎖，這是由cpu排程決定的。
下面給出一些示例程式碼來描述這個過程，例程中臨界區資源的處理非常簡單，只是對一個整型資料做+1處理，在實際的場景中，無論執行緒對臨界區資源的處理有多麼複雜的操作，生產者與消費者執行緒同步的原理與例程中的模型不會有任何差異。這裡我們抽象出一個關於生產者-消費者的執行緒模型，方便大家理解
1.2 例程編譯驗證
下面給出例程的編譯腳步，編譯過程非常簡單，在原始碼及Makefile編譯指令碼目錄執行[make]命令，即可完成例程編譯，執行下面的操作即可開始例程的執行。
Makefile編譯指令碼
test: main.c
gcc -o test -g3 main.c -l pthread
clean:
rm test
程式碼編譯執行
make
./test
程式碼執行結果
./test
consumer_before_lock--------------0
consumer_lock---------------------0
consumer_wait---------------------0
producer_before_lock++++++++++++++++++++++++++++++++0
producer_lock+++++++++++++++++++++++++++++++++++++++0
producer_process_critical_resource++++++++++++++++++0
producer_unlock+++++++++++++++++++++++++++++++++++++1
consumer_revive-------------------1
consumer_wait---------------------1
producer_before_lock++++++++++++++++++++++++++++++++1
producer_lock+++++++++++++++++++++++++++++++++++++++1
producer_process_critical_resource++++++++++++++++++1
producer_unlock+++++++++++++++++++++++++++++++++++++2
consumer_revive-------------------2
consumer_wait---------------------2
producer_before_lock++++++++++++++++++++++++++++++++2
producer_lock+++++++++++++++++++++++++++++++++++++++2
producer_process_critical_resource++++++++++++++++++2
producer_unlock+++++++++++++++++++++++++++++++++++++3
consumer_revive-------------------3
consumer_wait---------------------3
producer_before_lock++++++++++++++++++++++++++++++++3
producer_lock+++++++++++++++++++++++++++++++++++++++3
producer_process_critical_resource++++++++++++++++++3
producer_unlock+++++++++++++++++++++++++++++++++++++4
consumer_revive-------------------4
consumer_wait---------------------4
producer_before_lock++++++++++++++++++++++++++++++++4
producer_lock+++++++++++++++++++++++++++++++++++++++4
producer_process_critical_resource++++++++++++++++++4
producer_unlock+++++++++++++++++++++++++++++++++++++5
consumer_revive-------------------5
consumer_wait---------------------5
producer_before_lock++++++++++++++++++++++++++++++++5
producer_lock+++++++++++++++++++++++++++++++++++++++5
producer_process_critical_resource++++++++++++++++++5
producer_unlock+++++++++++++++++++++++++++++++++++++6
consumer_revive-------------------6
consumer_wait---------------------6
producer_before_lock++++++++++++++++++++++++++++++++6
producer_lock+++++++++++++++++++++++++++++++++++++++6
producer_process_critical_resource++++++++++++++++++6
producer_unlock+++++++++++++++++++++++++++++++++++++7
consumer_revive-------------------7
consumer_wait---------------------7
producer_before_lock++++++++++++++++++++++++++++++++7
producer_lock+++++++++++++++++++++++++++++++++++++++7
producer_process_critical_resource++++++++++++++++++7
producer_unlock+++++++++++++++++++++++++++++++++++++8
consumer_revive-------------------8
consumer_wait---------------------8
producer_before_lock++++++++++++++++++++++++++++++++8
producer_lock+++++++++++++++++++++++++++++++++++++++8
producer_process_critical_resource++++++++++++++++++8
producer_unlock+++++++++++++++++++++++++++++++++++++9
consumer_revive-------------------9
consumer_wait---------------------9
producer_before_lock++++++++++++++++++++++++++++++++9
producer_lock+++++++++++++++++++++++++++++++++++++++9
producer_process_critical_resource++++++++++++++++++9
producer_unlock+++++++++++++++++++++++++++++++++++++10
consumer_revive-------------------10
consumer_process_critical_resource------------------10
consumer_unlock-----------------------0
producer_before_lock++++++++++++++++++++++++++++++++0
producer_lock+++++++++++++++++++++++++++++++++++++++0
producer_process_critical_resource++++++++++++++++++0
producer_unlock+++++++++++++++++++++++++++++++++++++1
consumer_before_lock--------------0
consumer_lock---------------------1
consumer_wait---------------------1
producer_before_lock++++++++++++++++++++++++++++++++1
producer_lock+++++++++++++++++++++++++++++++++++++++1
producer_process_critical_resource++++++++++++++++++1
producer_unlock+++++++++++++++++++++++++++++++++++++2
consumer_revive-------------------2
consumer_wait---------------------2
producer_before_lock++++++++++++++++++++++++++++++++2
producer_lock+++++++++++++++++++++++++++++++++++++++2
producer_process_critical_resource++++++++++++++++++2
producer_unlock+++++++++++++++++++++++++++++++++++++3
consumer_revive-------------------3
consumer_wait---------------------3
可以看到，消費者執行緒不斷的被喚醒，在臨界區資源不滿足執行條件的情況下，又阻塞到pthread_cond_wait並進入睡眠狀態，重新等待生產者pthread_cond_signal的喚醒
更多執行緒同步的例程見[公眾號：斷點實驗室]的其他文章
1.3 例程原始碼清單
#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>
static int val=0;//臨界區資源，critical resource
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;//靜態初始化互斥鎖
pthread_cond_t ready=PTHREAD_COND_INITIALIZER;//靜態初始化條件變數
//生產者臨界區資源處理函式
void producer_proc(int *cr){
printf("producer_process_critical_resource++++++++++++++++++%d\n\n",val);
*cr+=1;//處理臨界區資源，做+1簡單處理
}
//消費者臨界區資源處理函式
void consumer_proc(int *cr){
printf("consumer_process_critical_resource------------------%d\n\n",val);
*cr=0;//處理臨界區資源，這裡直接重置為0
}
//生產者執行緒函式
void *producer_fun(void *p) {
while(1){
printf("producer_before_lock++++++++++++++++++++++++++++++++%d\n\n",val);
pthread_mutex_lock(&lock);//取得互斥鎖
printf("producer_lock+++++++++++++++++++++++++++++++++++++++%d\n\n",val);
//處理臨界區資源，做+1簡單處理
producer_proc(&val);
pthread_cond_signal(&ready);//喚醒等待執行緒
printf("producer_unlock+++++++++++++++++++++++++++++++++++++%d\n\n",val);
pthread_mutex_unlock(&lock);//釋放互斥鎖
sleep(1);
}//end for while
return NULL;
}
//消費者執行緒函式
void *consumer_fun(void *p) {
while(1){
printf("consumer_before_lock--------------%d\n\n",val);
pthread_mutex_lock(&lock);//取得互斥鎖
printf("consumer_lock---------------------%d\n\n",val);
while(val<10) {//檢查臨界區資源是否滿足執行條件
printf("consumer_wait---------------------%d\n\n",val);
//暫時釋放互斥鎖，休眠等待生產者執行緒傳送就緒訊號ready，該函式返回時互斥鎖再次被鎖住
pthread_cond_wait(&ready,&lock);
printf("consumer_revive-------------------%d\n\n",val);
}
//處理臨界區資源，這裡直接重置為0
consumer_proc(&val);
printf("consumer_unlock-----------------------%d\n\n",val);
pthread_mutex_unlock(&lock);
sleep(1);
}//end for while
return NULL;
}
int main(int argc,char *argv[]) {
pthread_t cid,pid;//執行緒id
//建立消費者執行緒
int ret=pthread_create(&cid,NULL,consumer_fun,NULL);
if(ret!=0) {//檢查執行緒建立結果
printf("create consumer thread failed\n");
exit(1);
}
//建立生成者執行緒
ret=pthread_create(&pid,NULL,producer_fun,NULL);
if(ret!=0) {//檢查執行緒建立結果
printf("create producer thread failed\n");
exit(1);
}
//等待執行緒結束
pthread_join(cid,NULL);
pthread_join(pid,NULL);
return 0;
}
// 著作權歸作者所有。商業轉載請聯絡作者獲得授權，非商業轉載請註明出處。
// 公眾號：斷點實驗室
// 掃描二維碼，關注更多優質原創，內容包括：音視訊開發、影象處理、網路、
// Linux，Windows、Android、嵌入式開發等