ah_parallel_test.cc

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.
*/
 
 
#include <gtest/gtest.h>
#include <cstdlib>
#include <ah-parallel.H>
#include <vector>
#include <list>
#include <deque>
#include <array>
#include <string>
#include <numeric>
#include <atomic>
#include <chrono>
#include <random>
#include <algorithm>
#include <cmath>
 
using namespace Aleph;
 
// =============================================================================
// Test Fixture
// =============================================================================
 
class ParallelTest : public ::testing::Test
{
protected:
  ThreadPool pool{4};
  
  std::vector<int> empty_vec;
  std::vector<int> single_vec = {42};
  std::vector<int> small_vec = {1, 2, 3, 4, 5};
  std::vector<int> large_vec;
  
  void SetUp() override
  {
    large_vec.resize(10000);
    std::iota(large_vec.begin(), large_vec.end(), 1);
  }
};
 
// =============================================================================
// pmaps Tests
// =============================================================================
 
TEST_F(ParallelTest, PmapsSquare)
{
  auto result = pmaps(pool, small_vec, [](int x) { return x * x; });
  ASSERT_EQ(result.size(), 5u);
  EXPECT_EQ(result, (std::vector<int>{1, 4, 9, 16, 25}));
}
 
TEST_F(ParallelTest, PmapsTypeConversion)
{
  auto result = pmaps<double>(pool, small_vec, [](int x) { return x * 1.5; });
  ASSERT_EQ(result.size(), 5u);
  EXPECT_NEAR(result[0], 1.5, 0.001);
  EXPECT_NEAR(result[4], 7.5, 0.001);
}
 
TEST_F(ParallelTest, PmapsToString)
{
  auto result = pmaps(pool, small_vec, [](int x) { return std::to_string(x); });
  ASSERT_EQ(result.size(), 5u);
  EXPECT_EQ(result[0], "1");
  EXPECT_EQ(result[4], "5");
}
 
TEST_F(ParallelTest, PmapsEmpty)
{
  auto result = pmaps(pool, empty_vec, [](int x) { return x * 2; });
  EXPECT_TRUE(result.empty());
}
 
TEST_F(ParallelTest, PmapsSingle)
{
  auto result = pmaps(pool, single_vec, [](int x) { return x * 2; });
  ASSERT_EQ(result.size(), 1u);
  EXPECT_EQ(result[0], 84);
}
 
TEST_F(ParallelTest, PmapsLargeData)
{
  auto result = pmaps(pool, large_vec, [](int x) { return x * 2; });
  ASSERT_EQ(result.size(), 10000u);
  for (size_t i = 0; i < result.size(); ++i)
    EXPECT_EQ(result[i], static_cast<int>((i + 1) * 2));
}
 
TEST_F(ParallelTest, PmapsWithList)
{
  std::list<int> lst = {1, 2, 3, 4, 5};
  auto result = pmaps(pool, lst, [](int x) { return x * x; });
  EXPECT_EQ(result, (std::vector<int>{1, 4, 9, 16, 25}));
}
 
TEST_F(ParallelTest, PmapsPreservesOrder)
{
  auto result = pmaps(pool, large_vec, [](int x) { return x; });
  for (size_t i = 0; i < result.size(); ++i)
    EXPECT_EQ(result[i], static_cast<int>(i + 1));
}
 
TEST_F(ParallelTest, PmapsAcceptsParallelOptions)
{
  ParallelOptions options;
  options.pool = &pool;
  options.min_size = large_vec.size() + 1;
  options.chunk_size = 7;
  options.max_tasks = 2;
 
  auto result = pmaps(large_vec, [](int x) { return x + 1; }, options);
  ASSERT_EQ(result.size(), large_vec.size());
  EXPECT_EQ(result.front(), 2);
  EXPECT_EQ(result.back(), 10001);
}
 
// =============================================================================
// pfilter Tests
// =============================================================================
 
TEST_F(ParallelTest, PfilterEvens)
{
  auto result = pfilter(pool, small_vec, [](int x) { return x % 2 == 0; });
  EXPECT_EQ(result, (std::vector<int>{2, 4}));
}
 
TEST_F(ParallelTest, PfilterOdds)
{
  auto result = pfilter(pool, small_vec, [](int x) { return x % 2 != 0; });
  EXPECT_EQ(result, (std::vector<int>{1, 3, 5}));
}
 
TEST_F(ParallelTest, PfilterNone)
{
  auto result = pfilter(pool, small_vec, [](int x) { return x > 100; });
  EXPECT_TRUE(result.empty());
}
 
TEST_F(ParallelTest, PfilterAll)
{
  auto result = pfilter(pool, small_vec, [](int x) { return x > 0; });
  EXPECT_EQ(result.size(), 5u);
}
 
TEST_F(ParallelTest, PfilterEmpty)
{
  auto result = pfilter(pool, empty_vec, [](int) { return true; });
  EXPECT_TRUE(result.empty());
}
 
TEST_F(ParallelTest, PfilterPreservesOrder)
{
  std::vector<int> nums = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
  auto result = pfilter(pool, nums, [](int x) { return x % 3 == 0; });
  EXPECT_EQ(result, (std::vector<int>{3, 6, 9}));
}
 
TEST_F(ParallelTest, PfilterLargeData)
{
  auto result = pfilter(pool, large_vec, [](int x) { return x % 100 == 0; });
  EXPECT_EQ(result.size(), 100u);
  EXPECT_EQ(result[0], 100);
  EXPECT_EQ(result[99], 10000);
}
 
TEST_F(ParallelTest, ParallelAlgorithmsHonorCancellation)
{
  CancellationSource source;
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 32;
  options.cancel_token = source.token();
 
  EXPECT_THROW(pmaps(large_vec,
                     [&source](int x) {
                       if (x == 128)
                         source.request_cancel();
                       return x * 2;
                     },
                     options),
               operation_canceled);
}
 
// =============================================================================
// pfoldl Tests
// =============================================================================
 
TEST_F(ParallelTest, PfoldlSum)
{
  int sum = pfoldl(pool, small_vec, 0, std::plus<int>());
  EXPECT_EQ(sum, 15);
}
 
TEST_F(ParallelTest, PfoldlProduct)
{
  int product = pfoldl(pool, small_vec, 1, std::multiplies<int>());
  EXPECT_EQ(product, 120);
}
 
