c++ - 线程池在一些循环后阻塞主线程

Question

我正在尝试了解线程如何在 C++ 上工作，并且我找到了一个用作指南的实现
进行我自己的实现，但是在一个循环或几个循环之后它会阻塞。

我有一个线程安全队列，我在其中检索分配给线程池的作业。

每个线程都运行这个函数:

// Declarations
std::vector<std::thread> m_threads;
JobQueue m_jobs; // A queue with locks
std::mutex m_mutex;
std::condition_variable m_condition;
std::atomic_bool m_active;
std::atomic_bool m_started;
std::atomic_int m_busy;
///...

[this, threadIndex] {
    int numThread = threadIndex;

    while(this->m_active) {
        std::unique_ptr<Job> currJob;
        bool dequeued = false;
        {
            std::unique_lock<std::mutex> lock { this->m_mutex };
            this->m_condition.wait(lock, [this, numThread]() {
                return (this->m_started && !this->m_jobs.empty()) || !this->m_active;
            });
            if (this->m_active) {
                m_busy++;
                dequeued = this->m_jobs.dequeue(currJob);
            }
        }

        if (dequeued) {
            currJob->execute();
            {
                std::lock_guard<std::mutex> lock { this->m_mutex };
                m_busy--;
            }
            m_condition.notify_all();
        } else {
            {
                std::lock_guard<std::mutex> lock { this->m_mutex };
                m_busy--;
            }
        }
    }
}

循环基本上是:

while(1) {
        int numJobs = rand() % 10000;
        std::cout << "Will do " << numJobs << " jobs." << std::endl;
        while(numJobs--) {
            pool.assign([](){
                // some heavy calculation
             });
        }
        pool.waitEmpty();
        std::cout << "Done!" << std::endl; // chrono removed for readability
    }

而 waitEmpty方法描述为:

std::unique_lock<std::mutex> lock { this->m_mutex };
this->m_condition.wait(lock, [this] {
    return this->empty();
});

并且在这种等待方法中，代码通常会挂起，因为内部的测试不再被调用。

我已经对其进行了调试，更改了 notification_one 和 all 的地方，但由于某种原因，在一些循环之后它总是阻塞。

通常，但不总是，它锁定 condition_variable.wait()锁定当前线程直到没有其他线程工作并且队列为空的方法，但我也看到它在我调用 condition_variable.notify_all() 时发生.

一些调试帮助我注意到，当我调用 notify_all()在从线程上， wait()在主线程中不再测试。

预期的行为是它在循环时不会阻塞。

我正在使用 G++ 8.1.0在 Windows .

输出是:

Will do 41 jobs.  
Done! Took 0ms!   
Will do 8467 jobs.


<main thread blocked>

编辑:我修复了 paddy 评论指出的问题:现在 m_busy--当作业没有出队时也会发生。

编辑 2:在 Linux 上运行它不会锁定主线程并按预期运行。 ( g++ (Ubuntu 7.4.0-1ubuntu1~18.04.1) 7.4.0)

编辑 3:如评论中所述，已更正 deadlock至 block , 因为它只涉及一个锁

编辑 4:正如 Jérôme Richard 评论的那样我可以通过创建 lock_guard 来改进它围绕 m_busy--;但现在 notify_all() 的代码块在assign方法中调用。这里是assign方法供引用:

template<class Func, class... Args>
    auto assign(Func&& func, Args&&... args) -> std::future<typename std::result_of<Func(Args...)>::type> {
        using jobResultType = typename std::result_of<Func(Args...)>::type;

        auto task = std::make_shared<std::packaged_task<jobResultType()>>(
            std::bind(std::forward<Func>(func), std::forward<Args>(args)...)
        );

        auto job = std::unique_ptr<Job>(new Job([task](){ (*task)(); }));
        std::future<jobResultType> result = task->get_future();

        m_jobs.enqueue(std::move(job));
        std::cout << " - enqueued";
        m_condition.notify_all();
        std::cout << " - ok!" << std::endl;

        return result;
    }

在其中一个循环中，最后一个输出是

//...
 - enqueued - ok!
 - enqueued - ok!
 - enqueued

<blocked again>

编辑 5:随着最新的变化，这不会发生在 msbuild编译器。

我的实现要点在这里: https://gist.github.com/GuiAmPm/4be7716b7f1ea62819e61ef4ad3beb02

这也是我基于我的实现的原始文章:
https://roar11.com/2016/01/a-platform-independent-thread-pool-using-c14/

任何帮助将不胜感激。

Answer 1

tl；博士:使用 std::lock_guard的 m_mutex约m_busy--以避免意外的等待条件阻塞。

分析

首先，请注意，池中的一个线程和几个作业可能会出现问题。这意味着提交作业的主线程和执行它们的主线程之间存在问题。
使用GDB进一步分析等待条件卡住时程序的状态，可以看到没有作业，m_busy设置为 0 并且两个线程都在等待通知。
这意味着 master 和最后一个要执行的作业的唯一 worker 之间的等待条件存在并发问题。
通过在代码中添加全局原子钟，可以看到(几乎在所有情况下)worker 在 master 可以等待作业完成和 worker 完成之前完成所有工作。

这是检索到的一个实际场景(项目符号是按顺序完成的):

主机启动等待调用，还有作业剩余

worker 执行m_busy++ ，将最后一个作业出列并执行(m_busy 现在设置为 1，作业队列为空)

主计算等待调用的谓词

主叫ThreadPool::empty由于busy，结果为假设置为 1

worker 执行m_busy-- (m_busy 现在设置为 0)

从那一刻起，主人可以等待条件回来(但怀疑不这样做)

worker 通知条件

怀疑主人现在只等待条件返回并且不受最后一个通知的影响(因为接下来不会发生等待)

此时master不再执行指令，将永远等待

worker 等待条件，也将永远等待

主人不受通知影响的事实很奇怪。
它似乎与内存围栏问题有关。更详细的解释可以看 here .引用文章:

Even if you make dataReady an atomic, it must be modified under the mutex; if not the modification to the waiting thread may be published, but not correctly synchronized.

所以一个解决方案是替换 m_busy--通过以下几行指令:

{
    std::lock_guard<std::mutex> lck {this->m_mutex};
    m_busy--;
}

它避免了以前的情况。确实，一方面 m_mutex在 wait 的谓词检查期间获得来电阻止 m_busy在此特定时刻进行修改；另一方面，它强制数据正确同步。

从理论上讲，还应该包含 m_jobs.dequeue 更安全。调用它，但会大大降低工作人员的并行度。实际上，当在工作线程中释放锁时，会进行有用的同步。

请注意，避免此类问题的一种通用解决方法是使用 wait_for 为等待调用添加超时。循环中的函数以强制执行谓词条件。但是，此解决方案的代价是等待调用的延迟较高，因此会显着减慢执行速度。