Linux多线程编程- 条件变量（Conditional variable）

基本概念

条件变量是用于线程同步的另一种工具，特别是当线程需要等待某个条件成立（或者说满足）时。条件变量常与互斥锁（mutex）配合使用，以安全地检查和等待条件的变化。

1. 目的：条件变量的主要目的是允许一个或多个线程在某些条件成立之前等待。一旦条件成立，一个或多个正在等待的线程可以被唤醒。

2. 与互斥锁的关系：虽然条件变量与互斥锁都是同步工具，但它们的用途是不同的。互斥锁主要用于保护对共享数据的访问，而条件变量则用于等待某个条件成立。在检查和修改条件时，通常需要使用互斥锁来保证操作的原子性。

3. 核心操作：

   • 等待：线程通过调用pthread_cond_wait来等待某个条件变量。这个函数会释放与条件变量关联的互斥锁，并让调用线程睡眠，直到另一个线程发出唤醒信号。
   • 唤醒：线程可以使用pthread_cond_signal来唤醒一个等待的线程，或使用pthread_cond_broadcast来唤醒所有等待的线程。

4. 使用模式：通常，线程在一个循环中检查条件，如下：

   pthread_mutex_lock(&mutex);
   while (condition_is_not_met()) {
       pthread_cond_wait(&cond_var, &mutex);
   }
   // do something when condition is met
   pthread_mutex_unlock(&mutex);
   1
   2
   3
   4
   5
   6

5. 注意事项：

   • 使用条件变量前，需要先初始化，可以使用pthread_cond_init。
   • 不再使用时，应使用pthread_cond_destroy销毁条件变量。
   • 虽然pthread_cond_wait在等待时会释放互斥锁，但在被唤醒并从pthread_cond_wait返回时，它会再次获得该锁。
   • 由于存在所谓的"虚假唤醒"（spurious wakeup），线程在被唤醒后应该再次检查条件是否真的满足，这也是为什么通常在while循环中检查条件。

总的来说，条件变量提供了一种方式，允许线程在某些条件成立前休眠，并在条件成立后被唤醒，从而实现复杂的同步需求。

常用API

下面我们详细介绍pthread库中条件变量的相关函数pthread_cond_wait，pthread_cond_signal和pthread_cond_broadcast。

pthread_cond_wait

原型：

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
1

功能：
该函数使调用线程在指定的条件变量cond上等待。为了等待条件变量，线程必须先获取与条件变量相关联的互斥锁mutex。当线程调用pthread_cond_wait后，它会自动释放这个mutex，并将自己置于条件变量的等待队列中。当条件变量被signal或广播时，线程被唤醒并重新尝试获取mutex。一旦成功获取，pthread_cond_wait返回，并且线程可以继续执行。

注意事项：

1. 在调用pthread_cond_wait之前，线程必须持有互斥锁mutex。
2. 当pthread_cond_wait返回时，线程将重新获得互斥锁，因此需要在返回后释放互斥锁。
3. 由于存在“伪唤醒”，即使没有明确的pthread_cond_signal或pthread_cond_broadcast调用，pthread_cond_wait也可能返回（这种行为可能是由于多种原因引起的，包括操作系统的干扰、系统中断或其他未明确指定的原因）。因此，通常建议在循环中检查等待的条件。

一个简化版的pthread_cond_wait实现。（注：这只是为了说明其基本工作原理，并不代表实际的pthread库中的实现）

typedef struct {
    // 链表或队列，用于存放等待此条件变量的线程
    // 这只是为了概念化，实际实现可能会更复杂
    ThreadQueue queue;  
    
    // ...其他可能的字段
} pthread_cond_t;

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex) {
    // 将当前线程加入到等待队列中
    add_thread_to_queue(&cond->queue, current_thread());

    // 释放提供的互斥锁，使其他线程有机会进入临界区
    pthread_mutex_unlock(mutex);
    
    // 挂起当前线程，直到它被pthread_cond_signal唤醒
    block_current_thread();

    // 当线程被唤醒并重新调度运行时，尝试重新获得互斥锁
    pthread_mutex_lock(mutex);
    
    return 0;
}

