Wiki

condition_variable

更新於 2023-10-05約 8 分鐘閱讀

本文由簡體中文內容確定性轉換，並受版本化術語表保護。

std::condition_variable 用來解決這樣一類問題：

某個條件現在還不滿足，我不想一直佔著 CPU 檢查，而是希望先睡眠，等條件可能發生變化時再醒來。

典型場景包括：

队列里暂时没有任务
→ worker 先睡眠

初始化还没完成
→ 其他线程先等待

消费者暂时没有数据
→ 等生产者放入数据

多个线程等待某个状态变为 ready
→ 状态变化后再继续

條件變數最重要的不是背：

wait()
notify_one()
notify_all()

而是理解它必須圍繞下面這套東西一起使用：

共享状态 + mutex + condition_variable + predicate

它們的分工是：

共享状态
→ 记录“事情到底有没有发生”

mutex
→ 保护共享状态的读写

condition_variable
→ 条件不满足时让线程睡眠
→ 条件可能变化时把等待线程叫醒

predicate
→ 被唤醒以后重新判断：
  “现在到底能不能继续执行？”

所以：

condition_variable 本身不儲存業務狀態，它只是一個“等待和喚醒機制”。

1.條件變數基礎

1.1.為什麼不能一直迴圈檢查

最直接的等待方法可能是：

while (!ready)
{
}

這種程式碼有兩個問題。

首先，如果：

bool ready = false;

是普通變數，而且多個執行緒同時讀寫它，那麼可能產生：

data race

即使改成：

std::atomic<bool> ready{false};

然後：

while (!ready.load())
{
}

雖然執行緒安全了，但執行緒會不斷執行：

load
load
load
load
load
...

這叫：

忙等（busy waiting）

也就是執行緒雖然什麼有用的事都沒幹，卻一直佔著 CPU 檢查條件。

如果可能要等幾毫秒、幾秒，甚至更久，這通常很浪費。

條件變數的思想是：

条件不满足
↓
线程睡眠，不占着 CPU 空转

其他线程修改状态
↓
发出通知

等待线程醒来
↓
重新检查条件

所以條件變數解決的不是：

怎么让 bool 线程安全

而是：

条件不满足时怎么高效等待

条件可能满足时怎么把线程叫醒

1.2.條件變數最基本的三個物件

通常會看到：

std::mutex mutex;
std::condition_variable cv;
bool ready = false;

這裡：

ready

是真正的共享狀態，表示：

“现在能不能继续执行？”

mutex：

std::mutex mutex;

負責保護：

ready

的讀寫。

cv：

std::condition_variable cv;

負責：

让线程睡眠
+
把线程唤醒

最重要的一點：

condition_variable 本身不是狀態。

也就是說，下面這種想法是錯的：

“我收到 notify 了，所以条件肯定成立了。”

真正決定執行緒能不能繼續的是：

ready

或者：

!queue.empty()

或者其他業務狀態。

1.3.最簡單的 wait / notify 示例

#include <condition_variable>
#include <mutex>
#include <print>
#include <thread>

std::mutex mutex;
std::condition_variable cv;

bool ready = false;

void worker()
{
    std::unique_lock lock(mutex);

    cv.wait(lock, [] {
        return ready;
    });

    std::println("worker start");
}

int main()
{
    std::thread thread(worker);

    {
        std::lock_guard lock(mutex);
        ready = true;
    }

    cv.notify_one();

    thread.join();
}

執行結果：

worker start

整個過程可以理解成：

worker 获得 mutex
        ↓
检查 ready
        ↓
发现 ready == false
        ↓
wait()：
释放 mutex
并让 worker 睡眠
        ↓
main 获得 mutex
        ↓
ready = true
        ↓
main 释放 mutex
        ↓
notify_one()
        ↓
worker 被唤醒
        ↓
worker 重新获得 mutex
        ↓
重新检查 ready
        ↓
ready == true
        ↓
wait() 返回
        ↓
打印 worker start

