线程（四）:条件变量

作者: 404Not_Found | 来源:发表于2021-04-24 00:45 被阅读0次

作者: 雪山肥鱼
时间：20210424 08:32
目的：整理OSTEP 的条件变量相关章节

# 为何需要条件变量
# 定义
  ##  一定需要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函数返回。
对于上面一段代码会有两种情况

情况1：
1. 线程1在create 出线程2后，cpu调度，继续执行线程1
2. 线程1进入thr_join()函数，拿到锁，检查done==0，线程1进入wait，睡眠
3. 线程2拿到锁，将done置1，通知条件变量c，唤醒线程1，释放锁
4. 线程1被唤醒，切拿到锁，此时done == 1, 则打印出 parent
情况2：
1. 线程1在create出线程2后，cpu调度，先执行了线程2
2. 线程2拿到锁，将done置1，通知条件变量c，c下但此时没有线程睡眠，直接返回，结束
3. 线程1中，dong != 0 直接打出parent:end 结束
疑问1:
一定需要全局变量done 和 while吗？以下的代码不行吗？

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, 让父线程睡了。

疑问2：
一定要锁吗？

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。

grep 将file.txt中含有foo字符串的行输入到standard output，标准输出
Linux 将结果 redirect 重定向到 pipe 中
另一个进程wc 的标准输出 standard output 对接到 pipe 中的另一端。
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个生产者以上代码没有什么问题。让我们看下多个消费者的线程和一个生产这的情况。
问题：

当Tc1 睡眠后，Tp 生产，signal condition，将Tc1 从sleeb'bp状态转换到 ready 状态。但是此时注意Tc1还没有运行。
就在这时候 Tc2 抢占了CPU，插入进来，将队列里data 偷走了。
当Tc1 再次运行起来，去get 数据，发现数据内容没了，走进了 get函数中的 assert，报错。

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

Tc1 Tc2 先于 Tp1 运行。均没有资源，进入sleep状态。Tp被调度，生产1个资源。同时signal 给 condition, Tc1 得到调用，处于ready状态（由睡眠队列的shedule进行调度，决定Tc1还是Tc2，从睡眠队列出来 Tc2 Tp 处于睡眠队列
Tc1 消费了资源后，signal（&cond），此时激活那个线程呢？Tc2, Tp? 同样此时由睡眠队列的调度决定。如果Tc2 处于ready，那么Tc2发现没有资源继续睡，那么很容易出现Tc1 Tc2 Tp三者同时sleep。同时没有 signal 函数促触发，全部睡死。

因为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. 进入producer函数，上锁，count = 0
2. 生产唤醒 cond fill 下面挂的消费者线程
  2.1 生产资源，返回。释放锁
  - 生产过程中消费者线程不可能打扰生产者进程，因为此时mutex在生产者手上
3. 此时mutex 自由
  - 被生产者抢到
    上锁，睡眠在 cond empty下的queue中，交出锁
  - 被消费者抢到
    上锁，check count ！= 0 ，消费，激活empty下的睡眠队列的生产者线程，（没有也直接返回），release lock.
4. 此时 mutex自由
  - 被生产者抢到
    则往复上面的流程
  - 被消费者抢到
    上锁，此时资源为0，睡在了cond fill下的queue中，同时交出锁，for也不执行了，因为一直睡，及时下次时间片来了，也不会去抢锁（暂时理解为上期队列锁的park()）。
消费者优先被调用
1. 进入消费者函数，上锁 cont =0
2. 交出锁，将自己挂在cond fill的睡眠队列下, 线程队列不会在出来抢锁
3. 父进程抢到锁，生产，激活fill 下的消费者线程。丢掉锁
4. 此时mutex自由
  4.1 被producer 抢到，count == 1 ，生产者线程谁在cond empty 的睡眠队列下
  4.2 被consumer抢到，recheck count != 0，消费，激活empty下的生产者线程，释放锁。

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

对于消费者来说，当cont == 0 也就是没有资源。挂在cond fill 下，等待被通知，直到有资源了，就通知这个挂在fill的消费者线程。就是在被动等（wait）资源被filled了的意思
对于生产者来说，当cont == 1 也就是说此时有资源了，挂在 cont empty下，等待通知，消费者消费了一个，通知生产者线程，你需要生产了。在被动等（wait）资源被消费(empted)没了的意思。

覆盖性条件

其实就是一个 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.

线程（四）:条件变量

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

定义

生产者与消费者问题

代码分析

A Bronken Solution - CV

Better, But still broken , while , NOT if

The Single Buffer Producer/Cosumer Solution

覆盖性条件

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读