thread_pool.H

Go to the documentation of this file.

 
/*
                          Aleph_w
 
  Data structures & Algorithms
  version 2.0.0b
  https://github.com/lrleon/Aleph-w
 
  This file is part of Aleph-w library
 
  Copyright (c) 2002-2026 Leandro Rabindranath Leon
 
  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.
*/
 
 
#ifndef ALEPH_THREAD_POOL_H
#define ALEPH_THREAD_POOL_H
 
#include <future>
#include <queue>
#include <functional>
#include <condition_variable>
#include <atomic>
#include <vector>
#include <thread>
#include <mutex>
#include <stdexcept>
#include <type_traits>
#include <tuple>
#include <memory>
#include <limits>
#include <optional>
#include <chrono>
#include <numeric>
#include <iterator>
 
namespace Aleph
{
 
class queue_overflow_error : public std::overflow_error
{
public:
  explicit queue_overflow_error(size_t current_size, size_t hard_limit)
    : std::overflow_error("ThreadPool queue overflow: " + 
                          std::to_string(current_size) + " >= hard_limit " +
                          std::to_string(hard_limit)),
      current_size_(current_size), hard_limit_(hard_limit) {}
 
  [[nodiscard]] size_t current_size() const noexcept { return current_size_; }
  
  [[nodiscard]] size_t hard_limit() const noexcept { return hard_limit_; }
 
private:
  size_t current_size_;
  size_t hard_limit_;
};
 
struct ThreadPoolStats
{
  size_t tasks_completed = 0;         
  size_t tasks_failed = 0;            
  size_t current_queue_size = 0;      
  size_t current_active = 0;          
  size_t num_workers = 0;             
  size_t peak_queue_size = 0;         
  
  [[nodiscard]] size_t total_processed() const noexcept 
  { 
    return tasks_completed + tasks_failed; 
  }
  
  [[nodiscard]] double queue_utilization(size_t soft_limit) const noexcept
  {
    if (soft_limit == 0 || soft_limit == std::numeric_limits<size_t>::max())
      return 0.0;
    return (100.0 * current_queue_size) / soft_limit;
  }
};
 
using ExceptionCallback = std::function<void(std::exception_ptr)>;
 
class ThreadPool
{
  struct TaskBase
  {
    virtual ~TaskBase() = default;
    virtual void execute() = 0;
  };
 
  template <typename F>
  struct Task : TaskBase
  {
    F func;
 
    explicit Task(F&& f) : func(std::move(f)) {}
 
    void execute() override { func(); }
  };
 
  std::vector<std::thread> workers;
  std::queue<std::unique_ptr<TaskBase>> tasks;
  mutable std::mutex queue_mutex;
  std::condition_variable condition;        
  std::condition_variable space_available;  
  std::condition_variable idle_condition;   
  std::atomic<bool> stop{false};
  std::atomic<size_t> active_tasks{0};
  
  // Queue limits for bounded enqueue
  size_t soft_limit_ = std::numeric_limits<size_t>::max();  
  size_t hard_limit_ = std::numeric_limits<size_t>::max();  
  
  // Statistics
  std::atomic<size_t> tasks_completed_{0};
  std::atomic<size_t> tasks_failed_{0};
  std::atomic<size_t> peak_queue_size_{0};
  
  // Exception callback for detached tasks
  ExceptionCallback exception_callback_;
 
  void worker_loop()
  {
    while (true)
      {
        std::unique_ptr<TaskBase> task;
        bool should_notify_space = false;
        {
          std::unique_lock<std::mutex> lock(queue_mutex);
          condition.wait(lock, [this] {
            return stop or not tasks.empty();
          });
 
          if (stop and tasks.empty())
            return;
 
          // Check if we should notify waiters after popping
          should_notify_space = (tasks.size() == soft_limit_);
          
          task = std::move(tasks.front());
          tasks.pop();
          ++active_tasks;
        }
 
        // Notify any blocked enqueuers that space is available
        if (should_notify_space)
          space_available.notify_all();
 
        task->execute();
        
        // Update stats and check for idle state
        --active_tasks;
        
        // Notify wait_all if we're now idle
        {
          std::unique_lock<std::mutex> lock(queue_mutex);
          if (tasks.empty() && active_tasks == 0)
            idle_condition.notify_all();
        }
      }
  }
 