TEST_F(ParallelTest, PfoldlMax)
{
  int max_val = pfoldl(pool, small_vec, INT_MIN,
                       [](int a, int b) { return std::max(a, b); });
  EXPECT_EQ(max_val, 5);
}
 
TEST_F(ParallelTest, PfoldlEmpty)
{
  int result = pfoldl(pool, empty_vec, 42, std::plus<int>());
  EXPECT_EQ(result, 42);
}
 
TEST_F(ParallelTest, PfoldlLargeData)
{
  long long sum = pfoldl(pool, large_vec, 0LL,
                         [](long long a, int b) { return a + b; });
  // Sum 1..10000 = 10000*10001/2 = 50005000
  EXPECT_EQ(sum, 50005000LL);
}
 
// =============================================================================
// pfor_each Tests
// =============================================================================
 
TEST_F(ParallelTest, PforEachModify)
{
  std::vector<int> data = {1, 2, 3, 4, 5};
  pfor_each(pool, data, [](int& x) { x *= 2; });
  EXPECT_EQ(data, (std::vector<int>{2, 4, 6, 8, 10}));
}
 
TEST_F(ParallelTest, PforEachAtomic)
{
  std::atomic<int> count{0};
  pfor_each(pool, large_vec, [&count](const int& x) {
    if (x % 2 == 0) ++count;
  });
  EXPECT_EQ(count.load(), 5000);
}
 
TEST_F(ParallelTest, PforEachEmpty)
{
  std::atomic<int> count{0};
  pfor_each(pool, empty_vec, [&count](int) { ++count; });
  EXPECT_EQ(count.load(), 0);
}
 
TEST_F(ParallelTest, PforEachAcceptsDefaultPoolThroughOptions)
{
  std::vector<int> data = {1, 2, 3, 4};
  ParallelOptions options;
  options.chunk_size = 2;
  options.max_tasks = 2;
 
  pfor_each(data, [](int & x) { x += 5; }, options);
  EXPECT_EQ(data, (std::vector<int>{6, 7, 8, 9}));
}
 
// =============================================================================
// pall, pexists, pnone Tests
// =============================================================================
 
TEST_F(ParallelTest, PallTrue)
{
  bool result = pall(pool, small_vec, [](int x) { return x > 0; });
  EXPECT_TRUE(result);
}
 
TEST_F(ParallelTest, PallFalse)
{
  bool result = pall(pool, small_vec, [](int x) { return x > 3; });
  EXPECT_FALSE(result);
}
 
TEST_F(ParallelTest, PallEmpty)
{
  bool result = pall(pool, empty_vec, [](int) { return false; });
  EXPECT_TRUE(result);  // Vacuous truth
}
 
TEST_F(ParallelTest, PexistsTrue)
{
  bool result = pexists(pool, small_vec, [](int x) { return x == 3; });
  EXPECT_TRUE(result);
}
 
TEST_F(ParallelTest, PexistsFalse)
{
  bool result = pexists(pool, small_vec, [](int x) { return x > 100; });
  EXPECT_FALSE(result);
}
 
TEST_F(ParallelTest, PexistsEmpty)
{
  bool result = pexists(pool, empty_vec, [](int) { return true; });
  EXPECT_FALSE(result);
}
 
TEST_F(ParallelTest, PnoneTrue)
{
  bool result = pnone(pool, small_vec, [](int x) { return x < 0; });
  EXPECT_TRUE(result);
}
 
TEST_F(ParallelTest, PnoneFalse)
{
  bool result = pnone(pool, small_vec, [](int x) { return x == 3; });
  EXPECT_FALSE(result);
}
 
// =============================================================================
// pcount_if Tests
// =============================================================================
 
TEST_F(ParallelTest, PcountIfEvens)
{
  size_t count = pcount_if(pool, small_vec, [](int x) { return x % 2 == 0; });
  EXPECT_EQ(count, 2u);
}
 
TEST_F(ParallelTest, PcountIfDivisibleByHundred)
{
  size_t count = pcount_if(pool, large_vec, [](int x) { return x % 100 == 0; });
  EXPECT_EQ(count, 100u);
}
 
TEST_F(ParallelTest, PcountIfAll)
{
  size_t count = pcount_if(pool, small_vec, [](int x) { return x > 0; });
  EXPECT_EQ(count, 5u);
}
 
TEST_F(ParallelTest, PcountIfNone)
{
  size_t count = pcount_if(pool, small_vec, [](int x) { return x > 100; });
  EXPECT_EQ(count, 0u);
}
 
TEST_F(ParallelTest, PcountIfLargeData)
{
  size_t count = pcount_if(pool, large_vec, [](int x) { return x % 7 == 0; });
  EXPECT_EQ(count, 1428u);  // floor(10000/7) = 1428
}
 
// =============================================================================
// pfind Tests
// =============================================================================
 
TEST_F(ParallelTest, PfindFound)
{
  auto idx = pfind(pool, small_vec, [](int x) { return x == 3; });
  ASSERT_TRUE(idx.has_value());
  EXPECT_EQ(*idx, 2u);
}
 
TEST_F(ParallelTest, PfindNotFound)
{
  auto idx = pfind(pool, small_vec, [](int x) { return x == 100; });
  EXPECT_FALSE(idx.has_value());
}
 
TEST_F(ParallelTest, PfindFirst)
{
  std::vector<int> data = {1, 2, 3, 3, 3, 4, 5};
  auto idx = pfind(pool, data, [](int x) { return x == 3; });
  ASSERT_TRUE(idx.has_value());
  EXPECT_EQ(*idx, 2u);  // First occurrence
}
 
TEST_F(ParallelTest, PfindEmpty)
{
  auto idx = pfind(pool, empty_vec, [](int) { return true; });
  EXPECT_FALSE(idx.has_value());
}
 
TEST_F(ParallelTest, PfindAcceptsParallelOptions)
{
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 64;
  options.max_tasks = 4;
 
  auto idx = pfind(large_vec, [](int x) { return x == 7777; }, options);
  ASSERT_TRUE(idx.has_value());
  EXPECT_EQ(*idx, 7776u);
}
 
