wqllyx的网站
0%

c++ 设计线程池（2）----条件变量

发表于

锁并不是并发设计的唯一原语。

在某些情况下，线程需要检查某一条件（condition）满足之后，才会继续运行。

例如：父线程需要检查子线程是否执行完毕（jion（））。如何实现这种等待？—- 信号量

线程等待条件满足的方法

方法1： 自旋知直到条件满足

1
while (done == 0); // spin

简单的方案是自旋直到条件满足，这是极其低效的，某些情况下甚至是错误的。

方法2： 条件变量

条件变量是一个显式队列，当某些执行状态（即条件，condition）不满足时，线程可以把自己加入队列，等待（waiting）该条件。另外某个线程，当它改变了上述状态时，就可以唤醒一个或者多个等待线程（通过在该条件上发信号），让它们继续执行。

要声明这样的条件变量，只要像这样写：pthread_cond_t c;，这里声明 c 是一个条件变量（注意：还需要适当的初始化）。条件变量有两种相关操作：wait()和 signal()。线程要睡眠的时候，调用 wait()。当线程想唤醒等待在某个条件变量上的睡眠线程时，调用 signal()。具体来说，POSIX 调用如图 30.3 所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
pthread_cond_wait(pthread_cond_t *c, pthread_mutex_t *m);
pthread_cond_signal(pthread_cond_t*c);
int done = 0;
pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t c = PTHREAD_COND_INITIALIZER;
void thr_exit() {
    Pthread_mutex_lock(&m);
    done = 1;
    Pthread_cond_signal(&c);
    Pthread_mutex_unlock(&m);
}

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

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;
}

条件变量是一个队列,声明一个条件变量就是声明一个存放线程的队列。这些线程等待某些条件。

本代码中的条件就是done的值（状态）。done叫做状态变量

以上状态变量和锁，对于条件变量来说都是必不可少的。

  • 状态变量记录了所有线程都关注（所有线程都在观察着这个变量的动向）的值。睡眠唤醒锁都离不开他。
1
2
3
4
5
6
7
8
9
10
11
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);
}
// 如果缺少状态变量，假设子线程立即执行，调用exit，子线程发送信号，此时没有在条件变量(队列）上睡眠的线程。所以不会唤醒任何人，父进程在执行，就会一直等待在条件变量上，因为已经没有人可以唤醒他了。
  • 锁也是必不可少的。

保证发信号时，总是持有锁

使用条件变量的一些基本要求：

  • 有锁
  • 有等待的条件。

信号量

信号量是一个有数值的整数对象。 信号量就代表资源量。

可以用两个函数来操作他。

在 POSIX 标准中，是sem_wait()【P操作】和 sem_post()【V操作】。因为信号量的初始值能够决定其行为，所以首先要初始化信号量，才能调用其他函数与之交互，如图 31.1 所示。

1
2
3
#include <semaphore.h>
sem_t s;
sem_init(&s, 0, 1);

其中申明了一个信号量 s，通过第三个参数，将它的值初始化为 1。sem_init()的第二个参数，在我们看到的所有例子中都设置为 0，表示信号量是在同一进程的多个线程共享的。

PV操作

1
2
3
4
5
6
7
8
9
int sem_wait(sem_t *s) {
	信号量值减1;
	如果值是负值，等待。
}

int sem_post(sem_t *s) {
   信号量值加1
   如果值大于等于1，唤醒。
}
  • 首先，sem_wait()要么立刻返回（调用 sem_wait()时，信号量的值大于等于 1），要么会让调用线程挂起，直到之后的一个 post 操作。当然，
    也可能多个调用线程都调用 sem_wait()，因此都在队列中等待被唤醒。
  • 其次，sem_post()并没有等待某些条件满足。它直接增加信号量的值，如果有等待线程，唤醒其中一个。
  • 最后，当信号量的值为负数时，这个值就是等待线程的个数[D68b]。虽然这个值通常不会暴露给信号量的使用者，但这个恒定的关系值得了解，可能有助于记住信号量的工作原理。

二值信号量（锁）

使用信号量作为锁。

1
2
3
4
5
sem_t m;
sem_init(&m, 0, X); // initialize semaphore to X; what should X be?
sem_wait(&m);
// critical section here
sem_post(&m);

信号量初始值应该为1

信号量用作条件变量

信号量也可以用在一个线程暂停执行，等待某一条件成立的场景。例如，一个线程要等待一个链表非空，然后才能删除一个元素。在这种场景下，通常一个线程等待条件成立，另外一个线程修改条件并发信号给等待线程，从而唤醒等待线程。因为等待线程在等待某些条件（condition）发生变化，所以我们将信号量作为条件变量（condition variable）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
下面是一个简单例子。假设一个线程创建另外一线程，并且等待它结束（见图 31.4）。
sem_t s;
void *child(void *arg) {
    printf("child\n");
    sem_post(&s); // signal here: child is done

return NULL;
}

int main(int argc, char *argv[]) {
    sem_init(&s, 0, X); // what should X be?
    printf("parent: begin\n");
    pthread_t c;
    Pthread_create(c, NULL, child, NULL);
    sem_wait(&s); // wait here for child
    printf("parent: end\n");
    return 0;
}

如何实现信号量

最后，我们用底层的同步原语（锁和条件变量），来实现自己的信号量，名字叫作Zemaphore。这个任务相当简单，如图 31.12 所示。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// 用锁和条件变量实现 Zemahpore
typedef struct _Zem_t {
    int value;
    pthread_cond_t cond;
	pthread_mutex_t lock;
} Zem_t;

// only one thread can call this
void Zem_init(Zem_t *s, int value) {
    s->value = value;
    Cond_init(&s->cond);
    Mutex_init(&s->lock);
}

// p操作
void Zem_wait(Zem_t *s) {
	Mutex_lock(&s->lock);
	while (s->value <= 0)
		Cond_wait(&s->cond, &s->lock);
	s->value--;
	Mutex_unlock(&s->lock);
}
// V操作
void Zem_post(Zem_t *s) {
	Mutex_lock(&s->lock);
	s->value++;
	Cond_signal(&s->cond);
	Mutex_unlock(&s->lock);
}

可以看到，我们只用了一把锁、一个条件变量和一个状态的变量来记录信号量的值。

我们实现的 Zemaphore 和 Dijkstra 定义的信号量有一点细微区别，就是我们没有保持当信号量的值为负数时，让它反映出等待的线程数。事实上，该值永远不会小于 0。这一行为更容易实现，并符合现有的 Linux 实现

小心泛化

我们可以把信号量当作锁和条件变量的泛化。但这种泛化有必要吗？考虑基于信号量去实现条件变量的难度，可能这种泛化并没有你想的那么通用。

很奇怪，利用信号量来实现锁和条件变量，是棘手得多的问题。某些富有经验的并发程序员曾经在 Windows 环境下尝试过，随之而来的是很多缺陷[B04]。你自己试一下，看看是否能明白为什么使用信号量实现条件变量比看起来更困难。

参考：

​ ostep