linux 条件变量与线程池

条件变量Condition Variables

概述

1. 条件变量提供了另外一种线程同步的方式。如果没有条件变量，程序需要使用线程连续轮询（可能在临界区critical section内）方式检查条件是否满足。由于线程连续忙于轮询检查，这会非常消耗资源，而条件变量是一种实现同样目标不需要轮询的方式。

2. 条件变量总是和互斥锁相结合使用。

3. 条件变量使用示例结构：

Main Thread

  • Declare and initialize global data/variables which require synchronization (such as "count")
  • Declare and initialize a condition variable object
  • Declare and initialize an associated mutex
  • Create threads A and B to do work

Thread A

  • Do work up to the point where a certain condition must occur (such as "count" must reach a specified value)
  • Lock associated mutex and check value of a global variable
  • Call pthread_cond_wait() to perform a blocking wait for signal from Thread-B. Note that a call to pthread_cond_wait() automatically and atomically unlocks the associated mutex variable so that it can be used by Thread-B.
  • When signalled, wake up. Mutex is automatically and atomically locked.
  • Explicitly unlock mutex
  • Continue

Thread B

  • Do work
  • Lock associated mutex
  • Change the value of the global variable that Thread-A is waiting upon.
  • Check value of the global Thread-A wait variable. If it fulfills the desired condition, signal Thread-A.
  • Unlock mutex.
  • Continue

Main Thread

Join / Continue

 

创建和销毁条件变量

pthread_cond_init (condition,attr)

pthread_cond_destroy (condition)

pthread_condattr_init (attr)

pthread_condattr_destroy (attr)

条件变量必须声明为pthread_cond_t，并且使用之前必须初始化。有两种方式初始化条件变量：
1）静态初始化：pthread_cond_t myconvar = PTHREAD_COND_INITIALIZER;

2）动态初始化： pthread_cond_init()。条件变量的id号通过条件变量参数返回于调用线程，这种方式允许设置条件变量的属性。然而，只有一种条件变量属性process-shared，这允许其他进程的线程可见该条件变量。如果使用条件变量属性，那么必须是pthread_condattr_t 类型（为了接受默认值可以指定为NULL）。需要注意的是，并非所有实现提供process-shared属性。

信号等待与信号通知

pthread_cond_wait (condition,mutex):阻塞调用线程直到特定的条件触发。当互斥量被锁住时该函数应当被调用；当它等待时它将自动释放互斥锁。接收到信号通知和线程被唤醒后，互斥量将自动地被线程锁住。当线程完成任务时，需要手动解锁互斥量。

pthread_cond_signal (condition)用于唤醒另外一个等待条件变量的线程。互斥量被锁住之后才可调用pthread_cond_signal并且按序解锁用于pthread_cond_wait完成。

pthread_cond_broadcast (condition)如果多于一个线程处于阻塞等待状态，那么应当使用pthread_cond_broadcast而不是pthread_cond_signal。

建议使用while循环而不是if，这样可以检查一些潜在的问题，例如：如果若干线程在等待同一个唤醒信号，它们将轮询捕获互斥量，它们中的任何一个可以修改条件；由于程序bug，线程接收到错误信号；线程库允许不违反标准的前提下虚假的唤醒一个等待线程。

使用这些函数时，必须正确地加锁解锁互斥变量。

调用pthread_cond_wait前锁定互斥量失败可能导致线程阻塞失败；

调用pthread_cond_signal后解锁互斥量失败可能不允许匹配的pthread_cond_wait完成（即阻塞掉）。

 

实际上pthread_cond_wait的返回不仅仅是pthread_cond_signal和pthread_cond_broadcast导致的，还会有一些假唤醒，也就是spurious wakeup。

pthread_cond_wait的通常使用方法：

pthread_mutex_lock();

while(condition_is_false)

    pthread_cond_wait();

pthread_mutex_unlock();

为什么在pthread_cond_wait()前要加一个while循环来判断条件是否为假呢？

 

APUE中写道:

传递给pthread_cond_wait的互斥量对条件进行保护，调用者把锁住的互斥量传给函数。函数把调用线程放到等待条件的线程列表上，然后对互斥量解锁，这两个操作是原子操作。

 

线程释放互斥量，等待其他线程发给该条件变量的信号（唤醒一个等待者）或广播该条件变量（唤醒所有等待者）。当等待条件变量时，互斥量必须始终为释放的，这样其他线程才有机会锁住互斥量，修改条件变量。当线程从条件变量等待中醒来时，它重新继续锁住互斥量，对临界资源进行处理。

 

条件变量的作用是发信号，而不是互斥。

wait前检查

对于多线程程序，不能够用常规串行的思路来思考它们，因为它们是完全异步的，会出现很多临界情况。比如：pthread_cond_signal的时间早于pthread_cond_wait的时间，这样pthread_cond_wait就会一直等下去，漏掉了之前的条件变化。

对于这种情况，解决的方法是在锁住互斥量之后和等待条件变量之前，检查条件变量是否已经发生变化。

if(condition_is_false)

    pthread_cond_wait();

 

这样在等待条件变量前检查一下条件变量的值，如果条件变量已经发生了变化，那么就没有必要进行等待了，可以直接进行处理。这种方法在并发系统中比较常见

