quickpool.hpp

// Copyright 2026 Thomas Nagler (MIT License)
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
 
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
 
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
 
#pragma once
 
#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <exception>
#include <functional>
#include <future>
#include <iterator>
#include <limits>
#include <memory>
#include <mutex>
#include <numeric>
#include <stdexcept>
#include <thread>
#include <tuple>
#include <type_traits>
#include <utility>
#include <vector>
 
#if (defined __linux__ || defined AFFINITY)
#include <pthread.h>
#endif
 
#if __cplusplus >= 201703L || (defined(_MSVC_LANG) && _MSVC_LANG >= 201703L)
#define QUICKPOOL_HAS_CPP17 1
#else
#define QUICKPOOL_HAS_CPP17 0
#endif
 
// Layout of quickpool.hpp
//
// 1. Memory related utilities.
//    - Memory order aliases
//    - Class for padding bytes
//    - Memory aligned allocation utilities
//    - Class for cache aligned atomics
//    - Class for load/assign atomics with relaxed order
// 2. Loop related utilities.
//    - Worker class for parallel for loops
// 3. Scheduling utilities.
//    - Ring buffer
//    - Task queue
//    - Task manager
// 4. Thread pool class
// 5. Free-standing functions (main API)
 
namespace quickpool {
 
namespace detail {
 
template<class Function, class... Args>
struct Task
{
#if QUICKPOOL_HAS_CPP17
    using type = std::invoke_result_t<Function, Args...>;
 
    static auto make(Function&& f, Args&&... args)
    {
        return [f = std::forward<Function>(f),
                args = std::make_tuple(std::forward<Args>(args)...)]() mutable {
            return std::apply(f, args);
        };
    }
#else
    using type = decltype(std::declval<Function>()(std::declval<Args>()...));
 
    static auto make(Function&& f, Args&&... args)
      -> decltype(std::bind(std::forward<Function>(f),
                            std::forward<Args>(args)...))
    {
        return std::bind(std::forward<Function>(f),
                         std::forward<Args>(args)...);
    }
#endif
};
 
} // namespace detail
 
// 1. --------------------------------------------------------------------------
 
namespace mem {
 
static constexpr std::memory_order relaxed = std::memory_order_relaxed;
static constexpr std::memory_order acquire = std::memory_order_acquire;
static constexpr std::memory_order release = std::memory_order_release;
static constexpr std::memory_order seq_cst = std::memory_order_seq_cst;
 
namespace padding_impl {
 
constexpr size_t
mod(size_t a, size_t b)
{
    return a - b * (a / b);
}
 
// Padding bytes from end of aligned object until next alignment point. char[]
// must hold at least one byte.
template<class T, size_t Align>
struct padding_bytes
{
    static constexpr size_t free_space =
      Align - mod(sizeof(std::atomic<T>), Align);
    static constexpr size_t required = free_space > 1 ? free_space : 1;
    char padding_[required];
};
 
struct empty_struct
{};
 
template<class T, size_t Align>
struct padding
  : std::conditional<mod(sizeof(std::atomic<T>), Align) != 0,
                     padding_bytes<T, Align>,
                     empty_struct>::type
{};
 
} // end namespace padding_impl
 
namespace aligned {
 
// The alloc/dealloc mechanism is pretty much
// https://www.boost.org/doc/libs/1_76_0/boost/align/detail/aligned_alloc.hpp
 
inline void*
alloc(size_t alignment, size_t size) noexcept
{
    // Make sure alignment is at least that of void*.
    alignment = (alignment >= alignof(void*)) ? alignment : alignof(void*);
 
    // Allocate enough space required for object and a void*.
    size_t space = size + alignment + sizeof(void*);
    void* p = std::malloc(space);
    if (p == nullptr) {
        return nullptr;
    }
 
    // Shift pointer to leave space for void*.
    void* p_algn = static_cast<char*>(p) + sizeof(void*);
    space -= sizeof(void*);
 
    // Shift pointer further to ensure proper alignment.
    (void)std::align(alignment, size, p_algn, space);
 
    // Store unaligned pointer with offset sizeof(void*) before aligned
    // location. Later we'll know where to look for the pointer telling
    // us where to free what we malloc()'ed above.
    *(static_cast<void**>(p_algn) - 1) = p;
 
    return p_algn;
}
 
inline void
free(void* ptr) noexcept
{
    if (ptr) {
        std::free(*(static_cast<void**>(ptr) - 1));
    }
}
 
template<class T, std::size_t Alignment = 64>
class allocator : public std::allocator<T>
{
  private:
    static constexpr size_t min_align =
      (Alignment >= alignof(void*)) ? Alignment : alignof(void*);
 