這裡最重要的一步是：

wait() 会在睡眠期间释放 mutex

否則程式會死鎖。

1.4.為什麼 wait() 睡眠時必須釋放 mutex

假設 worker 這樣幹：

拿到 mutex
↓
发现 ready == false
↓
拿着 mutex 睡觉

那 main 想執行：

ready = true;

之前也必須先：

lock(mutex);

但 mutex 一直被 worker 拿著。

於是：

worker：
等 ready 变成 true
但一直拿着 mutex

main：
想把 ready 改成 true
但拿不到 mutex

結果就是：

谁也进行不下去

所以條件變數的 wait() 必須做一件非常特殊的事情：

当前已经持有 mutex
        ↓
原子地释放 mutex + 进入等待
        ↓
收到唤醒
        ↓
重新获得 mutex
        ↓
再返回

1.5.為什麼 wait() 要配合 `std::unique_lock`

常見寫法：

std::unique_lock lock(mutex);

cv.wait(lock, predicate);

而不是：

std::lock_guard lock(mutex);

原因是：

wait() 需要暫時：

unlock()

然後以後再：

lock()

而 std::unique_lock 支援這種：

临时释放锁
↓
以后重新获得锁

的所有權管理。

std::lock_guard 不支援手動解鎖和重新加鎖，所以不能滿足普通 condition_variable::wait() 的要求。

可以把：

cv.wait(lock, predicate);

粗略理解成：

while (!predicate())
{
    cv.wait(lock);
}

而內部的：

cv.wait(lock);

會做：

释放 mutex
+
线程睡眠
+
被唤醒
+
重新获得 mutex

1.6.為什麼“釋放 mutex + 睡眠”必須是原子的

這裡還有一個很重要的問題。

假設 wait 的內部邏輯不是原子的，而是：

1. unlock mutex

2. 过一会儿才真正开始睡眠

那麼可能發生：

worker：
unlock mutex
↓
还没真正睡下去

main：
修改 ready = true
↓
notify_one()

worker：
此时才进入睡眠

這樣：

notify 已经发生了

worker 却刚刚开始睡

worker 就可能一直睡下去。

所以條件變數必須保證：

“釋放 mutex”和“進入等待”之間不會留下一個能丟通知的空檔。

這也是為什麼不要自己用：

unlock();
sleep();
lock();

去模擬 condition_variable::wait()。

2.wait 與 predicate

2.1.`wait(lock)` 和 `wait(lock, predicate)`

條件變數有兩種常見等待方式。

2.1.1.不帶謂詞

cv.wait(lock);

它的意思是：

先睡眠
↓
被唤醒后重新拿锁
↓
返回

問題是：

被喚醒並不代表你的業務條件一定成立。

所以如果用這種形式，一般必須自己寫迴圈：

while (!ready)
{
    cv.wait(lock);
}

2.1.2.帶謂詞

2.2.什麼是 predicate

predicate 就是：

判斷“現在能不能繼續”的條件。

例如：

[] {
    return ready;
}

或者：

[] {
    return !queue.empty();
}

或者：

[] {
    return !queue.empty() || finished;
}

所以：

condition_variable
→ 负责把你叫醒

predicate
→ 决定你醒来以后到底能不能继续

可以記成：

notify 只是說“起來看看”；predicate 才說“現在能不能走”。

2.3.虛假喚醒（spurious wakeup）

條件變數允許一種情況：

没有真正的业务事件发生
↓
wait() 却醒了

這叫：

虛假喚醒（spurious wakeup）

所以這種寫法不安全：

cv.wait(lock);

use_shared_data();

因為：

wait() 返回

並不一定意味著：

共享数据已经满足条件

正確寫法是：

cv.wait(lock, [] {
    return ready;
});

或者等價的：

while (!ready)
{
    cv.wait(lock);
}

也就是說：

每次醒來都必須重新檢查真正的共享狀態。

3.notify 與狀態

3.1.notify 不是狀態

條件變數還有一個非常容易誤解的地方：