  void start_workers(size_t n)
  {
    workers.reserve(n);
    for (size_t i = 0; i < n; ++i)
      workers.emplace_back(&ThreadPool::worker_loop, this);
  }
 
  void stop_workers()
  {
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      stop = true;
    }
    condition.notify_all();
 
    for (auto& worker : workers)
      if (worker.joinable())
        worker.join();
 
    workers.clear();
  }
 
  template <typename F, typename... Args>
  static auto make_invocable(F&& f, Args&&... args)
  {
    // Decay types to store copies/moves, not references
    return [func = std::forward<F>(f),
            args_tuple = std::make_tuple(std::forward<Args>(args)...)]() mutable
    {
      return std::apply([&func](auto&&... a) {
        return std::invoke(std::move(func), std::forward<decltype(a)>(a)...);
      }, std::move(args_tuple));
    };
  }
 
public:
  explicit ThreadPool(size_t n_threads = std::thread::hardware_concurrency())
  {
    if (n_threads == 0)
      n_threads = 1;  // Ensure at least one worker
 
    start_workers(n_threads);
  }
 
  ThreadPool(const ThreadPool&) = delete;
  ThreadPool& operator=(const ThreadPool&) = delete;
 
  ThreadPool(ThreadPool&&) = delete;
  ThreadPool& operator=(ThreadPool&&) = delete;
 
  ~ThreadPool()
  {
    stop_workers();
  }
 
  template <typename F, typename... Args>
  [[nodiscard]] auto enqueue(F&& f, Args&&... args)
    -> std::future<std::invoke_result_t<F, Args...>>
  {
    using return_type = std::invoke_result_t<F, Args...>;
 
    // Create the invocable that captures function and arguments
    auto invocable = make_invocable(std::forward<F>(f), std::forward<Args>(args)...);
 
    // Wrap in packaged_task for future support
    auto promise = std::make_shared<std::promise<return_type>>();
    std::future<return_type> result = promise->get_future();
 
    // Capture stats counter for task completion tracking
    auto completed_counter = &tasks_completed_;
    
    // Create the task that will execute and set the promise
    auto task_func = [invocable = std::move(invocable), promise, completed_counter]() mutable {
      try
        {
          if constexpr (std::is_void_v<return_type>)
            {
              invocable();
              promise->set_value();
            }
          else
            {
              promise->set_value(invocable());
            }
          ++(*completed_counter);
        }
      catch (...)
        {
          promise->set_exception(std::current_exception());
          ++(*completed_counter);  // Still counts as processed
        }
    };
 
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
 
      if (stop)
        throw std::runtime_error("enqueue on stopped ThreadPool");
 
      tasks.push(std::make_unique<Task<decltype(task_func)>>(std::move(task_func)));
      
      // Update peak queue size
      size_t current_size = tasks.size();
      size_t peak = peak_queue_size_.load();
      while (current_size > peak and
             not peak_queue_size_.compare_exchange_weak(peak, current_size))
        ;
    }
 
