I have a WorkDispatcher class which holds Worker class objects as properties and launches their function in new threads. Here is a example:

WorkDispatcher.h:

class WorkDispatcher
{

private:
    std::thread m_reconstructionThread;
    std::shared_ptr <Reconstruction::RGBDImageModel> m_currentRGBD;

public:
    WorkDispatcher();

    std::mutex m_rgbdMutex, m_commandMutex;
    std::deque<Network::NetworkCommandModel> m_CommandQueue;
    std::condition_variable m_RgbConditional, m_CommandConditional;

    Reconstruction::SceneReconstructor m_Reconstructor;

    void Initialize();
    void Work();
    void Stop();

};

WorkDispatcher.cpp:

void WorkDispatcher::Work()
{    
    m_reconstructionThread = std::thread(
        &Reconstruction::SceneReconstructor::Reconstruct, 
        std::ref(m_Reconstructor),
        std::ref(m_currentRGBD),
        std::ref(m_CommandQueue),
        std::ref(m_rgbdMutex), 
        std::ref(m_RgbConditional), 
        std::ref(m_commandMutex), 
        std::ref(m_CommandConditional)
    );
}

These functions hold infinite loops and I use the condition variables to wait until work is avaible. For example my Reconstruct function:

void SceneReconstructor::Reconstruct(
    std::shared_ptr<RGBDImageModel>& currentImage,
    std::deque<Network::NetworkCommandModel> commandQueue,
    std::mutex& rgbdMutex,
    std::condition_variable& rgbdCond,
    std::mutex& commandMutex,
    std::condition_variable& commandConditional)
{
    while (true)
    {
        std::unique_lock<std::mutex> rgbdLocker(rgbdMutex);
        rgbdCond.wait(rgbdLocker, [this] {return m_QuitReconstructionFlag; });
        
        // Quit flag to break out of loop
        if (m_QuitReconstructionFlag)
            break;

        // do stuff here

    }
}

So far so good, however if I want to quit the application I need to quit all of my worker threads. As seen above, for this these classes have a flag to quit, which I uses as follows:

void WorkDispatcher::Stop()
{
    // for all worker threads do this
    m_Reconstructor.m_QuitReconstructionFlag = true;
    if (m_reconstructionThread.joinable())
        m_reconstructionThread.join();
}

In theory this should stop the wait() function within a worker threads loop and then break out of the loop with the m_QuitReconstructionFlag, however this doesn't work.

What does work is the following:

1. remove the lambda from the wait functions
2. call notify_all() on the condition variables after settings the quit-flags to true;

This works fine for me, however the question is, why doesn't the lambda work?

CodePudding user response：

why doesn't the lambda work?

It works just fine, by itself.

However, C requires complex rules to be followed to properly synchronize multiple execution threads. Just because one execution thread sets a particular variable to a value does not guarantee you, in any way, that other execution threads will see the variable's new value. The synchronization rules govern that behavior.

So, this lambda works just fine. In its own execution thread. But if you want this lambda to observe changes to the value, made by other execution threads, this must be correctly synchronized.

Additionally, if you review your documentation of wait(), you should find a detailed explanation that says that if the condition function evaluates to false, it will not be called again until the condition variable is notified.

What does work is ... call notify_all()

Well, of course. Since wait() requires the condition variable to be notified, before it checks the waited-on condition again, then that's what you must do!

Finally, notifying the condition variable will work correctly in most cases, but, as I mentioned, synchronization rules (of which mutexes and condition variables play an important part of) have some edge cases where this, by itself, will not work. You must follow the following sequence of events strictly in order to have proper synchronization in all edge cases:

1. Lock the same mutex that another execution thread has locked before waiting on its condition variable.
2. Notify the condition variable.
3. Unlock the mutex

CodePudding user response：

Here the lambda returning true is not the condition to stop waiting, rather the lambda is to account for spurious wake ups. The notify or notify_all function of the conditional_variable is what is used to make the wait quit.

Rather than removing the lambda, you must simply change the stop function to

void WorkDispatcher::Stop()
{
    // for all worker threads do this
    m_Reconstructor.m_QuitReconstructionFlag = true;
    m_RgbConditional.notify_all() 
    if (m_reconstructionThread.joinable())
        m_reconstructionThread.join();
}

from here you can see that the wait with predicate passed to it (wait(predicate)) is equivalent to

if(!predicate())
    wait()

Hence when you call Stop(), It sets the predicate to true, so when the thread is woken up wait() returns, and the predicate is checked, if it is true, the wait(predicate) returns.

In the earlier case, the predicate was set to true but the function was not woken up