ThreadPoolAccessor.hpp

Go to the documentation of this file.

#pragma once
 
#include "logger.hpp"
#include "ThreadPool.hpp"
#include <functional>
#include <unordered_set>
 
class ThreadPoolAccessor
{
public:
    ThreadPoolAccessor();
    ~ThreadPoolAccessor();
 
    ThreadPoolAccessor(const ThreadPoolAccessor &) = delete;
    ThreadPoolAccessor & operator=(const ThreadPoolAccessor &) = delete;
    ThreadPoolAccessor(ThreadPoolAccessor &&) = delete;
    ThreadPoolAccessor & operator=(ThreadPoolAccessor &&) = delete;
 
    template<typename F>
    uint64_t submit(F && f)
    {
        std::lock_guard<std::mutex> lock(m_futures_mutex);
 
        typedef typename std::invoke_result<F>::type result_type;
        //create a packaged task with the function
        std::shared_ptr<std::packaged_task<result_type()>> task_ptr = std::make_shared<std::packaged_task<result_type()>>(std::move(f));
 
        //encapsulate inside another void function with error handling + id tracking
        uint64_t id = m_counter++;
        std::function<void()> wrapper_func = [task_ptr = std::move(task_ptr), this, id] () mutable
        {
            std::exception_ptr eptr = nullptr;
 
            (*task_ptr)();
 
            try
            {
                task_ptr->get_future().get();
            }
            catch (...)
            {
                eptr = std::current_exception();
            }
 
            {
                std::lock_guard<std::mutex> lock(m_finished_tasks_mutex);
                m_finished_tasks.insert(id);
            }
 
            if (eptr != nullptr)
                std::rethrow_exception(eptr);
        };
 
        //submit the task to the thread pool
        std::future<void> future_v = m_thread_pool.submit(wrapper_func);
        m_futures.insert(std::make_pair(id, std::move(future_v)));
 
        return id;
    }
 
    void waitForAll();
 
    void waitForFinishedTasks();
 
    void waitForTask(uint64_t id);
 
    void waitForTasks(const std::vector<uint64_t> & ids);
private:
    ThreadPool &                                        m_thread_pool;
    uint64_t                                            m_counter = 0;
 
    std::unordered_set<uint64_t>                        m_finished_tasks;
    std::mutex                                          m_finished_tasks_mutex;
    std::unordered_map<uint64_t, std::future<void>>     m_futures;
    std::mutex                                          m_futures_mutex;
 
    void waitTask(uint64_t id);
};