    condition.notify_one();
    return result;
  }
 
  template <typename F, typename... Args>
  void enqueue_detached(F&& f, Args&&... args)
  {
    auto invocable = make_invocable(std::forward<F>(f), std::forward<Args>(args)...);
 
    // Capture stats and callback
    auto completed_counter = &tasks_completed_;
    auto failed_counter = &tasks_failed_;
    ExceptionCallback callback;
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      callback = exception_callback_;
    }
 
    auto task_func = [invocable = std::move(invocable), 
                      completed_counter, failed_counter,
                      callback = std::move(callback)]() mutable {
      try
        {
          invocable();
          ++(*completed_counter);
        }
      catch (...)
        {
          ++(*failed_counter);
          // Call exception callback if set
          if (callback)
            {
              try { callback(std::current_exception()); }
              catch (...) { /* Ignore callback exceptions */ }
            }
        }
    };
 
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
 
      if (stop)
        throw std::runtime_error("enqueue_detached on stopped ThreadPool");
 
      tasks.push(std::make_unique<Task<decltype(task_func)>>(std::move(task_func)));
      
      // Update peak queue size
      size_t current_size = tasks.size();
      size_t peak = peak_queue_size_.load();
      while (current_size > peak and
             not peak_queue_size_.compare_exchange_weak(peak, current_size))
        ;
    }
 
    condition.notify_one();
  }
 
  template <typename F, typename Container>
  [[nodiscard]] auto enqueue_bulk(F&& f, const Container& args_container)
    -> std::vector<std::future<std::invoke_result_t<F, typename Container::value_type>>>
  {
    using return_type = std::invoke_result_t<F, typename Container::value_type>;
    std::vector<std::future<return_type>> results;
    results.reserve(args_container.size());
 
    for (const auto& arg : args_container)
      results.push_back(enqueue(f, arg));
 
    return results;
  }
 
  void set_queue_limits(size_t soft_limit, 
                        size_t hard_limit = std::numeric_limits<size_t>::max())
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    soft_limit_ = soft_limit;
    hard_limit_ = (hard_limit == std::numeric_limits<size_t>::max() && 
                   soft_limit != std::numeric_limits<size_t>::max())
                  ? soft_limit * 10  // Default hard = 10x soft
                  : hard_limit;
  }
 
  [[nodiscard]] std::pair<size_t, size_t> get_queue_limits() const
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    return {soft_limit_, hard_limit_};
  }
 
  template <typename F, typename... Args>
  [[nodiscard]] auto enqueue_bounded(F&& f, Args&&... args)
    -> std::future<std::invoke_result_t<F, Args...>>
  {
    using return_type = std::invoke_result_t<F, Args...>;
 
    auto invocable = make_invocable(std::forward<F>(f), std::forward<Args>(args)...);
 
    auto promise = std::make_shared<std::promise<return_type>>();
    std::future<return_type> result = promise->get_future();
 
    auto task_func = [invocable = std::move(invocable), promise]() mutable {
      try
        {
          if constexpr (std::is_void_v<return_type>)
            {
              invocable();
              promise->set_value();
            }
          else
            {
              promise->set_value(invocable());
            }
        }
      catch (...)
        {
          promise->set_exception(std::current_exception());
        }
    };
 
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
 
      // Check hard limit first (throws immediately)
      if (tasks.size() >= hard_limit_)
        throw queue_overflow_error(tasks.size(), hard_limit_);
 
      // Block if at soft limit (wait for space)
      space_available.wait(lock, [this] {
        return stop || tasks.size() < soft_limit_;
      });
 
      if (stop)
        throw std::runtime_error("enqueue_bounded on stopped ThreadPool");
 
      // Re-check hard limit after waking up
      if (tasks.size() >= hard_limit_)
        throw queue_overflow_error(tasks.size(), hard_limit_);
 
      tasks.push(std::make_unique<Task<decltype(task_func)>>(std::move(task_func)));
    }
 
    condition.notify_one();
    return result;
  }
 
  template <typename F, typename... Args>
  void enqueue_bounded_detached(F&& f, Args&&... args)
  {
    auto invocable = make_invocable(std::forward<F>(f), std::forward<Args>(args)...);
 
    auto task_func = [invocable = std::move(invocable)]() mutable {
      try
        {
          invocable();
        }
      catch (...)
        {
          // Silently ignore exceptions in detached tasks
        }
    };
 
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
 
      // Check hard limit first
      if (tasks.size() >= hard_limit_)
        throw queue_overflow_error(tasks.size(), hard_limit_);
 
      // Block if at soft limit
      space_available.wait(lock, [this] {
        return stop || tasks.size() < soft_limit_;
      });
 
      if (stop)
        throw std::runtime_error("enqueue_bounded_detached on stopped ThreadPool");
 
      // Re-check hard limit after waking up
      if (tasks.size() >= hard_limit_)
        throw queue_overflow_error(tasks.size(), hard_limit_);
 
      tasks.push(std::make_unique<Task<decltype(task_func)>>(std::move(task_func)));
    }
 
    condition.notify_one();
  }
 
  template <typename F, typename... Args>
  [[nodiscard]] auto try_enqueue(F&& f, Args&&... args)
    -> std::optional<std::future<std::invoke_result_t<F, Args...>>>
  {
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      
      if (stop)
        throw std::runtime_error("try_enqueue on stopped ThreadPool");
      
      // Check if queue is at or above soft limit
      if (tasks.size() >= soft_limit_)
        return std::nullopt;
    }
 
    // Queue has space, delegate to regular enqueue
    return enqueue(std::forward<F>(f), std::forward<Args>(args)...);
  }
 
  template <typename F, typename... Args>
  bool try_enqueue_detached(F&& f, Args&&... args)
  {
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      
      if (stop)
        throw std::runtime_error("try_enqueue_detached on stopped ThreadPool");
      
      if (tasks.size() >= soft_limit_)
        return false;
    }
 
    enqueue_detached(std::forward<F>(f), std::forward<Args>(args)...);
    return true;
  }
 
  [[nodiscard]] size_t num_threads() const noexcept
  {
    return workers.size();
  }
 
  [[nodiscard]] size_t pending_tasks() const
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    return tasks.size();
  }
 
  [[nodiscard]] size_t running_tasks() const noexcept
  {
    return active_tasks.load();
  }
 
  [[nodiscard]] bool is_idle() const
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    return tasks.empty() && active_tasks == 0;
  }
 
  [[nodiscard]] bool is_stopped() const noexcept
  {
    return stop.load();
  }
 
  void shutdown()
  {
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      if (stop.exchange(true))
        return;  // Already stopped
    }
    condition.notify_all();
 
    for (auto& worker : workers)
      if (worker.joinable())
        worker.join();
 
    workers.clear();
  }
 
  void resize(size_t new_size)
  {
    if (new_size == 0)
      new_size = 1;
 
    if (new_size == workers.size())
      return;
 
    if (stop)
      throw std::runtime_error("cannot resize a stopped ThreadPool");
 
    // Stop current workers (but don't clear tasks)
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      stop = true;
    }
    condition.notify_all();
 
    for (auto& worker : workers)
      if (worker.joinable())
        worker.join();
 
    workers.clear();
 
    // Restart with new size
    stop = false;
    start_workers(new_size);
  }
 
  void wait_all(std::chrono::milliseconds poll_interval = std::chrono::milliseconds(1))
  {
    while (not is_idle())
      std::this_thread::sleep_for(poll_interval);
  }
 
  template <typename Rep, typename Period>
  [[nodiscard]] bool wait_all_for(std::chrono::duration<Rep, Period> timeout)
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    return idle_condition.wait_for(lock, timeout, [this] {
      return tasks.empty() && active_tasks == 0;
    });
  }
 
  template <typename Clock, typename Duration>
  [[nodiscard]] bool wait_all_until(std::chrono::time_point<Clock, Duration> deadline)
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    return idle_condition.wait_until(lock, deadline, [this] {
      return tasks.empty() && active_tasks == 0;
    });
  }
 
  [[nodiscard]] ThreadPoolStats get_stats() const
  {
    ThreadPoolStats stats;
    stats.tasks_completed = tasks_completed_.load();
    stats.tasks_failed = tasks_failed_.load();
    stats.current_active = active_tasks.load();
    stats.num_workers = workers.size();
    stats.peak_queue_size = peak_queue_size_.load();
    
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      stats.current_queue_size = tasks.size();
    }
    
    return stats;
  }
 
  void reset_stats()
  {
    tasks_completed_ = 0;
    tasks_failed_ = 0;
    peak_queue_size_ = 0;
  }
 
  void set_exception_callback(ExceptionCallback callback)
  {
    std::unique_lock<std::mutex> lock(queue_mutex);
    exception_callback_ = std::move(callback);
  }
 
  template <typename F, typename Container>
  [[nodiscard]] auto enqueue_batch(F&& f, const Container& args_list)
    -> std::vector<std::future<decltype(std::apply(f, *std::begin(args_list)))>>
  {
    using ArgsTuple = typename Container::value_type;
    using return_type = decltype(std::apply(f, std::declval<ArgsTuple>()));
    
    std::vector<std::future<return_type>> results;
    results.reserve(std::distance(std::begin(args_list), std::end(args_list)));
    
    std::vector<std::unique_ptr<TaskBase>> batch_tasks;
    batch_tasks.reserve(results.capacity());
    
    // Capture stats counter
    auto completed_counter = &tasks_completed_;
    
    for (const auto& args : args_list)
      {
        auto promise = std::make_shared<std::promise<return_type>>();
        results.push_back(promise->get_future());
        
        auto task_func = [func = f, args, promise, completed_counter]() mutable {
          try
            {
              if constexpr (std::is_void_v<return_type>)
                {
                  std::apply(func, args);
                  promise->set_value();
                }
              else
                {
                  promise->set_value(std::apply(func, args));
                }
              ++(*completed_counter);
            }
          catch (...)
            {
              promise->set_exception(std::current_exception());
              ++(*completed_counter);
            }
        };
        
        batch_tasks.push_back(
          std::make_unique<Task<decltype(task_func)>>(std::move(task_func)));
      }
    
    {
      std::unique_lock<std::mutex> lock(queue_mutex);
      
      if (stop)
        throw std::runtime_error("enqueue_batch on stopped ThreadPool");
      
      for (auto& task : batch_tasks)
        tasks.push(std::move(task));
      
      // Update peak queue size
      size_t current_size = tasks.size();
      size_t peak = peak_queue_size_.load();
      while (current_size > peak and
             not peak_queue_size_.compare_exchange_weak(peak, current_size))
        ;
    }
    
    // Notify all workers for batch
    condition.notify_all();
    
    return results;
  }
};
 