1.等待函数里面要传入一个互斥量，这个互斥量会在这个函数调用时会发生如下变化：函数刚刚被调用时，会把这个互斥量解锁，然后让调用线程阻塞，解锁后其他线程才有机会获得这个锁。当某个线程调用通知函数时，这个函数收到通知后，又把互斥量加锁，然后继续向下操作临界区。可见这个设计是非常合理的！

 

2.条件变量的等待函数用while循环包围。原因：如果有多个线程都在等待这个条件变量关联的互斥量，当条件变量收到通知，它下一步就是要锁住这个互斥量，但在这个极小的时间差里面，其他线程抢先获取了这互斥量并进入临界区把某个状态改变了。此时这个条件变量应该继续判断别人刚刚抢先修改的状态，即继续执行while的判断。还有一个原因时防止虚假通知，收到虚假通知后，只要while里面的条件为真，就继续休眠.

参考资料：https://computing.llnl.gov/tutorials/pthreads/#ConVarSignal

http://www.cnblogs.com/leaven/archive/2010/06/03/1750973.html

https://www.cnblogs.com/yuuyuu/p/5140875.html

Linux C++线程池框架

Linux的多任务编程-线程池

 

 1 #include <pthread.h>
 2 #include <stdio.h>
 3 #include <stdlib.h>
 4 #include <unistd.h>
 5 #define NUM_THREADS  3
 6 #define TCOUNT 10
 7 #define COUNT_LIMIT 12
 8 
 9 int     count = 0;
10 int     thread_ids[3] = {0,1,2};
11 pthread_mutex_t count_mutex;
12 pthread_cond_t count_threshold_cv;
13 
14 void *inc_count(void *t)
15 {
16     int i;
17     long my_id = (long)t;
18 
19     for (i=0; i<TCOUNT; i++) {
20         pthread_mutex_lock(&count_mutex);
21         count++;
22 
23         /*
24      Check the value of count and signal waiting thread when condition is
25      reached.  Note that this occurs while mutex is locked.
26      */
27         if (count == COUNT_LIMIT) {
28             pthread_cond_signal(&count_threshold_cv);
29             printf("inc_count(): thread %ld, count = %d  Threshold reached.\n",
30                    my_id, count);
31         }
32         printf("inc_count(): thread %ld, count = %d, unlocking mutex\n",
33                my_id, count);
34         pthread_mutex_unlock(&count_mutex);
35 
36         /* Do some "work" so threads can alternate on mutex lock */
37         sleep(1);
38     }
39     pthread_exit(NULL);
40 }
41 
42 void *watch_count(void *t)
43 {
44     long my_id = (long)t;
45 
46     printf("Starting watch_count(): thread %ld\n", my_id);
47 
48     /*
49    Lock mutex and wait for signal.  Note that the pthread_cond_wait
50    routine will automatically and atomically unlock mutex while it waits.
51    Also, note that if COUNT_LIMIT is reached before this routine is run by
52    the waiting thread, the loop will be skipped to prevent pthread_cond_wait
53    from never returning.
54    */
55     pthread_mutex_lock(&count_mutex);
56     while (count<COUNT_LIMIT) {
57         pthread_cond_wait(&count_threshold_cv, &count_mutex);
58         printf("watch_count(): thread %ld Condition signal received.\n", my_id);
59     }
60     count += 125;
61     printf("watch_count(): thread %ld count now = %d.\n", my_id, count);
62     pthread_mutex_unlock(&count_mutex);
63     pthread_exit(NULL);
64 }
65 
66 int main (int argc, char *argv[])
67 {
68     int i, rc;
69     long t1=1, t2=2, t3=3;
70     pthread_t threads[3];
71     pthread_attr_t attr;
72 
73     /* Initialize mutex and condition variable objects */
74     pthread_mutex_init(&count_mutex, NULL);
75     pthread_cond_init (&count_threshold_cv, NULL);
76 
77     /* For portability, explicitly create threads in a joinable state */
78     pthread_attr_init(&attr);
79     pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
80 
81     pthread_create(&threads[0], &attr, watch_count, (void *)t1);
82     pthread_create(&threads[1], &attr, inc_count, (void *)t2);
83     pthread_create(&threads[2], &attr, inc_count, (void *)t3);
84 
85     /* Wait for all threads to complete */
86     for (i=0; i<NUM_THREADS; i++) {
87         pthread_join(threads[i], NULL);
88     }
89     printf ("Main(): Waited on %d  threads. Done.\n", NUM_THREADS);
90 
91     /* Clean up and exit */
92     pthread_attr_destroy(&attr);
93     pthread_mutex_destroy(&count_mutex);
94     pthread_cond_destroy(&count_threshold_cv);
95     pthread_exit(NULL);
96 
97 }
View Code

相关文章：

  • 2021-08-18
  • 2021-09-04
  • 2021-06-02
  • 2021-09-25
  • 2022-02-15
  • 2021-11-20
猜你喜欢
  • 2021-04-19
  • 2021-04-06
  • 2022-12-23
  • 2022-12-23
  • 2021-11-18
  • 2021-10-11
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode