第9章 线程编程（6）_线程同步3：条件变量

5.4 条件变量

5.4.1 条件变量简介

（1）互斥锁的缺点是它只有两种状态：锁定和非锁定。

（2）条件变量通过允许线程阻塞和等待另一个线程发送信号的方法弥补了互斥锁的不足。

（3）条件变量内部是一个等待队列，放置等待的线程，线程在条件变量上等待和通知。互斥锁用于pthread_cond_wait内部对“测试条件”访问控制，条件变量通常和互斥锁一起使用。

（4）条件变量允许线程等待特定条件发生，当条件不满足时，线程通常先进入阻塞状态，等待条件发生变化。一旦其它的某个线程改变了条件，可唤醒一个或多个阻塞的线程。

（5）具体的判断条件（测试条件）还需要用户给出。

（6）条件变量数据类型：pthread_cond_t

5.4.2 条件变量的操作

（1）条件变量的创建和锁毁

头文件	#include <pthread.h>
函数	int pthread_cond_init(pthread_cond_t* cond, pthread_condattr_t* attr); //初始化 int pthread_cond_destroy(pthread_cond_t* cond); //销毁锁
返回值	成功返回0，否则返回错误编号
参数	cond：条件变量 attr：条件变量属性

（2）条件变量的等待操作

头文件	#include <pthread.h>
函数	int pthread_cond_wait(pthread_cond_t* cond, pthread_mutex_t* mutex); int pthread_cond_timewait(pthread_cond_t* cond, pthread_mutex_t* mutex, const struct timespec* timeout);
返回值	成功返回0，否则返回错误编号
参数	cond：条件变量 mutex：条件变量属性 timeout：是struct timespec结构体变量。 struct timespec{ 　　time_v tv_sec;   //seconds 　　long   tv_nsec;  //nanoseconds }
备注	（1）互斥锁mutex是对条件变量cond和“测试条件”的保护 （2）线程由于调用wait函数阻塞，否则释放互斥锁。

（3）条件变量的通知操作

头文件	#include <pthread.h>
函数	int pthread_cond_signal(pthread_cond_t* cond); int pthread_cond_broadcast(pthread_cond_t* cond);
返回值	成功返回0，否则返回错误编号
参数	cond：条件变量
备注	（1）pthread_cond_signal函数通知单个线程 （2）pthread_cond_broadcast函数通知所有线程。

5.4.3 条件变量使用的注意事项

（1）pthread_cond_wait的内部实现

【附】pthread_cond_wait源码

int __pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)
{
  volatile pthread_descr self = thread_self();
  pthread_extricate_if extr;
  int already_canceled = 0;
  int spurious_wakeup_count;

  /* Check whether the mutex is locked and owned by this thread.  */
  if (mutex->__m_kind != PTHREAD_MUTEX_TIMED_NP
      && mutex->__m_kind != PTHREAD_MUTEX_ADAPTIVE_NP
      && mutex->__m_owner != self)
    return EINVAL;

  /* Set up extrication interface */
  extr.pu_object = cond;
  extr.pu_extricate_func = cond_extricate_func;

  /* Register extrication interface */
  THREAD_SETMEM(self, p_condvar_avail, 0);
  __pthread_set_own_extricate_if(self, &extr);

  /* Atomically enqueue thread for waiting, but only if it is not
     canceled. If the thread is canceled, then it will fall through the
     suspend call below, and then call pthread_exit without
     having to worry about whether it is still on the condition variable queue.
     This depends on pthread_cancel setting p_canceled before calling the
     extricate function. */

  __pthread_lock(&cond->__c_lock, self);
  if (!(THREAD_GETMEM(self, p_canceled)
      && THREAD_GETMEM(self, p_cancelstate) == PTHREAD_CANCEL_ENABLE))
    enqueue(&cond->__c_waiting, self);
  else
    already_canceled = 1;
  __pthread_unlock(&cond->__c_lock);

  if (already_canceled) {
    __pthread_set_own_extricate_if(self, 0);
    __pthread_do_exit(PTHREAD_CANCELED, CURRENT_STACK_FRAME);
  }

  pthread_mutex_unlock(mutex);

  spurious_wakeup_count = 0;
  while (1)
    {
      suspend(self);
      if (THREAD_GETMEM(self, p_condvar_avail) == 0
      && (THREAD_GETMEM(self, p_woken_by_cancel) == 0
          || THREAD_GETMEM(self, p_cancelstate) != PTHREAD_CANCEL_ENABLE))
    {
      /* Count resumes that don't belong to us. */
      spurious_wakeup_count++;
      continue;
    }
      break;
    }

  __pthread_set_own_extricate_if(self, 0);

  /* Check for cancellation again, to provide correct cancellation
     point behavior */

  if (THREAD_GETMEM(self, p_woken_by_cancel)
      && THREAD_GETMEM(self, p_cancelstate) == PTHREAD_CANCEL_ENABLE) {
    THREAD_SETMEM(self, p_woken_by_cancel, 0);
    pthread_mutex_lock(mutex);
    __pthread_do_exit(PTHREAD_CANCELED, CURRENT_STACK_FRAME);
  }

  /* Put back any resumes we caught that don't belong to us. */
  while (spurious_wakeup_count--)
    restart(self);

  pthread_mutex_lock(mutex);
  return 0;
}

 