TEST_F(ParallelTest, PfindValueFound)
{
  std::vector<std::string> words = {"apple", "banana", "cherry"};
  auto found = pfind_value(pool, words, 
                           [](const std::string& s) { return s.length() > 5; });
  ASSERT_TRUE(found.has_value());
  EXPECT_EQ(*found, "banana");
}
 
// =============================================================================
// psum, pproduct Tests
// =============================================================================
 
TEST_F(ParallelTest, Psum)
{
  auto sum = psum(pool, small_vec);
  EXPECT_EQ(sum, 15);
}
 
TEST_F(ParallelTest, PsumWithInit)
{
  auto sum = psum(pool, small_vec, 100);
  EXPECT_EQ(sum, 115);
}
 
TEST_F(ParallelTest, PsumEmpty)
{
  auto sum = psum(pool, empty_vec);
  EXPECT_EQ(sum, 0);
}
 
TEST_F(ParallelTest, PsumLargeData)
{
  auto sum = psum<std::vector<int>, long long>(pool, large_vec, 0LL);
  EXPECT_EQ(sum, 50005000LL);
}
 
TEST_F(ParallelTest, Pproduct)
{
  auto product = pproduct(pool, small_vec);
  EXPECT_EQ(product, 120);
}
 
TEST_F(ParallelTest, PproductEmpty)
{
  auto product = pproduct(pool, empty_vec);
  EXPECT_EQ(product, 1);  // Identity for multiplication
}
 
// =============================================================================
// pmin, pmax, pminmax Tests
// =============================================================================
 
TEST_F(ParallelTest, Pmin)
{
  auto min_val = pmin(pool, small_vec);
  ASSERT_TRUE(min_val.has_value());
  EXPECT_EQ(*min_val, 1);
}
 
TEST_F(ParallelTest, Pmax)
{
  auto max_val = pmax(pool, small_vec);
  ASSERT_TRUE(max_val.has_value());
  EXPECT_EQ(*max_val, 5);
}
 
TEST_F(ParallelTest, PminEmpty)
{
  auto min_val = pmin(pool, empty_vec);
  EXPECT_FALSE(min_val.has_value());
}
 
TEST_F(ParallelTest, PmaxEmpty)
{
  auto max_val = pmax(pool, empty_vec);
  EXPECT_FALSE(max_val.has_value());
}
 
TEST_F(ParallelTest, Pminmax)
{
  auto result = pminmax(pool, small_vec);
  ASSERT_TRUE(result.has_value());
  EXPECT_EQ(result->first, 1);
  EXPECT_EQ(result->second, 5);
}
 
TEST_F(ParallelTest, PminmaxShuffle)
{
  std::vector<int> data = {5, 1, 3, 2, 4};
  auto result = pminmax(pool, data);
  ASSERT_TRUE(result.has_value());
  EXPECT_EQ(result->first, 1);
  EXPECT_EQ(result->second, 5);
}
 
TEST_F(ParallelTest, PminmaxLargeData)
{
  auto result = pminmax(pool, large_vec);
  ASSERT_TRUE(result.has_value());
  EXPECT_EQ(result->first, 1);
  EXPECT_EQ(result->second, 10000);
}
 
// =============================================================================
// psort Tests
// =============================================================================
 
TEST_F(ParallelTest, PsortBasic)
{
  std::vector<int> data = {5, 2, 8, 1, 9, 3, 7, 4, 6};
  psort(pool, data);
  EXPECT_EQ(data, (std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9}));
}
 
TEST_F(ParallelTest, PsortDescending)
{
  std::vector<int> data = {5, 2, 8, 1, 9, 3, 7, 4, 6};
  psort(pool, data, std::greater<int>());
  EXPECT_EQ(data, (std::vector<int>{9, 8, 7, 6, 5, 4, 3, 2, 1}));
}
 
TEST_F(ParallelTest, PsortEmpty)
{
  std::vector<int> data;
  psort(pool, data);
  EXPECT_TRUE(data.empty());
}
 
TEST_F(ParallelTest, PsortSingle)
{
  std::vector<int> data = {42};
  psort(pool, data);
  EXPECT_EQ(data[0], 42);
}
 
TEST_F(ParallelTest, PsortAlreadySorted)
{
  std::vector<int> data = {1, 2, 3, 4, 5};
  psort(pool, data);
  EXPECT_EQ(data, (std::vector<int>{1, 2, 3, 4, 5}));
}
 
TEST_F(ParallelTest, PsortLargeData)
{
  std::vector<int> data = large_vec;
  std::mt19937 rng(42);  // Fixed seed for reproducibility
  std::shuffle(data.begin(), data.end(), rng);
  psort(pool, data);
  for (size_t i = 0; i < data.size(); ++i)
    EXPECT_EQ(data[i], static_cast<int>(i + 1));
}
 
TEST_F(ParallelTest, PsortAcceptsParallelOptions)
{
  std::vector<int> data = {7, 1, 5, 9, 3, 2};
  ParallelOptions options;
  options.pool = &pool;
  options.min_size = 2;
  options.max_tasks = 2;
  psort(data, std::less<int>{}, options);
  EXPECT_EQ(data, (std::vector<int>{1, 2, 3, 5, 7, 9}));
}
 
TEST_F(ParallelTest, PsortHonorsCancellation)
{
  std::vector<int> data(4096);
  std::iota(data.begin(), data.end(), 0);
  std::mt19937 rng(77);
  std::shuffle(data.begin(), data.end(), rng);
 
  CancellationSource source;
  ParallelOptions options;
  options.pool = &pool;
  options.min_size = 64;
  options.chunk_size = 64;
  options.cancel_token = source.token();
  source.request_cancel();
 
  EXPECT_THROW(psort(data, std::less<int>{}, options), operation_canceled);
}
 
// =============================================================================
// pzip_for_each Tests
// =============================================================================
 
TEST_F(ParallelTest, PzipForEachBasic)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 5, 6};
  std::atomic<int> sum{0};
  
  pzip_for_each(pool, a, b, [&sum](int x, int y) { sum += x * y; });
  EXPECT_EQ(sum.load(), 32);  // 1*4 + 2*5 + 3*6
}
 
