通用線程：POSIX 線程詳解，第 3 部分

條件變量詳解

在 ​​上一篇文章​​結束時，我描述了一個比較特殊的難題：如果線程正在等待某個特定條件發生，它應該如何處理這種情況？它可以重複對互斥對象鎖定和解鎖，每次都會檢查共享資料結構，以查找某個值。但這是在浪費時間和資源，而且這種繁忙查詢的效率非常低。解決這個問題的最佳方法是使用 pthread_cond_wait() 調用來等待特殊條件發生。

了解 pthread_cond_wait() 的作用非常重要 -- 它是 POSIX 線程信号發送系統的核心，也是最難以了解的部分。

首先，讓我們考慮以下情況：線程為檢視已連結清單而鎖定了互斥對象，然而該清單恰巧是空的。這一特定線程什麼也幹不了 -- 其設計意圖是從清單中除去節點，但是現在卻沒有節點。是以，它隻能：

鎖定互斥對象時，線程将調用 pthread_cond_wait(&mycond,&mymutex)。pthread_cond_wait() 調用相當複雜，是以我們每次隻執行它的一個操作。

pthread_cond_wait() 所做的第一件事就是同時對互斥對象解鎖（于是其它線程可以修改已連結清單），并等待條件 mycond 發生（這樣當 pthread_cond_wait() 接收到另一個線程的“信号”時，它将蘇醒）。現在互斥對象已被解鎖，其它線程可以通路和修改已連結清單，可能還會添加項。

此時，pthread_cond_wait() 調用還未傳回。對互斥對象解鎖會立即發生，但等待條件 mycond 通常是一個阻塞操作，這意味着線程将睡眠，在它蘇醒之前不會消耗 CPU 周期。這正是我們期待發生的情況。線程将一直睡眠，直到特定條件發生，在這期間不會發生任何浪費 CPU 時間的繁忙查詢。從線程的角度來看，它隻是在等待 pthread_cond_wait() 調用傳回。

現在繼續說明，假設另一個線程（稱作“2 号線程”）鎖定了 mymutex 并對已連結清單添加了一項。在對互斥對象解鎖之後，2 号線程會立即調用函數 pthread_cond_broadcast(&mycond)。此操作之後，2 号線程将使所有等待 mycond 條件變量的線程立即蘇醒。這意味着第一個線程（仍處于 pthread_cond_wait() 調用中）現在将蘇醒。

現在，看一下第一個線程發生了什麼。您可能會認為在 2 号線程調用 pthread_cond_broadcast(&mymutex) 之後，1 号線程的 pthread_cond_wait() 會立即傳回。不是那樣！實際上，pthread_cond_wait() 将執行最後一個操作：重新鎖定 mymutex。一旦 pthread_cond_wait() 鎖定了互斥對象，那麼它将傳回并允許 1 号線程繼續執行。那時，它可以馬上檢查清單，檢視它所感興趣的更改。

​​回頁首​​

停止并回顧！

那個過程非常複雜，是以讓我們先來回顧一下。第一個線程首先調用：

pthread_mutex_lock(&mymutex);

然後，它檢查了清單。沒有找到感興趣的東西，于是它調用：

pthread_cond_wait(&mycond, &mymutex);

然後，pthread_cond_wait() 調用在傳回前執行許多操作：

pthread_mutex_unlock(&mymutex);

它對 mymutex 解鎖，然後進入睡眠狀态，等待 mycond 以接收 POSIX 線程“信号”。一旦接收到“信号”（加引号是因為我們并不是在讨論傳統的 UNIX 信号，而是來自 pthread_cond_signal() 或 pthread_cond_broadcast() 調用的信号），它就會蘇醒。但 pthread_cond_wait() 沒有立即傳回 -- 它還要做一件事：重新鎖定 mutex：

pthread_mutex_lock(&mymutex);

pthread_cond_wait() 知道我們在查找 mymutex “背後”的變化，是以它繼續操作，為我們鎖定互斥對象，然後才傳回。

​​回頁首​​

pthread_cond_wait() 小測驗

現在已回顧了 pthread_cond_wait() 調用，您應該了解了它的工作方式。應該能夠叙述 pthread_cond_wait() 依次執行的所有操作。嘗試一下。如果了解了 pthread_cond_wait()，其餘部分就相當容易，是以請重新閱讀以上部分，直到記住為止。好，讀完之後，能否告訴我在調用 pthread_cond_wait() 之 前，互斥對象必須處于什麼狀态？pthread_cond_wait() 調用傳回之後，互斥對象處于什麼狀态？這兩個問題的答案都是“鎖定”。既然已經完全了解了 pthread_cond_wait() 調用，現在來繼續研究更簡單的東西 -- 初始化和真正的發送信号和廣播程序。到那時，我們将會對包含了多線程工作隊列的 C 代碼了如指掌。

