具有可变线程的c ++线程池：将任务分配给线程时出现奇怪的死锁

Question

希望你们都度过了愉快的假期。

这个问题与我之前的问题有关：std::condition_variable - Wait for several threads to notify observer

我正在尝试基于下面我自己的可变线程实现来实现一个线程池：

class MutableThread
{
private:
    std::thread m_Thread;
    std::function<void()> m_Function;
    bool m_bRun;
    std::mutex m_LockMutex;
    std::mutex m_WaitMutex;
    std::condition_variable m_CV;
    IAsyncTemplateObserver<MutableThread>* m_Observer = nullptr;

private:
    void Execute()
    {
        while (m_bRun)
        {
            {
                std::unique_lock<std::mutex> wait(m_WaitMutex);
                m_CV.wait(wait);
            }               

            std::lock_guard<std::mutex> lock(m_LockMutex);
            if (m_bRun && m_Function)
            {
                m_Function();
                m_Function = std::function<void()>();

                if (m_Observer != nullptr)
                {
                    m_Observer->Signal(this);
                }
            }
        }
    }

public:
    HDEBUGNAME(TEXT("MutableThread"));

    MutableThread(const MutableThread& thread) = delete;

    MutableThread(IAsyncTemplateObserver<MutableThread>* _Observer)
    {
        m_Observer = _Observer;
        m_bRun = true;
        m_Thread = std::thread(&MutableThread::Execute, this);
    }

    MutableThread()
    {
        m_Observer = nullptr;
        m_bRun = true;
        m_Thread = std::thread(&MutableThread::Execute, this);
    }       

    ~MutableThread()
    {
        m_bRun = false;

        m_CV.notify_one();

        try
        {
            if (m_Thread.joinable())
                m_Thread.join();
        }
        catch (std::system_error& ex)
        {
            HWARNINGD(TEXT("%s"), ex.what());
        }                           
    }

    inline bool Start(const std::function<void()>& f)
    {
        std::lock_guard<std::mutex> lock(m_LockMutex);

        if (m_Function != nullptr)
            return false;

        m_Function = f;

        m_CV.notify_one();

        return true;
    }

IAsyncTemplateObserver 只是从我在前面问题中发布的 IAsyncObserver 类派生而来，并添加了一个虚函数：

template <typename T>
class IAsyncTemplateObserver : public IAsyncObserver
{
public:
    virtual void Signal(T* _Obj) = 0;
};

我想要做的是，向 ThreadPool 发出信号，表明函数已完成执行，并且将新任务分配给可变线程：

class MutableThread;

struct Task
{
    std::function<void()> m_Function;
    uint32_t m_uPriority;

    Task(const std::function<void()>& _Function, uint32_t _uPriority)
    {
        m_Function = _Function;
        m_uPriority = _uPriority;
    }
};

inline bool operator<(const Task& lhs, const Task& rhs)
{
    return lhs.m_uPriority < rhs.m_uPriority;
}

class ThreadPool : public IAsyncTemplateObserver<MutableThread>
{
private:
    std::list<MutableThread* > m_FreeThreads;
    std::list<MutableThread* > m_UsedThreads;

    std::set<Task> m_Tasks;

    std::mutex m_LockMutex;     
public:

    ThreadPool()
    {
        //Grow(std::thread::hardware_concurrency() - 1);
    }

    ThreadPool(size_t n)
    {
        Grow(n);
    }

    ~ThreadPool()
    {
        //std::lock_guard<std::mutex> lock(m_Mutex);
        for (MutableThread* pUsed : m_UsedThreads)
        {
            HSAFE_DELETE(pUsed);
        }

        for (MutableThread* pFree : m_FreeThreads)
        {
            HSAFE_DELETE(pFree);
        }
    }

    inline void Grow(size_t n)
    {
        std::lock_guard<std::mutex> lock(m_LockMutex);

        for (size_t i = 0; i < n; i++)
        {
            m_FreeThreads.push_back(new MutableThread(this));
        }
    }

    inline void AddTask(const Task& _Task)
    {
        {
            std::lock_guard<std::mutex> lock(m_LockMutex);
            m_Tasks.insert(_Task);
        }

        AssignThreads();
    }

    virtual void Signal(MutableThread* _pThread)
    {
        {
            std::lock_guard<std::mutex> lock(m_LockMutex);
            m_UsedThreads.remove(_pThread);
            m_FreeThreads.push_back(_pThread);
        }

        AssignThreads();

        NotifyOne();
    }

    inline void WaitForAllThreads()
    {
        bool bWait = true;
        do
        {
            {
                //check if we have to wait
                std::lock_guard<std::mutex> lock(m_LockMutex);
                bWait = !m_UsedThreads.empty() || !m_Tasks.empty();
            }

            if (bWait)
            {                   
                std::unique_lock<std::mutex> wait(m_ObserverMutex);
                m_ObserverCV.wait(wait);
            }

        } while (bWait);
    }

private:

    inline void AssignThreads()
    {
        std::lock_guard<std::mutex> lock(m_LockMutex);

        if (m_FreeThreads.empty() || m_Tasks.empty())
            return;

        //Get free thread
        MutableThread* pThread = m_FreeThreads.back();
        m_FreeThreads.pop_back();

        //park thread in used list
        m_UsedThreads.push_back(pThread);

        //get task with highest priority
        std::set<Task>::iterator it = m_Tasks.end();
        --it; //last entry has highest priority

        //start the task
        pThread->Start(it->m_Function);

        //remove the task from the list
        m_Tasks.erase(it);          
    }

同一线程多次调用 AddTask 函数，但是当可变线程向线程池发出信号时（通过 m_Observer->Signal(this) ），应用程序在 AssignThreads() 函数的 lock_guard 处冻结。现在奇怪的是不像正常的死锁，Visual Studio 中的所有调用堆栈视图都是空的，只要我尝试使用 lock_guard 越界。

谁能解释这种行为？是否有任何重大的设计缺陷或只是简单的混淆？

感谢您的帮助！

您好， 费边

编辑：我添加了一个重现问题的最小视觉工作室解决方案：ThreadPoolTest.zip

Answer 1

感谢一位朋友，我能够通过将调用 m_Observer->Signal(this) 移到 MutableThread::Execute() 函数的 lock_guard 范围之外来解决问题。其次，我删除了 AssignThreads() 函数中的 lock_guard 并将其调用移至 Signal()/AddTask 函数中的 lock_guard 范围内。并不真正相关，但仍然是一个缺陷：所有 condition_variables.wait() 调用现在都处于 while(m_bNotified == false) 循环中。