  public:
    template<class U>
    struct rebind
    {
        typedef allocator<U, Alignment> other;
    };
 
    allocator() noexcept
      : std::allocator<T>()
    {}
 
    template<class U, std::size_t UAlignment>
    allocator(const allocator<U, UAlignment>& other) noexcept
      : std::allocator<T>(other)
    {}
 
    T* allocate(size_t size, const void* = 0)
    {
        if (size == 0) {
            return 0;
        }
        void* p = mem::aligned::alloc(min_align, sizeof(T) * size);
        if (!p) {
            throw std::bad_alloc();
        }
        return static_cast<T*>(p);
    }
 
#if defined(__cpp_lib_allocate_at_least)
    std::allocation_result<T*> allocate_at_least(std::size_t size)
    {
        return { allocate(size), size };
    }
#endif
 
    void deallocate(T* ptr, size_t) { mem::aligned::free(ptr); }
 
    template<class U, class... Args>
    void construct(U* ptr, Args&&... args)
    {
      ::new (static_cast<void *>(ptr)) U(std::forward<Args>(args)...);
    }
 
    template <class U>
    void destroy(U *ptr) noexcept
    {
        (void)ptr;
        ptr->~U();
    }
};
 
template<class T, size_t Align = 64>
struct alignas(Align) atomic
  : public std::atomic<T>
  , private padding_impl::padding<T, Align>
{
  public:
    atomic() noexcept = default;
 
    atomic(T desired) noexcept
      : std::atomic<T>(desired)
    {}
 
    // Assignment operators have been deleted, must redefine.
    T operator=(T x) noexcept { return std::atomic<T>::operator=(x); }
    T operator=(T x) volatile noexcept { return std::atomic<T>::operator=(x); }
 
    static void* operator new(size_t count) noexcept
    {
        return mem::aligned::alloc(Align, count);
    }
 
    static void operator delete(void* ptr) { mem::aligned::free(ptr); }
};
 
template<typename T>
struct relaxed_atomic : public mem::aligned::atomic<T>
{
    explicit relaxed_atomic(T value)
      : mem::aligned::atomic<T>(value)
    {}
 
    operator T() const noexcept { return this->load(mem::relaxed); }
 
    T operator=(T desired) noexcept
    {
        this->store(desired, mem::relaxed);
        return desired;
    }
};
 
template<class T, size_t Alignment = 64>
using vector = std::vector<T, mem::aligned::allocator<T, Alignment>>;
 
} // end namespace aligned
 
} // end namespace mem
 
// 2. --------------------------------------------------------------------------
 
namespace loop {
 
template<class Iterator, class UnaryFunction, class Pool>
void
parallel_for_each_impl(Iterator begin,
                       Iterator end,
                       int size,
                       UnaryFunction f,
                       Pool& pool,
                       std::random_access_iterator_tag)
{
    (void)end;
    pool.parallel_for(0, static_cast<int>(size), [=](int i) { f(begin[i]); });
}
 
template<class Iterator, class UnaryFunction, class Pool, class IteratorCategory>
void
parallel_for_each_impl(Iterator begin,
                       Iterator end,
                       int size,
                       UnaryFunction f,
                       Pool& pool,
                       IteratorCategory)
{
    auto iterators = std::make_shared<std::vector<Iterator>>();
    iterators->reserve(static_cast<size_t>(size));
    for (auto it = begin; it != end; ++it) {
        iterators->push_back(it);
    }
    pool.parallel_for(0, static_cast<int>(iterators->size()), [=](int i) {
        f(*iterators->at(static_cast<size_t>(i)));
    });
}
 
struct State
{
    int pos; 
    int end; 
};
 
template<typename Function>
struct Worker
{
    Worker() {}
    Worker(int begin, int end, Function fun)
      : state{ State{ begin, end } }
      , f{ fun }
    {}
 