int pthread_cond_signal(pthread_cond_t *cond) {
    // 从等待队列中选择一个线程并唤醒它
    // 如何选择线程（例如FIFO、LIFO或优先级）取决于具体实现
    Thread *t = remove_thread_from_queue(&cond->queue);
    if (t) {
        wake_up_thread(t);
    }
    return 0;
}
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
33

这里的实现使用了假设的ThreadQueue数据结构和与之相关的操作，如add_thread_to_queue, block_current_thread, 和wake_up_thread。这些都是为了说明目的而提出的，并不真实存在。真正的pthread_cond_wait实现可能会涉及许多与操作系统和硬件相关的细节。（请注意，这只是一个非常简化的模型，实际的实现要比这复杂得多，并且会因平台和实现而异）

pthread_cond_signal

原型：

int pthread_cond_signal(pthread_cond_t *cond);
1

功能：
该函数唤醒在指定条件变量cond上等待的一个线程。如果有多个线程在等待，选择哪个线程被唤醒是不确定的。

注意事项：

1. 调用pthread_cond_signal并不意味着与之相关的互斥锁mutex会被自动释放。信号仅仅表示等待条件的线程可以被唤醒。
2. 如果没有线程在条件变量上等待，调用pthread_cond_signal不会有任何副作用。也就是说，没有“积累”的效应；如果之后有线程开始等待，它不会因为之前的pthread_cond_signal调用而被立即唤醒。

总结：
在多线程环境中，条件变量提供了一种方式，使得一个线程可以等待某个特定条件成为真，而另一个线程在该条件成为真时可以通知等待线程。pthread_cond_wait用于等待条件，而pthread_cond_signal用于通知条件已经满足。使用这两个函数，线程可以在必要时进行精细的同步，从而确保正确的操作顺序和数据完整性。

一个简化的pthread_cond_signal的实现。（注：这只是为了说明其基本工作原理，并不代表实际的pthread库中的实现）

typedef struct {
    QueueType waiting_threads;  // 这是一个伪代码类型，代表等待该条件的线程队列
} pthread_cond_t;

