ah-parallel.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 AH_PARALLEL_H
#define AH_PARALLEL_H
 
#include <vector>
#include <atomic>
#include <optional>
#include <algorithm>
#include <numeric>
#include <type_traits>
#include <iterator>
#include <functional>
#include <thread_pool.H>
#include <ah-ranges.H>
 
namespace Aleph
{
  // =============================================================================
  // Implementation Details
  // =============================================================================
 
  namespace parallel_detail
  {
    struct chunk_bounds
    {
      size_t offset;
      size_t end;
    };
 
    inline ThreadPool &selected_parallel_pool(const ParallelOptions & options)
    {
      return options.pool == nullptr ? default_pool() : *options.pool;
    }
 
    inline size_t chunk_count(const size_t n, const size_t chunk_size) noexcept
    {
      return n == 0 ? 0 : (n + chunk_size - 1) / chunk_size;
    }
 
    inline chunk_bounds bounds_for_chunk(const size_t idx, const size_t n,
                                         const size_t chunk_size) noexcept
    {
      const size_t offset = idx * chunk_size;
      return {offset, std::min(offset + chunk_size, n)};
    }
 
    inline size_t chunk_size(const size_t n, const size_t num_threads, const size_t min_chunk = 64)
    {
      if (n == 0) return 1;
      // Use more chunks than threads for better load balancing
      const size_t chunks = num_threads * 4;
      const size_t size = (n + chunks - 1) / chunks;
      return std::max(size, min_chunk);
    }
 
    template <typename Container>
    constexpr bool has_random_access()
    {
      using It = decltype(std::begin(std::declval<Container &>()));
      return std::is_base_of_v<std::random_access_iterator_tag,
                               typename std::iterator_traits<It>::iterator_category>;
    }
 
    template <typename Container>
    auto ensure_random_access(const Container & c)
    {
      if constexpr (has_random_access<Container>())
        return &c; // Return pointer to original
      else
        return std::make_unique<std::vector<typename Container::value_type>>(std::begin(c), std::end(c));
    }
 
    template <typename T>
    decltype(auto) deref(T && ptr)
    {
      if constexpr (std::is_pointer_v<std::decay_t<T>>)
        return *ptr;
      else
        return *ptr; // unique_ptr also supports *
    }
 
    inline size_t effective_parallel_chunk_size(const size_t n,
                                                const ThreadPool & pool,
                                                const ParallelOptions & options,
                                                const size_t min_chunk = 64)
    {
      size_t effective = options.chunk_size == 0 ? chunk_size(n, pool.num_threads(), min_chunk) : options.chunk_size;
 
      if (options.max_tasks > 0)
        {
          const size_t min_for_cap = (n + options.max_tasks - 1) / options.max_tasks;
          effective = std::max(effective, std::max(min_chunk, min_for_cap));
        }
 
      return std::max<size_t>(1, effective);
    }
 
    inline bool use_sequential_parallel_path(const size_t n,
                                             const ThreadPool & pool,
                                             const ParallelOptions & options) noexcept
    {
      if (n == 0)
        return true;
      if (options.cancel_token.stop_requested())
        return true;
      if (options.max_tasks == 1)
        return true;
      if (options.min_size > 0 and n < options.min_size)
        return true;
      return pool.num_threads() <= 1;
    }
 
    inline void throw_if_parallel_canceled(const CancellationToken & token)
    {
      token.throw_if_cancellation_requested();
    }
  } // namespace parallel_detail
 
  // =============================================================================
  // Parallel Map
  // =============================================================================
 
  template <typename ResultT = void, typename Container, typename Op>
  [[nodiscard]] auto pmaps(ThreadPool & pool, const Container & c, Op op,
                           size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pmaps<ResultT>(c, op, options);
  }
 