    Worker(Worker&& other)
      : state{ other.state.load() }
      , f{ std::forward<Function>(other.f) }
    {}
 
    size_t tasks_left() const
    {
        State s = state.load();
        return (s.end > s.pos) ? static_cast<size_t>(s.end - s.pos)
                               : static_cast<size_t>(0);
    }
 
    bool done() const { return (tasks_left() == 0); }
 
    void run(std::shared_ptr<mem::aligned::vector<Worker>> others)
    {
        State s, s_old; // temporary state variables
        do {
            s = state.load();
            if (s.pos < s.end) {
                // Protect slot by trying to advance position before doing
                // work.
                s_old = s;
                s.pos++;
 
                // Another worker might have changed the end of the range in
                // the meanwhile. Check atomically if the state is unaltered
                // and, if so, replace by advanced state.
                if (state.compare_exchange_weak(s_old, s)) {
                    f(s_old.pos); // succeeded, do work
                } else {
                    continue; // failed, try again
                }
            }
            if (s.pos == s.end) {
                // Reached end of own range, steal range from others. Range
                // remains empty if all work is done, so we can leave the
                // loop.
                this->steal_range(*others);
            }
        } while (!this->done());
    }
 
    void steal_range(mem::aligned::vector<Worker>& workers)
    {
        do {
            Worker& other = find_victim(workers);
            State s = other.state.load();
            if (s.pos >= s.end) {
                continue; // other range is empty by now
            }
 
            // Remove second half of the range. Check atomically if the
            // state is unaltered and, if so, replace with reduced range.
            auto s_old = s;
            s.end -= (s.end - s.pos + 1) / 2;
            if (other.state.compare_exchange_weak(s_old, s)) {
                // succeeded, update own range
                state = State{ s.end, s_old.end };
                break;
            }
        } while (!all_done(workers)); // failed steal, try again
    }
 
    bool all_done(const mem::aligned::vector<Worker>& workers)
    {
        for (const auto& worker : workers) {
            if (!worker.done())
                return false;
        }
        return true;
    }
 
    Worker& find_victim(mem::aligned::vector<Worker>& workers)
    {
        size_t best = 0;
        size_t most_tasks_left = 0;
        for (size_t i = 0; i < workers.size(); ++i) {
            const auto tasks_left = workers[i].tasks_left();
            if (tasks_left > most_tasks_left) {
                best = i;
                most_tasks_left = tasks_left;
            }
        }
        return workers[best];
    }
 
    mem::aligned::relaxed_atomic<State> state; 
    Function f; //< function applied to the loop index
};
 
template<typename Function>
std::shared_ptr<mem::aligned::vector<Worker<Function>>>
create_workers(const Function& f, int begin, int end, size_t num_workers)
{
    auto num_tasks = std::max(end - begin, static_cast<int>(0));
    num_workers = std::max(num_workers, static_cast<size_t>(1));
    auto workers = std::make_shared<mem::aligned::vector<Worker<Function>>>();
    workers->reserve(num_workers);
    for (size_t i = 0; i < num_workers; i++) {
        const auto first =
          begin + static_cast<int>(static_cast<size_t>(num_tasks) * i /
                                   num_workers);
        const auto last =
          begin + static_cast<int>(static_cast<size_t>(num_tasks) * (i + 1) /
                                   num_workers);
        workers->emplace_back(first, last, f);
    }
    return workers;
}
 
} // end namespace loop
 
// 3. -------------------------------------------------------------------------
 
namespace sched {
 
template<typename T>
class RingBuffer
{
  public:
    explicit RingBuffer(size_t capacity)
      : buffer_{
          std::unique_ptr<std::atomic<T>[]>(new std::atomic<T>[capacity])
        }
      , capacity_{ capacity }
      , mask_{ capacity - 1 }
    {}
 