（2）pthread_cond_wait的使用：（用于等待线程中）

pthread_mutex_t qlock;
pthread_cond_t  qready；

/************pthread_cond_wait()的使用方法**********/
pthread_mutex_lock(&qlock);    /*lock*/

//等待某资源，并以qready作为条件通知我们
while (condition == FALSE){
    pthread_cond_wait(&qready, &qlock); /*enqueue-->unlock-->wait() -->lock*/
}

//do something

pthread_mutex_unlock(&qlock); /*unlock*/
/*****************************************************/

①为什么要在判断condition前后加锁？

　　在测试condition条件和阻塞等待(wait())之间有个“时间窗口”。假设线程A判断condition为假，并刚刚过入pthread_cond_wait函数。此时切换到其他线程B，B线程将condition改变为true。当重新切换回线程A时，由于线程A不知这种情况发生。仍会加入等待队列并阻塞等待通知。从而错过这次condition被改变的通知。因此，判断condition前后加锁，可防止此类事情的发生。

②为什么用while，而不用if来测试条件？

　　从线程被唤醒到pthread_cond_wait函数return之前，由于这个阶段mutex锁处于释放状态，因此是一个“时间窗口”。当有多个线程被同时唤醒情况下（如pthread_cond_signal在多处理上或pthread_cond_broadcast都可能同时唤醒多个线程），如果其他线程的优先级较高，可能先抢到mutex并上锁，然后return出去。如果这个优先级高的线程后来又改变了”测试条件”。那么这个优先级低的线程是无法知道这一改变的。解决的方案是在return出去以后，再次判断“测试条件”，这也是使用while而不用if来判断测试条件的原因了。当然如果每次只有一个线程被唤醒，则只需用if判断就可以了。

（3）pthread_cond_signal的使用（用于激活线程中）

pthread_mutex_lock
......
//改变测试条件
......
pthread_cond_signal或pthread_cond_broadcast
pthread_mutex_unlock

【编程实验】计算1+2+3+…+100

//pthread_cal.c

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

/*一个线程负责计算，多个线程获取结果*/

typedef struct
{
    int             res; //计算结果

    int             isfinished; //测试条件
    pthread_cond_t  cond;  //条件变量
    pthread_mutex_t mutex; //互斥锁。用于与条件变量配合使用的
}Result;

//计算并将结果放置在Result中的线程函数
void*  set_fn(void* arg)
{
    Result* r = (Result*)arg;

    int i = 1;
    int sum = 0;
    for(; i<=100; i++){
        sum += i;
    }

    //将计算结果放入Result中
    r->res = sum;

    //计算完毕，通知获取结果的线程
    pthread_mutex_lock(&r->mutex);
    r->isfinished = 1; //将“测试条件”设置1.表示计算完毕
    pthread_cond_broadcast(&r->cond);
    pthread_mutex_unlock(&r->mutex);

    return (void*)0;
}