cv.notify_one();

不是在：

condition_variable 里面存了一条消息

它更像是：

“現在正在等的人，可以起來重新檢查條件了。”

如果通知發生時：

根本没有线程在等待

那麼這次通知就結束了。

以後才來 wait 的執行緒：

不会补收到这次旧通知

例如：

线程 A：
notify_one()
↓
当时没人等待
↓
通知结束

过了一会儿

线程 B：
开始 wait()

執行緒 B 不會收到執行緒 A 之前那次通知。

這類現象通常稱為：

丟失喚醒（lost wakeup）

3.2.那為什麼不會輕易因為“通知先發生”而出錯

因為正確的程式碼不會依賴：

“通知有没有发生过”

而是依賴：

“共享状态现在是什么”

例如：

{
    std::lock_guard lock(mutex);
    ready = true;
}

cv.notify_one();

即使：

ready = true
notify_one()

都發生在 worker 開始 wait 之前，也沒問題。

因為 worker 後來執行：

cv.wait(lock, [] {
    return ready;
});

會先檢查：

ready

發現已經：

true

於是：

根本不会睡

所以：

共享状态
→ 保存事实

condition_variable
→ 只是优化等待方式

這是條件變數最核心的理解之一。

3.3.`notify_one()`

cv.notify_one();

表示：

喚醒一個當前正在等待這個 condition_variable 的執行緒。

例如有：

10 个 worker 都在等任务

但現在只放入了：

1 个任务

通常：

notify_one();

就夠了。

因為最終只有一個 worker 能消費這個任務。

3.4.`notify_all()`

cv.notify_all();

表示：

喚醒所有當前正在等待這個 condition_variable 的執行緒。

適合：

程序准备退出

全局状态发生变化

所有 worker 都应该重新检查退出条件

例如：

finished = true;
cv.notify_all();

所有執行緒都會被叫醒。

但：

notify_all()

不代表：

所有线程同时进入临界区

它們醒來以後仍然要：

竞争同一把 mutex
↓
一个一个拿锁
↓
检查 predicate

所以：

notify_all()
→ 所有人都醒来看看

不是
→ 所有人同时进入临界区

3.5.修改狀態以後什麼時候 notify

很常見的寫法是：

{
    std::lock_guard lock(mutex);
    ready = true;
}

cv.notify_one();

順序是：

持锁修改共享状态
↓
释放 mutex
↓
notify

這樣等待執行緒被叫醒以後：

更有机会直接拿到 mutex

而不是：

刚醒
↓
发现通知线程还拿着 mutex
↓
又继续阻塞

不過要注意：

notify() 並不是規定“必須永遠放在鎖外”。

例如：

{
    std::lock_guard lock(mutex);

    ready = true;
    cv.notify_one();
}

在很多場景下也完全正確。

真正決定正確性的核心是：

共享谓词状态必须正确同步

wait 方必须在锁保护下检查 predicate

把 notify 放鎖外，更多是常見的效能和排程最佳化習慣。

4.生產者消費者

4.1.生產者消費者模型

這是條件變數最經典的應用。

假設：

producer
→ 不断往 queue 塞数据

consumer
→ 没数据时睡眠
→ 有数据时取出来处理

同時還需要考慮：

producer 最终会结束

所以消費者真正等待的條件不能只是：

!queue.empty()

還要考慮：

finished

完整示例：

#include <condition_variable>
#include <mutex>
#include <print>
#include <queue>
#include <thread>

std::queue<int> queue;

std::mutex mutex;
std::condition_variable cv;

bool finished = false;

void producer()
{
    for (int value = 1; value <= 5; ++value)
    {
        {
            std::lock_guard lock(mutex);

            queue.push(value);
        }

        cv.notify_one();
    }

    {
        std::lock_guard lock(mutex);

        finished = true;
    }

    cv.notify_all();
}

