一次停止多个线程

Question

在下面的程序中，当线程等待 condition_variable_any 以确定何时停止时，我错过了什么？在下面列出的程序中，线程以不可预测的方式停止；有些在调用notify_all之前，有些根本不停止。

使用的条件变量定义如下：

static std::mutex interrupt_mutex;
static std::condition_variable_any interrupt_cv;

线程检查是否到了停止的时间，如下所示：

std::unique_lock<std::mutex> lock(interrupt_mutex);
const auto cv_status = interrupt_cv.wait_for(lock, std::chrono::milliseconds(1000));
const auto timeout_expired = cv_status == std::cv_status::timeout;
if (!timeout_expired)
{
    quit = true;
}

主线程向线程发出停止信号，如下所示：

std::unique_lock<std::mutex> lock(interrupt_mutex);
interrupt_cv.notify_all();

可能的输出如下例如：

Thread  1> Received interrupt signal at iteration 2
Thread  1> Terminate
Thread  2> Received interrupt signal at iteration 2
Thread  2> Terminate
Thread  4> Received interrupt signal at iteration 2
Thread  4> Terminate
**** Requesting all threads to stop ****
Waiting for all threads to complete...

在重现问题的完整代码下方：

#include <thread>
#include <vector>
#include <mutex>
#include <condition_variable>

static std::mutex interrupt_mutex;
static std::condition_variable_any interrupt_cv;

int main()
{
    std::vector<std::thread> thread_handles;
    for (int thread_idx = 0; thread_idx < 4; ++thread_idx)
    {
        thread_handles.emplace_back(std::thread([thread_idx](const int thread_id)
        {
            int num_iterations = 0;
            auto quit = false;
            while (!quit)
            {
                // Fake processing time during the lock for testing purpose
                std::this_thread::sleep_for(std::chrono::milliseconds(200));
                ++num_iterations;

                // Check if need to stop with a timeout of 200ms 
                {
                    std::unique_lock<std::mutex> lock(interrupt_mutex);
                    const auto cv_status = interrupt_cv.wait_for(lock, std::chrono::milliseconds(1000));
                    if (const auto timeout_expired = cv_status == std::cv_status::timeout; !timeout_expired)
                    {
                        printf("Thread %2d> Received interrupt signal at iteration %d\n", thread_id, num_iterations);
                        quit = true;
                    }
                }
            }

            printf("Thread %2d> Terminate\n", thread_id);
        }, thread_idx + 1));
    }

    std::this_thread::sleep_for(std::chrono::seconds(5));

    // Signals all threads to stop
    {
        printf("**** Requesting all threads to stop ****\n");
        std::unique_lock<std::mutex> lock(interrupt_mutex);
        interrupt_cv.notify_all();
    }

    // Wait until all threads stop
    printf("Waiting for all threads to complete...\n");
    std::ranges::for_each(thread_handles, [](std::thread& thread_handle)
    {
        thread_handle.join();
    });

    printf("Program ends\n");
    return 0;
}

Answer 1

code> condition_variable是指在条件更改时（例如，当a 共享变量更改值时）发出信号。但是您的代码没有条件。您正在尝试将condition_variable本身用作退出信号，而这并不是它的目的。 notify_all（）只能唤醒在此精确时刻在code> procention_variable上积极等待的线程。没有等待它的线程，因为他们忙于做其他事情，不会收到该信号以终止。但是，这些线程一旦准备好等待，就需要检测条件。因此，条件需要更持久。这就是为什么您的代码无法正常工作的原因。

在这种情况下，您可以简单地将退出变量移至全局范围，在code> procention_variable和mutex 。设置该退出变量将充当您的条件，您可以向等待线索发出信号。然后，您可以使用wait_for（）的超载版本，该版本可以让您检查退出的当前状态（忽略虚假觉醒）。

尝试更多这样的东西：

#include <thread>
#include <vector>
#include <mutex>
#include <condition_variable>

static std::mutex interrupt_mutex;
static std::condition_variable_any interrupt_cv;
static bool quit = false;

int main()
{
    std::vector<std::thread> thread_handles;
    for (int thread_idx = 0; thread_idx < 4; ++thread_idx)
    {
        thread_handles.emplace_back(std::thread([thread_idx](const int thread_id)
        {
            int num_iterations = 0;
            while (true)
            {
                // Fake processing time outside the lock for testing purpose
                std::this_thread::sleep_for(std::chrono::milliseconds(200));
                ++num_iterations;

                // Check if need to stop with a timeout of 1s 
                {
                    std::unique_lock<std::mutex> lock(interrupt_mutex);
                    const bool signaled = interrupt_cv.wait_for(lock, std::chrono::seconds(1), [](){ return quit; });
                    if (signaled) break;
                }
            }

            printf("Thread %2d> Received interrupt signal at iteration %d\n", thread_id, num_iterations);
            printf("Thread %2d> Terminate\n", thread_id);
        }, thread_idx + 1));
    }

    std::this_thread::sleep_for(std::chrono::seconds(5));

    // Signals all threads to stop
    printf("**** Requesting all threads to stop ****\n");
    {
        std::lock_guard<std::mutex> lock(interrupt_mutex);
        quit = true;
    }
    interrupt_cv.notify_all();

    // Wait until all threads stop
    printf("Waiting for all threads to complete...\n");
    std::ranges::for_each(thread_handles, [](std::thread& thread_handle)
    {
        thread_handle.join();
    });

    printf("Program ends\n");
    return 0;
}

Answer 2

您的代码有2个问题，并且都具有相同的解决方案。

1. 虚假的唤醒。如果您的Wait_for由于SW而结束，则您的病情会满足您的状况，尽管没有人实际要求它唤醒/结束。
2. 如果您的线程不握住锁，也没有睡觉，并且主线程通知全部呢？请注意，通知不会在任何地方存储，因此，如果您错过了一个通知，稍后再收到。因此，那些错过的线程永远不会终止。

为了解决这两个问题，您需要另一个标志，这将告诉您的线程工作已经完成，并且必须停止。

static bool stop = false;
//...
if (stop) // Instead of if (const auto timeout_expired = cv_status == std::cv_status::timeout; !timeout_expired)
//...
printf("**** Requesting all threads to stop ****\n");
std::unique_lock<std::mutex> lock(interrupt_mutex);
stop = true;
interrupt_cv.notify_all();