TEST_F(ParallelTest, PzipForEachDifferentLengths)
{
  std::vector<int> a = {1, 2, 3, 4, 5};
  std::vector<int> b = {10, 20, 30};
  std::atomic<int> sum{0};
  
  pzip_for_each(pool, a, b, [&sum](int x, int y) { sum += x + y; });
  EXPECT_EQ(sum.load(), 66);  // (1+10) + (2+20) + (3+30)
}
 
TEST_F(ParallelTest, PzipForEachEmpty)
{
  std::vector<int> a;
  std::vector<int> b = {1, 2, 3};
  std::atomic<int> count{0};
  
  pzip_for_each(pool, a, b, [&count](int, int) { ++count; });
  EXPECT_EQ(count.load(), 0);
}
 
// =============================================================================
// pzip_maps Tests
// =============================================================================
 
TEST_F(ParallelTest, PzipMapsBasic)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 5, 6};
  
  auto result = pzip_maps(pool, a, b, [](int x, int y) { return x * y; });
  EXPECT_EQ(result, (std::vector<int>{4, 10, 18}));
}
 
TEST_F(ParallelTest, PzipMapsTypeConversion)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<double> b = {1.5, 2.5, 3.5};
  
  auto result = pzip_maps(pool, a, b, [](int x, double y) { return x + y; });
  EXPECT_NEAR(result[0], 2.5, 0.001);
  EXPECT_NEAR(result[1], 4.5, 0.001);
  EXPECT_NEAR(result[2], 6.5, 0.001);
}
 
TEST_F(ParallelTest, ZipAlgorithmsAcceptParallelOptions)
{
  std::vector<int> a = {1, 2, 3, 4};
  std::vector<int> b = {10, 20, 30, 40};
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 2;
 
  auto mapped = pzip_maps(a, b, [](int x, int y) { return x + y; }, options);
  EXPECT_EQ(mapped, (std::vector<int>{11, 22, 33, 44}));
 
  auto folded = pzip_foldl(a, b, 0, [](int acc, int x, int y) {
    return acc + x * y;
  }, options);
  EXPECT_EQ(folded, 300);
 
  std::atomic<int> visits{0};
  pzip_for_each(a, b, [&visits](int, int) { ++visits; }, options);
  EXPECT_EQ(visits.load(), 4);
}
 
// =============================================================================
// ppartition Tests
// =============================================================================
 
TEST_F(ParallelTest, PpartitionBasic)
{
  auto [evens, odds] = ppartition(pool, small_vec, [](int x) { return x % 2 == 0; });
  EXPECT_EQ(evens, (std::vector<int>{2, 4}));
  EXPECT_EQ(odds, (std::vector<int>{1, 3, 5}));
}
 
TEST_F(ParallelTest, PpartitionAllTrue)
{
  auto [yes, no] = ppartition(pool, small_vec, [](int x) { return x > 0; });
  EXPECT_EQ(yes.size(), 5u);
  EXPECT_TRUE(no.empty());
}
 
TEST_F(ParallelTest, PpartitionAllFalse)
{
  auto [yes, no] = ppartition(pool, small_vec, [](int x) { return x < 0; });
  EXPECT_TRUE(yes.empty());
  EXPECT_EQ(no.size(), 5u);
}
 
TEST_F(ParallelTest, PpartitionEmpty)
{
  auto [yes, no] = ppartition(pool, empty_vec, [](int) { return true; });
  EXPECT_TRUE(yes.empty());
  EXPECT_TRUE(no.empty());
}
 
TEST_F(ParallelTest, PpartitionPreservesOrder)
{
  std::vector<int> data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
  auto [evens, odds] = ppartition(pool, data, [](int x) { return x % 2 == 0; });
  EXPECT_EQ(evens, (std::vector<int>{2, 4, 6, 8, 10}));
  EXPECT_EQ(odds, (std::vector<int>{1, 3, 5, 7, 9}));
}
 
TEST_F(ParallelTest, PpartitionAcceptsParallelOptions)
{
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 3;
 
  std::vector<int> data = {1, 2, 3, 4, 5, 6, 7, 8};
  auto [evens, odds] = ppartition(data, [](int x) { return x % 2 == 0; }, options);
  EXPECT_EQ(evens, (std::vector<int>{2, 4, 6, 8}));
  EXPECT_EQ(odds, (std::vector<int>{1, 3, 5, 7}));
}
 
TEST_F(ParallelTest, PpartitionSupportsNonDefaultConstructibleValues)
{
  struct NoDefault
  {
    int value;
 
    NoDefault() = delete;
    explicit NoDefault(int v) : value(v) {}
    NoDefault(const NoDefault &) = default;
    NoDefault(NoDefault &&) noexcept = default;
    NoDefault & operator=(const NoDefault &) = default;
    NoDefault & operator=(NoDefault &&) noexcept = default;
  };
 
  std::vector<NoDefault> data;
  for (int i = 1; i <= 6; ++i)
    data.emplace_back(i);
 
  auto [small, large] = ppartition(pool, data, [](const NoDefault & x) {
    return x.value <= 3;
  });
 
  ASSERT_EQ(small.size(), 3u);
  ASSERT_EQ(large.size(), 3u);
  EXPECT_EQ(small[0].value, 1);
  EXPECT_EQ(small[1].value, 2);
  EXPECT_EQ(small[2].value, 3);
  EXPECT_EQ(large[0].value, 4);
  EXPECT_EQ(large[1].value, 5);
  EXPECT_EQ(large[2].value, 6);
}
 
TEST_F(ParallelTest, PpartitionHonorsCancellation)
{
  CancellationSource source;
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 16;
  options.cancel_token = source.token();
  source.request_cancel();
 
  std::vector<int> data(512, 1);
  EXPECT_THROW((void) ppartition(data, [](int x) { return x % 2 == 0; }, options),
               operation_canceled);
}
 
TEST_F(ParallelTest, PfilterSupportsNonDefaultConstructibleValues)
{
  struct NoDefault
  {
    int value;
 
    NoDefault() = delete;
    explicit NoDefault(int v) : value(v) {}
    NoDefault(const NoDefault &) = default;
    NoDefault(NoDefault &&) noexcept = default;
    NoDefault & operator=(const NoDefault &) = default;
    NoDefault & operator=(NoDefault &&) noexcept = default;
  };
 
  std::vector<NoDefault> data;
  for (int i = 1; i <= 8; ++i)
    data.emplace_back(i);
 
  auto result = pfilter(pool, data, [](const NoDefault & x) {
    return x.value % 2 == 0;
  });
 
  ASSERT_EQ(result.size(), 4u);
  EXPECT_EQ(result[0].value, 2);
  EXPECT_EQ(result[1].value, 4);
  EXPECT_EQ(result[2].value, 6);
  EXPECT_EQ(result[3].value, 8);
}
 
