c++ - 延迟线程启动 - 通知所有不唤醒所有线程

Question

尝试:

在多个线程上延迟启动。

问题:

我创建了下面的示例来证明这个想法，并尝试在 x 上创建一个竞争条件来证明所有线程都将同时运行。

看起来事情是序列化的而不是并行运行——这是期望的行为，但也许每个线程运行的时间太短并且在另一个线程得到服务之前完成

有时一个线程会卡在 cv.wait --- 我在 GDB 中查看过这个并且可以看到其中一个线程正坐在第 0 帧中等待 --- 这意味着 notify_all 没有唤醒所有线程 ---(这是零星的行为，每隔几次尝试运行二进制文件就会发生)

提问:

1. 使用条件变量是否是延迟启动一组线程的有效方法，这些线程具有它们将全部并行运行的所需行为？

2. 为什么 notify_all() 没有唤醒所有线程？

代码:

// main.cpp
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <atomic>
#include <iostream>
#include <unistd.h>

int main()
{
    std::condition_variable cv;
    std::mutex cv_m;
    int x = 0;
    std::thread t1 = std::thread([&]{std::unique_lock<std::mutex> lk(cv_m); cv.wait(lk); std::cout << "t1 x:" << x++ << std::endl;});
    std::thread t2 = std::thread([&]{std::unique_lock<std::mutex> lk(cv_m); cv.wait(lk); std::cout << "t2 x:" << x++ << std::endl;});
    std::thread t3 = std::thread([&]{std::unique_lock<std::mutex> lk(cv_m); cv.wait(lk); std::cout << "t3 x:" << x++ << std::endl;});
    std::thread t4 = std::thread([&]{std::unique_lock<std::mutex> lk(cv_m); cv.wait(lk); std::cout << "t4 x:" << x++ << std::endl;});
    std::thread t5 = std::thread([&]{std::unique_lock<std::mutex> lk(cv_m); cv.wait(lk); std::cout << "t5 x:" << x++ << std::endl;});
    std::cout << "STARTING" << std::endl;
    cv.notify_all();
    t1.join();
    t2.join();
    t3.join();
    t4.join();
    t5.join();
    std::cout << "DONE" << std::endl;
    return 0;
}

编译:

g++ -std=c++14 main.cpp -lpthread

运行:

./a.out

Answer 1

条件变量是无状态的。如果没有服务员，通知就会丢失；可以发送虚假通知。您需要等待共享状态的更改，而不是来自条件变量的信号。

std::condition_variable :

When the condition variable is notified, a timeout expires, or a spurious wakeup occurs, the thread is awakened, and the mutex is atomically reacquired. The thread should then check the condition and resume waiting if the wake up was spurious.

此外，在通知条件变量时，如果需要保留服务员 FIFO 顺序，则必须保持互斥量。

修复:

int main()
{
    std::condition_variable cv;
    std::mutex cv_m;
    int x = 0;
    bool start = false;

    auto thread_fn = [&]{
        std::unique_lock<std::mutex> lk(cv_m);
        while(!start)
            cv.wait(lk);
        std::cout << "t1 x:" << x++ << std::endl;
    };

    std::thread t1 = std::thread(thread_fn);
    std::thread t2 = std::thread(thread_fn);
    std::thread t3 = std::thread(thread_fn);
    std::thread t4 = std::thread(thread_fn);
    std::thread t5 = std::thread(thread_fn);

    std::cout << "STARTING" << std::endl;
    {
        std::unique_lock<std::mutex> lock(cv_m);
        start = true;
        cv.notify_all();
    }

    t1.join();
    t2.join();
    t3.join();
    t4.join();
    t5.join();
    std::cout << "DONE" << std::endl;
}