    size_t capacity() const { return capacity_; }
 
    void set_entry(size_t i, T val)
    {
        buffer_[i & mask_].store(val, mem::relaxed);
    }
 
    T get_entry(size_t i) const
    {
        return buffer_[i & mask_].load(mem::relaxed);
    }
 
    RingBuffer<T>* enlarged_copy(size_t bottom, size_t top) const
    {
        RingBuffer<T>* new_buffer = new RingBuffer{ 2 * capacity_ };
        for (size_t i = top; i != bottom; ++i)
            new_buffer->set_entry(i, this->get_entry(i));
        return new_buffer;
    }
 
  private:
    std::unique_ptr<std::atomic<T>[]> buffer_;
    size_t capacity_;
    size_t mask_;
};
 
class TaskQueue
{
    using Task = std::function<void()>;
 
    struct TaskNode
    {
        Task task;
        TaskNode* next{ nullptr };
    };
 
  public:
    TaskQueue(size_t capacity = 256)
      : buffer_{ new RingBuffer<TaskNode*>(capacity) }
    {}
 
    ~TaskQueue() noexcept
    {
        delete buffer_.load();
    }
 
    TaskQueue(TaskQueue const& other) = delete;
    TaskQueue& operator=(TaskQueue const& other) = delete;
 
    bool empty() const
    {
        return (bottom_.load(mem::relaxed) <= top_.load(mem::relaxed));
    }
 
    void push(Task&& task)
    {
        // Must hold lock in case of multiple producers.
        std::unique_lock<std::mutex> lk(mutex_);
        auto b = bottom_.load(mem::relaxed);
        auto t = top_.load(mem::acquire);
        RingBuffer<TaskNode*>* buf_ptr = buffer_.load(mem::relaxed);
 
        const auto size = b - t;
        if (buf_ptr->capacity() < size + 1) {
            // Buffer is full, create enlarged copy before continuing.
            auto old_buf = buf_ptr;
            buf_ptr = std::move(buf_ptr->enlarged_copy(b, t));
            old_buffers_.emplace_back(old_buf);
            buffer_.store(buf_ptr, mem::release);
        }
 
        auto node = acquire_node();
        try {
            node->task = std::forward<Task>(task);
        } catch (...) {
            recycle_node(node);
            throw;
        }
        buf_ptr->set_entry(b, node);
        bottom_.store(b + 1, mem::release);
 
        lk.unlock(); // can release before signal
        cv_.notify_one();
    }
 
    bool try_pop(Task& task)
    {
        auto t = top_.load(mem::acquire);
        std::atomic_thread_fence(mem::seq_cst);
        auto b = bottom_.load(mem::acquire);
 
        if (t < b) {
            // Must load task pointer before acquiring the slot, because it
            // could be overwritten immediately after.
            auto node = buffer_.load(mem::acquire)->get_entry(t);
 
            // Atomically try to advance top.
            if (top_.compare_exchange_strong(
                  t, t + 1, mem::seq_cst, mem::relaxed)) {
                task = std::move(node->task); // won race, get task
                recycle_node(node);
                return true;
            }
        }
        return false; // queue is empty or lost race
    }
 
    void wait()
    {
        std::unique_lock<std::mutex> lk(mutex_);
        cv_.wait(lk, [this] { return !this->empty() || stopped_; });
    }
 
