一) 線程鎖
1) 只能用于"鎖"住臨界代碼區(qū)域
2) 一個(gè)線程加的鎖必須由該線程解鎖.
鎖幾乎是我們學(xué)習(xí)同步時(shí)最開始接觸到的一個(gè)策略,也是最簡單, 最直白的策略.
二) 條件變量,與鎖不同, 條件變量用于等待某個(gè)條件被觸發(fā)
1) 大體使用的偽碼:
// 線程一代碼
pthread_mutex_lock(&mutex);
// 設(shè)置條件為true
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
// 線程二代碼
pthread_mutex_lock(&mutex);
while (條件為false)
pthread_cond_wait(&cond, &mutex);
修改該條件
pthread_mutex_unlock(&mutex);
需要注意幾點(diǎn):
1) 第二段代碼之所以在pthread_cond_wait外面包含一個(gè)while循環(huán)不停測試條件是否成立的原因是, 在pthread_cond_wait被喚醒的時(shí)候可能該條件已經(jīng)不成立.UNPV2對這個(gè)的描述是:"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)用必須和某一個(gè)mutex一起調(diào)用, 這個(gè)mutex是在外部進(jìn)行加鎖的mutex,
在調(diào)用pthread_cond_wait時(shí), 內(nèi)部的實(shí)現(xiàn)將首先將這個(gè)mutex解鎖, 然后等待條件變量被喚醒, 如果沒有被喚醒,
該線程將一直休眠, 也就是說, 該線程將一直阻塞在這個(gè)pthread_cond_wait調(diào)用中, 而當(dāng)此線程被喚醒時(shí),
將自動(dòng)將這個(gè)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實(shí)際上可以看作是以下幾個(gè)動(dòng)作的合體:
解鎖線程鎖
等待條件為true
加鎖線程鎖.
這里是使用條件變量的經(jīng)典例子:
http://www.shnenglu.com/CppExplore/archive/2008/03/20/44949.html
之所以使用兩個(gè)條件變量,
是因?yàn)橛袃煞N情況需要進(jìn)行保護(hù),使用數(shù)組實(shí)現(xiàn)循環(huán)隊(duì)列,因此一個(gè)條件是在getq函數(shù)中判斷讀寫指針相同且可讀數(shù)據(jù)計(jì)數(shù)為0,此時(shí)隊(duì)列為空沒有數(shù)據(jù)可讀,因此獲取新數(shù)據(jù)的條件變量就一直等待,另一個(gè)條件是讀寫指針相同且可讀數(shù)據(jù)計(jì)數(shù)大于0,此時(shí)隊(duì)列滿了不能再添加數(shù)據(jù),
因此添加新數(shù)據(jù)的條件變量就一直等待,而nEmptyThreadNum和nFullThreadNum則是計(jì)數(shù),
只有這個(gè)計(jì)數(shù)大于0時(shí)才會喚醒相應(yīng)的條件變量,這樣可以減少調(diào)用pthread_cond_signal的次數(shù).
為了在下面的敘述方便, 我將這段代碼整理在下面, 是一個(gè)可以編譯運(yùn)行的代碼,但是注意需要在編譯時(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)判斷的原因如下：假設(shè)2個(gè)線程在getq阻塞，然后兩者都被激活，而其中一個(gè)線程運(yùn)行比較塊，快速消耗了2個(gè)數(shù)據(jù)，另一個(gè)線程醒來的時(shí)候已經(jīng)沒有新
數(shù)據(jù)可以消耗了。另一點(diǎn)，man
pthread_cond_wait可以看到，該函數(shù)可以被信號中斷返回，此時(shí)返回EINTR。為避免以上任何一點(diǎn)，都必須醒來后再次判斷睡眠條件。更
正：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) //必要時(shí)才進(jìn)行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) //必要時(shí)才進(jìn)行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)消息隊(duì)列
int sizeQueue; //隊(duì)列大小
int lput; //location put 放數(shù)據(jù)的指針偏移
int lget; //location get 取數(shù)據(jù)的指針偏移
int nFullThread; //隊(duì)列滿，阻塞在putq處的線程數(shù)
int nEmptyThread; //隊(duì)列空，阻塞在getq處的線程數(shù)
int nData; //隊(duì)列中的消息個(gè)數(shù)，主要用來判斷隊(duì)列空還是滿
};
CThreadQueue queue;//使用的時(shí)候給出稍大的CThreadQueue初始化參數(shù)，可以減少進(jìn)入內(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;
}
三) 信號量
信號量既可以作為二值計(jì)數(shù)器(即0,1),也可以作為資源計(jì)數(shù)器.
主要是兩個(gè)函數(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);則用于獲取信號量當(dāng)前的計(jì)數(shù).
可以用信號量模擬鎖和條件變量:
1) 鎖,在同一個(gè)線程內(nèi)同時(shí)對某個(gè)信號量先調(diào)用sem_wait再調(diào)用sem_post, 兩個(gè)函數(shù)調(diào)用其中的區(qū)域就是所要保護(hù)的臨界區(qū)代碼了,這個(gè)時(shí)候其實(shí)信號量是作為二值計(jì)數(shù)器來使用的.不過在此之前要初始化該信號量計(jì)數(shù)為1,見下面例子中的代碼.
2)
條件變量,在某個(gè)線程中調(diào)用sem_wait, 而在另一個(gè)線程中調(diào)用sem_post.
我們將上面例子中的線程鎖和條件變量都改成用信號量實(shí)現(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) //必要時(shí)才進(jìn)行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)消息隊(duì)列
int sizeQueue; //隊(duì)列大小
int lput; //location put 放數(shù)據(jù)的指針偏移
int lget; //location get 取數(shù)據(jù)的指針偏移
int nFullThread; //隊(duì)列滿，阻塞在putq處的線程數(shù)
int nEmptyThread; //隊(duì)列空，阻塞在getq處的線程數(shù)
int nData; //隊(duì)列中的消息個(gè)數(shù)，主要用來判斷隊(duì)列空還是滿
};
CThreadQueue queue;//使用的時(shí)候給出稍大的CThreadQueue初始化參數(shù)，可以減少進(jìn)入內(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;
}
不過, 信號量除了可以作為二值計(jì)數(shù)器用于模擬線程鎖和條件變量之外, 還有比它們更加強(qiáng)大的功能, 信號量可以用做資源計(jì)數(shù)器, 也就是說初始化信號量的值為某個(gè)資源當(dāng)前可用的數(shù)量, 使用了一個(gè)之后遞減, 歸還了一個(gè)之后遞增, 將前面的例子用資源計(jì)數(shù)器的形式再次改寫如下,注意在初始化的時(shí)候要將資源計(jì)數(shù)進(jìn)行初始化, 在下面代碼中的構(gòu)造函數(shù)中將put初始化為隊(duì)列的最大數(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) //必要時(shí)才進(jìn)行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)消息隊(duì)列
int sizeQueue; //隊(duì)列大小
int lput; //location put 放數(shù)據(jù)的指針偏移
int lget; //location get 取數(shù)據(jù)的指針偏移
};
CThreadQueue queue;//使用的時(shí)候給出稍大的CThreadQueue初始化參數(shù)，可以減少進(jìn)入內(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;
}
可以看見,采用信號量作為資源計(jì)數(shù)之后, 代碼變得"很直白",原來的一些保存隊(duì)列狀態(tài)的變量都不再需要了.
信號量與線程鎖,條件變量相比還有以下幾點(diǎn)不同:
1)鎖必須是同一個(gè)線程獲取以及釋放, 否則會死鎖.而條件變量和信號量則不必.
2)信號的遞增與減少會被系統(tǒng)自動(dòng)記住, 系統(tǒng)內(nèi)部有一個(gè)計(jì)數(shù)器實(shí)現(xiàn)信號量,不必?fù)?dān)心會丟失, 而喚醒一個(gè)條件變量時(shí),如果沒有相應(yīng)的線程在等待該條件變量, 這次喚醒將被丟失.