c++ - 条件变量的“等待”函数在提供谓词时导致意外行为

Question

作为一项教育练习，我正在使用条件变量实现一个线程池。控制器线程创建一个等待信号的线程池（一个原子变量被设置为大于零的值）。当通知线程唤醒时，执行它们的工作，当最后一个线程完成时，它通知主线程唤醒。控制器线程阻塞，直到最后一个线程完成。然后该池可用于后续重复使用。

时不时地，我在等待工作人员发出完成信号的控制器线程上超时（可能是因为减少活动工作计数器时的竞争条件），所以为了巩固池，我更换了“wait（lck )”形式的条件变量的等待方法，带有“wait(lck, predicate)”。由于这样做，线程池的行为似乎允许将活动工作计数器递减到 0 以下（这是重新唤醒控制器线程的条件） - 我有一个竞争条件。我在 stackoverflow 和其他各种网站上阅读了无数关于原子变量、同步、内存排序、虚假和丢失唤醒的文章，并尽我所能结合了我所学的知识，并且仍然无法为我的生活弄清楚为什么我编写谓词等待的方式不起作用。计数器只能与池中的线程数（例如 8）一样高，并且低至零。我开始对自己失去信心——做一些从根本上简单的事情不应该这么难。很明显，我还需要在这里学习其他东西 :)

当然考虑到存在竞争条件，我确保驱动池的唤醒和终止的两个变量都是原子的，并且只有在使用 unique_lock 保护时才会更改。具体来说，我确保当向池发起请求时，获取锁，活动线程计数器从 0 更改为 8，解锁互斥锁，然后“notified_all”。控制器线程只有在活动线程计数为零时才会被唤醒，一旦最后一个工作线程将其递减到那么远并且“notified_one”。

在工作线程中，条件变量只有在活动线程计数大于零时才会等待并唤醒，解锁互斥锁，并行执行创建池时预分配给处理器的工作，重新获取互斥锁，并以原子方式减少活动线程数。然后，虽然它仍然应该受到锁的保护，但测试它是否是最后一个仍然处于活动状态的线程，如果是，则再次解锁互斥锁和“notify_one”以唤醒控制器。

问题是 - 即使只有 1 或 2 次迭代，活动线程计数器也会反复低于零。如果我在新工作负载开始时测试活动线程数，我会发现活动线程数在 -6 左右减少 - 就好像允许池在工作完成之前重新唤醒控制器线程一样。

鉴于线程计数器和终止标志都是原子变量，并且只在同一个互斥锁的保护下被修改，我对所有更新使用顺序内存排序，我只是看不到这是如何发生的，我迷路了。

#include <stdafx.h>
#include <Windows.h>

#include <iostream>
#include <thread>
using std::thread;
#include <mutex>
using std::mutex;
using std::unique_lock;
#include <condition_variable>
using std::condition_variable;
#include <atomic>
using std::atomic;
#include <chrono>
#include <vector>
using std::vector;

class IWorkerThreadProcessor
{
public:
    virtual void Process(int) = 0;
};


class MyProcessor : public IWorkerThreadProcessor
{
    int index_ = 0;
public:
    MyProcessor(int index)
    {
        index_ = index;
    }

    void Process(int threadindex)
    {
    for (int i = 0; i < 5000000; i++);
        std::cout << '(' << index_ << ':' << threadindex << ") ";
    }
};


#define MsgBox(x) do{ MessageBox(NULL, x, L"", MB_OK ); }while(false)


class ThreadPool
{
private:
    atomic<unsigned int> invokations_ = 0;

     //This goes negative when using the wait_for with predicate
    atomic<int> threadsActive_ = 0;
    atomic<bool> terminateFlag_ = false;
    vector<std::thread> threads_;
    atomic<unsigned int> poolSize_ = 0;

    mutex mtxWorker_;
    condition_variable cvSignalWork_;
    condition_variable cvSignalComplete_;

public:

    ~ThreadPool()
    {
        TerminateThreads();
    }