void consumer()
{
    while (true)
    {
        int value;

        {
            std::unique_lock lock(mutex);

            cv.wait(lock, [] {
                return !queue.empty() || finished;
            });

            if (queue.empty() && finished)
            {
                break;
            }

            value = queue.front();
            queue.pop();
        }

        std::println("consume {}", value);
    }
}

int main()
{
    std::thread producer_thread(producer);
    std::thread consumer_thread(consumer);

    producer_thread.join();
    consumer_thread.join();
}

輸出：

consume 1
consume 2
consume 3
consume 4
consume 5

4.2.為什麼 predicate 是：

!queue.empty() || finished

而不是：

!queue.empty()

假設生產者已經徹底結束：

finished == true

同時佇列也空了：

queue.empty() == true

如果 consumer 只等待：

!queue.empty()

那麼：

以后永远不会再有数据
↓
consumer 却继续等待“队列非空”
↓
没人会再生产任务
↓
consumer 永远睡下去

所以正確的等待條件應該表達：

有任务了

或者

生产已经结束了

也就是：

!queue.empty() || finished

4.3.consumer 醒來後為什麼還要判斷

等待結束以後：

cv.wait(lock, [] {
    return !queue.empty() || finished;
});

這裡只說明：

队列非空
或者
生产结束

至少有一個是真的。

所以接下來要區分：

4.3.1.情況：佇列有資料

queue 非空
↓
取出一个任务

4.3.2.情況：佇列為空，而且 finished == true

queue.empty()
&&
finished

說明：

现在没任务
而且以后也不会再有任务

這時 consumer 才可以安全退出。

4.4.為什麼真正處理任務要放到鎖外

consumer 裡面：

{
    std::unique_lock lock(mutex);

    // 检查 queue
    // pop 一个任务
}

process_task();

一般比：

std::unique_lock lock(mutex);

queue.pop();

process_task(); // 耗时 2 秒

更好。

原因是：

如果你拿著 mutex 做耗時工作：

consumer 拿着 queue mutex 处理 2 秒

那麼這 2 秒內：

producer 不能 push

其他 consumer 不能 pop

所有執行緒都被這一把鎖堵住了。

所以鎖內儘量只做：

检查共享状态

读取 / 修改共享状态

取出任务

真正耗時的業務邏輯：

放到锁外

這是多執行緒程式裡非常重要的習慣。

5.超時與高階用法

5.1.`wait_for()`

可以讓執行緒：

最多等一段時間。

例如：

bool ok = cv.wait_for(
    lock,
    std::chrono::seconds(1),
    [] {
        return ready;
    });

true
→ 在等待结束前 predicate 成立

false
→ 超时以后 predicate 仍然不成立

例如：

if (!cv.wait_for(
        lock,
        std::chrono::seconds(1),
        [] {
            return ready;
        }))
{
    std::println("timeout");
}

注意：

timeout

不代表程式一定出錯。

它只表示：

这段时间里条件没有成立

至於超時以後：

重试

退出

打印日志

走备用逻辑

由業務決定。

5.2.`wait_until()`

wait_until() 用來等待到某個：

绝对时间点

例如：

auto deadline =
    std::chrono::steady_clock::now()
    + std::chrono::seconds(2);

cv.wait_until(
    lock,
    deadline,
    [] {
        return ready;
    });

可以理解成：

一直等
直到：

ready == true

或者

到达 deadline

如果只是：

等 2 秒

通常：

wait_for(...)

更直觀。

如果你本來就有：

明确截止时间

那麼：

wait_until(...)

更合適。

5.3.為什麼超時通常推薦 `steady_clock`

涉及“持續多久”時：

std::chrono::steady_clock

一般更合適。

因為它是單調時鐘，不會因為：

用户手动改系统时间

NTP 校时

系统时间向前 / 向后跳

而突然改變。

所以這種程式碼：

auto deadline =
    std::chrono::steady_clock::now()
    + 2s;

通常比基於牆上時間更適合：

超时
计时
等待

5.4.一個 condition_variable 能等待多個條件嗎

當然可以。