    void stop()
    {
        {
            std::lock_guard<std::mutex> lk(mutex_);
            stopped_ = true;
        }
        cv_.notify_one();
    }
 
    void wake_up()
    {
        {
            std::lock_guard<std::mutex> lk(mutex_);
        }
        cv_.notify_one();
    }
 
  private:
    mem::aligned::atomic<size_t> top_{ 0 };
    mem::aligned::atomic<size_t> bottom_{ 0 };
 
    std::atomic<RingBuffer<TaskNode*>*> buffer_{ nullptr };
 
    std::vector<std::unique_ptr<RingBuffer<TaskNode*>>> old_buffers_;
 
    std::vector<std::unique_ptr<TaskNode>> allocated_nodes_;
 
    std::atomic<TaskNode*> free_nodes_{ nullptr };
 
    // Only push() calls acquire_node(), and push() holds mutex_, while many
    // worker threads may recycle nodes concurrently.
    TaskNode* acquire_node()
    {
        auto node = free_nodes_.load(mem::acquire);
        while (node != nullptr) {
            auto next = node->next;
            if (free_nodes_.compare_exchange_weak(
                  node, next, mem::acquire, mem::acquire)) {
                node->next = nullptr;
                return node;
            }
        }
 
        allocated_nodes_.emplace_back(new TaskNode);
        return allocated_nodes_.back().get();
    }
 
    void recycle_node(TaskNode* node) noexcept
    {
        node->task = nullptr;
        auto head = free_nodes_.load(mem::relaxed);
        do {
            node->next = head;
        } while (!free_nodes_.compare_exchange_weak(
          head, node, mem::release, mem::relaxed));
    }
 
    std::mutex mutex_;
    std::condition_variable cv_;
    bool stopped_{ false };
};
 
class TaskManager
{
  public:
    explicit TaskManager(size_t num_queues)
      : queues_(std::max(num_queues, static_cast<size_t>(1)))
      , num_queues_(std::max(num_queues, static_cast<size_t>(1)))
      , owner_id_(std::this_thread::get_id())
    {}
 
    TaskManager& operator=(TaskManager&& other)
    {
        std::swap(queues_, other.queues_);
        num_queues_ = other.num_queues_;
        status_ = other.status_.load();
        num_waiting_ = other.num_waiting_.load();
        push_idx_ = other.push_idx_.load();
        todo_ = other.todo_.load();
        return *this;
    }
 
    void resize(size_t num_queues)
    {
        if (!done()) {
            throw std::logic_error("cannot resize with pending tasks");
        }
        num_queues_ = std::max(num_queues, static_cast<size_t>(1));
        if (num_queues_ > queues_.size()) {
            queues_ = mem::aligned::vector<TaskQueue>(num_queues_);
            // thread pool must have stopped the manager, reset
            num_waiting_ = 0;
            todo_ = 0;
            status_ = Status::running;
        }
    }
 
    template<typename Task>
    void push(Task&& task)
    {
        rethrow_exception(); // push() throws if a task has errored.
        if (is_running()) {
            todo_.fetch_add(1, mem::release);
            try {
                queues_[push_idx_++ % num_queues_].push(task);
            } catch (...) {
                report_success();
                throw;
            }
        }
    }
 
    template<typename Task>
    bool try_pop(Task& task, size_t worker_id = 0)
    {
        // Always start pop cycle at own queue to avoid contention.
        for (size_t k = 0; k < num_queues_; k++) {
            if (queues_[(worker_id + k) % num_queues_].try_pop(task)) {
                if (is_running()) {
                    return true;
                } else {
                    // Throw away task if pool has stopped or errored.
                    return false;
                }
            }
        }
 
        return false;
    }
 
    void wake_up_all_workers()
    {
        for (auto& q : queues_)
            q.wake_up();
    }
 
    void wait_for_jobs(size_t id)
    {
        if (has_errored()) {
            // Main thread may be waiting to reset the pool.
            std::lock_guard<std::mutex> lk(mtx_);
            if (++num_waiting_ == queues_.size())
                cv_.notify_all();
        } else {
            ++num_waiting_;
        }
 
        queues_[id].wait();
        --num_waiting_;
    }
 