int pthread_cond_signal(pthread_cond_t *cond) {
    if (isEmptyQueue(&cond->waiting_threads)) {
        // 如果没有线程在等待，则无需执行任何操作
        return 0;
    }

    // 从队列中取出一个线程，并将其标记为“可运行”或“唤醒”
    ThreadType* thread = dequeue(&cond->waiting_threads);
    markThreadAsRunnable(thread);  // 这是一个伪代码函数，代表将线程状态设置为可运行

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

这只是一个非常简化的版本。在实际的pthread库中，有许多其他的细节需要考虑，例如错误处理、多核心和多处理器调度、以及与其他同步原语的交互。

重要的是理解pthread_cond_signal的核心目的：从等待条件的线程队列中选择一个线程并唤醒它，使其可以再次运行。如果没有线程正在等待，那么pthread_cond_signal基本上不做任何事情。

pthread_cond_broadcast

pthread_cond_broadcast 是 Pthreads 库中的一个函数，用于操纵条件变量。条件变量通常与互斥锁一起使用，以实现线程间的同步。

功能与目的：
pthread_cond_broadcast 用于唤醒所有等待特定条件变量的线程。这与 pthread_cond_signal 不同，后者只唤醒一个等待该条件的线程。

原型：

int pthread_cond_broadcast(pthread_cond_t *cond);
1

参数：

• cond: 是指向目标条件变量的指针。

返回值：

• 如果成功，该函数返回0。
• 如果失败，它会返回一个非零的错误代码。

使用场景：
在生产者-消费者问题中，其中多个消费者线程等待数据可用。当生产者生产了一批数据时，而不仅仅是一个数据项，希望唤醒所有等待的消费者线程，而不仅仅是一个。这时，可以使用 pthread_cond_broadcast。

使用注意事项：

• 在调用 pthread_cond_broadcast（或任何与条件变量相关的函数）之前，通常需要持有与该条件变量相关联的互斥锁。
• 虽然 pthread_cond_broadcast 会唤醒所有等待的线程，但这并不意味着所有这些线程都会立即开始执行。哪个线程首先获得执行权取决于线程调度和优先级等因素。
• 使用 pthread_cond_broadcast 而不是 pthread_cond_signal 可能会导致更高的上下文切换开销，因为它可能唤醒多个线程。因此，只有在确实需要唤醒所有线程的情况下，才应使用它。

总的来说，pthread_cond_broadcast 是 Pthreads 同步机制的重要部分，用于在多线程环境中实现复杂的同步逻辑。正确使用它可以帮助避免死锁和竞态条件，从而提高程序的健壮性和性能。

一个简化的pthread_cond_broadcast的实现。（注：这只是为了说明其基本工作原理，并不代表实际的pthread库中的实现）

为了简化，我们假设存在一个可以管理线程的队列结构，以及一些虚构的函数如dequeue()和make_thread_ready()。

typedef struct {
    // ... 其他属性 ...

    // 等待此条件变量的线程队列
    Queue waiting_threads; 
} pthread_cond_t;

int pthread_cond_broadcast(pthread_cond_t *cond) {
    // 注意：这里我们并没有涉及互斥锁，因为外部调用者在调用此函数时
    // 应已经持有了与此条件变量相关的互斥锁

    while (!is_empty(&cond->waiting_threads)) {
        // 从队列中取出一个线程
        Thread *t = dequeue(&cond->waiting_threads);

        // 将该线程设置为就绪状态，以便调度器稍后可以重新调度它
        make_thread_ready(t);
    }

    return 0; // 返回成功
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

在这个模型中，线程使用pthread_cond_wait进入等待队列。当调用pthread_cond_broadcast时，所有在此队列中等待的线程都会被唤醒。

这是一个简化版，并没有考虑许多真实实现中需要处理的细节和边界情况。真实的实现可能涉及底层的系统调用、硬件指令和其他与平台相关的操作。如果想知道具体的C库如何实现pthread_cond_broadcast，建议查看其源代码。

示例1

让我们考虑一个简单的生产者-消费者问题，其中生产者线程生成数据并将其放入缓冲区，而消费者线程从缓冲区中取出数据并处理它。为了同步两个线程，我们将使用条件变量和互斥锁。

在此示例中，缓冲区可以存储一个整数。生产者生成一个整数并将其放入缓冲区，而消费者从缓冲区中读取并打印该整数。当缓冲区为空时，消费者会等待直到生产者生产一个整数。同样，当缓冲区满时，生产者会等待直到消费者消费一个整数。条件变量data_available用于同步生产者和消费者线程，确保每次只有一个整数在缓冲区中，并且每个整数都被消费一次。

#include 
#include 

#define EMPTY -1

int buffer = EMPTY;  // 缓冲区
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;  // 用于保护对缓冲区的访问
pthread_cond_t data_available = PTHREAD_COND_INITIALIZER;  // 表示缓冲区是否有数据

void* producer(void* arg) {
    for (int i = 0; i < 10; i++) {
        pthread_mutex_lock(&mutex);
        
        while (buffer != EMPTY) {
            pthread_cond_wait(&data_available, &mutex);
        }
        
        buffer = i;  // 放数据到缓冲区
        printf("Produced %d\n", i);
        
        pthread_cond_signal(&data_available);  // 唤醒消费者
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

void* consumer(void* arg) {
    for (int i = 0; i < 10; i++) {
        pthread_mutex_lock(&mutex);
        
        while (buffer == EMPTY) {
            pthread_cond_wait(&data_available, &mutex);
        }
        
        printf("Consumed %d\n", buffer);
        buffer = EMPTY;
        
        pthread_cond_signal(&data_available);  // 唤醒生产者
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

int main() {
    pthread_t prod, cons;
    pthread_create(&prod, NULL, producer, NULL);
    pthread_create(&cons, NULL, consumer, NULL);
    
    pthread_join(prod, NULL);
    pthread_join(cons, NULL);
    
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&data_available);
    return 0;
}
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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

运行结果如下：

Produced 0
Consumed 0
Produced 1
Consumed 1
Produced 2
Consumed 2
Produced 3
Consumed 3
Produced 4
Consumed 4
Produced 5
Consumed 5
Produced 6
Consumed 6
Produced 7
Consumed 7
Produced 8
Consumed 8
Produced 9
Consumed 9
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

对上述程序中的互斥锁和条件变量做以下分析：

在上述代码中，目的是确保“生产者”不会在“缓冲区”已满时继续“生产”数据。这需要一个同步机制来告知生产者何时停下来等待，以及何时可以继续生产。

当生产者线程进入for循环并尝试“生产”数据之前，它需要确保没有其他线程（可能还有其他生产者或消费者）正在与缓冲区交互。这就是为什么我们在for循环开始时调用pthread_mutex_lock(&mutex);的原因。

但是，当生产者线程在缓冲区已满的情况下等待其他线程（如消费者）来消费数据时，它不能一直持有互斥锁。这样做会阻止消费者线程访问缓冲区，从而导致死锁。

这就是pthread_cond_wait(&data_available, &mutex);的作用所在。当调用此函数时，互斥锁mutex会被自动释放，允许其他线程（如消费者）获取该锁并访问缓冲区。当条件变量data_available被signal时，原来的线程会醒来并再次尝试获取互斥锁。只有当该线程重新获得互斥锁时，pthread_cond_wait才会返回。

因此，尽管pthread_cond_wait在其内部释放并重新获得互斥锁，但在调用它之前和之后仍然需要明确地获取和释放锁，以确保在检查条件（例如buffer != EMPTY）和执行与缓冲区相关的其他操作时的线程安全性。

总之，互斥锁和条件变量通常结合使用，以提供两个层次的同步：一是确保一次只有一个线程访问共享数据，二是使线程能够在特定条件下等待或被唤醒。

示例2

以下是一个例子，其中多个读线程等待数据可用，而写线程在数据写入完毕后使用pthread_cond_broadcast来通知所有等待的读线程。

#include 
#include 
#include 
#include 

#define NUM_READERS 5

pthread_mutex_t data_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t data_cond = PTHREAD_COND_INITIALIZER;

int data_ready = 0;

void *reader(void *arg) {
    int num = *(int *)arg;

    pthread_mutex_lock(&data_mutex);
    while (!data_ready) {
        pthread_cond_wait(&data_cond, &data_mutex);
    }
    printf("Reader %d: Data is now ready!\n", num);
    pthread_mutex_unlock(&data_mutex);

    pthread_exit(0);
}

void *writer(void *arg) {
    sleep(2);  // Simulating some data generation delay

    pthread_mutex_lock(&data_mutex);
    data_ready = 1;
    printf("Writer: Data preparation complete. Broadcasting to readers...\n");
    pthread_cond_broadcast(&data_cond);
    pthread_mutex_unlock(&data_mutex);

    pthread_exit(0);
}

int main() {
    pthread_t readers[NUM_READERS];
    pthread_t write_thread;

    int ids[NUM_READERS];
    for (int i = 0; i < NUM_READERS; ++ i) {
        ids[i] = i + 1;
        pthread_create(&readers[i], NULL, reader, &ids[i]);
    }

    pthread_create(&write_thread, NULL, writer, NULL);

    for (int i = 0; i < NUM_READERS; ++ i) {
        pthread_join(write_thread, NULL);
    }

    pthread_mutex_destroy(&data_mutex);
    pthread_cond_destroy(&data_cond);

    return 0;
}
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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

运行结果如下：

Writer: Data preparation complete. Broadcasting to readers...
Reader 1: Data is now ready!
Reader 2: Data is now ready!
Reader 3: Data is now ready!
Reader 4: Data is now ready!
Reader 5: Data is now ready!
1
2
3
4
5
6

在这个例子中，当写线程完成数据准备后，它调用pthread_cond_broadcast来唤醒所有等待数据的读线程。所有的读线程都会同时收到通知，并开始处理数据。

---

注：在大多数情况下，如果只需要默认的属性，那么使用PTHREAD_MUTEX_INITIALIZER和PTHREAD_COND_INITIALIZER是初始化互斥锁和条件变量的简单方法。但是，如果需要更复杂的属性配置，例如进程间共享的互斥锁，则必须使用pthread_mutex_init()和pthread_cond_init()函数，并提供适当的属性。