parallel_filter.hpp

Go to the documentation of this file.

#pragma once
 
#include "ityr/common/util.hpp"
#include "ityr/pattern/parallel_invoke.hpp"
#include "ityr/pattern/parallel_loop.hpp"
 
namespace ityr {
 
namespace internal {
 
template <typename W, typename BidirectionalIterator, typename Predicate>
inline BidirectionalIterator
stable_partition_aux(const execution::parallel_policy<W>& policy,
                     BidirectionalIterator                first,
                     BidirectionalIterator                last,
                     Predicate                            pred) {
  std::size_t d = std::distance(first, last);
 
  if (d <= policy.cutoff_count) {
    // TODO: consider policy.checkout_count
    ITYR_CHECK(policy.cutoff_count == policy.checkout_count);
 
    auto&& [css, its] = checkout_global_iterators(d, first);
    auto&& first_ = std::get<0>(its);
    auto m = std::stable_partition(first_, std::next(first_, d), pred);
    return std::next(first, std::distance(first_, m));
  }
 
  auto mid = std::next(first, d / 2);
 
  auto [m1, m2] = parallel_invoke(
      [=] { return stable_partition_aux(policy, first, mid , pred); },
      [=] { return stable_partition_aux(policy, mid  , last, pred); });
 
  return rotate(policy, m1, mid, m2);
}
 
}
 
template <typename ExecutionPolicy, typename BidirectionalIterator, typename Predicate>
inline BidirectionalIterator stable_partition(const ExecutionPolicy& policy,
                                              BidirectionalIterator  first,
                                              BidirectionalIterator  last,
                                              Predicate              pred) {
  if constexpr (ori::is_global_ptr_v<BidirectionalIterator>) {
    return stable_partition(
        policy,
        internal::convert_to_global_iterator(first, checkout_mode::read_write),
        internal::convert_to_global_iterator(last , checkout_mode::read_write),
        pred);
 
  } else {
    return internal::stable_partition_aux(policy, first, last, pred);
  }
}
 
ITYR_TEST_CASE("[ityr::pattern::parallel_filter] stable_partition") {
  ito::init();
  ori::init();
 
  ITYR_SUBCASE("split half") {
    long n = 100000;
    ori::global_ptr<long> p = ori::malloc_coll<long>(n);
 
    ito::root_exec([=] {
      transform(
          execution::parallel_policy(100),
          count_iterator<long>(0), count_iterator<long>(n), p,
          [=](long i) { return i; });
 
      auto pp = stable_partition(
          execution::parallel_policy(100),
          p, p + n,
          [](long x) { return x % 2 == 0; });
 
      ITYR_CHECK(pp == p + n / 2);
 
      for_each(
          execution::parallel_policy(100),
          make_global_iterator(p , checkout_mode::read),
          make_global_iterator(pp, checkout_mode::read),
          count_iterator<long>(0),
          [](long x, long i) { ITYR_CHECK(x == i * 2); });
 
      for_each(
          execution::parallel_policy(100),
          make_global_iterator(pp   , checkout_mode::read),
          make_global_iterator(p + n, checkout_mode::read),
          count_iterator<long>(0),
          [](long x, long i) { ITYR_CHECK(x == i * 2 + 1); });
    });
 
    ori::free_coll(p);
  }
 
  ITYR_SUBCASE("split 1:2") {
    long n = 90000;
    ori::global_ptr<long> p = ori::malloc_coll<long>(n);
 
    ito::root_exec([=] {
      transform(
          execution::parallel_policy(100),
          count_iterator<long>(0), count_iterator<long>(n), p,
          [=](long i) { return i; });
 
      auto pp = stable_partition(
          execution::parallel_policy(100),
          p, p + n,
          [](long x) { return x % 3 == 0; });
 
      ITYR_CHECK(pp == p + n / 3);
 
      for_each(
          execution::parallel_policy(100),
          make_global_iterator(p , checkout_mode::read),
          make_global_iterator(pp, checkout_mode::read),
          count_iterator<long>(0),
          [](long x, long i) { ITYR_CHECK(x == i * 3); });
 
      for_each(
          execution::parallel_policy(100),
          make_global_iterator(pp   , checkout_mode::read),
          make_global_iterator(p + n, checkout_mode::read),
          count_iterator<long>(0),
          [](long x, long i) { ITYR_CHECK(x == (i / 2) * 3 + (i % 2) + 1); });
    });
 
    ori::free_coll(p);
  }
 
  ITYR_SUBCASE("corner cases") {
    long n = 100000;
    ori::global_ptr<long> p = ori::malloc_coll<long>(n);
 
    ito::root_exec([=] {
      transform(
          execution::parallel_policy(100),
          count_iterator<long>(0), count_iterator<long>(n), p,
          [=](long i) { return i; });
 
      auto pp1 = stable_partition(
          execution::parallel_policy(100),
          p, p + n,
          [](long) { return true; });
 
      ITYR_CHECK(pp1 == p + n);
 
      for_each(
          execution::parallel_policy(100),
          make_global_iterator(p,     checkout_mode::read),
          make_global_iterator(p + n, checkout_mode::read),
          count_iterator<long>(0),
          [](long x, long i) { ITYR_CHECK(x == i); });
 
      auto pp2 = stable_partition(
          execution::parallel_policy(100),
          p, p + n,
          [](long) { return false; });
 
      ITYR_CHECK(pp2 == p);
 
      for_each(
          execution::parallel_policy(100),
          make_global_iterator(p,     checkout_mode::read),
          make_global_iterator(p + n, checkout_mode::read),
          count_iterator<long>(0),
          [](long x, long i) { ITYR_CHECK(x == i); });
    });
 
    ori::free_coll(p);
  }
 
  ori::fini();
  ito::fini();
}
 
template <typename ExecutionPolicy, typename BidirectionalIterator, typename Predicate>
inline BidirectionalIterator partition(const ExecutionPolicy& policy,
                                       BidirectionalIterator  first,
                                       BidirectionalIterator  last,
                                       Predicate              pred) {
  // TODO: implement faster unstable partition
  return stable_partition(policy, first, last, pred);
}
 
}