    void wait_for_finish(size_t millis = 0)
    {
        if (called_from_owner_thread() && is_running()) {
            auto wake_up = [this] { return (todo_ <= 0) || !is_running(); };
            std::unique_lock<std::mutex> lk(mtx_);
            if (millis == 0) {
                cv_.wait(lk, wake_up);
            } else {
                cv_.wait_for(lk, std::chrono::milliseconds(millis), wake_up);
            }
        }
        rethrow_exception();
    }
 
    bool called_from_owner_thread() const
    {
        return (std::this_thread::get_id() == owner_id_);
    }
 
    void report_success()
    {
 
        auto n = todo_.fetch_sub(1, mem::release) - 1;
        if (n == 0) {
            // all jobs are done; lock before signal to prevent spurious
            // failure
            {
                std::lock_guard<std::mutex> lk{ mtx_ };
            }
            cv_.notify_all();
        }
    }
 
    void report_fail(std::exception_ptr err_ptr)
    {
        std::lock_guard<std::mutex> lk(mtx_);
        if (has_errored()) // only catch first exception
            return;
        err_ptr_ = err_ptr;
        status_ = Status::errored;
 
        // Some threads may change todo_ after we stop. The large
        // negative number forces them to exit the processing loop.
        todo_.store(std::numeric_limits<int>::min() / 2);
        cv_.notify_all();
    }
 
    void stop()
    {
        {
            std::lock_guard<std::mutex> lk(mtx_);
            status_ = Status::stopped;
        }
        // Worker threads wait on queue-specific mutex -> notify all queues.
        for (auto& q : queues_)
            q.stop();
    }
 
    void rethrow_exception()
    {
        // Exceptions are only thrown from the owner thread, not in workers.
        if (called_from_owner_thread() && has_errored()) {
            {
                // Wait for all threads to idle so we can clean up after
                // them.
                std::unique_lock<std::mutex> lk(mtx_);
                cv_.wait(lk, [this] { return num_waiting_ == queues_.size(); });
            }
            // Before throwing: restore defaults for potential future use of
            // the task manager.
            todo_ = 0;
            auto current_exception = err_ptr_;
            err_ptr_ = nullptr;
            status_ = Status::running;
 
            std::rethrow_exception(current_exception);
        }
    }
 
    bool is_running() const
    {
        return status_.load(mem::relaxed) == Status::running;
    }
 
    bool has_errored() const
    {
        return status_.load(mem::relaxed) == Status::errored;
    }
 
    bool stopped() const
    {
        return status_.load(mem::relaxed) == Status::stopped;
    }
 
    bool done() const { return (todo_.load(mem::relaxed) <= 0); }
 
  private:
    mem::aligned::vector<TaskQueue> queues_;
    size_t num_queues_;
 
    mem::aligned::relaxed_atomic<size_t> num_waiting_{ 0 };
    mem::aligned::relaxed_atomic<size_t> push_idx_{ 0 };
    mem::aligned::atomic<int> todo_{ 0 };
 
    const std::thread::id owner_id_;
    enum class Status
    {
        running,
        errored,
        stopped
    };
    mem::aligned::atomic<Status> status_{ Status::running };
    std::mutex mtx_;
    std::condition_variable cv_;
    std::exception_ptr err_ptr_{ nullptr };
};
 
// find out which cores are allowed for use by pthread
inline std::vector<size_t>
get_avail_cores()
{
    auto ncores = std::thread::hardware_concurrency();
    std::vector<size_t> avail_cores;
    avail_cores.reserve(ncores);
#if (defined __linux__)
    cpu_set_t cpuset;
    int rc = pthread_getaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
    if (rc != 0) {
        throw std::runtime_error("Error calling pthread_getaffinity_np");
    }
    for (size_t id = 0; id < ncores; id++) {
        if (CPU_ISSET(id, &cpuset)) {
            avail_cores.push_back(id);
        }
    }
#endif
    return avail_cores;
}
 
inline size_t
num_cores_avail()
{
#if (defined __linux__)
    return get_avail_cores().size();
#endif
    return std::thread::hardware_concurrency();
}
 
} // end namespace sched
 
// 4. ------------------------------------------------------------------------
 
class ThreadPool
{
  public:
    explicit ThreadPool(size_t threads = sched::num_cores_avail())
      : task_manager_{ threads }
    {
        set_active_threads(threads);
    }
 
