一) 線程鎖
1) 只能用于"鎖"住臨界代碼區(qū)域
2) 一個線程加的鎖必須由該線程解鎖.
鎖幾乎是我們學習同步時最開始接觸到的一個策略,也是最簡單, 最直白的策略.
二) 條件變量,與鎖不同, 條件變量用于等待某個條件被觸發(fā)
1) 大體使用的偽碼:
// 線程一代碼
pthread_mutex_lock(&mutex);
// 設置條件為true
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
// 線程二代碼
pthread_mutex_lock(&mutex);
while (條件為false)
pthread_cond_wait(&cond, &mutex);
修改該條件
pthread_mutex_unlock(&mutex);
需要注意幾點:
1) 第二段代碼之所以在pthread_cond_wait外面包含一個while循環(huán)不停測試條件是否成立的原因是, 在pthread_cond_wait被喚醒的時候可能該條件已經(jīng)不成立.UNPV2對這個的描述是:"Notice that when pthread_cond_wait returns, we always test the condition again, because spurious wakeups can occur: a wakeup when the desired condition is still not true.".
2)
pthread_cond_wait調(diào)用必須和某一個mutex一起調(diào)用, 這個mutex是在外部進行加鎖的mutex,
在調(diào)用pthread_cond_wait時, 內(nèi)部的實現(xiàn)將首先將這個mutex解鎖, 然后等待條件變量被喚醒, 如果沒有被喚醒,
該線程將一直休眠, 也就是說, 該線程將一直阻塞在這個pthread_cond_wait調(diào)用中, 而當此線程被喚醒時,
將自動將這個mutex加鎖.
man文檔中對這部分的說明是:
pthread_cond_wait atomically unlocks the mutex (as per pthread_unlock_mutex) and waits for the condition variable cond to be signaled. The thread execution is suspended and does not consume any CPU time until the condition variable is
signaled. The mutex must be locked by the calling thread on entrance to pthread_cond_wait. Before returning to the calling thread, pthread_cond_wait re-acquires mutex (as per pthread_lock_mutex).
也就是說pthread_cond_wait實際上可以看作是以下幾個動作的合體:
解鎖線程鎖
等待條件為true
加鎖線程鎖.
這里是使用條件變量的經(jīng)典例子:
http://www.shnenglu.com/CppExplore/archive/2008/03/20/44949.html
之所以使用兩個條件變量,
是因為有兩種情況需要進行保護,使用數(shù)組實現(xiàn)循環(huán)隊列,因此一個條件是在getq函數(shù)中判斷讀寫指針相同且可讀數(shù)據(jù)計數(shù)為0,此時隊列為空沒有數(shù)據(jù)可讀,因此獲取新數(shù)據(jù)的條件變量就一直等待,另一個條件是讀寫指針相同且可讀數(shù)據(jù)計數(shù)大于0,此時隊列滿了不能再添加數(shù)據(jù),
因此添加新數(shù)據(jù)的條件變量就一直等待,而nEmptyThreadNum和nFullThreadNum則是計數(shù),
只有這個計數(shù)大于0時才會喚醒相應的條件變量,這樣可以減少調(diào)用pthread_cond_signal的次數(shù).
為了在下面的敘述方便, 我將這段代碼整理在下面, 是一個可以編譯運行的代碼,但是注意需要在編譯時加上-pthread鏈接線程庫:
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
class CThreadQueue
{
public:
CThreadQueue(int queueSize=1024):
sizeQueue(queueSize),lput(0),lget(0),nFullThread(0),nEmptyThread(0),nData(0)
{
pthread_mutex_init(&mux,0);
pthread_cond_init(&condGet,0);
pthread_cond_init(&condPut,0);
buffer=new void *[sizeQueue];
}
virtual ~CThreadQueue()
{
delete[] buffer;
}
void * getq()
{
void *data;
pthread_mutex_lock(&mux);
/*
此
處循環(huán)判斷的原因如下：假設2個線程在getq阻塞，然后兩者都被激活，而其中一個線程運行比較塊，快速消耗了2個數(shù)據(jù)，另一個線程醒來的時候已經(jīng)沒有新
數(shù)據(jù)可以消耗了。另一點，man
pthread_cond_wait可以看到，該函數(shù)可以被信號中斷返回，此時返回EINTR。為避免以上任何一點，都必須醒來后再次判斷睡眠條件。更
正：pthread_cond_wait是信號安全的系統(tǒng)調(diào)用，不會被信號中斷。
*/
while(lget==lput&&nData==0)
{
nEmptyThread++;
pthread_cond_wait(&condGet,&mux);
nEmptyThread--;
}
data=buffer[lget++];
nData--;
if(lget==sizeQueue)
{
lget=0;
}
if(nFullThread) //必要時才進行signal操作，勿總是signal
{
pthread_cond_signal(&condPut);
}
pthread_mutex_unlock(&mux);
return data;
}
void putq(void *data)
{
pthread_mutex_lock(&mux);
while(lput==lget&&nData)
{
nFullThread++;
pthread_cond_wait(&condPut,&mux);
nFullThread--;
}
buffer[lput++]=data;
nData++;
if(lput==sizeQueue)
{
lput=0;
}
if(nEmptyThread) //必要時才進行signal操作，勿總是signal
{
pthread_cond_signal(&condGet);
}
pthread_mutex_unlock(&mux);
}
private:
pthread_mutex_t mux;
pthread_cond_t condGet;
pthread_cond_t condPut;
void * * buffer; //循環(huán)消息隊列
int sizeQueue; //隊列大小
int lput; //location put 放數(shù)據(jù)的指針偏移
int lget; //location get 取數(shù)據(jù)的指針偏移
int nFullThread; //隊列滿，阻塞在putq處的線程數(shù)
int nEmptyThread; //隊列空，阻塞在getq處的線程數(shù)
int nData; //隊列中的消息個數(shù)，主要用來判斷隊列空還是滿
};
CThreadQueue queue;//使用的時候給出稍大的CThreadQueue初始化參數(shù)，可以減少進入內(nèi)核態(tài)的操作。
void * produce(void * arg)
{
int i=0;
pthread_detach(pthread_self());
while(i++<100)
{
queue.putq((void *)i);
}
}
void *consume(void *arg)
{
int data;
while(1)
{
data=(int)(queue.getq());
printf("data=%d\n",data);
}
}
int main()
{
pthread_t pid;
int i=0;
while(i++<3)
pthread_create(&pid,0,produce,0);
i=0;
while(i++<3)
pthread_create(&pid,0,consume,0);
sleep(5);
return 0;
}
三) 信號量
信號量既可以作為二值計數(shù)器(即0,1),也可以作為資源計數(shù)器.
主要是兩個函數(shù):
sem_wait() decrements (locks) the semaphore pointed to by sem. If the semaphore's value is greater than zero, then
the decrement proceeds, and the function returns, immediately. If the semaphore currently has the value zero, then
the call blocks until either it becomes possible to perform the decrement (i.e., the semaphore value rises above
zero), or a signal handler interrupts the call.
sem_post() increments (unlocks) the semaphore pointed to by sem. If the semaphore's value consequently becomes
greater than zero, then another process or thread blocked in a sem_wait(3) call will be woken up and proceed to lock
the semaphore.
而函數(shù)int sem_getvalue(sem_t *sem, int *sval);則用于獲取信號量當前的計數(shù).
可以用信號量模擬鎖和條件變量:
1) 鎖,在同一個線程內(nèi)同時對某個信號量先調(diào)用sem_wait再調(diào)用sem_post, 兩個函數(shù)調(diào)用其中的區(qū)域就是所要保護的臨界區(qū)代碼了,這個時候其實信號量是作為二值計數(shù)器來使用的.不過在此之前要初始化該信號量計數(shù)為1,見下面例子中的代碼.
2)
條件變量,在某個線程中調(diào)用sem_wait, 而在另一個線程中調(diào)用sem_post.
我們將上面例子中的線程鎖和條件變量都改成用信號量實現(xiàn)以說明信號量如何模擬兩者:
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <semaphore.h>
#include <errno.h>
#include <string.h>
class CThreadQueue
{
public:
CThreadQueue(int queueSize=1024):
sizeQueue(queueSize),lput(0),lget(0),nFullThread(0),nEmptyThread(0),nData(0)
{
//pthread_mutex_init(&mux,0);
mux = sem_open("mutex", O_RDWR | O_CREAT);
get = sem_open("get", O_RDWR | O_CREAT);
put = sem_open("put", O_RDWR | O_CREAT);
sem_init(mux, 0, 1);
buffer=new void *[sizeQueue];
}
virtual ~CThreadQueue()
{
delete[] buffer;
sem_unlink("mutex");
sem_unlink("get");
sem_unlink("put");
}
void * getq()
{
void *data;
//pthread_mutex_lock(&mux);
sem_wait(mux);
while(lget==lput&&nData==0)
{
nEmptyThread++;
//pthread_cond_wait(&condGet,&mux);
sem_wait(get);
nEmptyThread--;
}
data=buffer[lget++];
nData--;
if(lget==sizeQueue)
{
lget=0;
}
if(nFullThread) //必要時才進行signal操作，勿總是signal
{
//pthread_cond_signal(&condPut);
sem_post(put);
}
//pthread_mutex_unlock(&mux);
sem_post(mux);
return data;
}
void putq(void *data)
{
//pthread_mutex_lock(&mux);
sem_wait(mux);
while(lput==lget&&nData)
{
nFullThread++;
//pthread_cond_wait(&condPut,&mux);
sem_wait(put);
nFullThread--;
}
buffer[lput++]=data;
nData++;
if(lput==sizeQueue)
{
lput=0;
}
if(nEmptyThread)
{
//pthread_cond_signal(&condGet);
sem_post(get);
}
//pthread_mutex_unlock(&mux);
sem_post(mux);
}
private:
//pthread_mutex_t mux;
sem_t* mux;
//pthread_cond_t condGet;
//pthread_cond_t condPut;
sem_t* get;
sem_t* put;
void * * buffer; //循環(huán)消息隊列
int sizeQueue; //隊列大小
int lput; //location put 放數(shù)據(jù)的指針偏移
int lget; //location get 取數(shù)據(jù)的指針偏移
int nFullThread; //隊列滿，阻塞在putq處的線程數(shù)
int nEmptyThread; //隊列空，阻塞在getq處的線程數(shù)
int nData; //隊列中的消息個數(shù)，主要用來判斷隊列空還是滿
};
CThreadQueue queue;//使用的時候給出稍大的CThreadQueue初始化參數(shù)，可以減少進入內(nèi)核態(tài)的操作。
void * produce(void * arg)
{
int i=0;
pthread_detach(pthread_self());
while(i++<100)
{
queue.putq((void *)i);
}
}
void *consume(void *arg)
{
int data;
while(1)
{
data=(int)(queue.getq());
printf("data=%d\n",data);
}
}
int main()
{
pthread_t pid;
int i=0;
while(i++<3)
pthread_create(&pid,0,produce,0);
i=0;
while(i++<3)
pthread_create(&pid,0,consume,0);
sleep(5);
return 0;
}
不過, 信號量除了可以作為二值計數(shù)器用于模擬線程鎖和條件變量之外, 還有比它們更加強大的功能, 信號量可以用做資源計數(shù)器, 也就是說初始化信號量的值為某個資源當前可用的數(shù)量, 使用了一個之后遞減, 歸還了一個之后遞增, 將前面的例子用資源計數(shù)器的形式再次改寫如下,注意在初始化的時候要將資源計數(shù)進行初始化, 在下面代碼中的構(gòu)造函數(shù)中將put初始化為隊列的最大數(shù)量, 而get為0:
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <semaphore.h>
class CThreadQueue
{
public:
CThreadQueue(int queueSize=1024):
sizeQueue(queueSize),lput(0),lget(0)
{
pthread_mutex_init(&mux,0);
get = sem_open("get", O_RDWR | O_CREAT);
put = sem_open("put", O_RDWR | O_CREAT);
sem_init(get, 0, 0);
sem_init(put, 0, sizeQueue);
buffer=new void *[sizeQueue];
}
virtual ~CThreadQueue()
{
sem_unlink("get");
sem_unlink("put");
delete[] buffer;
}
void * getq()
{
sem_wait(get);
void *data;
pthread_mutex_lock(&mux);
/*
while(lget==lput&&nData==0)
{
nEmptyThread++;
//pthread_cond_wait(&condGet,&mux);
nEmptyThread--;
}
*/
data=buffer[lget++];
//nData--;
if(lget==sizeQueue)
{
lget=0;
}
/*
if(nFullThread) //必要時才進行signal操作，勿總是signal
{
//pthread_cond_signal(&condPut);
sem_post(put);
}
*/
pthread_mutex_unlock(&mux);
sem_post(put);
return data;
}
void putq(void *data)
{
sem_wait(put);
pthread_mutex_lock(&mux);
/*
while(lput==lget&&nData)
{
nFullThread++;
//pthread_cond_wait(&condPut,&mux);
sem_wait(put);
nFullThread--;
}
*/
buffer[lput++]=data;
//nData++;
if(lput==sizeQueue)
{
lput=0;
}
/*
if(nEmptyThread)
{
//pthread_cond_signal(&condGet);
sem_post(get);
}
*/
pthread_mutex_unlock(&mux);
sem_post(get);
}
private:
pthread_mutex_t mux;
//pthread_cond_t condGet;
//pthread_cond_t condPut;
sem_t* get;
sem_t* put;
void * * buffer; //循環(huán)消息隊列
int sizeQueue; //隊列大小
int lput; //location put 放數(shù)據(jù)的指針偏移
int lget; //location get 取數(shù)據(jù)的指針偏移
};
CThreadQueue queue;//使用的時候給出稍大的CThreadQueue初始化參數(shù)，可以減少進入內(nèi)核態(tài)的操作。
void * produce(void * arg)
{
int i=0;
pthread_detach(pthread_self());
while(i++<100)
{
queue.putq((void *)i);
}
}
void *consume(void *arg)
{
int data;
while(1)
{
data=(int)(queue.getq());
printf("data=%d\n",data);
}
}
int main()
{
pthread_t pid;
int i=0;
while(i++<3)
pthread_create(&pid,0,produce,0);
i=0;
while(i++<3)
pthread_create(&pid,0,consume,0);
sleep(5);
return 0;
}
可以看見,采用信號量作為資源計數(shù)之后, 代碼變得"很直白",原來的一些保存隊列狀態(tài)的變量都不再需要了.
信號量與線程鎖,條件變量相比還有以下幾點不同:
1)鎖必須是同一個線程獲取以及釋放, 否則會死鎖.而條件變量和信號量則不必.
2)信號的遞增與減少會被系統(tǒng)自動記住, 系統(tǒng)內(nèi)部有一個計數(shù)器實現(xiàn)信號量,不必擔心會丟失, 而喚醒一個條件變量時,如果沒有相應的線程在等待該條件變量, 這次喚醒將被丟失.