​​回頁首​​

初始化和清除

條件變量是一個需要初始化的真實資料結構。以下就初始化的方法。首先，定義或配置設定一個條件變量，如下所示：

pthread_cond_t mycond;

然後，調用以下函數進行初始化：

pthread_cond_init(&mycond,NULL);

瞧，初始化完成了！在釋放或廢棄條件變量之前，需要毀壞它，如下所示：

pthread_cond_destroy(&mycond);

很簡單吧。接着讨論 pthread_cond_wait() 調用。

​​回頁首​​

等待

一旦初始化了互斥對象和條件變量，就可以等待某個條件，如下所示：

pthread_cond_wait(&mycond, &mymutex);

請注意，代碼在邏輯上應該包含 mycond 和 mymutex。一個特定條件隻能有一個互斥對象，而且條件變量應該表示互斥資料“内部”的一種特殊的條件更改。一個互斥對象可以用許多條件變量（例如，cond_empty、cond_full、cond_cleanup），但每個條件變量隻能有一個互斥對象。

​​回頁首​​

發送信号和廣播

對于發送信号和廣播，需要注意一點。如果線程更改某些共享資料，而且它想要喚醒所有正在等待的線程，則應使用 pthread_cond_broadcast 調用，如下所示：

pthread_cond_broadcast(&mycond);

在某些情況下，活動線程隻需要喚醒第一個正在睡眠的線程。假設您隻對隊列添加了一個工作作業。那麼隻需要喚醒一個工作程式線程（再喚醒其它線程是不禮貌的！）：

pthread_cond_signal(&mycond);

此函數隻喚醒一個線程。如果 POSIX 線程标準允許指定一個整數，可以讓您喚醒一定數量的正在睡眠的線程，那就更完美了。但是很可惜，我沒有被邀請參加會議。

​​回頁首​​

工作組

我将示範如何建立多線程工作組。在這個方案中，我們建立了許多工作程式線程。每個線程都會檢查 wq（“工作隊列”），檢視是否有需要完成的工作。如果有需要完成的工作，那麼線程将從隊列中除去一個節點，執行這些特定工作，然後等待新的工作到達。

與此同時，主線程負責建立這些工作程式線程、将工作添加到隊列，然後在它退出時收集所有工作程式線程。您将會遇到許多 C 代碼，好好準備吧！

​​回頁首​​

隊列

需要隊列是出于兩個原因。首先，需要隊列來儲存工作作業。還需要可用于跟蹤已終止線程的資料結構。還記得前幾篇文章（請參閱本文結尾處的 ​​參考資料​​）中，我曾提到過需要使用帶有特定程序辨別的 pthread_join 嗎？使用“清除隊列”（稱作 "cq"）可以解決無法等待 任何已終止線程的問題（稍後将詳細讨論這個問題）。以下是标準隊列代碼。将此代碼儲存到檔案 queue.h 和 queue.c：

queue.h

/* queue.h ** Copyright 2000 Daniel Robbins, Gentoo Technologies, Inc. ** Author: Daniel Robbins ** Date: 16 Jun 2000 */ typedef struct node { struct node *next; } node; typedef struct queue { node *head, *tail; } queue; void queue_init(queue *myroot); void queue_put(queue *myroot, node *mynode); node *queue_get(queue *myroot);

queue.c

/* queue.c ** Copyright 2000 Daniel Robbins, Gentoo Technologies, Inc. ** Author: Daniel Robbins ** Date: 16 Jun 2000 ** ** This set of queue functions was originally thread-aware. I ** redesigned the code to make this set of queue routines ** thread-ignorant (just a generic, boring yet very fast set of queue ** routines). Why the change? Because it makes more sense to have ** the thread support as an optional add-on. Consider a situation ** where you want to add 5 nodes to the queue. With the ** thread-enabled version, each call to queue_put() would ** automatically lock and unlock the queue mutex 5 times -- that's a ** lot of unnecessary overhead. However, by moving the thread stuff ** out of the queue routines, the caller can lock the mutex once at ** the beginning, then insert 5 items, and then unlock at the end. ** Moving the lock/unlock code out of the queue functions allows for ** optimizations that aren't possible otherwise. It also makes this ** code useful for non-threaded applications. ** ** We can easily thread-enable this data structure by using the ** data_control type defined in control.c and control.h. */ #include <stdio.h> #include "queue.h" void queue_init(queue *myroot) { myroot->head=NULL; myroot->tail=NULL; } void queue_put(queue *myroot,node *mynode) { mynode->next=NULL; if (myroot->tail!=NULL) myroot->tail->next=mynode; myroot->tail=mynode; if (myroot->:head==NULL) myroot->head=mynode; } node *queue_get(queue *myroot) { //get from root node *mynode; mynode=myroot->head; if (myroot->head!=NULL) myroot->head=myroot->head->next; return mynode; }