    void Init(std::vector<IWorkerThreadProcessor*>& processors)
    {
        unique_lock<mutex> lck2(mtxWorker_);
        threadsActive_ = 0;
        terminateFlag_ = false;

        poolSize_ = processors.size();
        for (int i = 0; i < poolSize_; ++i)
            threads_.push_back(thread(&ThreadPool::launchMethod, this, processors[i], i));
    }

    void ProcessWorkload(std::chrono::milliseconds timeout)
    {
        //Only used to see how many invocations I was getting through before experiencing the issue - sadly it's only one or two
        invocations_++;
        try
        {
        unique_lock<mutex> lck(mtxWorker_);

        //!!!!!! If I use the predicated wait this break will fire !!!!!!
        if (threadsActive_.load() != 0)
        __debugbreak();

        threadsActive_.store(poolSize_);
        lck.unlock();
        cvSignalWork_.notify_all();

        lck.lock();
        if (!cvSignalComplete_.wait_for(
                lck,
                timeout,
                [this] { return threadsActive_.load() == 0; })
                )
        {
            //As you can tell this has taken me through a journey trying to characterise the issue...
            if (threadsActive_ > 0)
                MsgBox(L"Thread pool timed out with still active threads");
            else if (threadsActive_ == 0)
                MsgBox(L"Thread pool timed out with zero active threads");
            else
                MsgBox(L"Thread pool timed out with negative active threads");
            }
        }
        catch (std::exception e)
        {
            __debugbreak();
        }
    }

    void launchMethod(IWorkerThreadProcessor* processor, int threadIndex)
    {
        do
        {
            unique_lock<mutex> lck(mtxWorker_);

            //!!!!!! If I use this predicated wait I see the failure !!!!!!
            cvSignalWork_.wait(
                lck,
                [this] {
                return
                    threadsActive_.load() > 0 ||
                    terminateFlag_.load();
            });


            //!!!!!!!! Does not cause the failure but obviously will not handle
            //spurious wake-ups !!!!!!!!!!
            //cvSignalWork_.wait(lck);

            if (terminateFlag_.load())
                return;

            //Unlock to parallelise the work load
            lck.unlock();
            processor->Process(threadIndex);

            //Re-lock to decrement the work count
            lck.lock();
            //This returns the value before the subtraction so theoretically if the previous value was 1 then we're the last thread going and we can now signal the controller thread to wake.  This is the only place that the decrement happens so I don't know how it could possibly go negative
            if (threadsActive_.fetch_sub(1, std::memory_order_seq_cst) == 1)
            {
                lck.unlock();
                cvSignalComplete_.notify_one();
            }
            else
                lck.unlock();

        } while (true);
    }

    void TerminateThreads()
    {
        try
        {
            unique_lock<mutex> lck(mtxWorker_);
            if (!terminateFlag_)
            {
                terminateFlag_ = true;
                lck.unlock();
                cvSignalWork_.notify_all();

                for (int i = 0; i < threads_.size(); i++)
                    threads_[i].join();
            }
        }
        catch (std::exception e)
        {
            __debugbreak();
        }
    }
};


int main()
{
    std::vector<IWorkerThreadProcessor*> processors;
    for (int i = 0; i < 8; i++)
        processors.push_back(new MyProcessor(i));


    std::cout << "Instantiating thread pool\n";
    auto pool = new ThreadPool;
    std::cout << "Initialisting thread pool\n";
    pool->Init(processors);
    std::cout << "Thread pool initialised\n";

    for (int i = 0; i < 200; i++)
    {
        std::cout << "Workload " << i << "\n";
        pool->ProcessWorkload(std::chrono::milliseconds(500));
        std::cout << "Workload " << i << " complete." << "\n";
    }

    for (auto a : processors)
        delete a;

    delete pool;

    return 0;
}

标签： c++multithreadingcondition-variablesynchronisationmemory-barriers

Answer 1

