在 上一篇文章 結束時，我描述了一個比較特殊的難題：如果線程正在等待某個特定條件發生，它應該如何處理這種情況？它可以重復對互斥對象鎖定和解鎖，每次都會檢查共享數據結構，以查找某個值。但這是在浪費時間和資源，而且這種繁忙查詢的效率非常低。解決這個問題的最佳方法是使用 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() 調用，現在來繼續研究更簡單的東西 -- 初始化和真正的發送信號和廣播進程。到那時，我們將會對包含了多線程工作隊列的 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
** 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
** 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;
}
|
我編寫的并不是線程安全的隊列例程，事實上我創建了一個“數據包裝”或“控制”結構，它可以是任何線程支持的數據結構?？匆幌?control.h：
#include
typedef struct data_control {
pthread_mutex_t mutex;
pthread_cond_t cond;
int active;
} data_control;
|
現在您看到了 data_control 結構定義，以下是它的視覺表示：
圖像中的鎖代表互斥對象，它允許對數據結構進行互斥訪問。黃色的星代表條件變量，它可以睡眠，直到所討論的數據結構改變為止。on/off 開關表示整數 "active"，它告訴線程此數據是否是活動的。在代碼中，我使用整數 active 作為標志，告訴工作隊列何時應該關閉。以下是 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：
#define dabort() /
{ printf("Aborting at line %d in source file %s/n",__LINE__,__FILE__); abort(); }
|
此代碼用于處理工作組代碼中的不可糾正錯誤。
說到工作組代碼，以下就是：
#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()，這是所有工作程序線程都要執行的代碼。當工作程序線程啟動時，它會立即鎖定工作隊列互斥對象，獲取一個工作節點（如果有的話），然后對它進行處理。如果沒有工作，則調用 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() 會一直執行，等待清除隊列中新的清除節點。如果有新節點，我們會將此節點移出隊列、對清除隊列解鎖（從而使工作程序可以添加清除節點）、聯接新的工作程序線程（使用存儲在清除節點中的 TID）、釋放清除節點、減少“現有”線程的數量，然后繼續。
現在已經到了“POSIX 線程詳解”系列的尾聲，希望您已經準備好開始將多線程代碼添加到您自己的應用程序中。有關詳細信息，請參閱 參考資料 部分，這部分內容還包含了本文中使用的所有源碼的 tar 文件。下一個系列中再見！
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