​​回頁首​​

data_control 代碼

我編寫的并不是線程安全的隊列例程，事實上我建立了一個“資料包裝”或“控制”結構，它可以是任何線程支援的資料結構。看一下 control.h：

control.h

#include typedef struct data_control { pthread_mutex_t mutex; pthread_cond_t cond; int active; } data_con

現在您看到了 data_control 結構定義，以下是它的視覺表示：

所使用的 data_control 結構

圖像中的鎖代表互斥對象，它允許對資料結構進行互斥通路。黃色的星代表條件變量，它可以睡眠，直到所讨論的資料結構改變為止。on/off 開關表示整數 "active"，它告訴線程此資料是否是活動的。在代碼中，我使用整數 active 作為标志，告訴工作隊列何時應該關閉。以下是 control.c：

control.c

/* control.c ** Copyright 2000 Daniel Robbins, Gentoo Technologies, Inc. ** Author: Daniel Robbins ** Date: 16 Jun 2000 ** ** These routines provide an easy way to make any type of ** data-structure thread-aware. Simply associate a data_control ** structure with the data structure (by creating a new struct, for ** example). Then, simply lock and unlock the mutex, or ** wait/signal/broadcast on the condition variable in the data_control ** structure as needed. ** ** data_control structs contain an int called "active". This int is ** intended to be used for a specific kind of multithreaded design, ** where each thread checks the state of "active" every time it locks ** the mutex. If active is 0, the thread knows that instead of doing ** its normal routine, it should stop itself. If active is 1, it ** should continue as normal. So, by setting active to 0, a ** controlling thread can easily inform a thread work crew to shut ** down instead of processing new jobs. Use the control_activate() ** and control_deactivate() functions, which will also broadcast on ** the data_control struct's condition variable, so that all threads ** stuck in pthread_cond_wait() will wake up, have an opportunity to ** notice the change, and then terminate. */ #include "control.h" int control_init(data_control *mycontrol) { int mystatus; if (pthread_mutex_init(&(mycontrol->mutex),NULL)) return 1; if (pthread_cond_init(&(mycontrol->cond),NULL)) return 1; mycontrol->active=0; return 0; } int control_destroy(data_control *mycontrol) { int mystatus; if (pthread_cond_destroy(&(mycontrol->cond))) return 1; if (pthread_cond_destroy(&(mycontrol->cond))) return 1; mycontrol->active=0; return 0; } int control_activate(data_control *mycontrol) { int mystatus; if (pthread_mutex_lock(&(mycontrol->mutex))) return 0; mycontrol->active=1; pthread_mutex_unlock(&(mycontrol->mutex)); pthread_cond_broadcast(&(mycontrol->cond)); return 1; } int control_deactivate(data_control *mycontrol) { int mystatus; if (pthread_mutex_lock(&(mycontrol->mutex))) return 0; mycontrol->active=0; pthread_mutex_unlock(&(mycontrol->mutex)); pthread_cond_broadcast(&(mycontrol->cond)); return 1; }

​​回頁首​​

調試時間

在開始調試之前，還需要一個檔案。以下是 dbug.h：

dbug.h

#define dabort() / { printf("Aborting at line %d in source file %s/n",__LINE__,__FILE__); abort(); }

此代碼用于處理工作組代碼中的不可糾正錯誤。

​​回頁首​​

工作組代碼

說到工作組代碼，以下就是：

workcrew.c