// =============================================================================
// Parallel Algorithms
// =============================================================================
 
template <typename Iterator, typename F>
void parallel_for(ThreadPool& pool, Iterator begin, Iterator end, F&& f, 
                  size_t chunk_size = 0)
{
  const size_t total = std::distance(begin, end);
  if (total == 0)
    return;
  
  // Auto-calculate chunk size if not specified
  if (chunk_size == 0)
    chunk_size = std::max(size_t(1), total / (pool.num_threads() * 4));
  
  std::vector<std::future<void>> futures;
  futures.reserve((total + chunk_size - 1) / chunk_size);
  
  // Check if F accepts (Iterator, Iterator) or (T&)
  if constexpr (std::is_invocable_v<F, Iterator, Iterator>)
    {
      // Chunk-based function
      for (Iterator chunk_begin = begin; chunk_begin < end; )
        {
          Iterator chunk_end = chunk_begin;
          std::advance(chunk_end, std::min(chunk_size, 
                       static_cast<size_t>(std::distance(chunk_begin, end))));
          
          futures.push_back(pool.enqueue([f, chunk_begin, chunk_end]() {
            f(chunk_begin, chunk_end);
          }));
          
          chunk_begin = chunk_end;
        }
    }
  else
    {
      // Per-element function
      for (Iterator chunk_begin = begin; chunk_begin < end; )
        {
          Iterator chunk_end = chunk_begin;
          std::advance(chunk_end, std::min(chunk_size, 
                       static_cast<size_t>(std::distance(chunk_begin, end))));
          
          futures.push_back(pool.enqueue([f, chunk_begin, chunk_end]() {
            for (Iterator it = chunk_begin; it != chunk_end; ++it)
              f(*it);
          }));
          
          chunk_begin = chunk_end;
        }
    }
  
  // Wait for all chunks to complete
  for (auto& fut : futures)
    fut.get();
}
 
