ThreadPool.hpp

/*
 * Copyright (C) 2021 Alexander Kornilov (akornilov.82@gmail.com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#ifndef CPP_MATE_THREAD_POOL_HPP
#define CPP_MATE_THREAD_POOL_HPP
 
#include <CppMate/Checkers.hpp>
#include <CppMate/SharedState.hpp>
 
#include <memory>
#include <functional>
#include <future>
#include <vector>
#include <queue>
#include <thread>
 
namespace CppMate {
 
template<typename R = int>
class ThreadPool final
{
public:
 
    explicit ThreadPool(unsigned maxThreads = 0): _state(maxThreads, [this] { return std::thread(&ThreadPool::worker, this); }) {}
 
    ThreadPool(const ThreadPool& other) = delete;
    ThreadPool(ThreadPool&& other) = default;
 
    ThreadPool& operator=(const ThreadPool& other) = delete;
    ThreadPool& operator=(ThreadPool&& other) = default;
 
    ~ThreadPool() {
        for (auto& worker: _state.modify().template notifyAll<std::vector<std::thread>>([](auto& state) { return state.finish(); })) {
            if (worker.joinable()) { worker.join(); }
        }
    }
 
    unsigned getMaxThreads() const { return _state.template view<unsigned>([](auto& state) { return state.getMaxThreads(); }); }
 
    unsigned getWorkersCount() const { return _state.template view<unsigned>([](auto& state) { return state.getWorkersCount(); }); }
 
    unsigned getPendingTasksCount() const { return _state.template view<unsigned>([](auto& state) { return state.getPendingTasksCount(); }); }
 
    bool hasPendingTasks() const { return getPendingTasksCount() > 0; }
 
    unsigned getWaitingCount() const { return _state.template view<unsigned>([](auto& state) { return state.getWaitingCount(); }); }
 
    template<typename F, typename... A>
    std::future<R> addTask(F&& task, A&&... args) {
        return _state.modify().template notifyOne<std::future<R>>([&](auto& state) {
            return state.addTask(std::bind(std::forward<F>(task), std::forward<A>(args)...));
        });
    }
 
private:
    class State;
 
    void worker() {
        auto exit = false;
        auto isWorking = true;
        while(!exit) {
            auto task = _state.modify().when([&isWorking](auto& state) {
                const auto wakeupIf = state.isFinished() || state.hasTasks();
                if (!wakeupIf && isWorking) { isWorking = false; state.waitingBegin(); }
                return wakeupIf;
 
            }).template extract<std::packaged_task<R()>>([&isWorking, &exit](auto& state) {
                exit = state.isFinished();
                if (!exit && !isWorking) {
                    isWorking = true;
                    state.waitingFinish();
                }
                return state.takeTask();
            });
            if (!exit) { task(); }
        }
    }
 
    SharedState<State> _state;
};
 
template<typename R>
class ThreadPool<R>::State final
{
public:
    using CreateWorker = std::function<std::thread()>;
 
    State(unsigned maxThreads, CreateWorker createWorker):
        _createWorker(std::move(createWorker)),
        _workers(detectMaxThreads(maxThreads)),
        _workersCount(0),
        _waitingCount(0),
        _exit(false)
    {
    }
 
    unsigned getMaxThreads() const { return _workers.size(); }
 
    unsigned getWorkersCount() const { return _workersCount; }
 
    unsigned getPendingTasksCount() const { return _tasks.size(); }
 
    bool hasTasks() const { return !_tasks.empty(); }
 
    unsigned getWaitingCount() const { return _waitingCount; }
 
    bool isFinished() const { return _exit; }
 
    template<typename F, typename... A>
    inline std::future<R> addTask(F&& task, A&&... args) {
        checkLogic(!isFinished());
        _tasks.emplace(std::bind(std::forward<F>(task), std::forward<A>(args)...));
        reviewWorkers();
        return _tasks.back().get_future();
    }
 
    void waitingBegin() {
        _waitingCount++;
    }
 
    void waitingFinish() {
        if (_waitingCount > 0) { _waitingCount--; }
    }
 
    std::packaged_task<R()> takeTask() {
        if (_tasks.empty()) { return std::packaged_task<R()>(); }
        auto task = std::move(_tasks.front());
        _tasks.pop();
        return task;
    }
 
    std::vector<std::thread> finish() {
        _exit = true;
        clearTasks();
        std::vector<std::thread> result(_workersCount);
        for (auto i = 0u; i < _workersCount; i++) {
            result[i] = std::move(_workers[i]);
        }
        return result;
    }
 
private:
    void reviewWorkers() {
        if (_workersCount < _workers.size() && !_waitingCount) { _workers[_workersCount++] = _createWorker(); }
    }
 
    void clearTasks() {
        std::queue<std::packaged_task<R()>> empty;
        std::swap(_tasks, empty);
    }
 
    static unsigned defaultMaxThreads() {
        constexpr auto DefaultMaxThreads = 4u;
        const auto systemCpuCount = std::thread::hardware_concurrency();
        return systemCpuCount ? systemCpuCount : DefaultMaxThreads;
    }
 
    static unsigned detectMaxThreads(unsigned desiredMaxThreads) {
        return desiredMaxThreads > 0 ? desiredMaxThreads : defaultMaxThreads();
    }
 
    std::function<std::thread()> _createWorker;
    std::vector<std::thread> _workers;
    std::queue<std::packaged_task<R()>> _tasks;
    unsigned _workersCount;
    unsigned _waitingCount;
    bool _exit;
};
 
template<typename R, typename... A>
class Pipeline
{
public:
 
    using Signature = std::function<R(A...)>;
 
    virtual ~Pipeline() = default;
 
    virtual std::future<R> addTask(Signature task, A&&... args) = 0;
};
 
template<typename R, typename... A>
class PipelineThreadPool: public Pipeline<R, A...>
{
public:
    PipelineThreadPool(const PipelineThreadPool& other) = default;
    PipelineThreadPool(PipelineThreadPool&& other) = default;
 
    PipelineThreadPool& operator=(const PipelineThreadPool& other) = default;
    PipelineThreadPool& operator=(PipelineThreadPool&& other) = default;
 
    std::future<R> addTask(typename Pipeline<R, A...>::Signature task, A&&... args) override {
        return _pool->addTask(std::forward<decltype(task)>(task), std::forward<A>(args)...);
    }
 
    explicit PipelineThreadPool(unsigned maxThreads = 0): _pool(std::make_shared<ThreadPool<R>>(maxThreads)) {}
 
    explicit PipelineThreadPool(std::shared_ptr<ThreadPool<R>> pool): _pool(std::move(pool)) {}
 
    std::shared_ptr<ThreadPool<R>> getPool() const { return _pool; }
 
    static std::unique_ptr<Pipeline<R, A...>> create(unsigned maxThreads = 0) {
        return std::make_unique<PipelineThreadPool>(maxThreads);
    }
 
    static std::unique_ptr<Pipeline<R, A...>> create(std::shared_ptr<ThreadPool<R>> pool) {
        return std::make_unique<PipelineThreadPool>(pool);
    }
 
private:
    std::shared_ptr<ThreadPool<R>> _pool;
};
 
} // namespace CppMate
 
#endif // CPP_MATE_THREAD_POOL_HPP