#pragma once #include "unbounded_queue.hpp" #include #include #include #include #include #include #include #include #include class ThreadPool { public: explicit ThreadPool(const std::size_t thread_count_ = std::thread::hardware_concurrency()) : _queues(thread_count_), _count(thread_count_) { auto worker = [this](std::size_t i) { while(true) { Proc f; for(std::size_t n = 0; n < (_count * K); ++n) { if(_queues[(i + n) % _count].try_pop(f)) break; } if(!f && !_queues[i].pop(f)) break; f(); } }; _threads.reserve(thread_count_); for(std::size_t i = 0; i < thread_count_; ++i) _threads.emplace_back(worker, i); } ~ThreadPool() { for(auto& queue : _queues) queue.unblock(); for(auto& thread : _threads) thread.join(); } template void enqueue_work(F&& f_) { auto i = _index++; for(std::size_t n = 0; n < (_count * K); ++n) { if(_queues[(i + n) % _count].try_push(f_)) return; } _queues[i % _count].push(std::move(f_)); } template [[nodiscard]] std::future::type> enqueue_task(F&& f_) { using TaskReturnType = typename std::result_of::type; using Promise = std::promise; auto i = _index++; auto promise = std::make_shared(); auto future = promise->get_future(); auto work = [=]() { auto rv = f_(); promise->set_value(rv); }; for(std::size_t n = 0; n < (_count * K); ++n) { if(_queues[(i + n) % _count].try_push(work)) return future; } _queues[i % _count].push(std::move(work)); return future; } private: using Proc = std::function; using Queue = UnboundedQueue; using Queues = std::vector; Queues _queues; private: std::vector _threads; private: const std::size_t _count; std::atomic_uint _index; static const unsigned int K = 2; };