线程（四）:条件变量

2021-04-24  本文已影响0人  404Not_Found
# 为何需要条件变量
# 定义
  ##  一定需要while 和 全局变量 done吗
  ## 一定需要锁吗
# 生产者和与消费者
  ## 代码分析
    ### A Bronken Solution - CV
    ### Better, But still broken , while , NOT if
    ### The Single Buffer Producer/Cosumer Solution
# 覆盖性条件

为何需要条件变量(Condition Variables)

锁并不是唯一的多线程通信的方案。在其他一些case中，比如父进程在执行后续操作之前，要检查子进程是否结束。也就是说父进程被block，子进程complete后，需要通知父进程，然后父进程才能继续运行下去。

volatile int done = 0;

void * child(void * arg) {
  printf("child\n");
  done = 1;
  return NULL:
}

int main(int argc, char ** argv) {
  printf("parent: begin\n");
  pthread_t c;
  pthread_create(&c, NULL, child, NULL); // create child
  while (done == 0)
    ; // spin
  printf("parent:end \n");
  return 0;
}

在父进程中采用自旋的方式，其实非常低效，浪费CPU周期。而且有时候正确性也不能保证。此时在这种A线程需要等待B线程的通知才能进行下去的情况，我们可以使用条件变量，condition variable.

定义

条件变量是一个队列，线程可以将他们自己放入其中，睡眠，等待条件满足被唤醒（当然被唤醒可以不止一个）。
变量类型：pthread_cond_t c
操作动作（Posix call）：**pthread_cond_wait(pthread_cond_t c, pthread_mutex_t m)
其实就是wait + signal 的操作

int done = 0;
pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t c = PTHREAD_COND_INITIALIZER;

void * child(void * arg) {
  printf("child\n");
  thr_exit();
  return NULL;
}

void thr_exit() {
  pthread_mutex_lock(&m);
  done = 1;
  pthread_cond_signal(&c);
  pthread_mutex_unlock(&m);
}

void thr_join() {
  pthread_mutex_lock(&m);
  while(done == 0)
      pthread_cond_wait(&c, &m);
  pthread_mutex_unlock(&m);
}

