ThreadPool.h

Go to the documentation of this file.

 
#ifndef LLU_ASYNC_THREADPOOL_H
#define LLU_ASYNC_THREADPOOL_H
 
#include <atomic>
#include <deque>
#include <functional>
#include <future>
#include <thread>
#include <type_traits>
#include <vector>
 
#include "LLU/Async/Queue.h"
#include "LLU/Async/Utilities.h"
#include "LLU/Async/WorkStealingQueue.h"
 
namespace LLU::Async {
 
    template<typename Queue>
    class BasicThreadPool {
    public:
        using TaskType = typename Queue::value_type;
 
    public:
        BasicThreadPool() : BasicThreadPool(std::thread::hardware_concurrency()) {};
 
        explicit BasicThreadPool(unsigned threadCount) : joiner(threads) {
            try {
                for (unsigned i = 0; i < threadCount; ++i) {
                    threads.emplace_back(&BasicThreadPool::workerThread, this);
                }
            } catch (...) {
                done = true;
                throw;
            }
        }
 
        // Thread pool is non-copyable
        BasicThreadPool(const BasicThreadPool&) = delete;
        BasicThreadPool& operator=(const BasicThreadPool&) = delete;
        BasicThreadPool(BasicThreadPool&&) = delete;
        BasicThreadPool& operator=(BasicThreadPool&&) = delete;
 
        ~BasicThreadPool() {
            done = true;
            for ([[maybe_unused]] auto& t : threads) {
                workQueue.push(TaskType {[] {}});
            }
        }
 
        template<typename FunctionType, typename... Args>
        std::future<std::invoke_result_t<FunctionType, Args...>> submit(FunctionType&& f, Args&&... args) {
            auto task = Async::getPackagedTask(std::forward<FunctionType>(f), std::forward<Args>(args)...);
            auto res = task.get_future();
            workQueue.push(TaskType {std::move(task)});
            return res;
        }
 
        void runPendingTask() {
            TaskType task;
            workQueue.waitPop(task);
            task();
        }
 
    private:
        std::atomic_bool done = false;
        Queue workQueue;
        std::vector<std::thread> threads;
        Async::ThreadJoiner joiner;
 
        void workerThread() {
            while (!done) {
                runPendingTask();
            }
        }
    };
 
    template<typename PoolQueue, typename LocalQueue>
    class GenericThreadPool : public Async::Pausable {
    public:
        using TaskType = typename PoolQueue::value_type;
 
    public:
        GenericThreadPool() : GenericThreadPool(std::thread::hardware_concurrency()) {};
 
        explicit GenericThreadPool(unsigned threadCount) : joiner(threads) {
            try {
                for (unsigned i = 0; i < threadCount; ++i) {
                    queues.emplace_back(std::make_unique<LocalQueue>());
                }
                for (unsigned i = 0; i < threadCount; ++i) {
                    threads.emplace_back(&GenericThreadPool::workerThread, this, i);
                }
            } catch (...) {
                done = true;
                throw;
            }
        }
 
        // Thread pool is non-copyable
        GenericThreadPool(const GenericThreadPool&) = delete;
        GenericThreadPool& operator=(const GenericThreadPool&) = delete;
        GenericThreadPool(GenericThreadPool&&) = delete;
        GenericThreadPool& operator=(GenericThreadPool&&) = delete;
 
        ~GenericThreadPool() {
            done = true;
            resume();
        }
 
        template<typename FunctionType, typename... Args>
        std::future<std::invoke_result_t<FunctionType, Args...>> submit(FunctionType&& f, Args&&... args) {
            auto task = Async::getPackagedTask(std::forward<FunctionType>(f), std::forward<Args>(args)...);
            auto res = task.get_future();
            if (localWorkQueue) {
                localWorkQueue->push(TaskType {std::move(task)});
            } else {
                poolWorkQueue.push(TaskType {std::move(task)});
            }
            return res;
        }
 
        void runPendingTask() {
            TaskType task;
            if (popTaskFromLocalQueue(task) || popTaskFromPoolQueue(task) || popTaskFromOtherThreadQueue(task)) {
                task();
            } else {
                std::this_thread::yield();
            }
        }
 
    private:
        std::atomic_bool done = false;
        PoolQueue poolWorkQueue;
        std::vector<std::unique_ptr<LocalQueue>> queues;
        std::vector<std::thread> threads;
        Async::ThreadJoiner joiner;
        inline static thread_local LocalQueue* localWorkQueue = nullptr;
        inline static thread_local unsigned myIndex = 0;
 
        void workerThread(unsigned my_index_) {
            myIndex = my_index_;
            localWorkQueue = queues[myIndex].get();
            while (!done) {
                runPendingTask();
                checkPause();
            }
        }
        bool popTaskFromLocalQueue(TaskType& task) {
            return localWorkQueue && localWorkQueue->tryPop(task);
        }
        bool popTaskFromPoolQueue(TaskType& task) {
            return poolWorkQueue.tryPop(task);
        }
        bool popTaskFromOtherThreadQueue(TaskType& task) {
            for (unsigned i = 0; i < queues.size(); ++i) {
                unsigned const index = (myIndex + i + 1) % queues.size();
                if (queues[index]->trySteal(task)) {
                    return true;
                }
            }
            return false;
        }
    };
 
}  // namespace LLU::Async
 
namespace LLU {
    using BasicPool = Async::BasicThreadPool<Async::ThreadsafeQueue<Async::FunctionWrapper>>;
 
    using ThreadPool = Async::GenericThreadPool<Async::ThreadsafeQueue<Async::FunctionWrapper>, Async::WorkStealingQueue<std::deque<Async::FunctionWrapper>>>;
}// namespace LLU
 
#endif    // LLU_ASYNC_THREADPOOL_H