真正決定等待條件的是：

predicate

例如：

cv.wait(lock, [] {
    return !queue.empty()
        || stopped
        || error;
});

表示：

只要下面任何一个发生：

队列有数据
或
程序停止
或
发生错误

就结束等待并继续处理

然後醒來後再判斷：

if (error)
{
    // 处理错误
}
else if (stopped && queue.empty())
{
    // 退出
}
else
{
    // 处理队列任务
}

所以：

condition_variable

並不關心你到底在等幾個條件。

它只負責：

睡
+
醒

複雜條件都由：

predicate

決定。

5.5.但 predicate 不要寫得太亂

如果你開始寫：

cv.wait(lock, [] {
    return
        a && b
        || c && !d
        || error_code != 0
        || queue.size() > 3
        || ...
});

說明共享狀態設計可能已經開始變複雜了。

這時候應該考慮：

能不能定义更清楚的状态对象

能不能使用 enum 表示状态

能不能拆分不同等待条件

能不能简化线程之间的协议

而不是把所有業務邏輯全部堆進一個巨大 predicate。

5.6.`condition_variable_any`

標準庫還有：

std::condition_variable_any

普通：

std::condition_variable

主要和：

std::unique_lock<std::mutex>

配合。

而：

std::condition_variable_any

可以配合更廣泛的鎖型別。

所以它：

更灵活

但通常：

也可能有更多开销

如果只是：

std::mutex

普通場景優先：

std::condition_variable

就夠了。

5.7.`condition_variable_any` 和 `stop_token`

C++20 中：

std::condition_variable_any

還有一個很實用的地方：

可以和 std::stop_token 配合等待。

這對於：

std::jthread

很方便。

例如 worker 正在：

等待队列任务

同時程式又希望：

request_stop()

以後能把它從等待中喚醒並退出。

這時：

condition_variable_any
+
stop_token

就可以把：

业务条件成立

或者

收到停止请求

組合進同一個等待過程裡。

5.8.condition_variable 和 atomic::wait 怎麼選

C++20 以後：

std::atomic

也支援：

wait()
notify_one()
notify_all()

如果只是：

等待一个 atomic 值变化

例如：

std::atomic<bool> ready{false};

ready.wait(false);

這種場景：

atomic::wait

會非常方便。

如果等待條件是：

queue 非空

或者 finished == true

或者 error != 0

這類涉及：

多个共享变量

容器

复杂业务状态

通常：

mutex
+
condition_variable
+
predicate

更加自然。

可以簡單記成：

只围绕一个 atomic 值等待
→ atomic::wait()

围绕一组共享业务状态等待
→ condition_variable

6.常見錯誤

6.1.錯誤：把 notify 當成狀態

錯誤想法：

“notify 已经调用了，
所以以后 wait 一定知道。”

不對。

cv.notify_one();

不會永久儲存一條訊息。

真正的狀態必須放在：

ready

或者：

queue

或者：

finished

中。

6.2.錯誤：不用 predicate 處理虛假喚醒

不推薦：

cv.wait(lock);

use_data();

應該：

cv.wait(lock, [] {
    return ready;
});

6.3.錯誤：等待方加鎖，修改方卻裸寫共享變數

例如：

bool ready;

worker：

std::unique_lock lock(mutex);

cv.wait(lock, [] {
    return ready;
});

main 卻：

ready = true; // 没有锁

這樣仍然可能產生資料競爭。

對於這種普通共享狀態：

读和写

都應該遵守同一套 mutex 同步規則。

6.4.錯誤：拿著鎖做耗時工作

錯誤：

std::unique_lock lock(mutex);

auto task = queue.front();
queue.pop();

do_heavy_work(task);

更好：

Task task;

{
    std::unique_lock lock(mutex);

    task = queue.front();
    queue.pop();
}

do_heavy_work(task);

儘快釋放鎖。

6.5.錯誤：停止時只修改 finished，卻忘了 notify

例如：

{
    std::lock_guard lock(mutex);
    finished = true;
}