//获取结果的线程运行函数
void* get_fn(void* arg)
{
    Result* r = (Result*)arg;
    
    //对两个线程共享的“测试条件”进行保护
    pthread_mutex_lock(&r->mutex);
   
    while(!r->isfinished){
        pthread_cond_wait(&r->cond, &r->mutex); //等待计算结果
                                                //因wait期间mutex锁是被释放的。因此，即使该
                                                //线程先拿到锁，在休眠期间也会让出锁出来。
    }

    //运行到这里，说明计算结果己出来
    pthread_mutex_unlock(&r->mutex);

    //输出结果
    int res = r->res;
    printf("0x%lx get sum is %d
", pthread_self(), res);

    return (void*)0;
}

int main(void)
{
    int err = 0;
    pthread_t  cal, get1, get2;
    Result  r;
    r.isfinished = 0;
    pthread_cond_init(&r.cond, NULL);
    pthread_mutex_init(&r.mutex, NULL);

    //启动获取结果的线程1
    if((err = pthread_create(&get1, NULL, get_fn, (void*)&r)) !=0 ){
        perror("pthread create error");
    }
    //启动获取结果的线程2
    if((err = pthread_create(&get2, NULL, get_fn, (void*)&r)) !=0 ){
        perror("pthread create error");
    }

    //启动计算的线程
    if((err = pthread_create(&cal, NULL, set_fn, (void*)&r)) !=0 ){
        perror("pthread create error");
    }

    //等待子线程结束
    pthread_join(cal, NULL);
    pthread_join(get1, NULL);
    pthread_join(get2, NULL);

    //销毁条件变量和互斥锁
    pthread_cond_destroy(&r.cond);
    pthread_mutex_destroy(&r.mutex);

    return 0;
}
/*输出结果
 0xb6df1b70 get sum is 5050
 0xb77f2b70 get sum is 5050
 */

【编程实验】读者、写者问题

//reader_writer.c

 

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

/*一个读者、一个写者的多线程同步问题*/

#define DATANUM 10 //产生DATANUM个数据

//共享资源
typedef struct
{
    int    value; //读写的数据

    //用于读数据的同步
    pthread_cond_t   readCond;
    pthread_mutex_t  readMutex;
    int              isReadFinished;

    //用于写数据的同步
    pthread_cond_t   writeCond;
    pthread_mutex_t  writeMutex;
    int              isWriteFinished;
}Storage;

//将数据写入共享资源
void   set_data(Storage* s, int value)
{
    s->value = value;
}
//从共享资源中读取数据
int get_data(Storage* s)
{    
    return s->value;
}

//写者线程
void*  set_th(void* arg)
{
    Storage* s = (Storage*)arg;
    int i=1, data = 0;
    for(; i<=DATANUM; i++){
        data = i + 100;
        set_data(s, data);
        printf("0x%lx(%-3d) write data:%d
", pthread_self(), i, data);

        //写完一个数据，通知读者线程
        pthread_mutex_lock(&s->writeMutex);
        s->isWriteFinished = 1;
        pthread_cond_broadcast(&s->writeCond);
        pthread_mutex_unlock(&s->writeMutex);

        //等待读者线程读取完数据的通知
        pthread_mutex_lock(&s->readMutex);
        while(!s->isReadFinished){
            pthread_cond_wait(&s->readCond, &s->readMutex);
        }
        s->isReadFinished = 0; //为读者线程的下一次读取做准备
        pthread_mutex_unlock(&s->readMutex);
    }
    return (void*)0;
}

//读者线程
void* get_th(void* arg)
{
    Storage* s = (Storage*)arg;
    int i = 1;
    for(; i<=DATANUM; i++){
        //等待写者线程写入数据完毕
        pthread_mutex_lock(&s->writeMutex);
        while(!s->isWriteFinished){
            pthread_cond_wait(&s->writeCond, &s->writeMutex);
        }

        //接到写者的通知，可以获取数据
        printf("0x%lx(%-3d) read data: %d
", pthread_self(), i, get_data(s));

        s->isWriteFinished = 0; //为写者下一次写入数据作准备
        pthread_mutex_unlock(&s->writeMutex);
        
        //通知写者可以继续写入数据
        pthread_mutex_lock(&s->readMutex);
        s->isReadFinished = 1;
        pthread_cond_broadcast(&s->readCond);
        pthread_mutex_unlock(&s->readMutex);
    }
 
    return (void*)0;
}

int main(void)
{
    int err = 0;
    pthread_t rth, wth;

    Storage s;
    s.isReadFinished = 0;
    s.isWriteFinished = 0;
    pthread_mutex_init(&s.readMutex, NULL);
    pthread_mutex_init(&s.writeMutex, NULL);
    pthread_cond_init(&s.readCond, NULL);
    pthread_cond_init(&s.writeCond, NULL);
    
    //创建读写进程
    if((err = pthread_create(&rth, NULL, get_th, (void*)&s)) != 0){
        perror("pthread create error");
    }
    if((err = pthread_create(&wth, NULL, set_th, (void*)&s)) != 0){
        perror("pthread create error");
    }
   
    //等待子线程结果
    pthread_join(rth, NULL);
    pthread_join(wth, NULL);

    pthread_mutex_destroy(&s.readMutex);
    pthread_mutex_destroy(&s.writeMutex);
    pthread_cond_destroy(&s.readCond);
    pthread_cond_destroy(&s.writeCond);    

    return 0;
}
/*输出结果
 0xb6dcfb70(1  ) write data:101
 0xb77d0b70(1  ) read data: 101
 0xb6dcfb70(2  ) write data:102
 0xb77d0b70(2  ) read data: 102
 0xb6dcfb70(3  ) write data:103
 0xb77d0b70(3  ) read data: 103
 0xb6dcfb70(4  ) write data:104
 0xb77d0b70(4  ) read data: 104
 0xb6dcfb70(5  ) write data:105
 0xb77d0b70(5  ) read data: 105
 0xb6dcfb70(6  ) write data:106
 0xb77d0b70(6  ) read data: 106
 0xb6dcfb70(7  ) write data:107
 0xb77d0b70(7  ) read data: 107
 0xb6dcfb70(8  ) write data:108
 0xb77d0b70(8  ) read data: 108
 0xb6dcfb70(9  ) write data:109
 0xb77d0b70(9  ) read data: 109
 0xb6dcfb70(10 ) write data:110
 0xb77d0b70(10 ) read data: 110
*/