c++ - C++中的线程池 - 如何结束程序

Question

我已经根据 Kerrek SB 的回答实现了线程池在 this question .

我已经实现了 MPMC queue for the functions和线程的 vector 线程。

一切都很完美，除了我不知道如何终止程序，最后如果我只是做 thread.join 因为线程还在等待更多的任务来做，它不会加入，主线程也不会继续。

知道如何正确结束程序吗？

为了完整起见，这是我的代码:

function_pool.h

#pragma once
#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>

class Function_pool
    {

private:
    std::queue<std::function<void()>> m_function_queue;
    std::mutex m_lock;
    std::condition_variable m_data_condition;

public: 
    Function_pool();
    ~Function_pool();
    void push(std::function<void()> func);
    std::function<void()> pop();
};

function_pool.cpp

#include "function_pool.h"

Function_pool::Function_pool() : m_function_queue(), m_lock(), m_data_condition()
{
}

Function_pool::~Function_pool()
{
} 

void Function_pool::push(std::function<void()> func)
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_function_queue.push(func);
    // when we send the notification immediately, the consumer will try to 
get the lock , so unlock asap
    lock.unlock();
    m_data_condition.notify_one();
}

std::function<void()> Function_pool::pop()
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_data_condition.wait(lock, [this]() {return !m_function_queue.empty(); 
});
    auto func = m_function_queue.front();
    m_function_queue.pop();
    return func;
    // Lock will be released
}

main.cpp

#include "function_pool.h"
#include <string>
#include <iostream>
#include <mutex>
#include <functional>
#include <thread>
#include <vector>

Function_pool func_pool;

void example_function()
{
    std::cout << "bla" << std::endl;
}

void infinite_loop_func()
{
    while (true)
    {
        std::function<void()> func = func_pool.pop();
        func();
    }
}

int main()
{
    std::cout << "stating operation" << std::endl;
    int num_threads = std::thread::hardware_concurrency();
    std::cout << "number of threads = " << num_threads << std::endl;
    std::vector<std::thread> thread_pool;
    for (int i = 0; i < num_threads; i++)
    {
        thread_pool.push_back(std::thread(infinite_loop_func));
    }

    //here we should send our functions
    func_pool.push(example_function);

    for (int i = 0; i < thread_pool.size(); i++)
    {
        thread_pool.at(i).join();
    }
    int i;
    std::cin >> i;
}

Answer 1

您的问题位于 infinite_loop_func 中，这是一个无限循环，结果不会终止。我已经阅读了之前的建议抛出异常的答案，但是，我不喜欢它，因为异常不应该用于常规控制流。

解决这个问题的最好方法是显式处理停止条件。例如:

std::atomic<bool> acceptsFunctions;

将此添加到函数池中可以让您清楚地拥有状态并断言在您析构时没有添加新函数。

std::optional<std::function<void()>> Function_pool::pop()

返回一个空的可选(或 C++14 及之前的函数)，允许您处理一个空队列。您必须这样做，因为 condition_variable 可以进行虚假唤醒。

有了这个，m_data_condition.notify_all() 可以用来唤醒所有线程。

最后我们必须修复无限循环，因为它不包括过度使用，同时允许您执行仍在队列中的所有函数:

while (func_pool.acceptsFunctions || func_pool.containsFunctions())
{
    auto f = func_pool.pop();
    If (!f)
    {
           func_pool.m_data_condition.wait_for(1s);
            continue;
     }

    auto &function = *f;
    function ();
}

我会留给你来实现 containsFunctions() 并清理代码(infinite_loop_func 作为成员函数？)请注意，使用计数器，你甚至可以处理生成的后台任务.