class ThreadPool
{
private:
    atomic<unsigned int> invokations_ = 0;
    std::atomic<unsigned int> awakenings_ = 0;
    std::atomic<unsigned int> startedWorkloads_ = 0;
    std::atomic<unsigned int> completedWorkloads_ = 0;
    atomic<bool> terminate_ = false;
    atomic<bool> stillFiring_ = false;
    vector<std::thread> threads_;
    atomic<unsigned int> poolSize_ = 0;

    mutex mtx_;
    condition_variable cvSignalWork_;
    condition_variable cvSignalComplete_;

public:

    ~ThreadPool()
    {
        TerminateThreads();
    }


    void Init(std::vector<IWorkerThreadProcessor*>& processors)
    {
        unique_lock<mutex> lck2(mtx_);
        //threadsActive_ = 0;
        terminate_ = false;

        poolSize_ = processors.size();
        for (int i = 0; i < poolSize_; ++i)
            threads_.push_back(thread(&ThreadPool::launchMethod, this, processors[i], i));

        awakenings_ = 0;
        completedWorkloads_ = 0;
        startedWorkloads_ = 0;
        invokations_ = 0;
    }


    void ProcessWorkload(std::chrono::milliseconds timeout)
    {
        try
        {
            unique_lock<mutex> lck(mtx_);
            invokations_++;

            if (startedWorkloads_ != 0)
                __debugbreak();

            if (completedWorkloads_ != 0)
                __debugbreak();

            if (awakenings_ != 0)
                __debugbreak();

            if (stillFiring_)
                __debugbreak();

            stillFiring_ = true;
            lck.unlock();
            cvSignalWork_.notify_all();

            lck.lock();
            if (!cvSignalComplete_.wait_for(
                lck,
                timeout,
                //[this] { return this->threadsActive_.load() == 0; })
                [this] { return completedWorkloads_ == poolSize_ && !stillFiring_; })
                )
            {
                if (completedWorkloads_ < poolSize_)
                {
                    if (startedWorkloads_ < poolSize_)
                        MsgBox(L"Thread pool timed out with some threads unstarted");
                    else if (startedWorkloads_ == poolSize_)
                        MsgBox(L"Thread pool timed out with all threads started but not all completed");
                }
                else
                    __debugbreak();
            }


            if (completedWorkloads_ != poolSize_)
                __debugbreak();

            if (awakenings_ != poolSize_)
                __debugbreak();

            awakenings_ = 0;
            completedWorkloads_ = 0;
            startedWorkloads_ = 0;

        }
        catch (std::exception e)
        {
            __debugbreak();
        }

    }



    void launchMethod(IWorkerThreadProcessor* processor, int threadIndex)
    {
        do
        {
            unique_lock<mutex> lck(mtx_);
            cvSignalWork_.wait(
                lck,
                [this] {
                return
                    (stillFiring_ && (startedWorkloads_ < poolSize_)) ||
                    terminate_;
            });
            awakenings_++;

            if (startedWorkloads_ == 0 && terminate_)
                return;

            if (stillFiring_ && startedWorkloads_ < poolSize_) //guard against spurious wakeup
            {
                startedWorkloads_++;
                if (startedWorkloads_ == poolSize_)
                    stillFiring_ = false;

                lck.unlock();
                processor->Process(threadIndex);
                lck.lock();
                completedWorkloads_++;

                if (completedWorkloads_ == poolSize_)
                {
                    lck.unlock();
                    cvSignalComplete_.notify_one();
                }
                else
                    lck.unlock();
            }
            else
                lck.unlock();

        } while (true);
    }



    void TerminateThreads()
    {
        try
        {
            unique_lock<mutex> lck(mtx_);
            if (!terminate_) //Don't attempt to double-terminate
            {
                terminate_ = true;
                lck.unlock();
                cvSignalWork_.notify_all();

                for (int i = 0; i < threads_.size(); i++)
                    threads_[i].join();
            }
        }
        catch (std::exception e)
        {
            __debugbreak();
        }
    }
};