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
", *my_id, count); 
        pthread_mutex_unlock(&count_mutex); 
        sleep(1);
    }   
    printf("inc_count(): thread %d, Threshold reached.
", *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
", *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.
", *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
", *my_id, count);
        pthread_mutex_unlock(&count_mutex);
        sleep(1);
    }
    printf("inc_count(): thread %d, Threshold reached.
", *my_id);

    pthread_exit(NULL);
}

void *watch_count(void *idp)
{
    int *my_id = (int*)idp;
    printf("Starting watch_count(): thread %d
", *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.
", *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.
", 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;
}