// =============================================================================
// pscan / pexclusive_scan / pmerge Tests
// =============================================================================
 
TEST_F(ParallelTest, PscanContainerWrapperMatchesPartialSum)
{
  std::vector<int> data(256);
  std::iota(data.begin(), data.end(), 1);
  std::vector<int> expected(data.size());
  std::partial_sum(data.begin(), data.end(), expected.begin());
 
  auto result = pscan(pool, data, std::plus<int>{});
  EXPECT_EQ(result, expected);
}
 
TEST_F(ParallelTest, PexclusiveScanContainerWrapperWorks)
{
  std::vector<int> data = {1, 2, 3, 4, 5};
  auto result = pexclusive_scan(pool, data, 0, std::plus<int>{});
  EXPECT_EQ(result, (std::vector<int>{0, 1, 3, 6, 10}));
}
 
TEST_F(ParallelTest, ScanAndMergeAcceptParallelOptions)
{
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 8;
 
  std::vector<int> data = {3, 1, 4, 1, 5, 9};
  auto inclusive = pscan(data, std::plus<int>{}, options);
  EXPECT_EQ(inclusive, (std::vector<int>{3, 4, 8, 9, 14, 23}));
 
  auto exclusive = pexclusive_scan(data, 10, std::plus<int>{}, options);
  EXPECT_EQ(exclusive, (std::vector<int>{10, 13, 14, 18, 19, 24}));
 
  std::vector<int> left = {1, 3, 5, 7};
  std::vector<int> right = {2, 4, 6, 8};
  auto merged = pmerge(left, right, std::less<int>{}, options);
  EXPECT_EQ(merged, (std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8}));
}
 
TEST_F(ParallelTest, ScanAndMergeHonorCancellation)
{
  CancellationSource source;
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 16;
  options.cancel_token = source.token();
  source.request_cancel();
 
  std::vector<int> data(512, 1);
  EXPECT_THROW((void) pscan(data, std::plus<int>{}, options), operation_canceled);
  EXPECT_THROW((void) pexclusive_scan(data, 0, std::plus<int>{}, options),
               operation_canceled);
 
  std::vector<int> left(256), right(256);
  std::iota(left.begin(), left.end(), 0);
  std::iota(right.begin(), right.end(), 0);
  EXPECT_THROW((void) pmerge(left, right, std::less<int>{}, options),
               operation_canceled);
}
 
TEST_F(ParallelTest, ScanAndMergeSupportNonDefaultConstructibleValues)
{
  struct NoDefault
  {
    int value;
 
    NoDefault() = delete;
    explicit NoDefault(int v) : value(v) {}
    NoDefault(const NoDefault &) = default;
    NoDefault(NoDefault &&) noexcept = default;
    NoDefault & operator=(const NoDefault &) = default;
    NoDefault & operator=(NoDefault &&) noexcept = default;
  };
 
  std::vector<NoDefault> data;
  for (int i = 1; i <= 4; ++i)
    data.emplace_back(i);
 
  auto inclusive = pscan(data,
                         [](const NoDefault & a, const NoDefault & b) {
                           return NoDefault(a.value + b.value);
                         });
  ASSERT_EQ(inclusive.size(), 4u);
  EXPECT_EQ(inclusive[0].value, 1);
  EXPECT_EQ(inclusive[1].value, 3);
  EXPECT_EQ(inclusive[2].value, 6);
  EXPECT_EQ(inclusive[3].value, 10);
 
  auto exclusive = pexclusive_scan(data, NoDefault(0),
                                   [](const NoDefault & a, const NoDefault & b) {
                                     return NoDefault(a.value + b.value);
                                   });
  ASSERT_EQ(exclusive.size(), 4u);
  EXPECT_EQ(exclusive[0].value, 0);
  EXPECT_EQ(exclusive[1].value, 1);
  EXPECT_EQ(exclusive[2].value, 3);
  EXPECT_EQ(exclusive[3].value, 6);
 
  std::vector<NoDefault> left, right;
  left.emplace_back(1);
  left.emplace_back(3);
  left.emplace_back(5);
  right.emplace_back(2);
  right.emplace_back(4);
  right.emplace_back(6);
 
  auto merged = pmerge(left, right,
                       [](const NoDefault & a, const NoDefault & b) {
                         return a.value < b.value;
                       });
  ASSERT_EQ(merged.size(), 6u);
  for (size_t i = 0; i < merged.size(); ++i)
    EXPECT_EQ(merged[i].value, static_cast<int>(i + 1));
}
 
// =============================================================================
// Correctness: Sequential vs Parallel Comparison
// =============================================================================
 
TEST_F(ParallelTest, CorrectnessMapVsSequential)
{
  // Sequential
  std::vector<int> seq_result;
  for (int x : large_vec)
    seq_result.push_back(x * 2 + 1);
  
  // Parallel
  auto par_result = pmaps(pool, large_vec, [](int x) { return x * 2 + 1; });
  
  EXPECT_EQ(seq_result, par_result);
}
 
TEST_F(ParallelTest, CorrectnessFilterVsSequential)
{
  // Sequential
  std::vector<int> seq_result;
  for (int x : large_vec)
    if (x % 17 == 0)
      seq_result.push_back(x);
  
  // Parallel
  auto par_result = pfilter(pool, large_vec, [](int x) { return x % 17 == 0; });
  
  EXPECT_EQ(seq_result, par_result);
}
 
TEST_F(ParallelTest, CorrectnessFoldVsSequential)
{
  // Sequential
  long long seq_sum = 0;
  for (int x : large_vec)
    seq_sum += x;
  
  // Parallel
  long long par_sum = pfoldl(pool, large_vec, 0LL,
                             [](long long a, int b) { return a + b; });
  
  EXPECT_EQ(seq_sum, par_sum);
}
 
// =============================================================================
// Performance Benchmarks
// =============================================================================
 
