多线程共用std::condition_variable_any::wait_for与std::stop_token触发死锁

线程模型与死锁场景 #

• 应用采用以下简化线程模型：
  1. 生成器线程m_WorkerGenerator（std::jthread）负责启动异步任务；
  2. 多个异步任务持续运行至生成器线程停止，它们通过引用std::stop_token（m_token），在同一std::mutex（m_mut）锁定下等待同一个std::condition_variable_any（m_cv）。
• 死锁发生在调用std::jthread::request_stop()终止m_WorkerGenerator之前。

死锁代码示例 #

#include <iostream>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <future>

using namespace std::chrono_literals;

class Foo {
std::condition_variable_any m_cv;
std::mutex m_mut;
std::stop_token m_token;
std::jthread m_WorkerGenerator;

void worker() {
    std::cout << "Worker thread start" << std::endl;

    while (true) {
        std::unique_lock lck{ m_mut };
        if (m_cv.wait_for(lck, m_token, 5ms, [=]() { return m_token.stop_requested(); })) {
            break;
        }
    }

    std::cout << "Worker thread stop" << std::endl;
}

public:

Foo() {
    m_WorkerGenerator = std::jthread{ [&](std::stop_token t) {
        m_token = t;

        std::vector<std::future<void>> futures;
        while (!t.stop_requested()) {

            auto fut = std::async(std::launch::async, [=]() {
                worker();
                });
            futures.emplace_back(std::move(fut));

            std::this_thread::sleep_for(5ms);
        }
    } };
}
};

int main()
{
    Foo f;
    std::this_thread::sleep_for(50ms); // 无法复现可延长此时间
}

修复方案及代码示例 #

移除wait_for中的std::stop_token参数，改用std::stop_callback手动触发条件变量通知，即可避免死锁。修复后的代码如下：

#include <iostream>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <future>

using namespace std::chrono_literals;

class Foo {
std::condition_variable_any m_cv;
std::mutex m_mut;
std::stop_token m_token;
std::jthread m_WorkerGenerator;

void worker() {
    std::cout << "Worker thread start" << std::endl;

    while (true) {
        std::unique_lock lck{ m_mut };
        if (m_cv.wait_for(lck, 5ms, [=]() { return m_token.stop_requested(); })) {
            break;
        }
    }

    std::cout << "Worker thread stop" << std::endl;
}

public:

Foo() {
    m_WorkerGenerator = std::jthread{ [&](std::stop_token t) {
        m_token = t;

        std::stop_callback(t, [=]() {
            m_cv.notify_all();
            });

        std::vector<std::future<void>> futures;
        while (!t.stop_requested()) {

            auto fut = std::async(std::launch::async, [=]() {
                worker();
                });
            futures.emplace_back(std::move(fut));

            std::this_thread::sleep_for(5ms);
        }
    } };
}
};

int main()
{
    Foo f;
    std::this_thread::sleep_for(5000ms);
}

内容的提问来源于stack exchange，提问作者Liviu Stancu