template <typename InputIt, typename OutputIt, typename F>
OutputIt parallel_transform(ThreadPool& pool, InputIt first, InputIt last, 
                            OutputIt d_first, F&& f, size_t chunk_size = 0)
{
  const size_t total = std::distance(first, last);
  if (total == 0)
    return d_first;
  
  if (chunk_size == 0)
    chunk_size = std::max(size_t(1), total / (pool.num_threads() * 4));
  
  std::vector<std::future<void>> futures;
  futures.reserve((total + chunk_size - 1) / chunk_size);
  
  InputIt chunk_in = first;
  OutputIt chunk_out = d_first;
  
  while (chunk_in < last)
    {
      size_t chunk_len = std::min(chunk_size, 
                                  static_cast<size_t>(std::distance(chunk_in, last)));
      InputIt chunk_in_end = chunk_in;
      std::advance(chunk_in_end, chunk_len);
      
      futures.push_back(pool.enqueue([f, chunk_in, chunk_in_end, chunk_out]() {
        InputIt in = chunk_in;
        OutputIt out = chunk_out;
        while (in != chunk_in_end)
          {
            *out = f(*in);
            ++in;
            ++out;
          }
      }));
      
      chunk_in = chunk_in_end;
      std::advance(chunk_out, chunk_len);
    }
  
  for (auto& fut : futures)
    fut.get();
  
  return d_first + total;
}
 
