在同一个 std::condition_variable_any::wait_for 和 std::std::stop_token 上等待多个线程时死锁

Question

在我正在编写的应用程序中，我有一个简化如下的线程模型：

• 生成器 jthread (

  m_WorkerGenerator

  ) 正在启动异步任务。
• 多个异步任务一直工作到生成器线程停止为止。为此，他们使用 std::stop_token (

  m_token

  ) 的引用，并等待相同的 condition_variable_any (

  m_cv

  )，锁定在相同的 mutex (

  m_mut

  ) 下。死锁发生在 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); // Increase here if you can't reproduce
}

如果我使用 condition_variable_any::wait_for 重写而没有 stop_token 并从 stop_callback 手动发出信号，死锁是 not 发生。

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

Answer 1

这是一个典型的Lost Wakeup.

如果文章还不够，你想要更多细节，请参阅：

为什么'wait with predicate'解决了条件变量的'lost wakeup'？

只是为了自我检查，每次问问自己，是什么让你认为在 Foo::Foo 中无休止地创建新的工人时，任何工人至少会被执行一次直到“wait_for”点？他们可以随机地做，但是，再一次，是什么让你认为他们必须这样做？即使使用这个 sleep_for，也没有人能保证这一点。所以，从形式上讲，这段代码在某些时候可能会溢出内存。