namespace
{
  [[nodiscard]] bool parallel_benchmarks_enabled()
  {
    return std::getenv("ALEPH_RUN_PARALLEL_BENCHMARKS") != nullptr;
  }
}
 
TEST_F(ParallelTest, BenchmarkMapSpeedup)
{
  if (not parallel_benchmarks_enabled())
    GTEST_SKIP() << "Skipping timing-sensitive parallel benchmarks; "
                 << "set ALEPH_RUN_PARALLEL_BENCHMARKS=1 to enable";
 
  std::vector<int> data(1000000);
  std::iota(data.begin(), data.end(), 0);
  
  // Sequential
  auto seq_start = std::chrono::high_resolution_clock::now();
  std::vector<int> seq_result;
  seq_result.reserve(data.size());
  for (int x : data)
    seq_result.push_back(x * x + 2 * x + 1);
  auto seq_end = std::chrono::high_resolution_clock::now();
  auto seq_duration = std::chrono::duration_cast<std::chrono::microseconds>(
    seq_end - seq_start).count();
  
  // Parallel
  auto par_start = std::chrono::high_resolution_clock::now();
  auto par_result = pmaps(pool, data, [](int x) { return x * x + 2 * x + 1; });
  auto par_end = std::chrono::high_resolution_clock::now();
  auto par_duration = std::chrono::duration_cast<std::chrono::microseconds>(
    par_end - par_start).count();
  
  EXPECT_EQ(seq_result, par_result);
  
  double speedup = static_cast<double>(seq_duration) / par_duration;
  std::cout << "\n=== Benchmark: pmaps ===\n";
  std::cout << "Data size: 1M elements\n";
  std::cout << "Sequential: " << seq_duration << " μs\n";
  std::cout << "Parallel:   " << par_duration << " μs\n";
  std::cout << "Speedup:    " << speedup << "x\n";
  std::cout << "========================\n\n";
  
  // With multiple threads, parallel overhead may exceed benefits for trivial ops
  // On CI machines with limited cores or high load, speedup can be < 1.0
  // Use very relaxed threshold - this test validates correctness, not performance
  // Performance is validated by filter/sort benchmarks which have higher per-element cost
  if (pool.num_threads() > 1)
    EXPECT_GT(speedup, 0.3);  // Very relaxed: just ensure parallelization isn't broken
}
 
TEST_F(ParallelTest, BenchmarkFilterSpeedup)
{
  if (not parallel_benchmarks_enabled())
    GTEST_SKIP() << "Skipping timing-sensitive parallel benchmarks; "
                 << "set ALEPH_RUN_PARALLEL_BENCHMARKS=1 to enable";
 
  std::vector<int> data(1000000);
  std::iota(data.begin(), data.end(), 0);
  
  auto pred = [](int x) { return x % 7 == 0 || x % 11 == 0; };
  
  // Sequential
  auto seq_start = std::chrono::high_resolution_clock::now();
  std::vector<int> seq_result;
  for (int x : data)
    if (pred(x))
      seq_result.push_back(x);
  auto seq_end = std::chrono::high_resolution_clock::now();
  auto seq_duration = std::chrono::duration_cast<std::chrono::microseconds>(
    seq_end - seq_start).count();
  
  // Parallel
  auto par_start = std::chrono::high_resolution_clock::now();
  auto par_result = pfilter(pool, data, pred);
  auto par_end = std::chrono::high_resolution_clock::now();
  auto par_duration = std::chrono::duration_cast<std::chrono::microseconds>(
    par_end - par_start).count();
  
  EXPECT_EQ(seq_result, par_result);
  
  double speedup = static_cast<double>(seq_duration) / par_duration;
  std::cout << "\n=== Benchmark: pfilter ===\n";
  std::cout << "Data size: 1M elements\n";
  std::cout << "Sequential: " << seq_duration << " μs\n";
  std::cout << "Parallel:   " << par_duration << " μs\n";
  std::cout << "Speedup:    " << speedup << "x\n";
  std::cout << "==========================\n\n";
}
 
TEST_F(ParallelTest, BenchmarkSortSpeedup)
{
  if (not parallel_benchmarks_enabled())
    GTEST_SKIP() << "Skipping timing-sensitive parallel benchmarks; "
                 << "set ALEPH_RUN_PARALLEL_BENCHMARKS=1 to enable";
 
  std::vector<int> data(100000);
  std::iota(data.begin(), data.end(), 0);
  std::mt19937 rng(123);  // Fixed seed for reproducibility
  std::shuffle(data.begin(), data.end(), rng);
  
  std::vector<int> data_copy = data;
  
  // Sequential
  auto seq_start = std::chrono::high_resolution_clock::now();
  std::sort(data.begin(), data.end());
  auto seq_end = std::chrono::high_resolution_clock::now();
  auto seq_duration = std::chrono::duration_cast<std::chrono::microseconds>(
    seq_end - seq_start).count();
  
  // Parallel
  auto par_start = std::chrono::high_resolution_clock::now();
  psort(pool, data_copy);
  auto par_end = std::chrono::high_resolution_clock::now();
  auto par_duration = std::chrono::duration_cast<std::chrono::microseconds>(
    par_end - par_start).count();
  
  EXPECT_EQ(data, data_copy);
  
  double speedup = static_cast<double>(seq_duration) / par_duration;
  std::cout << "\n=== Benchmark: psort ===\n";
  std::cout << "Data size: 100K elements\n";
  std::cout << "std::sort: " << seq_duration << " μs\n";
  std::cout << "psort:     " << par_duration << " μs\n";
  std::cout << "Speedup:   " << speedup << "x\n";
  std::cout << "========================\n\n";
}
 
// =============================================================================
// Thread Safety Tests
// =============================================================================
 
TEST_F(ParallelTest, ThreadSafetyConcurrentOperations)
{
  // Run multiple parallel operations simultaneously
  auto future1 = std::async(std::launch::async, [this]() {
    return pmaps(pool, large_vec, [](int x) { return x * 2; });
  });
  
  auto future2 = std::async(std::launch::async, [this]() {
    return pfilter(pool, large_vec, [](int x) { return x % 2 == 0; });
  });
  
  auto future3 = std::async(std::launch::async, [this]() {
    return psum<std::vector<int>, long long>(pool, large_vec, 0LL);
  });
  
  auto result1 = future1.get();
  auto result2 = future2.get();
  auto result3 = future3.get();
  
  EXPECT_EQ(result1.size(), 10000u);
  EXPECT_EQ(result2.size(), 5000u);
  EXPECT_EQ(result3, 50005000LL);
}
 
