1

我一直在尝试创建一个线程池类（用于个人实验/使用/乐趣）。我找到了一种通过使用参数包（见下面的代码）并将函数绑定到 std::function 来接受具有任何参数/返回类型的任何函数的方法。这没有问题。问题是我想尝试创建一个成员函数来从正在执行的作业中检索返回值。在尝试执行此操作时，我不知道如何使代码通用。

到目前为止，我已经尝试制作一个地图，它使用作业 ID 作为键，并将该作业的返回值存储在其中。我的挣扎是 1. 我不知道如何确定类型（我可以使用“typename std::invoke_result::type”，但如果类型为 void，这会崩溃） 2. 如何制作可以拥有 Job 的地图ID 作为键和任何类型，以便我可以将返回类型放在那里。

class ThreadPool{
public:

    //getInstance to allow the second constructor to be called
    static ThreadPool& getInstance(int numThreads){
        static ThreadPool instance(numThreads);

        return instance;
    }

    //add any arg # function to queue
    template <typename Func, typename... Args >
    inline uint64_t push(Func& f, Args&&... args){
        auto funcToAdd = std::bind(f, args...);



        uint64_t newID = currentID++;
        std::unique_lock<std::mutex> lock(JobMutex);

        JobQueue.push(std::make_pair(funcToAdd, newID));
        thread.notify_one();
        return newID; //return the ID of the job in the queue
    }


    /* utility functions will go here*/
    inline void resize(int newTCount){

        int tmp = MAX_THREADS;
        if(newTCount > tmp || newTCount < 1){
            throw bad_thread_alloc("Cannot allocate " + std::to_string(newTCount) + " threads because it is greater than your systems maximum of " + std::to_string(tmp), __FILE__, __LINE__);
        }

        numThreads = (uint8_t)newTCount;
        Pool.resize(newTCount);
        DEBUG("New size is: " + std::to_string(Pool.size()));
    }

    inline uint8_t getThreadCount(){
        return numThreads;
    }

        //how i want the user to interact with this class is
        // int id = push(func, args);
        // auto value = getReturnValue(id); //blocks until return value is returned 
    auto getReturnValue(uint64_t jobID) {
        //Not sure how to handle this
    }

private:

    uint64_t currentID;
    uint8_t numThreads;
    std::vector<std::thread> Pool; //the actual thread pool
    std::queue<std::pair<std::function<void()>, uint64_t>> JobQueue; //the jobs with their assigned ID
    std::condition_variable thread;
    std::mutex JobMutex;

    /* infinite loop function */
    void threadManager();

    /*  Constructors */
    ThreadPool(); //prevent default constructor from being called

    //real constructor that is used
    inline ThreadPool(uint8_t numThreads) : numThreads(numThreads) {
        currentID = 0; //initialize currentID
        int tmp = MAX_THREADS;
        if(numThreads > tmp){
            throw bad_thread_alloc("Cannot allocate " + std::to_string(numThreads) + " threads because it is greater than your systems maximum of " + std::to_string(tmp), __FILE__, __LINE__);
        }
        for(int i = 0; i != numThreads; ++i){
            Pool.push_back(std::thread(&ThreadPool::threadManager, this));
            Pool.back().detach();
            DEBUG("Thread " + std::to_string(i) + " allocated");
        }
        DEBUG("Number of threads being allocated " + std::to_string(numThreads));
    }
    /* end constructors */


NULL_COPY_AND_ASSIGN(ThreadPool);
}; /* end ThreadPool Class */


void ThreadPool::threadManager(){
    while (true) {

        std::unique_lock<std::mutex> lock(JobMutex);
        thread.wait(lock, [this] {return !JobQueue.empty(); });

        //strange bug where it will continue even if the job queue is empty
        if (JobQueue.size() < 1)
            continue;

        auto job = JobQueue.front().first;
        JobQueue.pop();
        job();
    }
}

我对这一切都错了吗？我不知道任何其他方法可以通用地存储任何类型的函数，同时还能从中获取返回类型。

4

1 回答 1

0

提供将返回值存储到创建作业的函数的位置。如果我们将返回位置与其他元素一起提供，那么就不会不确定作业的存储位置。

我们可以比较简单地设计这部分界面。如果我们将参数与函数打包在一起，我们可以将所有排队的作业存储为std::function<void()>. 这简化了线程队列的实现，因为线程队列只需要关心一种类型std::function，而根本不需要关心返回值。

using job_t = std::function<void()>; 

template<class Return, class Func, class... Args>
job_t createJob(Return& dest, Func&& func, Args&&... args) {
    return job_t([=]() {
        dest = func(std::forward<Args>(args)...); 
    });
};

这种实现可以轻松地随意异步启动作业，而无需线程池担心返回值的存储位置。例如，我编写了一个函数来运行作业。该函数在新线程上启动作业，并在函数完成时将其标记atomic_bool为。true因为我们不必担心返回值，所以函数只需要返回一个我们shared_ptr用来atomic_bool检查函数是否完成的函数。 

using std::shared_ptr; 
using std::atomic_bool; 

shared_ptr<atomic_bool> run_job_async(std::function<void()> func) {
    shared_ptr<atomic_bool> is_complete = std::make_shared<atomic_bool>(false); 
    std::thread t([f = std::move(func), =]() {
        f();
        *is_complete = true;
    }); 
    t.detach(); 
    return is_complete; 
}
于 2019-06-01T07:22:21.497 回答