    ~ThreadPool()
    {
        task_manager_.stop();
        join_threads();
    }
 
    ThreadPool(ThreadPool&&) = delete;
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
    ThreadPool& operator=(ThreadPool&& other) = delete;
 
    static ThreadPool& global_instance()
    {
#ifdef _WIN32
        // Must leak resource, because windows + R deadlock otherwise.
        // Memory is released on shutdown.
        static auto ptr = new ThreadPool;
        return *ptr;
#else
        static ThreadPool instance_;
        return instance_;
#endif
    }
 
    void set_active_threads(size_t threads)
    {
        if (!task_manager_.called_from_owner_thread())
            return;
 
        if (threads == active_threads_.load(mem::relaxed)) {
            return;
        }
 
        this->wait();
        if (workers_.size() > 0) {
            task_manager_.stop();
            join_threads();
            workers_.clear();
        }
 
        task_manager_ = quickpool::sched::TaskManager{ threads };
        workers_ = std::vector<std::thread>{ threads };
        for (size_t id = 0; id < threads; ++id) {
            add_worker(id);
        }
#if (defined __linux__)
        if (threads > 0) {
            set_thread_affinity();
        }
#endif
        active_threads_ = threads;
    }
 
    size_t get_active_threads() const { return active_threads_; }
 
    template<class Function, class... Args>
    void push(Function&& f, Args&&... args)
    {
        if (active_threads_ == 0) {
            std::forward<Function>(f)(std::forward<Args>(args)...);
            return;
        }
        task_manager_.push(detail::Task<Function, Args...>::make(
          std::forward<Function>(f), std::forward<Args>(args)...));
    }
 
    template<class Function, class... Args>
    auto async(Function&& f, Args&&... args)
      -> std::future<typename detail::Task<Function, Args...>::type>
    {
        auto pack = detail::Task<Function, Args...>::make(
          std::forward<Function>(f), std::forward<Args>(args)...);
        using result_t = typename detail::Task<Function, Args...>::type;
        using pack_t = std::packaged_task<result_t()>;
        auto task_ptr = std::make_shared<pack_t>(std::move(pack));
        this->push([task_ptr] { (*task_ptr)(); });
        return task_ptr->get_future();
    }
 
    template<class UnaryFunction>
    void parallel_for(int begin, int end, UnaryFunction f)
    {
        if (end <= begin) {
            return;
        }
        const auto active_threads = active_threads_.load(mem::relaxed);
        if (active_threads == 0) {
            for (auto i = begin; i < end; ++i) {
                f(i);
            }
            return;
        }
 
        // each worker has its dedicated range, but can steal part of
        // another worker's ranges when done with own
        const auto num_tasks = static_cast<size_t>(end - begin);
        const auto n =
          std::min(std::max(active_threads, static_cast<size_t>(1)), num_tasks);
        auto workers = loop::create_workers<UnaryFunction>(f, begin, end, n);
        for (size_t k = 0; k < n; k++) {
            this->push([=] { workers->at(k).run(workers); });
        }
        this->wait();
    }
 
