ExecutorAccessor.hpp

Go to the documentation of this file.

#pragma once
 
#include "logger.hpp"
#include "Executor.hpp"
#include <functional>
#include <unordered_set>
 
 
class ExecutorAccessor
{
public:
    ExecutorAccessor();
    ~ExecutorAccessor();
 
    ExecutorAccessor(const ExecutorAccessor &) = delete;
    ExecutorAccessor & operator=(const ExecutorAccessor &) = delete;
    ExecutorAccessor(ExecutorAccessor &&) = delete;
    ExecutorAccessor & operator=(ExecutorAccessor &&) = 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;
            auto future = task_ptr->get_future();
            (*task_ptr)();
 
            try
            {
                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_executor.run(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:
    task::Executor &                                    m_executor;
    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);
};