#include <stdio.h> #include <stdlib.h> #include "control.h" #include "queue.h" #include "dbug.h" /* the work_queue holds tasks for the various threads to complete. */ struct work_queue { data_control control; queue work; } wq; /* I added a job number to the work node. Normally, the work node would contain additional data that needed to be processed. */ typedef struct work_node { struct node *next; int jobnum; } wnode; /* the cleanup queue holds stopped threads. Before a thread terminates, it adds itself to this list. Since the main thread is waiting for changes in this list, it will then wake up and clean up the newly terminated thread. */ struct cleanup_queue { data_control control; queue cleanup; } cq; /* I added a thread number (for debugging/instructional purposes) and a thread id to the cleanup node. The cleanup node gets passed to the new thread on startup, and just before the thread stops, it attaches the cleanup node to the cleanup queue. The main thread monitors the cleanup queue and is the one that performs the necessary cleanup. */ typedef struct cleanup_node { struct node *next; int threadnum; pthread_t tid; } cnode; void *threadfunc(void *myarg) { wnode *mywork; cnode *mynode; mynode=(cnode *) myarg; pthread_mutex_lock(&wq.control.mutex); while (wq.control.active) { while (wq.work.head==NULL && wq.control.active) { pthread_cond_wait(&wq.control.cond, &wq.control.mutex); } if (!wq.control.active) break; //we got something! mywork=(wnode *) queue_get(&wq.work); pthread_mutex_unlock(&wq.control.mutex); //perform processing... printf("Thread number %d processing job %d/n",mynode->threadnum,mywork->jobnum); free(mywork); pthread_mutex_lock(&wq.control.mutex); } pthread_mutex_unlock(&wq.control.mutex); pthread_mutex_lock(&cq.control.mutex); queue_put(&cq.cleanup,(node *) mynode); pthread_mutex_unlock(&cq.control.mutex); pthread_cond_signal(&cq.control.cond); printf("thread %d shutting down.../n",mynode->threadnum); return NULL; } #define NUM_WORKERS 4 int numthreads; void join_threads(void) { cnode *curnode; printf("joining threads.../n"); while (numthreads) { pthread_mutex_lock(&cq.control.mutex); /* below, we sleep until there really is a new cleanup node. This takes care of any false wakeups... even if we break out of pthread_cond_wait(), we don't make any assumptions that the condition we were waiting for is true. */ while (cq.cleanup.head==NULL) { pthread_cond_wait(&cq.control.cond,&cq.control.mutex); } /* at this point, we hold the mutex and there is an item in the list that we need to process. First, we remove the node from the queue. Then, we call pthread_join() on the tid stored in the node. When pthread_join() returns, we have cleaned up after a thread. Only then do we free() the node, decrement the number of additional threads we need to wait for and repeat the entire process, if necessary */ curnode = (cnode *) queue_get(&cq.cleanup); pthread_mutex_unlock(&cq.control.mutex); pthread_join(curnode->tid,NULL); printf("joined with thread %d/n",curnode->threadnum); free(curnode); numthreads--; } } int create_threads(void) { int x; cnode *curnode; for (x=0; x<NUM_WORKERS; x++) { curnode=malloc(sizeof(cnode)); if (!curnode) return 1; curnode->threadnum=x; if (pthread_create(&curnode->tid, NULL, threadfunc, (void *) curnode)) return 1; printf("created thread %d/n",x); numthreads++; } return 0; } void initialize_structs(void) { numthreads=0; if (control_init(&wq.control)) dabort(); queue_init(&wq.work); if (control_init(&cq.control)) { control_destroy(&wq.control); dabort(); } queue_init(&wq.work); control_activate(&wq.control); } void cleanup_structs(void) { control_destroy(&cq.control); control_destroy(&wq.control); } int main(void) { int x; wnode *mywork; initialize_structs(); /* CREATION */ if (create_threads()) { printf("Error starting threads... cleaning up./n"); join_threads(); dabort(); } pthread_mutex_lock(&wq.control.mutex); for (x=0; x<16000; x++) { mywork=malloc(sizeof(wnode)); if (!mywork) { printf("ouch! can't malloc!/n"); break; } mywork->jobnum=x; queue_put(&wq.work,(node *) mywork); } pthread_mutex_unlock(&wq.control.mutex); pthread_cond_broadcast(&wq.control.cond); printf("sleeping.../n"); sleep(2); printf("deactivating work queue.../n"); control_deactivate(&wq.control); /* CLEANUP */ join_threads(); cleanup_structs(); }

​​回頁首​​

代碼初排

現在來快速初排代碼。定義的第一個結構稱作 "wq"，它包含了 data_control 和隊列頭。data_control 結構用于仲裁對整個隊列的通路，包括隊列中的節點。下一步工作是定義實際的工作節點。要使代碼符合本文中的示例，此處所包含的都是作業号。