template <typename Iterator, typename T, typename BinaryOp>
T parallel_reduce(ThreadPool& pool, Iterator first, Iterator last, 
                  T init, BinaryOp op, size_t chunk_size = 0)
{
  const size_t total = std::distance(first, last);
  if (total == 0)
    return init;
  
  if (chunk_size == 0)
    chunk_size = std::max(size_t(1), total / (pool.num_threads() * 4));
  
  std::vector<std::future<T>> futures;
  futures.reserve((total + chunk_size - 1) / chunk_size);
  
  for (Iterator chunk_begin = first; chunk_begin < last; )
    {
      Iterator chunk_end = chunk_begin;
      std::advance(chunk_end, std::min(chunk_size, 
                   static_cast<size_t>(std::distance(chunk_begin, last))));
      
      futures.push_back(pool.enqueue([op, chunk_begin, chunk_end]() {
        T local_result = *chunk_begin;
        for (Iterator it = chunk_begin + 1; it != chunk_end; ++it)
          local_result = op(local_result, *it);
        return local_result;
      }));
      
      chunk_begin = chunk_end;
    }
  
  // Reduce partial results
  T result = init;
  for (auto& fut : futures)
    result = op(result, fut.get());
  
  return result;
}
 
template <typename F>
void parallel_for_index(ThreadPool& pool, size_t start, size_t end, F&& f,
                        size_t chunk_size = 0)
{
  if (start >= end)
    return;
  
  const size_t total = end - start;
  
  if (chunk_size == 0)
    chunk_size = std::max(size_t(1), total / (pool.num_threads() * 4));
  
  std::vector<std::future<void>> futures;
  futures.reserve((total + chunk_size - 1) / chunk_size);
  
  for (size_t chunk_start = start; chunk_start < end; )
    {
      size_t chunk_end = std::min(chunk_start + chunk_size, end);
      
      futures.push_back(pool.enqueue([f, chunk_start, chunk_end]() {
        for (size_t i = chunk_start; i < chunk_end; ++i)
          f(i);
      }));
      
      chunk_start = chunk_end;
    }
  
  for (auto& fut : futures)
    fut.get();
}
 
inline ThreadPool& default_pool()
{
  static ThreadPool pool;
  return pool;
}
 
} // end namespace Aleph
 
#endif // ALEPH_THREAD_POOL_H