c++ 条件变量

1.条件变量创建
静态创建：pthread_cond_t cond=PTHREAD_COND_INITIALIZER;
动态创建：pthread_cond _t cond;
  pthread_cond_init(&cond,NULL);
其中的第二个参数NULL表示条件变量的属性，虽然POSIX中定义了条件变量的属性，但在LinuxThread中并没有实现，因此常常忽略。
2.条件等待
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_lock(&mutex);
while(条件1)
  pthread_cond_wait(&cond,&mutex);
函数操作1;
pthread_mutex_unlock(&mutex);
当条件1成立的时候，执行pthread_cond_wait(&cond,&mutex)这一句，开放互斥锁，然后线程被挂起。当条件1不成立的时候，跳过while循环体，执行函数操作1，然后开放互斥锁。
3.条件激发
pthread_mutex_lock(&mutex);
函数操作2;
if(条件1不成立)
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
先执行函数操作2，改变条件状态，使得条件1不成立的时候,执行pthread_cond_signal(&cond)这句话。这句话的意思是激发条件变量cond，使得被挂起的线程被唤醒。
pthread_cond_broadcast(&cond);
这句话也是激发条件变量cond，但是，这句话是激发所有由于cond条件被挂起的线程。而signal的函数则是激发一个由于条件变量cond被挂起的线程。
4.条件变量的销毁
pthread_cond_destroy(&cond);
在linux中，由于条件变量不占用任何资源，所以，这句话除了检查有没有等待条件变量cond的线程外，不做任何操作。

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

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int count = 5;

//pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
//pthread_mutex_lock(&mutex);
//while(条件1)
//  pthread_cond_wait(&cond, &mutex);
//函数操作1
//pthread_mutex_unlock(&mutex);
//解释：当条件1成立的时候，执行pthread_cond_wait(&cond, &mutex)这一句，开放互斥锁，然后线程被挂起。
//      当条件1不成立的时候，跳过while循环体，执行函数操作1，然后开放互斥锁
//      即无论走哪个分支，都会放开互斥锁