  template <typename ResultT = void, typename Container, typename Op>
  [[nodiscard]] auto pmaps(const Container & c, Op op,
                           const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    using InputT = std::decay_t<decltype(*std::begin(c))>;
    using ActualResultT = std::conditional_t<
      std::is_void_v<ResultT>,
      std::invoke_result_t<Op, const InputT &>,
      ResultT>;
 
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::vector<ActualResultT>{};
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        std::vector<ActualResultT> result;
        result.reserve(n);
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            result.push_back(op(*it));
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return result;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    // Ensure random access for parallel processing
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::optional<ActualResultT>> slots(n);
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
 
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&slots, &data, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto in_it = std::begin(data);
                                           std::advance(in_it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++in_it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               slots[i] = op(*in_it);
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    {
      std::exception_ptr ep;
      for (auto & f: futures)
        try { f.get(); } catch (...) { if (not ep) ep = std::current_exception(); }
      if (ep)
        std::rethrow_exception(ep);
    }
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
 
    std::vector<ActualResultT> result;
    result.reserve(n);
    for (auto & slot : slots)
      result.push_back(std::move(*slot));
 
    return result;
  }
 
  // =============================================================================
  // Parallel Filter
  // =============================================================================
 
  template <typename Container, typename Pred>
  [[nodiscard]] auto pfilter(ThreadPool & pool, const Container & c, Pred pred,
                             size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pfilter(c, pred, options);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] auto pfilter(const Container & c, Pred pred,
                             const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    using T = std::decay_t<decltype(*std::begin(c))>;
 
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::vector<T>{};
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        std::vector<T> result;
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (pred(*it))
              result.push_back(*it);
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return result;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<std::vector<T>>> futures;
    const size_t num_chunks = parallel_detail::chunk_count(n, chunk_size);
    futures.reserve(num_chunks);
 
    for (size_t chunk_idx = 0; chunk_idx < num_chunks; ++chunk_idx)
      {
        const auto bounds = parallel_detail::bounds_for_chunk(chunk_idx, n, chunk_size);
 
        futures.push_back(pool.enqueue([&data, pred,
                                        offset = bounds.offset,
                                        chunk_end = bounds.end,
                                        token = options.cancel_token]()
          {
            std::vector<T> chunk_result;
            auto it = std::begin(data);
            std::advance(it, offset);
            for (size_t i = offset; i < chunk_end; ++i, ++it)
              {
                parallel_detail::throw_if_parallel_canceled(token);
                if (pred(*it))
                  chunk_result.push_back(*it);
              }
            return chunk_result;
          }));
      }
 
    std::vector<T> result;
    for (auto & f: futures)
      {
        auto chunk_result = f.get();
        result.insert(result.end(),
                      std::make_move_iterator(chunk_result.begin()),
                      std::make_move_iterator(chunk_result.end()));
      }
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
 
    return result;
  }
 
  // =============================================================================
  // Parallel Fold/Reduce
  // =============================================================================
 
  template <typename T, typename Container, typename BinaryOp>
  [[nodiscard]] T pfoldl(ThreadPool & pool, const Container & c, T init, BinaryOp op,
                         size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pfoldl(c, init, op, options);
  }
 
  template <typename T, typename Container, typename BinaryOp>
  [[nodiscard]] T pfoldl(const Container & c, T init, BinaryOp op,
                         const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return init;
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        T result = init;
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            result = op(result, *it);
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return result;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<T>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           T local = *it++;
                                           for (size_t i = offset + 1; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               local = op(local, *it);
                                             }
                                           return local;
                                         }));
 
        offset = chunk_end;
      }
 
    // Combine partial results
    T result = init;
    for (auto & f: futures)
      result = op(result, f.get());
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
 
    return result;
  }
 
  // =============================================================================
  // Parallel For Each
  // =============================================================================
 
  template <typename Container, typename Op>
  void pfor_each(ThreadPool & pool, Container & c, Op op, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    pfor_each(c, op, options);
  }
 
  template <typename Container, typename Op>
  void pfor_each(Container & c, Op op, const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return;
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            op(*it);
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return;
      }
 
    std::vector<std::future<void>> futures;
    size_t offset = 0;
 
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&c, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(c);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               op(*it);
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
  }
 
  template <typename Container, typename Op>
  void pfor_each(ThreadPool & pool, const Container & c, Op op, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    pfor_each(c, op, options);
  }
 
  template <typename Container, typename Op>
  void pfor_each(const Container & c, Op op, const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return;
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            op(*it);
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               op(*it);
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
  }
 
  // =============================================================================
  // Parallel Predicates (all, exists, none)
  // =============================================================================
 
  template <typename Container, typename Pred>
  [[nodiscard]] bool pall(ThreadPool & pool, const Container & c, Pred pred,
                          size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pall(c, pred, options);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] bool pall(const Container & c, Pred pred,
                          const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return true;
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (not pred(*it))
              return false;
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return true;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::atomic<bool> found_false{false};
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, pred, &found_false, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           if (found_false.load(std::memory_order_relaxed))
                                             return; // Short-circuit
 
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               if (not pred(*it))
                                                 {
                                                   found_false.store(true, std::memory_order_relaxed);
                                                   return;
                                                 }
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               if (found_false.load(std::memory_order_relaxed))
                                                 return;
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
    return not found_false.load();
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] bool pexists(ThreadPool & pool, const Container & c, Pred pred,
                             size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pexists(c, pred, options);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] bool pexists(const Container & c, Pred pred,
                             const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return false;
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (pred(*it))
              return true;
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return false;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::atomic<bool> found{false};
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, pred, &found, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           if (found.load(std::memory_order_relaxed))
                                             return; // Short-circuit
 
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               if (pred(*it))
                                                 {
                                                   found.store(true, std::memory_order_relaxed);
                                                   return;
                                                 }
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               if (found.load(std::memory_order_relaxed))
                                                 return;
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
    return found.load();
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] bool pnone(ThreadPool & pool, const Container & c, Pred pred,
                           size_t chunk_size = 0)
  {
    return not pexists(pool, c, pred, chunk_size);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] bool pnone(const Container & c, Pred pred,
                           const ParallelOptions & options = {})
  {
    return not pexists(c, pred, options);
  }
 
  // =============================================================================
  // Parallel Count
  // =============================================================================
 
  template <typename Container, typename Pred>
  [[nodiscard]] size_t pcount_if(ThreadPool & pool, const Container & c, Pred pred,
                                 size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pcount_if(c, pred, options);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] size_t pcount_if(const Container & c, Pred pred,
                                 const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return 0;
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        size_t total = 0;
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (pred(*it))
              ++total;
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return total;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<size_t>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, pred, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           size_t count = 0;
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               if (pred(*it))
                                                 ++count;
                                             }
                                           return count;
                                         }));
 
        offset = chunk_end;
      }
 
    size_t total = 0;
    for (auto & f: futures)
      total += f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
    return total;
  }
 
  // =============================================================================
  // Parallel Find
  // =============================================================================
 
  template <typename Container, typename Pred>
  [[nodiscard]] std::optional<size_t> pfind(ThreadPool & pool, const Container & c,
                                            Pred pred, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pfind(c, pred, options);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] std::optional<size_t> pfind(const Container & c,
                                            Pred pred,
                                            const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::nullopt;
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        size_t idx = 0;
        for (auto it = std::begin(c); it != std::end(c); ++it, ++idx)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (pred(*it))
              return idx;
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return std::nullopt;
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    // Track minimum found index
    std::atomic<size_t> min_index{n}; // n means not found
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, pred, &min_index, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           // Skip if we already found something earlier
                                           if (min_index.load(std::memory_order_relaxed) <= offset)
                                             return;
 
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               // Stop if earlier match found
                                               if (min_index.load(std::memory_order_relaxed) <= i)
                                                 return;
 
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               if (pred(*it))
                                                 {
                                                   // Atomically update minimum
                                                   size_t expected = min_index.load(std::memory_order_relaxed);
                                                   while (i < expected and
                                                          not min_index.compare_exchange_weak(expected, i,
                                                               std::memory_order_relaxed));
                                                   return;
                                                 }
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
    if (size_t result = min_index.load(); result < n)
      return result;
    return std::nullopt;
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] auto pfind_value(ThreadPool & pool, const Container & c,
                                 Pred pred, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pfind_value(c, pred, options);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] auto pfind_value(const Container & c,
                                 Pred pred,
                                 const ParallelOptions & options = {})
  {
    using T = std::decay_t<decltype(*std::begin(c))>;
 
    auto idx = pfind(c, pred, options);
    if (not idx)
      return std::optional<T>{std::nullopt};
 
    auto it = std::begin(c);
    std::advance(it, *idx);
    return std::optional<T>{*it};
  }
 
  // =============================================================================
  // Parallel Numeric Operations
  // =============================================================================
 
  template <typename Container,
            typename T = std::decay_t<decltype(*std::begin(std::declval<Container>()))>>
  [[nodiscard]] T psum(ThreadPool & pool, const Container & c, T init = T{},
                       size_t chunk_size = 0)
  {
    return pfoldl(pool, c, init, std::plus<T>{}, chunk_size);
  }
 
  template <typename Container,
            typename T = std::decay_t<decltype(*std::begin(std::declval<Container>()))>>
  [[nodiscard]] T psum(const Container & c, T init, const ParallelOptions & options)
  {
    return pfoldl(c, init, std::plus<T>{}, options);
  }
 
  template <typename Container,
            typename T = std::decay_t<decltype(*std::begin(std::declval<Container>()))>>
  [[nodiscard]] T pproduct(ThreadPool & pool, const Container & c, T init = T{1},
                           size_t chunk_size = 0)
  {
    return pfoldl(pool, c, init, std::multiplies<T>{}, chunk_size);
  }
 
  template <typename Container,
            typename T = std::decay_t<decltype(*std::begin(std::declval<Container>()))>>
  [[nodiscard]] T pproduct(const Container & c, T init, const ParallelOptions & options)
  {
    return pfoldl(c, init, std::multiplies<T>{}, options);
  }
 
  template <typename Container>
  [[nodiscard]] auto pmin(ThreadPool & pool, const Container & c, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pmin(c, options);
  }
 
  template <typename Container>
  [[nodiscard]] auto pmin(const Container & c, const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    using T = std::decay_t<decltype(*std::begin(c))>;
 
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::optional<T>{std::nullopt};
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        auto it = std::begin(c);
        T result = *it++;
        for (; it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (*it < result)
              result = *it;
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return std::optional<T>{result};
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<T>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           T local_min = *it++;
                                           for (size_t i = offset + 1; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               if (*it < local_min)
                                                 local_min = *it;
                                             }
                                           return local_min;
                                         }));
 
        offset = chunk_end;
      }
 
    T result = futures[0].get();
    for (size_t i = 1; i < futures.size(); ++i)
      {
        T val = futures[i].get();
        if (val < result)
          result = val;
      }
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
    return std::optional<T>{result};
  }
 
  template <typename Container>
  [[nodiscard]] auto pmax(ThreadPool & pool, const Container & c, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pmax(c, options);
  }
 
  template <typename Container>
  [[nodiscard]] auto pmax(const Container & c, const ParallelOptions & options = {})
  {
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
    using T = std::decay_t<decltype(*std::begin(c))>;
 
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::optional<T>{std::nullopt};
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        auto it = std::begin(c);
        T result = *it++;
        for (; it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (*it > result)
              result = *it;
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return std::optional<T>{result};
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<T>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           T local_max = *it++;
                                           for (size_t i = offset + 1; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               if (*it > local_max)
                                                 local_max = *it;
                                             }
                                           return local_max;
                                         }));
 
        offset = chunk_end;
      }
 
    T result = futures[0].get();
    for (size_t i = 1; i < futures.size(); ++i)
      {
        T val = futures[i].get();
        if (val > result)
          result = val;
      }
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
    return std::optional<T>{result};
  }
 
  template <typename Container>
  [[nodiscard]] auto pminmax(ThreadPool & pool, const Container & c, size_t chunk_size = 0)
  {
    using T = std::decay_t<decltype(*std::begin(c))>;
 
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::optional<std::pair<T, T>>{std::nullopt};
 
    if (chunk_size == 0)
      chunk_size = parallel_detail::chunk_size(n, pool.num_threads());
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<std::pair<T, T>>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, offset, chunk_end]()
                                         {
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           T local_min = *it;
                                           T local_max = *it++;
                                           for (size_t i = offset + 1; i < chunk_end; ++i, ++it)
                                             {
                                               if (*it < local_min) local_min = *it;
                                               if (*it > local_max) local_max = *it;
                                             }
                                           return std::make_pair(local_min, local_max);
                                         }));
 
        offset = chunk_end;
      }
 
    auto result = futures[0].get();
    for (size_t i = 1; i < futures.size(); ++i)
      {
        auto [mi, ma] = futures[i].get();
        if (mi < result.first) result.first = mi;
        if (ma > result.second) result.second = ma;
      }
 
    return std::optional<std::pair<T, T>>{result};
  }
 
  // =============================================================================
  // Parallel Sort
  // =============================================================================
 
  template <typename Container, typename Compare = std::less<>>
  void psort(ThreadPool & pool, Container & c, Compare cmp = Compare{},
             const size_t min_parallel_size = 1024)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.min_size = min_parallel_size;
    psort(c, std::move(cmp), options);
  }
 
  template <typename Container, typename Compare = std::less<>>
  void psort(Container & c, Compare cmp = Compare{},
             const ParallelOptions & options = {})
  {
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n <= 1)
      return;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    const size_t min_parallel_size = options.min_size == 0 ? 1024 : options.min_size;
 
    // For small sizes, use regular sort
    if (parallel_detail::use_sequential_parallel_path(n, pool, options)
        or n <= min_parallel_size)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        sort_range(c, cmp);
        return;
      }
 
    // Split into chunks, sort each in parallel, then merge
    size_t num_chunks = std::min(pool.num_threads() * 2, n / min_parallel_size);
    if (options.max_tasks > 0)
      num_chunks = std::min(num_chunks, options.max_tasks);
    num_chunks = std::max<size_t>(1, num_chunks);
    const size_t chunk_size = (n + num_chunks - 1) / num_chunks;
 
    // Sort chunks in parallel
    std::vector<std::future<void>> futures;
    for (size_t i = 0; i < n; i += chunk_size)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t end = std::min(i + chunk_size, n);
        auto begin_it = std::begin(c);
        std::advance(begin_it, i);
        auto end_it = std::begin(c);
        std::advance(end_it, end);
 
        futures.push_back(pool.enqueue([begin_it, end_it, cmp, token = options.cancel_token]()
                                         {
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           std::sort(begin_it, end_it, cmp);
                                         }));
      }
 
    for (auto & f: futures)
      f.get();
 
    // Merge sorted chunks
    using T = std::decay_t<decltype(*std::begin(c))>;
    std::vector<std::optional<T>> buffer(n);
 
    ParallelOptions merge_options = options;
    merge_options.pool = &pool;
 
    for (size_t width = chunk_size; width < n; width *= 2)
      {
        for (size_t i = 0; i < n; i += 2 * width)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            size_t mid = std::min(i + width, n);
            size_t end = std::min(i + 2 * width, n);
 
            if (mid < end)
              {
                auto begin_it = std::begin(c);
                std::advance(begin_it, i);
                auto mid_it = std::begin(c);
                std::advance(mid_it, mid);
                auto end_it = std::begin(c);
                std::advance(end_it, end);
 
                Aleph::pmerge(begin_it, mid_it, mid_it, end_it,
                              buffer.begin() + i, cmp, merge_options);
              }
            else
              {
                // Copy remaining elements
                auto begin_it = std::begin(c);
                std::advance(begin_it, i);
                auto end_it = std::begin(c);
                std::advance(end_it, mid);
                std::copy(begin_it, end_it, buffer.begin() + i);
              }
          }
 
        // Swap buffer back to container
        auto it = std::begin(c);
        for (size_t i = 0; i < n; ++i, ++it)
          *it = std::move(*buffer[i]);
      }
  }
 
  // =============================================================================
  // Parallel Zip Operations
  // =============================================================================
 
  template <typename Container1, typename Container2, typename Op>
  void pzip_for_each(ThreadPool & pool, const Container1 & c1, const Container2 & c2,
                     Op op, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    pzip_for_each(c1, c2, op, options);
  }
 
  template <typename Container1, typename Container2, typename Op>
  void pzip_for_each(const Container1 & c1, const Container2 & c2, Op op,
                     const ParallelOptions & options = {})
  {
    const size_t n1 = std::distance(std::begin(c1), std::end(c1));
    const size_t n2 = std::distance(std::begin(c2), std::end(c2));
    const size_t n = std::min(n1, n2);
 
    if (n == 0)
      return;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        auto it1 = std::begin(c1);
        auto it2 = std::begin(c2);
        for (size_t i = 0; i < n; ++i, ++it1, ++it2)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            op(*it1, *it2);
          }
        return;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    auto h1 = parallel_detail::ensure_random_access(c1);
    auto h2 = parallel_detail::ensure_random_access(c2);
    const auto & d1 = parallel_detail::deref(h1);
    const auto & d2 = parallel_detail::deref(h2);
 
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&d1, &d2, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it1 = std::begin(d1);
                                           auto it2 = std::begin(d2);
                                           std::advance(it1, offset);
                                           std::advance(it2, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it1, ++it2)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               op(*it1, *it2);
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
  }
 
  template <typename Container1, typename Container2, typename Op>
  [[nodiscard]] auto pzip_maps(ThreadPool & pool, const Container1 & c1,
                               const Container2 & c2, Op op, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pzip_maps(c1, c2, op, options);
  }
 
  template <typename Container1, typename Container2, typename Op>
  [[nodiscard]] auto pzip_maps(const Container1 & c1, const Container2 & c2, Op op,
                               const ParallelOptions & options = {})
  {
    using T1 = std::decay_t<decltype(*std::begin(c1))>;
    using T2 = std::decay_t<decltype(*std::begin(c2))>;
    using ResultT = std::invoke_result_t<Op, const T1 &, const T2 &>;
 
    const size_t n1 = std::distance(std::begin(c1), std::end(c1));
    const size_t n2 = std::distance(std::begin(c2), std::end(c2));
    const size_t n = std::min(n1, n2);
 
    if (n == 0)
      return std::vector<ResultT>{};
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        std::vector<ResultT> result;
        result.reserve(n);
        auto it1 = std::begin(c1);
        auto it2 = std::begin(c2);
        for (size_t i = 0; i < n; ++i, ++it1, ++it2)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            result.push_back(op(*it1, *it2));
          }
        return result;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    auto h1 = parallel_detail::ensure_random_access(c1);
    auto h2 = parallel_detail::ensure_random_access(c2);
    const auto & d1 = parallel_detail::deref(h1);
    const auto & d2 = parallel_detail::deref(h2);
 
    std::vector<std::optional<ResultT>> slots(n);
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&slots, &d1, &d2, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it1 = std::begin(d1);
                                           auto it2 = std::begin(d2);
                                           std::advance(it1, offset);
                                           std::advance(it2, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it1, ++it2)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               slots[i] = op(*it1, *it2);
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    std::vector<ResultT> result;
    result.reserve(n);
    for (auto & slot : slots)
      result.push_back(std::move(*slot));
 
    return result;
  }
 
  template <typename Container1, typename Container2, typename T, typename Op>
  [[nodiscard]] T pzip_foldl(ThreadPool & pool, const Container1 & c1,
                             const Container2 & c2, T init, Op op,
                             size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pzip_foldl(c1, c2, init, op, options);
  }
 
  template <typename Container1, typename Container2, typename T, typename Op>
  [[nodiscard]] T pzip_foldl(const Container1 & c1, const Container2 & c2, T init, Op op,
                             const ParallelOptions & options = {})
  {
    const size_t n1 = std::distance(std::begin(c1), std::end(c1));
    const size_t n2 = std::distance(std::begin(c2), std::end(c2));
    const size_t n = std::min(n1, n2);
 
    if (n == 0)
      return init;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        auto it1 = std::begin(c1);
        auto it2 = std::begin(c2);
        T result = init;
        for (size_t i = 0; i < n; ++i, ++it1, ++it2)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            result = op(result, *it1, *it2);
          }
        return result;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    auto h1 = parallel_detail::ensure_random_access(c1);
    auto h2 = parallel_detail::ensure_random_access(c2);
    const auto & d1 = parallel_detail::deref(h1);
    const auto & d2 = parallel_detail::deref(h2);
 
    std::vector<std::future<T>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&d1, &d2, &init, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it1 = std::begin(d1);
                                           auto it2 = std::begin(d2);
                                           std::advance(it1, offset);
                                           std::advance(it2, offset);
 
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           T local = op(init, *it1++, *it2++);
                                           for (size_t i = offset + 1; i < chunk_end; ++i, ++it1, ++it2)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               local = op(local, *it1, *it2);
                                             }
                                           return local;
                                         }));
 
        offset = chunk_end;
      }
 
    // Binary reduce the partial results
    // We need a binary op for this - derive it from the ternary op
    T result = futures[0].get();
    for (size_t i = 1; i < futures.size(); ++i)
      {
        T val = futures[i].get();
        // Combine using addition - user should use pfoldl + pzip_maps for complex cases
        result = result + val - init; // Compensate for init being added in each chunk
      }
 
    return result;
  }
 
  // =============================================================================
  // Parallel Partition
  // =============================================================================
 
  template <typename Container, typename Pred>
  [[nodiscard]] auto ppartition(ThreadPool & pool, const Container & c, Pred pred,
                                size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return ppartition(c, pred, options);
  }
 
  template <typename Container, typename Pred>
  [[nodiscard]] auto ppartition(const Container & c, Pred pred,
                                const ParallelOptions & options = {})
  {
    using T = std::decay_t<decltype(*std::begin(c))>;
    ThreadPool & pool = parallel_detail::selected_parallel_pool(options);
 
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::make_pair(std::vector<T>{}, std::vector<T>{});
 
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        std::vector<T> yes_result, no_result;
        for (auto it = std::begin(c); it != std::end(c); ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            if (pred(*it))
              yes_result.push_back(*it);
            else
              no_result.push_back(*it);
          }
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        return std::make_pair(std::move(yes_result), std::move(no_result));
      }
 
    const size_t chunk_size =
        parallel_detail::effective_parallel_chunk_size(n, pool, options);
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    const size_t num_chunks = parallel_detail::chunk_count(n, chunk_size);
    std::vector<size_t> yes_counts(num_chunks, 0);
    std::vector<size_t> no_counts(num_chunks, 0);
    std::vector<std::future<void>> count_futures;
    count_futures.reserve(num_chunks);
 
    for (size_t chunk_idx = 0; chunk_idx < num_chunks; ++chunk_idx)
      {
        const auto bounds = parallel_detail::bounds_for_chunk(chunk_idx, n, chunk_size);
 
        count_futures.push_back(pool.enqueue([&data, &yes_counts, &no_counts, pred,
                                              chunk_idx,
                                              offset = bounds.offset,
                                              chunk_end = bounds.end,
                                              token = options.cancel_token]()
          {
            size_t yes = 0;
            size_t no = 0;
            auto it = std::begin(data);
            std::advance(it, offset);
            for (size_t i = offset; i < chunk_end; ++i, ++it)
              {
                parallel_detail::throw_if_parallel_canceled(token);
                if (pred(*it))
                  ++yes;
                else
                  ++no;
              }
            yes_counts[chunk_idx] = yes;
            no_counts[chunk_idx] = no;
          }));
      }
 
    for (auto & f: count_futures)
      f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
 
    std::vector<size_t> yes_offsets(num_chunks, 0);
    std::vector<size_t> no_offsets(num_chunks, 0);
    if (num_chunks > 0)
      {
        Aleph::pexclusive_scan(yes_counts.begin(), yes_counts.end(), yes_offsets.begin(),
                               size_t{0}, std::plus<size_t>{}, options);
        Aleph::pexclusive_scan(no_counts.begin(), no_counts.end(), no_offsets.begin(),
                               size_t{0}, std::plus<size_t>{}, options);
      }
 
    const size_t yes_total = num_chunks == 0 ? 0 : yes_offsets.back() + yes_counts.back();
    const size_t no_total = num_chunks == 0 ? 0 : no_offsets.back() + no_counts.back();
 
    std::vector<std::optional<T>> yes_slots(yes_total);
    std::vector<std::optional<T>> no_slots(no_total);
    std::vector<std::future<void>> fill_futures;
    fill_futures.reserve(num_chunks);
 
    for (size_t chunk_idx = 0; chunk_idx < num_chunks; ++chunk_idx)
      {
        const auto bounds = parallel_detail::bounds_for_chunk(chunk_idx, n, chunk_size);
        const size_t yes_offset = yes_offsets[chunk_idx];
        const size_t no_offset = no_offsets[chunk_idx];
 
        fill_futures.push_back(pool.enqueue([&data, &yes_slots, &no_slots, pred,
                                             offset = bounds.offset,
                                             chunk_end = bounds.end,
                                             yes_offset, no_offset,
                                             token = options.cancel_token]()
          {
            auto it = std::begin(data);
            std::advance(it, offset);
            size_t yes_pos = yes_offset;
            size_t no_pos = no_offset;
            for (size_t i = offset; i < chunk_end; ++i, ++it)
              {
                parallel_detail::throw_if_parallel_canceled(token);
                if (pred(*it))
                  yes_slots[yes_pos++].emplace(*it);
                else
                  no_slots[no_pos++].emplace(*it);
              }
          }));
      }
 
    for (auto & f: fill_futures)
      f.get();
 
    parallel_detail::throw_if_parallel_canceled(options.cancel_token);
 
    std::vector<T> yes_result, no_result;
    yes_result.reserve(yes_total);
    no_result.reserve(no_total);
 
    for (auto & slot: yes_slots)
      yes_result.push_back(std::move(*slot));
    for (auto & slot: no_slots)
      no_result.push_back(std::move(*slot));
 
    return std::make_pair(std::move(yes_result), std::move(no_result));
  }
 
  // =============================================================================
  // Parallel Scan and Merge Wrappers
  // =============================================================================
 
  template <typename Container, typename BinaryOp>
  [[nodiscard]] auto pscan(ThreadPool & pool, const Container & c, BinaryOp op,
                           size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pscan(c, op, options);
  }
 
  template <typename Container, typename BinaryOp>
  [[nodiscard]] auto pscan(const Container & c, BinaryOp op,
                           const ParallelOptions & options = {})
  {
    using T = std::decay_t<decltype(*std::begin(c))>;
 
    auto holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(holder);
    const size_t n = static_cast<size_t>(std::distance(std::begin(data), std::end(data)));
    std::vector<std::optional<T>> slots(n);
 
    Aleph::pscan(std::begin(data), std::end(data), slots.begin(), op, options);
 
    std::vector<T> result;
    result.reserve(n);
    for (auto & slot: slots)
      result.push_back(std::move(*slot));
    return result;
  }
 
  template <typename Container, typename T, typename BinaryOp>
  [[nodiscard]] auto pexclusive_scan(ThreadPool & pool, const Container & c, T init,
                                     BinaryOp op, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pexclusive_scan(c, init, op, options);
  }
 
  template <typename Container, typename T, typename BinaryOp>
  [[nodiscard]] auto pexclusive_scan(const Container & c, T init, BinaryOp op,
                                     const ParallelOptions & options = {})
  {
    auto holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(holder);
    const size_t n = static_cast<size_t>(std::distance(std::begin(data), std::end(data)));
    std::vector<std::optional<T>> slots(n);
 
    Aleph::pexclusive_scan(std::begin(data), std::end(data), slots.begin(),
                           init, op, options);
 
    std::vector<T> result;
    result.reserve(n);
    for (auto & slot: slots)
      result.push_back(std::move(*slot));
    return result;
  }
 
  template <typename Container1, typename Container2, typename Compare = std::less<>>
  [[nodiscard]] auto pmerge(ThreadPool & pool, const Container1 & c1, const Container2 & c2,
                            Compare comp = Compare{}, size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return pmerge(c1, c2, std::move(comp), options);
  }
 
  template <typename Container1, typename Container2, typename Compare = std::less<>>
  [[nodiscard]] auto pmerge(const Container1 & c1, const Container2 & c2,
                            Compare comp = Compare{},
                            const ParallelOptions & options = {})
  {
    using T1 = std::decay_t<decltype(*std::begin(c1))>;
    using T2 = std::decay_t<decltype(*std::begin(c2))>;
    using ResultT = std::common_type_t<T1, T2>;
 
    auto h1 = parallel_detail::ensure_random_access(c1);
    auto h2 = parallel_detail::ensure_random_access(c2);
    const auto & d1 = parallel_detail::deref(h1);
    const auto & d2 = parallel_detail::deref(h2);
 
    const size_t total = static_cast<size_t>(std::distance(std::begin(d1), std::end(d1))
                                             + std::distance(std::begin(d2), std::end(d2)));
    std::vector<std::optional<ResultT>> slots(total);
 
    Aleph::pmerge(std::begin(d1), std::end(d1),
                  std::begin(d2), std::end(d2),
                  slots.begin(), comp, options);
 
    std::vector<ResultT> result;
    result.reserve(total);
    for (auto & slot: slots)
      result.push_back(std::move(*slot));
    return result;
  }
 
  // =============================================================================
  // Variadic Parallel Zip Operations (N containers)
  // =============================================================================
 
  namespace parallel_zip_detail
  {
    template <typename Container>
    struct ContainerHolder
    {
      using value_type = std::decay_t<decltype(*std::begin(std::declval<Container &>()))>;
      using holder_type = std::conditional_t<
        parallel_detail::has_random_access<Container>(),
        const Container *,
        std::unique_ptr<std::vector<value_type>>>;
 
      holder_type data;
      size_t cached_size; 
 
      explicit ContainerHolder(const Container & c)
      {
        if constexpr (parallel_detail::has_random_access<Container>())
          {
            data = &c;
            // For random access, std::distance is O(1)
            cached_size = static_cast<size_t>(std::distance(std::begin(c), std::end(c)));
          }
        else
          {
            // Copy to vector (O(n) - unavoidable), then get size from vector (O(1))
            data = std::make_unique<std::vector<value_type>>(std::begin(c), std::end(c));
            cached_size = data->size();
          }
      }
 
      decltype(auto) get() const
      {
        if constexpr (parallel_detail::has_random_access<Container>())
          return *data;
        else
          return *data;
      }
 
      [[nodiscard]] size_t size() const noexcept { return cached_size; }
 
      auto begin() const { return std::begin(get()); }
      auto end() const { return std::end(get()); }
    };
 
    template <typename... Holders, size_t... Is>
    size_t min_holder_size_impl(const std::tuple<Holders...> & holders,
                                std::index_sequence<Is...>)
    {
      return std::min({std::get<Is>(holders).size()...});
    }
 
    template <typename... Holders>
    size_t min_holder_size(const std::tuple<Holders...> & holders)
    {
      return min_holder_size_impl(holders, std::make_index_sequence<sizeof...(Holders)>{});
    }
 
    template <typename... Holders, size_t... Is>
    auto make_iterators_at(size_t offset, const std::tuple<Holders...> & holders,
                           std::index_sequence<Is...>)
    {
      return std::make_tuple([&]()
                               {
                                 auto it = std::get<Is>(holders).begin();
                                 std::advance(it, offset);
                                 return it;
                               }()...);
    }
 
    template <typename... Iters, size_t... Is>
    void advance_all_iters(std::tuple<Iters...> & iters, std::index_sequence<Is...>)
    {
      (++std::get<Is>(iters), ...);
    }
 
    template <typename... Iters, size_t... Is>
    auto deref_all_iters(const std::tuple<Iters...> & iters, std::index_sequence<Is...>)
    {
      return std::make_tuple(*std::get<Is>(iters)...);
    }
  } // namespace parallel_zip_detail
 
  template <typename Op, typename... Containers>
  void pzip_for_each_n(ThreadPool & pool, Op op, const Containers &... cs)
  {
    ParallelOptions options;
    options.pool = &pool;
    pzip_for_each_n(op, options, cs...);
  }
 
  template <typename Op, typename... Containers>
  void pzip_for_each_n(Op op, const ParallelOptions & options, const Containers &... cs)
  {
    static_assert(sizeof...(Containers) >= 2,
                  "pzip_for_each requires at least 2 containers");
 
    // Convert all containers to random access FIRST
    // This is O(n) for non-RA containers, but unavoidable
    auto holders = std::make_tuple(parallel_zip_detail::ContainerHolder<Containers>(cs)...);
 
    // Now get min size - O(1) because all holders have cached sizes
    const size_t n = parallel_zip_detail::min_holder_size(holders);
    if (n == 0)
      return;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        constexpr size_t N = sizeof...(Containers);
        auto iters = parallel_zip_detail::make_iterators_at(
                                                            0, holders, std::make_index_sequence<N>{});
        for (size_t i = 0; i < n; ++i)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            std::apply(op, parallel_zip_detail::deref_all_iters(
                                                                iters, std::make_index_sequence<N>{}));
            parallel_zip_detail::advance_all_iters(iters, std::make_index_sequence<N>{});
          }
        return;
      }
 
    const size_t chunk_size = parallel_detail::effective_parallel_chunk_size(
                                                                             n, pool, options, pool.num_threads());
 
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&holders, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           constexpr size_t N = sizeof...(Containers);
                                           auto iters = parallel_zip_detail::make_iterators_at(
                                              offset, holders, std::make_index_sequence<N>{});
 
                                           for (size_t i = offset; i < chunk_end; ++i)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               std::apply(op, parallel_zip_detail::deref_all_iters(
                                                             iters, std::make_index_sequence<N>{}));
                                               parallel_zip_detail::advance_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
  }
 
  template <typename Op, typename... Containers>
  [[nodiscard]] auto pzip_maps_n(ThreadPool & pool, Op op, const Containers &... cs)
  {
    ParallelOptions options;
    options.pool = &pool;
    return pzip_maps_n(op, options, cs...);
  }
 
  template <typename Op, typename... Containers>
  [[nodiscard]] auto pzip_maps_n(Op op, const ParallelOptions & options,
                                 const Containers &... cs)
  {
    static_assert(sizeof...(Containers) >= 2,
                  "pzip_maps requires at least 2 containers");
 
    // Deduce result type from operation
    using ResultT = std::invoke_result_t<Op,
                                         std::decay_t<decltype(*std::begin(cs))>...>;
 
    // Convert all containers to random access FIRST
    auto holders = std::make_tuple(parallel_zip_detail::ContainerHolder<Containers>(cs)...);
 
    // Now get min size - O(1) because all holders have cached sizes
    const size_t n = parallel_zip_detail::min_holder_size(holders);
    if (n == 0)
      return std::vector<ResultT>{};
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        std::vector<std::optional<ResultT>> slots(n);
        constexpr size_t N = sizeof...(Containers);
        auto iters = parallel_zip_detail::make_iterators_at(
                                                            0, holders, std::make_index_sequence<N>{});
        for (size_t i = 0; i < n; ++i)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            slots[i] = std::apply(op, parallel_zip_detail::deref_all_iters(
                                                                            iters, std::make_index_sequence<N>{}));
            parallel_zip_detail::advance_all_iters(iters, std::make_index_sequence<N>{});
          }
 
        std::vector<ResultT> result;
        result.reserve(n);
        for (auto & slot: slots)
          result.push_back(std::move(*slot));
        return result;
        }
 
        size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                           pool.num_threads());
 
        std::vector<std::optional<ResultT>> slots(n);
        std::vector<std::future<void>> futures;
 
        size_t offset = 0;
        while (offset < n)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            size_t chunk_end = std::min(offset + chunk_size, n);
 
            futures.push_back(pool.enqueue([&slots, &holders, op, offset, chunk_end,
                                             token = options.cancel_token]()
                                             {
                                               constexpr size_t N = sizeof...(Containers);
                                               auto iters = parallel_zip_detail::make_iterators_at(
                                                  offset, holders, std::make_index_sequence<N>{});
 
                                               for (size_t i = offset; i < chunk_end; ++i)
                                                 {
                                                   parallel_detail::throw_if_parallel_canceled(token);
                                                   slots[i] = std::apply(op, parallel_zip_detail::deref_all_iters(
                                                      iters, std::make_index_sequence<N>{}));
                                                   parallel_zip_detail::advance_all_iters(iters, std::make_index_sequence<N>{});
                                                 }
                                             }));
            offset = chunk_end;
          }
 
        for (auto & f: futures)
          f.get();
 
        std::vector<ResultT> result;
        result.reserve(n);
        for (auto & slot: slots)
          result.push_back(std::move(*slot));
        return result;
 
  }
 
  template <typename T, typename Op, typename Combiner, typename... Containers>
  [[nodiscard]] T pzip_foldl_n(ThreadPool & pool, T init, Op op, Combiner combiner,
                               const Containers &... cs)
  {
    ParallelOptions options;
    options.pool = &pool;
    return pzip_foldl_n(init, op, combiner, options, cs...);
  }
 
  template <typename T, typename Op, typename Combiner, typename... Containers>
  [[nodiscard]] T pzip_foldl_n(T init, Op op, Combiner combiner,
                               const ParallelOptions & options, const Containers &... cs)
  {
    static_assert(sizeof...(Containers) >= 2,
                  "pzip_foldl requires at least 2 containers");
 
    // Convert all containers to random access FIRST
    auto holders = std::make_tuple(parallel_zip_detail::ContainerHolder<Containers>(cs)...);
 
    // Now get min size - O(1) because all holders have cached sizes
    const size_t n = parallel_zip_detail::min_holder_size(holders);
    if (n == 0)
      return init;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        constexpr size_t N = sizeof...(Containers);
        auto iters = parallel_zip_detail::make_iterators_at(
                                                            0, holders, std::make_index_sequence<N>{});
        T result = init;
        for (size_t i = 0; i < n; ++i)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            auto tuple = parallel_zip_detail::deref_all_iters(iters, std::make_index_sequence<N>{});
            result = std::apply([&op, &result](auto &&... args)
                                  {
                                    return op(result, std::forward<decltype(args)>(args)...);
                                  }, tuple);
            parallel_zip_detail::advance_all_iters(iters, std::make_index_sequence<N>{});
          }
        return result;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    std::vector<std::future<T>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&holders, init, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           constexpr size_t N = sizeof...(Containers);
                                           auto iters = parallel_zip_detail::make_iterators_at(
                                              offset, holders, std::make_index_sequence<N>{});
 
                                           // First element
                                           parallel_detail::throw_if_parallel_canceled(token);
                                           auto first_tuple = parallel_zip_detail::deref_all_iters(
                                              iters, std::make_index_sequence<N>{});
                                           T local = std::apply([&op, &init](auto &&... args)
                                                                  {
                                                                    return op(init, std::forward<decltype(args)>(args)
                                                                              ...);
                                                                  }, first_tuple);
                                           parallel_zip_detail::advance_all_iters(
                                              iters, std::make_index_sequence<N>{});
 
                                           // Remaining elements
                                           for (size_t i = offset + 1; i < chunk_end; ++i)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               auto tuple = parallel_zip_detail::deref_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                               local = std::apply([&op, &local](auto &&... args)
                                                                    {
                                                                      return op(local,
                                                                                  std::forward<decltype(args)>(args)
                                                                                  ...);
                                                                    }, tuple);
                                               parallel_zip_detail::advance_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                             }
 
                                           return local;
                                         }));
 
        offset = chunk_end;
      }
 
    // Combine partial results using the combiner
    T result = futures[0].get();
    for (size_t i = 1; i < futures.size(); ++i)
      result = combiner(result, futures[i].get());
 
    return result;
  }
 
  template <typename Pred, typename... Containers>
  [[nodiscard]] bool pzip_all_n(ThreadPool & pool, Pred pred, const Containers &... cs)
  {
    ParallelOptions options;
    options.pool = &pool;
    return pzip_all_n(pred, options, cs...);
  }
 
  template <typename Pred, typename... Containers>
  [[nodiscard]] bool pzip_all_n(Pred pred, const ParallelOptions & options,
                                const Containers &... cs)
  {
    static_assert(sizeof...(Containers) >= 2,
                  "pzip_all requires at least 2 containers");
 
    // Convert all containers to random access FIRST
    auto holders = std::make_tuple(
                                   parallel_zip_detail::ContainerHolder<Containers>(cs)...);
 
    // Now get min size - O(1) because all holders have cached sizes
    const size_t n = parallel_zip_detail::min_holder_size(holders);
    if (n == 0)
      return true; // Vacuous truth
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        constexpr size_t N = sizeof...(Containers);
        auto iters = parallel_zip_detail::make_iterators_at(
                                                            0, holders, std::make_index_sequence<N>{});
        for (size_t i = 0; i < n; ++i)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            auto tuple = parallel_zip_detail::deref_all_iters(iters, std::make_index_sequence<N>{});
            if (! std::apply(pred, tuple))
              return false;
            parallel_zip_detail::advance_all_iters(iters, std::make_index_sequence<N>{});
          }
        return true;
      }
 
    const size_t chunk_size = parallel_detail::effective_parallel_chunk_size(
                                                                             n, pool, options, pool.num_threads());
 
    std::atomic<bool> found_false{false};
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&holders, pred, &found_false, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           if (found_false.load(std::memory_order_relaxed))
                                             return;
 
                                           constexpr size_t N = sizeof...(Containers);
                                           auto iters = parallel_zip_detail::make_iterators_at(
                                              offset, holders, std::make_index_sequence<N>{});
 
                                           for (size_t i = offset; i < chunk_end; ++i)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               auto tuple = parallel_zip_detail::deref_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                               if (! std::apply(pred, tuple))
                                                 {
                                                   found_false.store(true, std::memory_order_relaxed);
                                                   return;
                                                 }
                                               if (found_false.load(std::memory_order_relaxed))
                                                 return;
                                               parallel_zip_detail::advance_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    return not found_false.load();
  }
 
  template <typename Pred, typename... Containers>
  [[nodiscard]] bool pzip_exists_n(ThreadPool & pool, Pred pred, const Containers &... cs)
  {
    ParallelOptions options;
    options.pool = &pool;
    return pzip_exists_n(pred, options, cs...);
  }
 
  template <typename Pred, typename... Containers>
  [[nodiscard]] bool pzip_exists_n(Pred pred, const ParallelOptions & options,
                                   const Containers &... cs)
  {
    static_assert(sizeof...(Containers) >= 2,
                  "pzip_exists requires at least 2 containers");
 
    // Convert all containers to random access FIRST
    auto holders = std::make_tuple(parallel_zip_detail::ContainerHolder<Containers>(cs)...);
 
    // Now get min size - O(1) because all holders have cached sizes
    const size_t n = parallel_zip_detail::min_holder_size(holders);
    if (n == 0)
      return false;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        constexpr size_t N = sizeof...(Containers);
        auto iters = parallel_zip_detail::make_iterators_at(
                                                            0, holders, std::make_index_sequence<N>{});
        for (size_t i = 0; i < n; ++i)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            auto tuple = parallel_zip_detail::deref_all_iters(iters, std::make_index_sequence<N>{});
            if (std::apply(pred, tuple))
              return true;
            parallel_zip_detail::advance_all_iters(iters, std::make_index_sequence<N>{});
          }
        return false;
      }
 
    const size_t chunk_size = parallel_detail::effective_parallel_chunk_size(
                                                                             n, pool, options, pool.num_threads());
 
    std::atomic<bool> found{false};
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&holders, pred, &found, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           if (found.load(std::memory_order_relaxed))
                                             return;
 
                                           constexpr size_t N = sizeof...(Containers);
                                           auto iters = parallel_zip_detail::make_iterators_at(
                                              offset, holders, std::make_index_sequence<N>{});
 
                                           for (size_t i = offset; i < chunk_end; ++i)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               auto tuple = parallel_zip_detail::deref_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                               if (std::apply(pred, tuple))
                                                 {
                                                   found.store(true, std::memory_order_relaxed);
                                                   return;
                                                 }
                                               if (found.load(std::memory_order_relaxed))
                                                 return;
                                               parallel_zip_detail::advance_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    return found.load();
  }
 
  template <typename Pred, typename... Containers>
  [[nodiscard]] size_t pzip_count_if_n(ThreadPool & pool, Pred pred,
                                       const Containers &... cs)
  {
    ParallelOptions options;
    options.pool = &pool;
    return pzip_count_if_n(pred, options, cs...);
  }
 
  template <typename Pred, typename... Containers>
  [[nodiscard]] size_t pzip_count_if_n(Pred pred, const ParallelOptions & options,
                                       const Containers &... cs)
  {
    static_assert(sizeof...(Containers) >= 2,
                  "pzip_count_if requires at least 2 containers");
 
    // Convert all containers to random access FIRST
    auto holders = std::make_tuple(parallel_zip_detail::ContainerHolder<Containers>(cs)...);
 
    // Now get min size - O(1) because all holders have cached sizes
    const size_t n = parallel_zip_detail::min_holder_size(holders);
    if (n == 0)
      return 0;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        constexpr size_t N = sizeof...(Containers);
        auto iters = parallel_zip_detail::make_iterators_at(
                                                            0, holders, std::make_index_sequence<N>{});
        size_t count = 0;
        for (size_t i = 0; i < n; ++i)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            auto tuple = parallel_zip_detail::deref_all_iters(iters, std::make_index_sequence<N>{});
            if (std::apply(pred, tuple))
              ++count;
            parallel_zip_detail::advance_all_iters(iters, std::make_index_sequence<N>{});
          }
        return count;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    std::vector<std::future<size_t>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&holders, pred, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           constexpr size_t N = sizeof...(Containers);
                                           auto iters = parallel_zip_detail::make_iterators_at(
                                              offset, holders, std::make_index_sequence<N>{});
 
                                           size_t count = 0;
                                           for (size_t i = offset; i < chunk_end; ++i)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               auto tuple = parallel_zip_detail::deref_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                               if (std::apply(pred, tuple))
                                                 ++count;
                                               parallel_zip_detail::advance_all_iters(
                                                  iters, std::make_index_sequence<N>{});
                                             }
                                           return count;
                                         }));
 
        offset = chunk_end;
      }
 
    size_t total = 0;
    for (auto & f: futures)
      total += f.get();
 
    return total;
  }
 
  // =============================================================================
  // Parallel Enumerate
  // =============================================================================
 
  template <typename Container, typename Op>
  void penumerate_for_each(ThreadPool & pool, Container & c, Op op,
                           size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    penumerate_for_each(c, op, options);
  }
 
  template <typename Container, typename Op>
  void penumerate_for_each(Container & c, Op op,
                           const ParallelOptions & options = {})
  {
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        auto it = std::begin(c);
        for (size_t i = 0; i < n; ++i, ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            op(i, *it);
          }
        return;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&c, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(c);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               op(i, *it);
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
  }
 
  template <typename Container, typename Op>
  void penumerate_for_each(ThreadPool & pool, const Container & c, Op op,
                           size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    penumerate_for_each(c, op, options);
  }
 
  template <typename Container, typename Op>
  void penumerate_for_each(const Container & c, Op op,
                           const ParallelOptions & options = {})
  {
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return;
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        auto it = std::begin(c);
        for (size_t i = 0; i < n; ++i, ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            op(i, *it);
          }
        return;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&data, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               op(i, *it);
                                             }
                                         }));
 
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
  }
 
  template <typename Container, typename Op>
  [[nodiscard]] auto penumerate_maps(ThreadPool & pool, const Container & c, Op op,
                                     size_t chunk_size = 0)
  {
    ParallelOptions options;
    options.pool = &pool;
    options.chunk_size = chunk_size;
    return penumerate_maps(c, op, options);
  }
 
  template <typename Container, typename Op>
  [[nodiscard]] auto penumerate_maps(const Container & c, Op op,
                                     const ParallelOptions & options = {})
  {
    using T = std::decay_t<decltype(*std::begin(c))>;
    using ResultT = std::invoke_result_t<Op, size_t, const T &>;
 
    const size_t n = std::distance(std::begin(c), std::end(c));
    if (n == 0)
      return std::vector<ResultT>{};
 
    auto & pool = parallel_detail::selected_parallel_pool(options);
    if (parallel_detail::use_sequential_parallel_path(n, pool, options))
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        std::vector<ResultT> result;
        result.reserve(n);
        auto it = std::begin(c);
        for (size_t i = 0; i < n; ++i, ++it)
          {
            parallel_detail::throw_if_parallel_canceled(options.cancel_token);
            result.push_back(op(i, *it));
          }
        return result;
      }
 
    size_t chunk_size = parallel_detail::effective_parallel_chunk_size(n, pool, options,
                                                                       pool.num_threads());
 
    auto data_holder = parallel_detail::ensure_random_access(c);
    const auto & data = parallel_detail::deref(data_holder);
 
    std::vector<std::optional<ResultT>> slots(n);
    std::vector<std::future<void>> futures;
 
    size_t offset = 0;
    while (offset < n)
      {
        parallel_detail::throw_if_parallel_canceled(options.cancel_token);
        size_t chunk_end = std::min(offset + chunk_size, n);
 
        futures.push_back(pool.enqueue([&slots, &data, op, offset, chunk_end,
                                         token = options.cancel_token]()
                                         {
                                           auto it = std::begin(data);
                                           std::advance(it, offset);
                                           for (size_t i = offset; i < chunk_end; ++i, ++it)
                                             {
                                               parallel_detail::throw_if_parallel_canceled(token);
                                               slots[i] = op(i, *it);
                                             }
                                         }));
        offset = chunk_end;
      }
 
    for (auto & f: futures)
      f.get();
 
    std::vector<ResultT> result;
    result.reserve(n);
    for (auto & slot: slots)
      result.push_back(std::move(*slot));
    return result;
 
  }
 
  // =============================================================================
  // Convenience: Default Pool Variants
  // =============================================================================
 
  inline ThreadPool &parallel_default_pool()
  {
    return default_pool();
  }
 
  // Convenience macros for using default pool (optional)
#ifdef AH_PARALLEL_USE_DEFAULT_POOL
 
#define PMAP(c, op) pmaps(parallel_default_pool(), c, op)
#define PFILTER(c, pred) pfilter(parallel_default_pool(), c, pred)
#define PFOLD(c, init, op) pfoldl(parallel_default_pool(), c, init, op)
#define PFOR_EACH(c, op) pfor_each(parallel_default_pool(), c, op)
#define PALL(c, pred) pall(parallel_default_pool(), c, pred)
#define PEXISTS(c, pred) pexists(parallel_default_pool(), c, pred)
#define PSUM(c) psum(parallel_default_pool(), c)
 
#endif // AH_PARALLEL_USE_DEFAULT_POOL
} // namespace Aleph
 
#endif // AH_PARALLEL_H