接着，建立清除隊列。注釋說明了它的工作方式。好，現在讓我們跳過 threadfunc()、join_threads()、create_threads() 和 initialize_structs() 調用，直接跳到 main()。所做的第一件事就是初始化結構 -- 這包括初始化 data_controls 和隊列，以及激活工作隊列。

​​回頁首​​

有關清除的注意事項

現在初始化線程。如果看一下 create_threads() 調用，似乎一切正常 -- 除了一件事。請注意，我們正在配置設定清除節點，以及初始化它的線程号和 TID 元件。我們還将清除節點作為初始自變量傳遞給每一個新的工作程式線程。為什麼這樣做？

因為當某個工作程式線程退出時，它會将其清除節點連接配接到清除隊列，然後終止。那時，主線程會在清除隊列中檢測到這個節點（利用條件變量），并将這個節點移出隊列。因為 TID（線程辨別）存儲在清除節點中，是以主線程可以确切知道哪個線程已終止了。然後，主線程将調用 pthread_join(tid)，并聯接适當的工作程式線程。如果沒有做記錄，那麼主線程就需要按任意順序聯接工作程式線程，可能是按它們的建立順序。由于線程不一定按此順序終止，那麼主線程可能會在已經聯接了十個線程時，等待聯接另一個線程。您能了解這種設計決策是如何使關閉代碼加速的嗎（尤其在使用幾百個工作程式線程的情況下）？

​​回頁首​​

建立工作

我們已啟動了工作程式線程（它們已經完成了執行 threadfunc()，稍後将讨論此函數），現在主線程開始将工作節點插入工作隊列。首先，它鎖定 wq 的控制互斥對象，然後配置設定 16000 個工作包，将它們逐個插入隊列。完成之後，将調用 pthread_cond_broadcast()，于是所有正在睡眠的線程會被喚醒，并開始執行工作。此時，主線程将睡眠兩秒鐘，然後釋放工作隊列，并通知工作程式線程終止活動。接着，主線程會調用 join_threads() 函數來清除所有工作程式線程。

​​回頁首​​

threadfunc()

現在來讨論 threadfunc()，這是所有工作程式線程都要執行的代碼。當工作程式線程啟動時，它會立即鎖定工作隊列互斥對象，擷取一個工作節點（如果有的話），然後對它進行處理。如果沒有工作，則調用 pthread_cond_wait()。您會注意到這個調用在一個非常緊湊的 while() 循環中，這是非常重要的。當從 pthread_cond_wait() 調用中蘇醒時，決不能認為條件肯定發生了 -- 它 可能發生了，也可能沒有發生。如果發生了這種情況，即錯誤地喚醒了線程，而清單是空的，那麼 while 循環将再次調用 pthread_cond_wait()。

如果有一個工作節點，那麼我們隻列印它的作業号，釋放它并退出。然而，實際代碼會執行一些更實質性的操作。在 while() 循環結尾，我們鎖定了互斥對象，以便檢查 active 變量，以及在循環頂部檢查新的工作節點。如果執行完此代碼，就會發現如果 wq.control.active 是 0，while 循環就會終止，并會執行 threadfunc() 結尾處的清除代碼。

工作程式線程的清除代碼部件非常有趣。首先，由于 pthread_cond_wait() 傳回了鎖定的互斥對象，它會對 work_queue 解鎖。然後，它鎖定清除隊列，添加清除代碼（包含了 TID，主線程将使用此 TID 來調用 pthread_join()），然後再對清除隊列解鎖。此後，它發信号給所有 cq 等待者 (pthread_cond_signal(&cq.control.cond))，于是主線程就知道有一個待處理的新節點。我們不使用 pthread_cond_broadcast()，因為沒有這個必要 -- 隻有一個線程（主線程）在等待清除隊列中的新節點。當它調用 join_threads() 時，工作程式線程将列印關閉消息，然後終止，等待主線程發出的 pthread_join() 調用。

​​回頁首​​

join_threads()

如果要檢視關于如何使用條件變量的簡單示例，請參考 join_threads() 函數。如果還有工作程式線程，join_threads() 會一直執行，等待清除隊列中新的清除節點。如果有新節點，我們會将此節點移出隊列、對清除隊列解鎖（進而使工作程式可以添加清除節點）、聯接新的工作程式線程（使用存儲在清除節點中的 TID）、釋放清除節點、減少“現有”線程的數量，然後繼續。