    template<class Items, class UnaryFunction>
    inline void parallel_for_each(Items& items, UnaryFunction f)
    {
        auto begin = std::begin(items);
        auto size = std::distance(begin, std::end(items));
        if (size <= 0) {
            return;
        }
        if (size > std::numeric_limits<int>::max()) {
            throw std::length_error("parallel_for_each range is too large");
        }
        typedef typename std::iterator_traits<decltype(begin)>::iterator_category
          iterator_category;
        loop::parallel_for_each_impl(begin,
                                     std::end(items),
                                     static_cast<int>(size),
                                     f,
                                     *this,
                                     iterator_category{});
    }
 
    void wait(size_t millis = 0) { task_manager_.wait_for_finish(millis); }
 
    // state, and rethrows an exception if one is pending.
    void stop_and_reset()  
    { 
        task_manager_.report_fail(std::current_exception());
        task_manager_.wait_for_finish();
        task_manager_.rethrow_exception();
    }
 
    bool done() const { return task_manager_.done(); }
 
    static void* operator new(size_t count)
    {
        return mem::aligned::alloc(alignof(ThreadPool), count);
    }
 
    static void operator delete(void* ptr) { mem::aligned::free(ptr); }
 
  private:
    void join_threads()
    {
        for (auto& worker : workers_) {
            if (worker.joinable())
                worker.join();
        }
    }
 
    void add_worker(size_t id)
    {
        workers_[id] = std::thread([&, id] {
            std::function<void()> task;
            while (!task_manager_.stopped()) {
                task_manager_.wait_for_jobs(id);
                do {
                    // inner while to save some time calling done()
                    while (task_manager_.try_pop(task, id))
                        this->execute_safely(task);
                } while (!task_manager_.done());
            }
        });
    }
 
#if (defined __linux__)
    void set_thread_affinity()
    {
        cpu_set_t cpuset;
        auto avail_cores = sched::get_avail_cores();
        for (size_t id = 0; id < workers_.size(); id++) {
            CPU_ZERO(&cpuset);
            CPU_SET(avail_cores[id % avail_cores.size()], &cpuset);
            int rc = pthread_setaffinity_np(
              workers_.at(id).native_handle(), sizeof(cpu_set_t), &cpuset);
            if (rc != 0) {
                throw std::runtime_error(
                  "Error calling pthread_setaffinity_np");
            }
        }
    }
#endif
 
    void execute_safely(std::function<void()>& task)
    {
        try {
            task();
            task_manager_.report_success();
        } catch (...) {
            task_manager_.report_fail(std::current_exception());
        }
    }
 
    sched::TaskManager task_manager_;
    std::vector<std::thread> workers_;
    std::atomic_size_t active_threads_{ 0 };
};
 
// 5. ---------------------------------------------------
 
 
template<class Function, class... Args>
inline void
push(Function&& f, Args&&... args)
{
    ThreadPool::global_instance().push(std::forward<Function>(f),
                                       std::forward<Args>(args)...);
}
 
template<class Function, class... Args>
inline auto
async(Function&& f, Args&&... args)
  -> std::future<typename detail::Task<Function, Args...>::type>
{
    return ThreadPool::global_instance().async(std::forward<Function>(f),
                                               std::forward<Args>(args)...);
}
 
inline void
wait()
{
    ThreadPool::global_instance().wait();
}
 
inline bool
done()
{
    return ThreadPool::global_instance().done();
}
 
inline void
set_active_threads(size_t threads)
{
    ThreadPool::global_instance().set_active_threads(threads);
}
 
inline size_t
get_active_threads()
{
    return ThreadPool::global_instance().get_active_threads();
}
 
template<class UnaryFunction>
inline void
parallel_for(int begin, int end, UnaryFunction&& f)
{
    ThreadPool::global_instance().parallel_for(
      begin, end, std::forward<UnaryFunction>(f));
}
 
template<class Items, class UnaryFunction>
inline void
parallel_for_each(Items& items, UnaryFunction&& f)
{
    ThreadPool::global_instance().parallel_for_each(
      items, std::forward<UnaryFunction>(f));
}
 
} // end namespace quickpool
 
#undef QUICKPOOL_HAS_CPP17