将 std::condition_variable 与 atomic<bool> 一起使用

Question

关于 SO 处理 atomic 和其他处理 std::condition_variable 有几个问题。但是我的问题是我在下面的使用是否正确？

三个线程，一个 ctrl 线程在取消暂停其他两个线程之前进行准备工作。 ctrl 线程还能够在工作线程（发送方/接收方）处于紧密的发送/接收循环时暂停它们。 使用原子的想法是在未设置暂停布尔值的情况下使紧密循环更快。

class SomeClass
{

public:
    //...                                                                                                                                                                                                                                                                                                                                                                                   
    // Disregard that data is public...                                                                                                                                                                                                                                                                                                                                                     

    std::condition_variable cv; // UDP threads will wait on this cv until allowed                                                                                                                                                                                                                                                                                                           
                                // to run by ctrl thread.                                                                                                                                                                                                                                                                                                                                   
    std::mutex cv_m;
    std::atomic<bool> pause_test_threads;
};

void do_pause_test_threads(SomeClass *someclass)
{
    if (!someclass->pause_test_threads)
    {
        // Even though we use an atomic, mutex must be held during                                                                                                                                                                                                                                                                                                                          
        // modification. See documentation of condition variable                                                                                                                                                                                                                                                                                                                            
        // notify_all/wait. Mutex does not need to be held for the actual                                                                                                                                                                                                                                                                                                                   
        // notify call.                                                                                                                                                                                                                                                                                                                                                                     
        std::lock_guard<std::mutex> lk(someclass->cv_m);
        someclass->pause_test_threads = true;
    }
}

void unpause_test_threads(SomeClass *someclass)
{
    if (someclass->pause_test_threads)
    {
        {
            // Even though we use an atomic, mutex must be held during                                                                                                                                                                                                                                                                                                                      
            // modification. See documentation of condition variable                                                                                                                                                                                                                                                                                                                        
            // notify_all/wait. Mutex does not need to be held for the actual                                                                                                                                                                                                                                                                                                               
            // notify call.                                                                                                                                                                                                                                                                                                                                                                 
            std::lock_guard<std::mutex> lk(someclass->cv_m);
            someclass->pause_test_threads = false;
        }
        someclass->cv.notify_all(); // Allow send/receive threads to run.                                                                                                                                                                                                                                                                                                                   
    }
}

void wait_to_start(SomeClass *someclass)
{
    std::unique_lock<std::mutex> lk(someclass->cv_m); // RAII, no need for unlock.                                                                                                                                                                                                                                                                                                          
    auto not_paused = [someclass](){return someclass->pause_test_threads == false;};
    someclass->cv.wait(lk, not_paused);
}

void ctrl_thread(SomeClass *someclass)
{
    // Do startup work                                                                                                                                                                                                                                                                                                                                                                      
    // ...                                                                                                                                                                                                                                                                                                                                                                                  
    unpause_test_threads(someclass);

    for (;;)
    {
        // ... check for end-program etc, if so, break;                                                                                                                                                                                                                                                                                                                                     
        if (lost ctrl connection to other endpoint)
        {
            pause_test_threads();
        }
        else
        {
            unpause_test_threads();
        }
        sleep(SLEEP_INTERVAL);

    }

    unpause_test_threads(someclass);
}

void sender_thread(SomeClass *someclass)
{
    wait_to_start(someclass);
    ...
    for (;;)
    {
        // ... check for end-program etc, if so, break;                                                                                                                                                                                                                                                                                                                                     
        if (someclass->pause_test_threads) wait_to_start(someclass);
        ...
    }
}

void receiver_thread(SomeClass *someclass)
{
    wait_to_start(someclass);
    ...
    for (;;)
    {
        // ... check for end-program etc, if so, break;                                                                                                                                                                                                                                                                                                                                     
        if (someclass->pause_test_threads) wait_to_start(someclass);
        ...
    }

Answer 1

我查看了你操作条件变量和原子的代码，似乎是正确的，不会引起问题。

为什么你应该保护对共享变量的写入，即使它是原子的：

如果在谓词中检查共享变量和等待条件之间发生写入共享变量可能会出现问题。考虑以下：

1. 等待线程虚假唤醒，获取互斥体，检查谓词并将其评估为false，因此它必须再次等待 cv。

2. 控制线程将共享变量设置为true。

3. 控制线程发送通知，任何人都没有收到，因为没有线程在等待条件变量。

4. 等待线程等待条件变量。由于通知已经发送，它会等到下一次虚假唤醒，或者下次控制线程发送通知时。可能会无限期地等待。

从共享原子变量中读取而不加锁通常是安全的，除非它引入了TOCTOU problems。

在您的情况下，您正在读取共享变量以避免不必要的锁定，然后在锁定后再次检查它（在有条件的wait 调用中）。这是一个有效的优化，称为双重检查锁定，我在这里看不到任何潜在问题。

您可能想检查atomic<bool> 是否无锁。否则，您将拥有更多没有它的锁。

Answer 2

一般来说，您希望独立于变量与条件变量一起工作的方式来对待变量是原子的这一事实。

如果与条件变量交互的所有代码都遵循在查询/修改之前锁定互斥锁的通常模式，并且与条件变量交互的代码不依赖于不与条件变量交互的代码，它将继续即使它包装了一个原子互斥体也是正确的。

通过快速阅读您的伪代码，这似乎是正确的。但是，对于多线程代码，伪代码通常不能很好地替代真实代码。

当原子读取表明您可能想要也可能不想要优化时，仅等待条件变量（并锁定互斥锁）的“优化”。您需要分析吞吐量。

Answer 3

原子数据不需要再次同步，它是无锁算法和数据结构的基础。

void do_pause_test_threads(SomeClass *someclass)
{
    if (!someclass->pause_test_threads)
    {
        /// your pause_test_threads might be changed here by other thread
        /// so you have to acquire mutex before checking and changing
        /// or use atomic methods - compare_exchange_weak/strong,
        /// but not all together                                                                                                                                                                                                                                                                                                                                                             
        std::lock_guard<std::mutex> lk(someclass->cv_m);
        someclass->pause_test_threads = true;
    }
}