thread.hpp

/****************************************************************************
 *  hipipe library
 *  Copyright (c) 2017, Cognexa Solutions s.r.o.
 *  Copyright (c) 2018, Iterait a.s.
 *  Author(s) Filip Matzner
 *
 *  This file is distributed under the MIT License.
 *  See the accompanying file LICENSE.txt for the complete license agreement.
 ****************************************************************************/
 
#ifndef HIPIPE_CORE_THREAD_HPP
#define HIPIPE_CORE_THREAD_HPP
 
#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>
 
#include <cmath>
#include <functional>
#include <future>
#include <experimental/optional>
#include <thread>
 
namespace hipipe {
 
class thread_pool {
private:
    boost::asio::io_service service_;
    std::experimental::optional<boost::asio::io_service::work> work_{service_};
    boost::thread_group threads_;
 
public:
 
    thread_pool(unsigned n_threads = std::thread::hardware_concurrency())
    {
        // always use at least a single thread
        n_threads = std::max(1u, n_threads);
        // populate the thread pool
        for (unsigned i = 0; i < n_threads; ++i) {
            threads_.create_thread([this]() { return this->service_.run(); });
        }
    }
 
    template<typename Fun, typename... Args>
    std::future<std::result_of_t<Fun(Args...)>> enqueue(Fun fun, Args... args)
    {
        using Ret = std::result_of_t<Fun(Args...)>;
        std::packaged_task<Ret(Args...)> task{fun};
        std::future<Ret> future = task.get_future();
        // Packaged task is non-copyable, so ASIO's post() method cannot handle it.
        // So we build a shared_ptr of the task and post a lambda
        // dereferencing and running the task stored in the pointer.
        auto shared_task = std::make_shared<std::packaged_task<Ret(Args...)>>(std::move(task));
        auto shared_args = std::make_shared<std::tuple<Args...>>(std::move(args)...);
        auto asio_task = [task = std::move(shared_task), args = std::move(shared_args)]() {
            return std::apply(std::move(*task), std::move(*args));
        };
        service_.post(std::move(asio_task));
        return future;
    }
 
    ~thread_pool()
    {
        work_ = std::experimental::nullopt;
        threads_.join_all();
    }
};
 
static thread_pool global_thread_pool;
 
} // namespace hipipe
#endif