//此线程先启动
void* decrement(void* arg) {
    while(1) {
        //条件等待
        pthread_mutex_lock(&mutex);
        while(count <= 0) {
            printf("count<=0,thread is hanging!
");
            pthread_cond_wait(&cond, &mutex); 
            sleep(1);
            printf("sleep!
");
        }
        count -= 1;
        pthread_mutex_unlock(&mutex);

        if (count == 9) {
            printf("count==9,thread1 is over!
");
            return NULL;
        }
    }
}

//条件激发
//pthread_mutex_lock(&mutex);
//函数操作2；
//if(条件1不成立)
//  pthread_cond_signal(&cond);
//  pthread_mutex_unlock(&mutex);
//解释：先执行函数操作2，改变条件变量，使得条件1不成立的时候，执行pthread_cond_signal(&cond)这句话的意思是激发条件变量cond，使得被挂起的线程被唤醒。
//      pthread_cond_broadcast(&cond); 激发所有由于cond条件而被挂起的线程。而signal的函数则是激发一个由于条件变量cond被挂起的线程

void *increment(void *arg) {
    sleep(1);
    while(1) {
        pthread_mutex_lock(&mutex);
        count += 1;
        if(count > 0) {
            printf("count=%d, change cond state!
", count);
            pthread_cond_signal(&cond);
        }
        pthread_mutex_unlock(&mutex);
        
        if (count == 10) {
            printf("count=10,thread is over!
");
            return NULL;
        }
    }
}

int main(void) {
    int i1=1, i2=1;
    pthread_t id1, id2;
    pthread_mutex_init(&mutex, NULL);
    pthread_cond_init(&cond, NULL);

    pthread_create(&id1, NULL, decrement, NULL);
    pthread_create(&id2, NULL, increment, NULL);

    i2 = pthread_join(id2, NULL);
    i1 = pthread_join(id1, NULL);
    if ( (i2==0) && (i1==0) ) {
        printf("count=%d, the main thread!
", count);
        pthread_cond_destroy(&cond);
        pthread_mutex_destroy(&mutex);
        return 0;
    }
    return -1;
}

#include <stdio.h>
#include <pthread.h>
#include <queue>
#include <string>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
std::queue<std::string> que;

void* consumer(void* arg) {
    while (true) {
        pthread_mutex_lock(&mutex);
        while ( que.empty() ) {
            printf("que is empty , thread consumer hanging!
");
            pthread_cond_wait(&cond, &mutex);
        }
        while (!que.empty()) {
            printf("%s
", que.front().c_str());
            que.pop();
        }
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

void* producer(void* arg) {
    int count = 0;
    while (true) {
        pthread_mutex_lock(&mutex);
        for (int i=0; i<10; i++) {
            char arr[20];
            snprintf(arr, sizeof(arr), "my num is: %d", i);
            que.push(arr);
            pthread_cond_signal(&cond);
        }
        pthread_mutex_unlock(&mutex);
        count++;
        if (count > 10) {
            printf("stop producer %d s
", 5);
            sleep(15);
            count = 0;
        }
    }
    return NULL;
}


int main(void) {
    pthread_t arr[2];
    pthread_create(&arr[0], NULL, consumer, NULL);
    sleep(3);
    printf("sleep 3 s
");
    pthread_create(&arr[1], NULL, producer, NULL);

    for(int i=0; i< 2; i++) {
        pthread_join(arr[i], NULL);
    }
    
    pthread_cond_destroy(&cond);
    pthread_mutex_destroy(&mutex);
}

http://blog.csdn.net/hemmanhui/article/details/4417433

互斥锁：用来上锁。

条件变量：用来等待，当条件变量用来自动阻塞一个线程，直到某特殊情况发生为止。通常条件变量和互斥锁同时使用。

函数介绍：

1．

名称：	pthread_cond_init
目标：	条件变量初始化
头文件：	#include < pthread.h>
函数原形：	int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr);
参数:	cptr  条件变量 attr  条件变量属性
返回值：	成功返回0，出错返回错误编号。

     

  pthread_cond_init函数可以用来初始化一个条件变量。他使用变量attr所指定的属性来初始化一个条件变量，如果参数attr为空，那么它将使用缺省的属性来设置所指定的条件变量。

2.

名称：	pthread_cond_destroy
目标：	条件变量摧毁
头文件：	#include < pthread.h>
函数原形：	int pthread_cond_destroy(pthread_cond_t *cond);
参数:	cptr  条件变量
返回值：	成功返回0，出错返回错误编号。

      

 pthread_cond_destroy函数可以用来摧毁所指定的条件变量，同时将会释放所给它分配的资源。调用该函数的进程也并不要求等待在参数所指定的条件变量上。

3.

名称：	pthread_cond_wait/pthread_cond_timedwait
目标：	条件变量等待
头文件：	#include < pthread.h>
函数原形：	int pthread_cond_wait(pthread_cond_t *cond,pthread_mutex_t *mutex); int pthread_cond_timedwait(pthread_cond_t *cond,pthread_mutex_t mytex,const struct timespec *abstime);
参数:	cond 条件变量 mutex 互斥锁
返回值：	成功返回0，出错返回错误编号。

      

 第一个参数*cond是指向一个条件变量的指针。第二个参数*mutex则是对相关的互斥锁的指针。函数pthread_cond_timedwait函数类型与函数pthread_cond_wait,区别在于，如果达到或是超过所引用的参数*abstime,它将结束并返回错误ETIME.pthread_cond_timedwait函数的参数*abstime指向一个timespec结构。该结构如下：

typedef struct timespec{

       time_t tv_sec;

       long tv_nsex;

}timespec_t;

3.

 名称：

	pthread_cond_signal/pthread_cond_broadcast
目标：	条件变量通知
头文件：	#include < pthread.h>
函数原形：	int pthread_cond_signal(pthread_cond_t *cond); int pthread_cond_broadcast(pthread_cond_t *cond);
参数:	cond 条件变量
返回值：	成功返回0，出错返回错误编号。

      

 参数*cond是对类型为pthread_cond_t 的一个条件变量的指针。当调用pthread_cond_signal时一个在相同条件变量上阻塞的线程将被解锁。如果同时有多个线程阻塞，则由调度策略确定接收通知的线程。如果调用pthread_cond_broadcast,则将通知阻塞在这个条件变量上的所有线程。一旦被唤醒，线程仍然会要求互斥锁。如果当前没有线程等待通知，则上面两种调用实际上成为一个空操作。如果参数*cond指向非法地址，则返回值EINVAL。

下面是一个简单的例子，我们可以从程序的运行来了解条件变量的作用。

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

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;/*初始化互斥锁*/
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;/*初始化条件变量*/

void *thread1(void *);
void *thread2(void *);

int i=1;
int main(void)
{
    pthread_t t_a;
    pthread_t t_b;

    pthread_create(&t_a,NULL,thread2,(void *)NULL);/*创建进程t_a*/
    pthread_create(&t_b,NULL,thread1,(void *)NULL); /*创建进程t_b*/
    pthread_join(t_b, NULL);/*等待进程t_b结束*/
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
    exit(0);
}

void *thread1(void *junk)
{
    for(i=1;i<=9;i++)
    {
        pthread_mutex_lock(&mutex);/*锁住互斥量*/
        if(i%3==0)
             pthread_cond_signal(&cond);/*条件改变，发送信号，通知t_b进程*/
        else       
             printf("thead1:%d/n",i);
        pthread_mutex_unlock(&mutex);/*解锁互斥量*/

sleep(1);
}

}

void *thread2(void *junk)
{
    while(i<9)
    {
        pthread_mutex_lock(&mutex);

if(i%3!=0)
            pthread_cond_wait(&cond,&mutex);/*等待*/
        printf("thread2:%d/n",i);
        pthread_mutex_unlock(&mutex);

sleep(1);
}

}

程序创建了2个新线程使他们同步运行，实现进程t_b打印20以内3的倍数，t_a打印其他的数，程序开始线程t_b不满足条件等待，线程t_a运行使a循环加1并打印。直到i为3的倍数时，线程t_a发送信号通知进程t_b，这时t_b满足条件，打印i值。

下面是运行结果：

#cc –lpthread –o cond cond.c

#./cond

thread1:1

thread1:2

thread2:3

thread1:4

thread1:5

thread2:6

thread1:7

thread1:8

thread2:9

备注：

 pthread_cond_wait 执行的流程首先将这个mutex解锁, 然后等待条件变量被唤醒, 如果没有被唤醒, 该线程将一直休眠, 也就是说, 该线程将一直阻塞在这个pthread_cond_wait调用中, 而当此线程被唤醒时, 将自动将这个mutex加锁，然后再进行条件变量判断（原因是“惊群效应”，如果是多个线程都在等待这个条件，而同时只能有一个线程进行处理，此时就必须要再次条件判断，以使只有一个线程进入临界区处理。），如果满足，则线程继续执行，最后解锁，

也就是说pthread_cond_wait实际上可以看作是以下几个动作的合体:
解锁线程锁
等待线程唤醒，并且条件为true
加锁线程锁.

 pthread_cond_signal仅仅负责唤醒正在阻塞在同一条件变量上的一个线程，如果存在多个线程，系统自动根据调度策略决定唤醒其中的一个线程，在多处理器上，该函数是可能同时唤醒多个线程，同时该函数与锁操作无关，解锁是由pthread_mutex_unlock(&mutex)完成

唤醒丢失问题
在线程并没有阻塞在条件变量上时，调用pthread_cond_signal或pthread_cond_broadcast函数可能会引起唤醒丢失问题。

唤醒丢失往往会在下面的情况下发生：

一个线程调用pthread_cond_signal或pthread_cond_broadcast函数； 
另一个线程正处在测试条件变量和调用pthread_cond_wait函数之间； 
没有线程正在处在阻塞等待的状态下

由来：

最近一直在想怎么高效率的在IO线程接收到数据时通知逻辑线程（基于线程池）工作的问题，像网络编程的服务器模型的一些模型都需要用到这个实现，下面我这里简单的罗列一个多线程的网络服务器模型

 

半同步/半异步（half-sync/half-async）： 

许多餐厅使用 半同步/半异步 模式的变体。例如，餐厅常常雇佣一个领班负责迎接顾客，并在餐厅繁忙时留意给顾客安排桌位，为等待就餐的顾客按序排队是必要的。领班由所有顾客“共享”，不能被任何特定顾客占用太多时间。当顾客在一张桌子入坐后，有一个侍应生专门为这张桌子服务。 

 

对于上面罗列的这种模型，本文讨论的问题是当领班接到客人时，如何高效率的通知侍应生去服务顾客. 

在我们使用很广泛的线程池实现中，也会有一样的问题

 

方法实现：

1.使用锁+轮询

使用这种方法可以很简单的实现，但是会有一定的性能消耗，其还有一个点要好好把握，就是一次轮询没有结果后相隔多久进行下一次的轮询，间隔时间太短，消耗的CPU资源较多，间隔时间太长，不能很及时的响应请求。这就相当于上面的这个例子，侍应生时不时的取询问领班有没有顾客到来

 

2.使用条件变量的线程同步

线程条件变量pthread_cond_t

线程等待某个条件

int pthread_cond_timedwait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex,const struct timespec *restrict abstime);
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
通知函数

通知所有的线程

int pthread_cond_broadcast(pthread_cond_t *cond); 

只通知一个线程
int pthread_cond_signal(pthread_cond_t *cond); 

 

正确的使用方法

 　　pthread_cond_wait用法：

pthread_mutex_lock(&mutex);

while(condition_is_false)

  {

 pthread_cond_wait(&cond,&mutex);

 }

condition_is_false=true;  //此操作是带锁的，也就是说只有一个线程同时进入这块

pthread_mutex_unlock(&mutex);

 

pthread_cond_signal用法： 

pthread_mutex_lock(&mutex);

condition_is_false=false;

pthread_cond_signal(&cond)

pthread_mutex_unlock(&mutex)

我刚初用的时候，觉得非常的奇怪，为什么要这样用，加了mutex后还需要一个condition_is_false变量来表示有没有活干。其实这样子的一个操作主要是为了解决“假激活”问题，因为我么您这里的使用场景，只需要激活一个线程，因为一个线程干一个活，而不是多个线程干一个活，所以为了避免线程被激活了，但实际又没有事情干，所以使用了这么一套机制。

实际上，信号和pthread_cond_broadcast是两个常见的导致假唤醒的情况。假如条件变量上有多个线程在等待,pthread_cond_broadcast会唤醒所有的等待线程，而pthread_cond_signal只会唤醒其中一个等待线程。这样,pthread_cond_broadcast的情况也许要在pthread_cond_wait前使用while循环来检查条件变量。 

 

来个例子：
复制代码

 1 #include <pthread.h>
 2 #include <stdio.h>
 3 #include<stdlib.h>
 4 #include<unistd.h>
 5 
 6 /* For safe condition variable usage, must use a boolean predicate and  */
 7 /* a mutex with the condition.                                          */
 8 int                 workToDo = 0;
 9 pthread_cond_t      cond  = PTHREAD_COND_INITIALIZER;
10 pthread_mutex_t     mutex = PTHREAD_MUTEX_INITIALIZER;
11 
12 #define NTHREADS      20
13 
14  static void checkResults(char *string, int rc) {
15    if (rc) {
16      printf("Error on : %s, rc=%d",
17             string, rc);
18      exit(EXIT_FAILURE);
19    }
20    return;
21 }
22 
23 void *threadfunc(void *parm)
24 {
25   int           rc;
26 
27   while (1) {
28     /* Usually worker threads will loop on these operations */
29     rc = pthread_mutex_lock(&mutex);
30     checkResults("pthread_mutex_lock()
", rc);
31 
32     while (!workToDo) {
33       printf("Thread blocked
");
34       rc = pthread_cond_wait(&cond, &mutex);
35       checkResults("pthread_cond_wait()
", rc);
36     }
37     printf("Thread awake, finish work!
");
38     sleep(2);
39     /* Under protection of the lock, complete or remove the work     */
40     /* from whatever worker queue we have. Here it is simply a flag  */
41     workToDo = 0;
42     printf("In  mutex lock
");
43     rc = pthread_mutex_unlock(&mutex);
44     sleep(2);
45     printf("Out mutex lock
");
46     checkResults("pthread_mutex_lock()
", rc);
47   }
48   return NULL;
49 }
50 
51 int main(int argc, char **argv)
52 {
53   int                   rc=0;
54   int                   i;
55   pthread_t             threadid[NTHREADS];
56 
57   printf("Enter Testcase - %s
", argv[0]);
58 
59   printf("Create %d threads
", NTHREADS);
60   for(i=0; i<NTHREADS; ++i) {
61     rc = pthread_create(&threadid[i], NULL, threadfunc, NULL);
62     checkResults("pthread_create()
", rc);
63   }
64 
65   sleep(5);  /* Sleep is not a very robust way to serialize threads   */
66 
67   for(i=0; i<5; ++i) {
68     printf("Wake up a worker, work to do...
");
69 
70     rc = pthread_mutex_lock(&mutex);
71     checkResults("pthread_mutex_lock()
", rc);
72 
73     /* In the real world, all the threads might be busy, and        */
74     /* we would add work to a queue instead of simply using a flag  */
75     /* In that case the boolean predicate might be some boolean     */
76     /* statement like: if (the-queue-contains-work)                 */
77     if (workToDo) {
78        printf("Work already present, likely threads are busy
");
79     }
80     workToDo = 1;
81     rc = pthread_cond_broadcast(&cond);
82    // rc = pthread_cond_signal(&cond);
83     checkResults("pthread_cond_broadcast()
", rc);
84 
85     rc = pthread_mutex_unlock(&mutex);
86     checkResults("pthread_mutex_unlock()
", rc);
87     sleep(5);  /* Sleep is not a very robust way to serialize threads */
88   }
89 
90   printf("Main completed
");
91   exit(0);
92   return 0;
93 }

复制代码

事实上上面的例子无论是使用pthread_cond_signal还是pthread_cond_broadcast，都只会打印Thread awake, finish work5次，大家可能会觉得非常奇怪，但是实际情况就是这样的。 为了明白其pthread_cont_wait内部干了什么工作，有必要深入一下其内部实现。

 

关于其内部实现伪代码如下：
复制代码

 1 pthread_cond_wait(mutex, cond):
 2     value = cond->value; /* 1 */
 3     pthread_mutex_unlock(mutex); /* 2 */
 4     pthread_mutex_lock(cond->mutex); /* 10 */    pthread_cond_t自带一个mutex来互斥对waiter等待链表的操作
 5     if (value == cond->value) { /* 11 */    检查一次是不是cond有被其他线程设置过，相当于单例模式的第二次检测是否为NULL
 6         me->next_cond = cond->waiter;
 7         cond->waiter = me;//链表操作
 8         pthread_mutex_unlock(cond->mutex);
 9         unable_to_run(me);
10     } else
11         pthread_mutex_unlock(cond->mutex); /* 12 */
12     pthread_mutex_lock(mutex); /* 13 */    
13     
14 pthread_cond_signal(cond):
15     pthread_mutex_lock(cond->mutex); /* 3 */
16     cond->value++; /* 4 */
17     if (cond->waiter) { /* 5 */
18         sleeper = cond->waiter; /* 6 */
19         cond->waiter = sleeper->next_cond; /* 7 */        //链表操作
20         able_to_run(sleeper); /* 8 */    运行sleep的线程，即上面的me
21     }
22     pthread_mutex_unlock(cond->mutex); /* 9 */

复制代码

pthread_cond_broadcast虽然能够激活所有的线程，但是激活之后会有mutex锁，也就是说他的激活是顺序进行的，只有第一个激活的线程调用pthread_mutex_unlock(&mutex)后，后一个等待的线程才会继续运行.因为从pthread_cond_wait(&cond,&mutex)到pthread_mutex_unlock(&mutex)区间是加的独占锁，从wait激活后的第一个线程占用了这个锁，所以其他的线程不能运行，只能等待。所以当第一个被激活的线程修改了condition_is_false后（上面测试代码的workToDo），接着调用pthread_mutex_unlock(&mutex)后，此时其他等待在cond的一个线程会激活，但是此时condition_is_false已经被设置，所以他跑不出while循环，当调用pthread_cond_wait时，其内部pthread_mutex_unlock(mutex)调用会导致另一个在它后面的等待在cond的线程被激活。　　

 

所以，通过这种方式，即便是误调用了pthread_cond_broadcast或者由于信号中断的原因激活了所有在等待条件的线程，也能保证其结果是正确的。

另外说一句题外话，很多人写的基于条件变量线程同步的框架，说自己是无锁的，其实这是不对的，只是内部锁的机制在pthread_cond_wait实现了而已，其还是基于互斥锁的实现。真正想要达到无锁的可以关注一下lockfree相关的CAS算法，其内部使用一个intel CPU的cmpxchg8指令完成的，其实这种实现个人认为和传统锁相比只是一个非阻塞锁和阻塞锁的区别。

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

// for safe condition variable usage, must use a boolean predicate and
// a mutex with the condition

int work_to_do = 0;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

#define NUM_THREADS 20

static void check_result(char *str, int rc) {
    if (rc) {
        printf("Error on: %s, rc=%d", str, rc);
        exit(EXIT_FAILURE); 
    }
    return;
}

void *threadfunc(void *param) {
    int rc;
    while (1) {
        //usually woker threads will loop on these operations
        rc = pthread_mutex_lock(&mutex);
        check_result("pthread_mutex_lock()
", rc);

        while (!work_to_do) {
            printf("thread hanging
");
            rc = pthread_cond_wait(&cond, &mutex);
            check_result("pthread_cond_wait()
", rc);
        }
        printf("thread awake, finish work!
");
        sleep(2);
        // under protection of the lock, complete or remove the work
        // from whatever worker queue we have, here it is simply a flag
        work_to_do = 0;
        printf("in mutex lock
");
        rc = pthread_mutex_unlock(&mutex);
        sleep(2);
        printf("out mutex lock
");
        check_result("pthread_mutex_unlock()
", rc);
    }
    return NULL;
}

int main(void) {
    int rc = 0;
    int i;
    pthread_t threadid[NUM_THREADS];
    
    printf("Enter %d threads
", NUM_THREADS);
    for (i=0; i<NUM_THREADS; ++i) {
        rc = pthread_create(&threadid[i], NULL, threadfunc, NULL);
        check_result("pthread_create()
", rc);
    }

    sleep(5); //sleep is not a very robust way to serialize threads

    for(i=0; i<5; ++i) {
        printf("wake up a worker, work to do...
");

        rc = pthread_mutex_lock(&mutex);
        check_result("pthread_mutex_lock()
", rc);
        
        // in the real world, all the threads might be busy, and
        // we would add work to a queue instead of simply using a flag
        // in that case the boolean predicate might be some boolean
        // statement like: if (the-queue-contains-work)

        if(work_to_do) {
            printf("work already present, likely threads are busy
");
        }
        work_to_do = 1;
        //rc = pthread_cond_broadcast(&cond);
        rc = pthread_cond_signal(&cond);
        check_result("pthread_cond_broadcast()
", rc);
        
        rc = pthread_mutex_unlock(&mutex);
        check_result("pthread_mutex_unlock()
", rc);
        sleep(5); // sleep is not a very robust way to serialize threads
    }

    printf("mian complete
");
    exit(0);
    return 0;
}