如果 worker 正在：

cv.wait(...)

它可能仍然睡著。

通常還需要：

cv.notify_all();

把等待執行緒叫醒，讓它們重新檢查：

finished

6.6.錯誤：只考慮“現在沒任務”，沒考慮“以後也不會有任務”

生產者消費者模型必須設計：

什么时候表示生产彻底结束

否則 consumer 很容易：

队列空
↓
继续 wait
↓
但 producer 已经退出
↓
以后永远没人再 notify

所以通常需要：

bool finished;

或者其他明確的結束狀態。

7.最後總結

std::condition_variable 最核心的不是：

notify_one()

而是這一整套關係：

共享状态
+
mutex
+
condition_variable
+
predicate

可以這樣記：

共享状态
→ 记录事实

mutex
→ 保护事实

condition_variable
→ 负责睡眠和唤醒

predicate
→ 决定醒来后到底能不能继续

wait() 的核心流程：

拿着 mutex
↓
predicate 不成立
↓
释放 mutex + 睡眠
↓
收到 notify
↓
重新获得 mutex
↓
重新检查 predicate
↓
条件成立才真正返回

另外幾個重點：

notify_one()
→ 唤醒一个等待者

notify_all()
→ 唤醒所有等待者

wait_for()
→ 最多等一段时间

wait_until()
→ 最多等到某个时间点

最後一定記住：

condition_variable 沒有“記憶”，真正儲存狀態的是共享變數。

以及：

被 notify 喚醒不代表條件成立，最終永遠要重新檢查 predicate。

1.條件變數基礎

1.1.為什麼不能一直迴圈檢查

1.2.條件變數最基本的三個物件

1.3.最簡單的 wait / notify 示例

1.4.為什麼 wait() 睡眠時必須釋放 mutex

1.5.為什麼 wait() 要配合 std::unique_lock

1.6.為什麼“釋放 mutex + 睡眠”必須是原子的

2.wait 與 predicate

2.1.wait(lock) 和 wait(lock, predicate)

2.1.1.不帶謂詞

2.1.2.帶謂詞

2.2.什麼是 predicate

2.3.虛假喚醒（spurious wakeup）

3.notify 與狀態

3.1.notify 不是狀態

3.2.那為什麼不會輕易因為“通知先發生”而出錯

3.3.notify_one()

3.4.notify_all()

3.5.修改狀態以後什麼時候 notify

4.生產者消費者

4.1.生產者消費者模型

4.2.為什麼 predicate 是：

4.3.consumer 醒來後為什麼還要判斷

4.3.1.情況：佇列有資料

4.3.2.情況：佇列為空，而且 finished == true

4.4.為什麼真正處理任務要放到鎖外

5.超時與高階用法

5.1.wait_for()

5.2.wait_until()

5.3.為什麼超時通常推薦 steady_clock

5.4.一個 condition_variable 能等待多個條件嗎

5.5.但 predicate 不要寫得太亂

5.6.condition_variable_any

5.7.condition_variable_any 和 stop_token

5.8.condition_variable 和 atomic::wait 怎麼選

6.常見錯誤

6.1.錯誤：把 notify 當成狀態

6.2.錯誤：不用 predicate 處理虛假喚醒

6.3.錯誤：等待方加鎖，修改方卻裸寫共享變數

6.4.錯誤：拿著鎖做耗時工作

6.5.錯誤：停止時只修改 finished，卻忘了 notify

6.6.錯誤：只考慮“現在沒任務”，沒考慮“以後也不會有任務”

7.最後總結

1.5.為什麼 wait() 要配合 `std::unique_lock`

2.1.`wait(lock)` 和 `wait(lock, predicate)`

3.3.`notify_one()`

3.4.`notify_all()`

5.1.`wait_for()`

5.2.`wait_until()`

5.3.為什麼超時通常推薦 `steady_clock`

5.6.`condition_variable_any`

5.7.`condition_variable_any` 和 `stop_token`