// =============================================================================
// Edge Cases
// =============================================================================
 
TEST_F(ParallelTest, EdgeCaseSingleThread)
{
  ThreadPool single_pool(1);
  
  auto result = pmaps(single_pool, small_vec, [](int x) { return x * x; });
  EXPECT_EQ(result, (std::vector<int>{1, 4, 9, 16, 25}));
}
 
TEST_F(ParallelTest, EdgeCaseManyThreads)
{
  ThreadPool many_pool(16);
  
  auto result = pmaps(many_pool, small_vec, [](int x) { return x * x; });
  EXPECT_EQ(result, (std::vector<int>{1, 4, 9, 16, 25}));
}
 
TEST_F(ParallelTest, EdgeCaseVeryLargeData)
{
  std::vector<int> huge(100000);
  std::iota(huge.begin(), huge.end(), 0);
  
  auto sum = psum<std::vector<int>, long long>(pool, huge, 0LL);
  // Sum 0..99999 = 99999*100000/2 = 4999950000
  EXPECT_EQ(sum, 4999950000LL);
}
 
// =============================================================================
// Variadic Zip Tests (N containers)
// =============================================================================
 
TEST_F(ParallelTest, PzipForEachN3Containers)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 5, 6};
  std::vector<int> c = {7, 8, 9};
  std::atomic<int> sum{0};
 
  pzip_for_each_n(pool, [&sum](int x, int y, int z) {
    sum += x + y + z;
  }, a, b, c);
  
  EXPECT_EQ(sum.load(), 45);  // (1+4+7) + (2+5+8) + (3+6+9)
}
 
TEST_F(ParallelTest, PzipForEachN4Containers)
{
  std::vector<int> a = {1, 2};
  std::vector<int> b = {3, 4};
  std::vector<int> c = {5, 6};
  std::vector<int> d = {7, 8};
  std::atomic<int> sum{0};
 
  pzip_for_each_n(pool, [&sum](int w, int x, int y, int z) {
    sum += w * x * y * z;
  }, a, b, c, d);
  
  EXPECT_EQ(sum.load(), 1*3*5*7 + 2*4*6*8);  // 105 + 384 = 489
}
 
TEST_F(ParallelTest, PzipForEachNDifferentLengths)
{
  std::vector<int> a = {1, 2, 3, 4, 5};
  std::vector<int> b = {10, 20, 30};
  std::vector<int> c = {100, 200, 300, 400};
  std::atomic<int> count{0};
 
  pzip_for_each_n(pool, [&count](int, int, int) {
    ++count;
  }, a, b, c);
  
  EXPECT_EQ(count.load(), 3);  // Min length
}
 
TEST_F(ParallelTest, PzipMapsN3Containers)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 5, 6};
  std::vector<int> c = {7, 8, 9};
 
  auto result = pzip_maps_n(pool, [](int x, int y, int z) {
    return x + y + z;
  }, a, b, c);
  
  EXPECT_EQ(result, (std::vector<int>{12, 15, 18}));
}
 
TEST_F(ParallelTest, PzipMapsNEmpty)
{
  std::vector<int> a;
  std::vector<int> b = {1, 2, 3};
 
  auto result = pzip_maps_n(pool, [](int x, int y) {
    return x + y;
  }, a, b);
  
  EXPECT_TRUE(result.empty());
}
 
TEST_F(ParallelTest, PzipFoldlNVariadic)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 5, 6};
 
  // Dot product using variadic version with combiner
  int dot = pzip_foldl_n(pool, 0, 
    [](int acc, int x, int y) { return acc + x * y; },
    std::plus<int>(),
    a, b);
  
  EXPECT_EQ(dot, 32);  // 1*4 + 2*5 + 3*6
}
 
TEST_F(ParallelTest, PzipAllN3ContainersTrue)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {2, 3, 4};
  std::vector<int> c = {3, 4, 5};
 
  bool result = pzip_all_n(pool, [](int x, int y, int z) {
    return x < y && y < z;
  }, a, b, c);
  
  EXPECT_TRUE(result);
}
 
TEST_F(ParallelTest, PzipAllN3ContainersFalse)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {2, 3, 4};
  std::vector<int> c = {3, 2, 5};  // 2 < 3 is true, but 3 < 2 is false
 
  bool result = pzip_all_n(pool, [](int x, int y, int z) {
    return x < y && y < z;
  }, a, b, c);
  
  EXPECT_FALSE(result);
}
 
TEST_F(ParallelTest, PzipExistsNTrue)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 2, 6};
 
  bool result = pzip_exists_n(pool, [](int x, int y) {
    return x == y;
  }, a, b);
  
  EXPECT_TRUE(result);  // 2 == 2
}
 
TEST_F(ParallelTest, PzipExistsNFalse)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 5, 6};
 
  bool result = pzip_exists_n(pool, [](int x, int y) {
    return x == y;
  }, a, b);
  
  EXPECT_FALSE(result);
}
 
TEST_F(ParallelTest, PzipCountIfN)
{
  std::vector<int> a = {1, 2, 3, 4, 5};
  std::vector<int> b = {5, 4, 3, 2, 1};
 
  size_t count = pzip_count_if_n(pool, [](int x, int y) {
    return x + y == 6;
  }, a, b);
  
  EXPECT_EQ(count, 5u);  // All pairs sum to 6
}
 
TEST_F(ParallelTest, PzipCountIfNPartial)
{
  std::vector<int> a = {1, 2, 3, 4, 5};
  std::vector<int> b = {1, 2, 3, 4, 5};
 
  size_t count = pzip_count_if_n(pool, [](int x, int y) {
    return x == y && x % 2 == 0;
  }, a, b);
  
  EXPECT_EQ(count, 2u);  // 2 and 4
}
 
