pthread_cond_wait()
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <time.h>
#include <pthread.h>
void* testThreadPool(int *t);
pthread_mutex_t clifd_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t clifd_cond = PTHREAD_COND_INITIALIZER;
int a = 0;
int main() {
int sock_fd, conn_fd;
int optval;
socklen_t cli_len;
struct sockaddr_in cli_addr, serv_addr;
sock_fd = socket(AF_INET, SOCK_STREAM, 0);
if (sock_fd < 0) {
?? printf("socket\n");
}
optval = 1;
if (setsockopt(sock_fd, SOL_SOCKET, SO_REUSEADDR, (void *) &optval,
??? sizeof(int)) < 0) {
?? printf("setsockopt\n");
}
memset(&serv_addr, 0, sizeof(struct sockaddr_in));
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(4507);
serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
if (bind(sock_fd, (struct sockaddr *) &serv_addr,
??? sizeof(struct sockaddr_in)) < 0) {
?? printf("bind\n");
}
if (listen(sock_fd, 100) < 0) {
?? printf("listen\n");
}
cli_len = sizeof(struct sockaddr_in);
int t;
pthread_t * mythread;
mythread = (pthread_t*) malloc(100 * sizeof(pthread_t));
for (t = 0; t < 5; t++) {
?? int *i=(int*)malloc(sizeof(int));
?? *i=t;
?? if (pthread_create(&mythread[t], NULL, (void*)testThreadPool, (void*)i) != 0) {
??? printf("pthread_create");
?? }
}
while (1) {
?? conn_fd = accept(sock_fd, (struct sockaddr *) &cli_addr, &cli_len);
?? if (conn_fd < 0) {
??? printf("accept\n");
?? }
?? printf("accept a new client, ip:%s\n", inet_ntoa(cli_addr.sin_addr));
?? pthread_mutex_lock(&clifd_mutex);
?? a=conn_fd;
?? pthread_cond_signal(&clifd_cond);
?? pthread_mutex_unlock(&clifd_mutex);
}
return 0;
}
void* testThreadPool(int *t) {
printf("t is %d\n", *t);
for (;;) {
?? pthread_mutex_lock(&clifd_mutex);
?? pthread_cond_wait(&clifd_cond, &clifd_mutex);
?? printf("a is %d\n", a);
?? printf("t is %d\n", *t);
?? pthread_mutex_unlock(&clifd_mutex);
?? sleep(100);
}
return (void*) 0;
}
了解 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 “背后”的變化，因此它繼續操作，為我們鎖定互斥對象，然后才返回。
1.Linux“線程”
?????進程與線程之間是有區別的，不過Linux內核只提供了輕量進程的支持，未實現線程模型。Linux是一種“多進程單線程”的操作系統。Linux本身只有進程的概念，而其所謂的“線程”本質上在內核里仍然是進程。
?????大家知道，進程是資源分配的單位，同一進程中的多個線程共享該進程的資源（如作為共享內存的全局變量）。Linux中所謂的“線程”只是在被創建時clone了父進程的資源，因此clone出來的進程表現為“線程”，這一點一定要弄清楚。因此，Linux“線程”這個概念只有在打冒號的情況下才是最準確的。
?????目前Linux中最流行的線程機制為LinuxThreads，所采用的就是線程－進程“一對一”模型，調度交給核心，而在用戶級實現一個包括信號處理在內的線程管理機制。LinuxThreads由Xavier Leroy (Xavier.Leroy@inria.fr)負責開發完成，并已綁定在GLIBC中發行，它實現了一種BiCapitalized面向Linux的Posix 1003.1c “pthread”標準接口。Linuxthread可以支持Intel、Alpha、MIPS等平臺上的多處理器系統。
按照POSIX 1003.1c 標準編寫的程序與Linuxthread 庫相鏈接即可支持Linux平臺上的多線程，在程序中需包含頭文件pthread. h，在編譯鏈接時使用命令：
| gcc -D -REENTRANT -lpthread xxx. c |
其中-REENTRANT宏使得相關庫函數(如stdio.h、errno.h中函數) 是可重入的、線程安全的(thread-safe)，-lpthread則意味著鏈接庫目錄下的libpthread.a或libpthread.so文件。使用Linuxthread庫需要2.0以上版本的Linux內核及相應版本的C庫(libc 5.2.18、libc 5.4.12、libc 6)。
?????2.“線程”控制
線程創建
進程被創建時，系統會為其創建一個主線程，而要在進程中創建新的線程，則可以調用pthread_create：
| pthread_create(pthread_t *thread, const pthread_attr_t *attr, void * (start_routine)(void*), void *arg); |
start_routine為新線程的入口函數，arg為傳遞給start_routine的參數。
每個線程都有自己的線程ID，以便在進程內區分。線程ID在pthread_create調用時回返給創建線程的調用者；一個線程也可以在創建后使用pthread_self()調用獲取自己的線程ID：
| pthread_self (void) ; |
線程退出
線程的退出方式有三：
（1）執行完成后隱式退出；
（2）由線程本身顯示調用pthread_exit 函數退出；
| pthread_exit (void * retval) ; |
（3）被其他線程用pthread_cance函數終止：
| pthread_cance (pthread_t thread) ; |
在某線程中調用此函數，可以終止由參數thread 指定的線程。
如果一個線程要等待另一個線程的終止，可以使用pthread_join函數，該函數的作用是調用pthread_join的線程將被掛起直到線程ID為參數thread的線程終止：
| pthread_join (pthread_t thread, void** threadreturn); |
3.線程通信
線程互斥
互斥意味著“排它”，即兩個線程不能同時進入被互斥保護的代碼。Linux下可以通過pthread_mutex_t 定義互斥體機制完成多線程的互斥操作，該機制的作用是對某個需要互斥的部分，在進入時先得到互斥體，如果沒有得到互斥體，表明互斥部分被其它線程擁有，此時欲獲取互斥體的線程阻塞，直到擁有該互斥體的線程完成互斥部分的操作為止。
下面的代碼實現了對共享全局變量x 用互斥體mutex 進行保護的目的：
| int x; // 進程中的全局變量 pthread_mutex_t mutex; pthread_mutex_init(&mutex, NULL); //按缺省的屬性初始化互斥體變量mutex pthread_mutex_lock(&mutex); // 給互斥體變量加鎖 … //對變量x 的操作 phtread_mutex_unlock(&mutex); // 給互斥體變量解除鎖 |
線程同步
同步就是線程等待某個事件的發生。只有當等待的事件發生線程才繼續執行，否則線程掛起并放棄處理器。當多個線程協作時，相互作用的任務必須在一定的條件下同步。
Linux下的C語言編程有多種線程同步機制，最典型的是條件變量(condition variable)。pthread_cond_init用來創建一個條件變量，其函數原型為：
| pthread_cond_init (pthread_cond_t *cond, const pthread_condattr_t *attr); |
pthread_cond_wait和pthread_cond_timedwait用來等待條件變量被設置，值得注意的是這兩個等待調用需要一個已經上鎖的互斥體mutex，這是為了防止在真正進入等待狀態之前別的線程有可能設置該條件變量而產生競爭。pthread_cond_wait的函數原型為：
| pthread_cond_wait (pthread_cond_t *cond, pthread_mutex_t *mutex); |
pthread_cond_broadcast用于設置條件變量，即使得事件發生，這樣等待該事件的線程將不再阻塞：
| pthread_cond_broadcast (pthread_cond_t *cond) ; |
pthread_cond_signal則用于解除某一個等待線程的阻塞狀態：
| pthread_cond_signal (pthread_cond_t *cond) ; |
pthread_cond_destroy 則用于釋放一個條件變量的資源。
在頭文件semaphore.h 中定義的信號量則完成了互斥體和條件變量的封裝，按照多線程程序設計中訪問控制機制，控制對資源的同步訪問，提供程序設計人員更方便的調用接口。
| sem_init(sem_t *sem, int pshared, unsigned int val); |
這個函數初始化一個信號量sem 的值為val，參數pshared 是共享屬性控制，表明是否在進程間共享。
| sem_wait(sem_t *sem); |
調用該函數時，若sem為無狀態，調用線程阻塞，等待信號量sem值增加(post )成為有信號狀態；若sem為有狀態，調用線程順序執行，但信號量的值減一。
| sem_post(sem_t *sem); |
調用該函數，信號量sem的值增加，可以從無信號狀態變為有信號狀態。
?????
?
下面我們還是以名的生產者/消費者問題為例來闡述Linux線程的控制和通信。一組生產者線程與一組消費者線程通過緩沖區發生聯系。生產者線程將生產的產品送入緩沖區，消費者線程則從中取出產品。緩沖區有N 個，是一個環形的緩沖池。
| #include <stdio.h> #include <pthread.h> #define BUFFER_SIZE 16 // 緩沖區數量 struct prodcons { // 緩沖區相關數據結構 int buffer[BUFFER_SIZE]; /* 實際數據存放的數組*/ pthread_mutex_t lock; /* 互斥體lock 用于對緩沖區的互斥操作 */ int readpos, writepos; /* 讀寫指針*/ pthread_cond_t notempty; /* 緩沖區非空的條件變量 */ pthread_cond_t notfull; /* 緩沖區未滿的條件變量 */ }; /* 初始化緩沖區結構 */ void init(struct prodcons *b) { pthread_mutex_init(&b->lock, NULL); pthread_cond_init(&b->notempty, NULL); pthread_cond_init(&b->notfull, NULL); b->readpos = 0; b->writepos = 0; } /* 將產品放入緩沖區,這里是存入一個整數*/ void put(struct prodcons *b, int data) { pthread_mutex_lock(&b->lock); /* 等待緩沖區未滿*/ if ((b->writepos + 1) % BUFFER_SIZE == b->readpos) { pthread_cond_wait(&b->notfull, &b->lock); } /* 寫數據,并移動指針 */ b->buffer[b->writepos] = data; b->writepos++; if (b->writepos > = BUFFER_SIZE) b->writepos = 0; /* 設置緩沖區非空的條件變量*/ pthread_cond_signal(&b->notempty); pthread_mutex_unlock(&b->lock); } /* 從緩沖區中取出整數*/ int get(struct prodcons *b) { int data; pthread_mutex_lock(&b->lock); /* 等待緩沖區非空*/ if (b->writepos == b->readpos) { pthread_cond_wait(&b->notempty, &b->lock); } /* 讀數據,移動讀指針*/ data = b->buffer[b->readpos]; b->readpos++; if (b->readpos > = BUFFER_SIZE) b->readpos = 0; /* 設置緩沖區未滿的條件變量*/ pthread_cond_signal(&b->notfull); pthread_mutex_unlock(&b->lock); return data; } /* 測試:生產者線程將1 到10000 的整數送入緩沖區,消費者線 程從緩沖區中獲取整數,兩者都打印信息*/ #define OVER ( - 1) struct prodcons buffer; void *producer(void *data) { int n; for (n = 0; n < 10000; n++) { printf("%d --->\n", n); put(&buffer, n); } put(&buffer, OVER); return NULL; } void *consumer(void *data) { int d; while (1) { d = get(&buffer); if (d == OVER) break; printf("--->%d \n", d); } return NULL; } int main(void) { pthread_t th_a, th_b; void *retval; init(&buffer); /* 創建生產者和消費者線程*/ pthread_create(&th_a, NULL, producer, 0); pthread_create(&th_b, NULL, consumer, 0); /* 等待兩個線程結束*/ pthread_join(th_a, &retval); pthread_join(th_b, &retval); return 0; } |
目前為止，筆者已經創作了《基于嵌入式操作系統VxWorks的多任務并發程序設計》（《軟件報》2006年5~12期連載）、《深入淺出Win32多線程程序設計》（天極網技術專題）系列，我們來找出這兩個系列文章與本文的共通點。
看待技術問題要瞄準其本質，不管是Linux、VxWorks還是WIN32，其涉及到多線程的部分都是那些內容，無非就是線程控制和線程通信，它們的許多函數只是名稱不同，其實質含義是等價的，下面我們來列個三大操作系統共同點詳細表單：
| 事項 | WIN32 | VxWorks | Linux |
| 線程創建 | CreateThread | taskSpawn | pthread_create |
| 線程終止 | 執行完成后退出；線程自身調用ExitThread函數即終止自己；被其他線程調用函數TerminateThread函數 | 執行完成后退出；由線程本身調用exit退出；被其他線程調用函數taskDelete終止 | 執行完成后退出；由線程本身調用pthread_exit 退出；被其他線程調用函數pthread_cance終止 |
| 獲取線程ID | GetCurrentThreadId | taskIdSelf | pthread_self |
| 創建互斥 | CreateMutex | semMCreate | pthread_mutex_init |
| 獲取互斥 | WaitForSingleObject、WaitForMultipleObjects | semTake | pthread_mutex_lock |
| 釋放互斥 | ReleaseMutex | semGive | phtread_mutex_unlock |
| 創建信號量 | CreateSemaphore | semBCreate、semCCreate | sem_init |
| 等待信號量 | WaitForSingleObject | semTake | sem_wait |
| 釋放信號量 | ReleaseSemaphore | semGive | sem_post |
6.小結
本章講述了Linux下多線程的控制及線程間通信編程方法，給出了一個生產者/消費者的實例，并將Linux的多線程與WIN32、VxWorks多線程進行了類比，總結了一般規律。鑒于多線程編程已成為開發并發應用程序的主流方法，學好本章的意義也便不言自明。
![[TOOLS] 移動端調試進行時 - whistle](http://pic.xiahunao.cn/[TOOLS] 移動端調試進行時 - whistle)