linux C++ 多線程使用pthread_cond 條件變量

1. 背景

多線程中經常需要使用到鎖(pthread_mutex_t)來完成多個線程之間的互斥操作。

但是互斥鎖有一個明顯到缺點: 只有兩種狀態，鎖定和非鎖定。

而條件變量則通過允許線程阻塞並等待另一個線程發送喚醒信號的方法彌補了互斥鎖的不足，它常和互斥鎖一起使用。

 

2. 條件變量涉及到的主要函數

2.1 pthread_cond_wait 線程阻塞在條件變量

   int pthread_cond_wait(pthread_cond_t *cv, pthread_mutex_t *mutex);

函數將解鎖mutex參數指向的互斥鎖，並使當前線程阻塞在cv參數指向的條件變量上。
被阻塞的線程可以被pthread_cond_signal函數，pthread_cond_broadcast函數喚醒，也可能在被信號中斷后被喚醒。
pthread_cond_wait函數返回時，相應的互斥鎖將被當前線程鎖定，即使是函數出錯返回。
pthread_cond_wait函數的返回並不意味着條件的值一定發生了變化，必須重新檢查條件的值。
最好的測試方法是循環調用pthread_cond_wait函數，並把滿足條件的表達式置為循環的終止條件。

pthread_mutex_lock();
while (condition_is_false)
    pthread_cond_wait();
pthread_mutex_unlock();

阻塞在同一個條件變量上的不同線程被喚醒的次序是不一定的。

 2.2 pthread_cond_signal 線程被喚醒

int pthread_cond_signal(pthread_cond_t *cv);
函數被用來釋放被阻塞在指定條件變量上的一個線程。
必須在互斥鎖的保護下使用相應的條件變量。否則對條件變量的解鎖有可能發生在鎖定條件變量之前，從而造成死鎖。
喚醒阻塞在條件變量上的所有線程的順序由調度策略決定，如果線程的調度策略是SCHED_OTHER類型的，系統將根據線程的優先級喚醒線程。
如果沒有線程被阻塞在條件變量上，那么調用pthread_cond_signal()將沒有作用。

更多函數可以看: http://blog.csdn.net/icechenbing/article/details/7662026

 3. 實例代碼

實現功能: 2個線程對count每次分別加1, 第三個線程等count大於10后一次加100.

3.1 加1線程函數

void *inc_count(void *idp)  
{ 
    int i = 0;
    int taskid = 0;  
    int *my_id = (int*)idp; 

    for (i=0; i<TCOUNT; i++) { 
        pthread_mutex_lock(&count_mutex); 
        taskid = count;
        count++; 

        /*  
          喚醒一個阻塞在該條件變量到線程
          如果沒有線程被阻塞在條件變量上，那么調用pthread_cond_signal()將沒有作用
        */
        pthread_cond_signal(&count_threshold_cv); 

        printf("inc_count(): thread %d, count = %d, unlocking mutex\n", *my_id, count); 
        pthread_mutex_unlock(&count_mutex); 
        sleep(1);
    }   
    printf("inc_count(): thread %d, Threshold reached.\n", *my_id); 

    pthread_exit(NULL); 
} 

3.2 count滿足條件后, 單次加100函數

void *watch_count(void *idp)  
{ 
    int *my_id = (int*)idp; 
    printf("Starting watch_count(): thread %d\n", *my_id); 

    pthread_mutex_lock(&count_mutex); 
    while(count<COUNT_LIMIT) { 
        sleep(3);
        /*  
          函數將自動/原子的解鎖count_mutex參數指向的互斥鎖，並使當前線程阻塞在cv參數指向的條件變量上
          被阻塞的線程可以被pthread_cond_signal函數，pthread_cond_broadcast函數喚醒，也可能在被信號中斷后被喚醒
          pthread_cond_wait函數的返回並不意味着條件的值一定發生了變化，必須重新檢查條件的值. 
          本例子中線程被喚醒后, 仍然在while內會再次判斷COUNT_LIMIT是否滿足條件的值
          pthread_cond_wait函數返回時，相應的互斥鎖將被當前線程鎖定，即使是函數出錯返回
        */
        pthread_cond_wait(&count_threshold_cv, &count_mutex); 
        printf("watch_count(): thread %d Condition signal received.\n", *my_id); 
    }   
 
    count += 100;
    pthread_mutex_unlock(&count_mutex); 
    pthread_exit(NULL); 
}

3.3 整體代碼

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>

#define NUM_THREADS 3 
#define TCOUNT      10 
#define COUNT_LIMIT 10 

int count = 0;
int thread_ids[3] = {0,1,2};
pthread_mutex_t count_mutex;
pthread_cond_t  count_threshold_cv;

void *inc_count(void *idp)
{
    int i = 0;
    int taskid = 0;
    int *my_id = (int*)idp;

    for (i=0; i<TCOUNT; i++) {
        pthread_mutex_lock(&count_mutex);
        taskid = count;
        count++;

        /*
          喚醒一個阻塞在該條件變量到線程
          如果沒有線程被阻塞在條件變量上，那么調用pthread_cond_signal()將沒有作用
        */
        pthread_cond_signal(&count_threshold_cv);

        printf("inc_count(): thread %d, count = %d, unlocking mutex\n", *my_id, count);
        pthread_mutex_unlock(&count_mutex);
        sleep(1);
    }
    printf("inc_count(): thread %d, Threshold reached.\n", *my_id);

    pthread_exit(NULL);
}

void *watch_count(void *idp)
{
    int *my_id = (int*)idp;
    printf("Starting watch_count(): thread %d\n", *my_id);

    pthread_mutex_lock(&count_mutex);
    while(count<COUNT_LIMIT) {
        sleep(3);
        /*
          函數將自動/原子到解鎖mutex參數指向的互斥鎖，並使當前線程阻塞在cv參數指向的條件變量上
          被阻塞的線程可以被pthread_cond_signal函數，pthread_cond_broadcast函數喚醒，也可能在被信號中斷后被喚醒
          pthread_cond_wait函數的返回並不意味着條件的值一定發生了變化，必須重新檢查條件的值. 
          本例子中使用類COUNT_LIMIT最為滿足條件的值
          pthread_cond_wait函數返回時，相應的互斥鎖將被當前線程鎖定，即使是函數出錯返回
        */
        pthread_cond_wait(&count_threshold_cv, &count_mutex);
        printf("watch_count(): thread %d Condition signal received.\n", *my_id);
    }

    count += 100;
    pthread_mutex_unlock(&count_mutex);
    pthread_exit(NULL);
}

int main (int argc, char *argv[])
{
    int i, rc;
    pthread_t threads[3];
    pthread_attr_t attr;

    /* Initialize mutex and condition variable objects */
    pthread_mutex_init(&count_mutex, NULL);
    pthread_cond_init (&count_threshold_cv, NULL);

    /* For portability, explicitly create threads in a joinable state */
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
    pthread_create(&threads[0], &attr, inc_count,   (void *)&thread_ids[0]);
    pthread_create(&threads[1], &attr, inc_count,   (void *)&thread_ids[1]);
    pthread_create(&threads[2], &attr, watch_count, (void *)&thread_ids[2]);

    /* Wait for all threads to complete */
    for (i=0; i<NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }
    printf ("Main(): Waited on %d  threads. Done.\n", NUM_THREADS);

    /* Clean up and exit */
    pthread_attr_destroy(&attr);
    pthread_mutex_destroy(&count_mutex);
    pthread_cond_destroy(&count_threshold_cv);
    pthread_exit(NULL);

    return 0;
}