TEST_F(ParallelTest, VariadicZipAlgorithmsAcceptParallelOptions)
{
  std::vector<int> a = {1, 2, 3};
  std::vector<int> b = {4, 5, 6};
  std::vector<int> c = {7, 8, 9};
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 1;
 
  auto mapped = pzip_maps_n([](int x, int y, int z) { return x + y + z; },
                            options, a, b, c);
  EXPECT_EQ(mapped, (std::vector<int>{12, 15, 18}));
 
  auto folded = pzip_foldl_n(0,
                             [](int acc, int x, int y, int z) {
                               return acc + x + y + z;
                             },
                             std::plus<int>{},
                             options, a, b, c);
  EXPECT_EQ(folded, 45);
 
  EXPECT_TRUE(pzip_all_n([](int x, int y, int z) { return x < y && y < z; },
                         options, a, b, c));
  EXPECT_TRUE(pzip_exists_n([](int x, int y, int z) { return x + y + z == 15; },
                            options, a, b, c));
  EXPECT_EQ(pzip_count_if_n([](int x, int y, int z) { return x + y + z > 14; },
                            options, a, b, c),
            2u);
}
 
// =============================================================================
// Enumerate Tests
// =============================================================================
 
TEST_F(ParallelTest, PenumerateForEachModify)
{
  std::vector<int> data(100, 0);
  
  penumerate_for_each(pool, data, [](size_t i, int& x) {
    x = static_cast<int>(i * 2);
  });
  
  for (size_t i = 0; i < data.size(); ++i)
    EXPECT_EQ(data[i], static_cast<int>(i * 2));
}
 
TEST_F(ParallelTest, PenumerateForEachConst)
{
  const std::vector<int> data = {10, 20, 30, 40, 50};
  std::atomic<int> weighted_sum{0};
  
  penumerate_for_each(pool, data, [&weighted_sum](size_t i, int x) {
    weighted_sum += static_cast<int>(i) * x;
  });
  
  // 0*10 + 1*20 + 2*30 + 3*40 + 4*50 = 0 + 20 + 60 + 120 + 200 = 400
  EXPECT_EQ(weighted_sum.load(), 400);
}
 
TEST_F(ParallelTest, PenumerateMaps)
{
  std::vector<std::string> words = {"a", "bb", "ccc"};
  
  auto result = penumerate_maps(pool, words, 
    [](size_t i, const std::string& s) {
      return std::to_string(i) + ":" + s;
    });
  
  EXPECT_EQ(result[0], "0:a");
  EXPECT_EQ(result[1], "1:bb");
  EXPECT_EQ(result[2], "2:ccc");
}
 
TEST_F(ParallelTest, PenumerateMapsEmpty)
{
  std::vector<int> empty;
  
  auto result = penumerate_maps(pool, empty, 
    [](size_t i, int x) { return static_cast<int>(i + x); });
  
  EXPECT_TRUE(result.empty());
}
 
TEST_F(ParallelTest, PenumerateAcceptsParallelOptions)
{
  std::vector<int> data(6, 0);
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 2;
 
  penumerate_for_each(data, [](size_t i, int & x) {
    x = static_cast<int>(i + 10);
  }, options);
  EXPECT_EQ(data, (std::vector<int>{10, 11, 12, 13, 14, 15}));
 
  auto labels = penumerate_maps(data, [](size_t i, int x) {
    return std::to_string(i) + "=" + std::to_string(x);
  }, options);
  EXPECT_EQ(labels[0], "0=10");
  EXPECT_EQ(labels[5], "5=15");
}
 
TEST_F(ParallelTest, ZipAndEnumerateHonorCancellation)
{
  std::vector<int> a(256, 1), b(256, 2), c(256, 3);
  CancellationSource source;
  ParallelOptions options;
  options.pool = &pool;
  options.chunk_size = 16;
  options.cancel_token = source.token();
  source.request_cancel();
 
  EXPECT_THROW((void) pzip_maps(a, b, [](int x, int y) { return x + y; }, options),
               operation_canceled);
  EXPECT_THROW((void) pzip_maps_n([](int x, int y, int z) { return x + y + z; },
                                  options, a, b, c),
               operation_canceled);
  EXPECT_THROW((void) penumerate_maps(a, [](size_t i, int x) {
                 return static_cast<int>(i) + x;
               }, options),
               operation_canceled);
}
 
// =============================================================================
// Large Data Variadic Tests
// =============================================================================
 
TEST_F(ParallelTest, PzipForEachNLargeData)
{
  std::vector<int> a(10000), b(10000), c(10000);
  std::iota(a.begin(), a.end(), 0);
  std::iota(b.begin(), b.end(), 10000);
  std::iota(c.begin(), c.end(), 20000);
  
  std::atomic<long long> sum{0};
  
  pzip_for_each_n(pool, [&sum](int x, int y, int z) {
    sum += x + y + z;
  }, a, b, c);
  
  // Sum of 0..9999 + 10000..19999 + 20000..29999
  // = 3 * (0+1+...+9999) + 10000*10000 + 20000*10000
  // = 3 * 49995000 + 100000000 + 200000000
  // = 149985000 + 300000000 = 449985000
  EXPECT_EQ(sum.load(), 449985000LL);
}
 
TEST_F(ParallelTest, PzipMapsNLargeData)
{
  std::vector<int> a(10000), b(10000);
  std::iota(a.begin(), a.end(), 0);
  std::iota(b.begin(), b.end(), 0);
  
  auto result = pzip_maps_n(pool, [](int x, int y) {
    return x * y;
  }, a, b);
  
  for (size_t i = 0; i < result.size(); ++i)
    EXPECT_EQ(result[i], static_cast<int>(i * i));
}
 
TEST_F(ParallelTest, CorrectnessZipNVsSequential)
{
  std::vector<int> a(1000), b(1000), c(1000);
  std::iota(a.begin(), a.end(), 0);
  std::iota(b.begin(), b.end(), 1000);
  std::iota(c.begin(), c.end(), 2000);
  
  // Sequential
  std::vector<int> seq_result;
  for (size_t i = 0; i < a.size(); ++i)
    seq_result.push_back(a[i] + b[i] + c[i]);
  
  // Parallel
  auto par_result = pzip_maps_n(pool, [](int x, int y, int z) {
    return x + y + z;
  }, a, b, c);
  
  EXPECT_EQ(seq_result, par_result);
}
 
int main(int argc, char **argv)
{
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}