int main(int argc, char ** argv) {
  printf("parent: begin\n");
  pthread_t p ;
  pthread_create(&p, NULL, child, NULL
  thr_join();
  printf("parent:end\n");
  return 0;
}

wait 函数是携带一把锁的，在该线程wait之前，该线程是拿到这把锁的。调用wait时，该线程释放这把锁后，自行进入sleep队列。
如果该线程被唤醒，那么该线程一定已经再次 re-acuqire这把锁了。然后才从wait函数返回。
对于上面一段代码会有两种情况

void thr_exit() {
  pthread_mutex_lock(&m);
  pthread_cond_signal(&c);
  pthread_mutex_unlock(&m);
}

void thr_join() {
  pthread_mutex_lock(&m);
  pthread_cond_wait(&c, &m);
  pthread_mutex_unlock(&m);
}

缺陷：子线程先被调用后，无睡眠signal,该条件变量没有下挂的睡眠现成，则子线程立刻返回，父线程拿到锁，进入condi睡眠，则无人再将其唤醒。增加while 和 done 其实就是为了过滤这种情况。此时，就不用再进入condition, 让父线程睡了。

void thr_exit() {
  done = 1;
  pthread_cond_signal(&c);
}

void thr_joine() {
  if(done == 0)
    pthread_cond_wait(&c);
}

这里会有一个狡猾的race condition.
在thr_join函数中，当父线程检查了done的值后，在调用wait之前，被中断，执行了子线程。子线程中将done 置1。因为无线程sleep，直接返回
回到父线程，父线程进入睡眠，且done = 1,无人再次唤醒父进程。

条件变量中signal别人，还是自己wait 必带锁，来保护 变量 done，上述代码的c，只是下挂了一个 queue 起到通知的作用。我的理解是这样的，由程序员决定done是否满足条件，将queue里的线程唤醒。
Hold the lock when calling signal or wait

生产者与消费者问题

生产者可以多个线程，消费者也可以多个线程。我们经常用的linux grep命令也是生产者和消费者模型。

grep foo file.txt | wc -l

在linux中涉及两个进程 grep 与 wc。

  1. grep 将file.txt中含有foo字符串的行 输入到standard output，标准输出
  2. Linux 将 结果 redirect 重定向到 pipe 中
  3. 另一个进程wc 的 标准输出 standard output 对接到 pipe 中的另一端。
  4. grep 负责生产，wc 负责消费

代码分析

A Bronken Solution - CV

int buffer;
int count = 0; // initially, empty

cond_t cond;
mutex_t mutex;

void put(int value) {
  assert(count == 0);
  count = 1;
  buffer = value;
}

int get() {
  assert(count == 1);
  count = 0;
  return buffer;
}

void * producer(void *arg) {
  int i;
  for( i = 0; i < loops; i++) {
    pthread_mutex_lock(&mutex);
    if(count == 1)
         pthread_cond_wait(&cond, &mutex);
    put(i);
    pthread_cond_signal(&cond);
    pthread_mutex_unlock(&mutex);
  }
}

void * consumer(void * arg) {
  int i ;
  for(i = 0; i< loops; i++) {
    pthread_mutex_lock(&mutex);
    if(count == 0)
        pthread_cond_wait(&cond, &mutex);
    int temp = get();
    pthread_cond_signal(&cond);
    pthread_mutex_unlock(&mutex);
    printf("%d\n", tmp)
  }
}

只有1个消费者和1个生产者以上代码没有什么问题。让我们看下多个消费者的线程和一个生产这的情况。
问题：

Better, But still broken , while , NOT if

解决方式很简单，将if 换成 while

while(count == 1)
    pthread_cond_wait(&cond, &mutex);

while(count == 0)
    pthread_cond_wait(&cond, &mutex);

condition variable is to always use while loops! Sometimes you don't have to recheck the condition, buut it is always safe to do so

即使修改成while以后，依然存在bug. 假设2个Tc线程，1个Tp线程，会造成 all sleep的现象。或者Tp的生产速度很慢。注意，这里条件变量只有一个 cond

Broken Soluton 2.png

因为pthread_cond_signal唤醒的是相关条件变量cond，cond下挂的睡眠队列，谁先被唤醒，是基于这个队列的管理方式。同时，线程被唤醒，然后才是拿到锁！
which is definitely possible, depending on how the wait queue is managed.
问题的结症在于，Consumer 线程 不应该去 signal 其他 Consumer线程，只能signal Producer线程

The Single Buffer Producer/Cosumer Solution

cond_t empty, fill;
mutex_t mutex;

void * producer(void * arg) {
  int i;
  for(i = 0; i< loops; i++) {
    pthread_mutex_lock(&mutex);
    while(count == 1)
      pthread_cond_wait(&empty, &mutex);
    put(i);
    pthread_cond_signal(&fill);
    pthread_mutex_unlock(&mutex);
  }
}

void * consumer(void * arg) {
  int i ;
  for(i = 0;i < loops; i++) {
    pthread_mutex_lock(&mutex);
    while(count == 0)
        pthread_cond_wait(&fill, &mutex);
    int tmp = get();
    pthread_cond_signal(&empty);
    pthread_mutex_unlock(&mutex);
    printf("%d\n",tmp);
  }
}

代码解读：

1. 由上述分析得知，signal负责激活cond下挂的睡眠队列中的线程，而wait 负责将线程睡眠在cond下的睡眠队列
2. 锁 针对临界区而言，可以理解为线程从睡眠队列激活后，才去争抢的
3. 对于cond fill 和 cond empty的理解

覆盖性条件

其实就是一个 cond 变量中 的条件模糊，覆盖范围多个线程都满足的情况亦或者条件都不满足，wait无法被唤醒。

 cond_t c;
 mutex_t m;
void * allocated(int size) {
  pthread_mutex_lock(&m);
  while(byteleft < size)
      pthread_cond_wait(&c, &m);
  void * ptr = ....; // get mem on heap
  bytelfet -= size;
  pthread_mutex_unlock(&m);
  return ptr;
}

void free(void *ptr, int size) {
  pthread_mutex_lock(&m);
  byteleft+= size;
  pthread_cond_signal(&c);// whom to signal?
  pthread_mutex_unlock(&c);
}

比如一个线程Ta申请allocate(100), 另一个线程申请Tb allocate(10).此时可用内存为0.
那么两个线程就都睡了。
此时 Tc 线程 free(50)，但是可能没有连续的空间，Tb 依然没有被唤醒（为什么没被唤醒呢？）。Ta也肯定没有被唤醒。此时大家都在睡觉。Tc线程不知道该唤醒睡。
解决方案 pthread_cond_broadcast(),唤醒所有线程，然后枪锁，最后大部分都回去继续sleep. 对性能影响很大，造成类似惊群的效果。大量的上下文切换。
一般只有在分配内存的情况下会调用broadcast.

上一篇下一篇

猜你喜欢

热点阅读