Есть ли способ написать универсальный код для любой функции, такой, чтобы она могла выполняться (асинхронно), а возвращаемое значение было получено из пула потоков?

Question

Я пытался создать класс пула потоков (для личного эксперимента / использования / веселья). Я нашел способ принять любую функцию с любым аргументом / типом возврата, используя пакеты параметров (см. Код ниже) и привязав функции к std :: function. Это работает без проблем. Проблема в том, что я хочу попытаться сделать функцию-член для получения возвращенных значений из выполняемых заданий. Я не знаю, как сделать код универсальным при попытке сделать это.

До сих пор я пытался создать карту, которая использует идентификатор задания в качестве ключа и в нем будет храниться возвращаемое значение этого задания. Моя борьба 1. Я не могу понять, как определить тип (я могу с "typename std :: invoke_result :: type", но это ломается, если тип void) 2. Как создать карту, которая может иметь идентификатор задания в качестве ключа и любой тип, чтобы я мог поместить туда возвращаемые типы.

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();
    }
}

Я все об этом говорю? Я не знал ни одного другого способа общего хранения функций любого типа, но также мог бы получать из них возвращаемые типы.

Answer 1

Укажите место для хранения возвращаемого значения для функции, которая создает задание. Если мы предоставим место возврата вместе с другими элементами, то нет никакой неопределенности относительно того, где задание будет сохранено.

Мы можем разработать эту часть интерфейса относительно просто.Если мы упаковываем аргументы вместе с функцией, мы можем сохранить все поставленные в очередь задания как 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; 
}