parallel_reduce.hpp

Go to the documentation of this file.

#pragma once
 
#include "ityr/common/util.hpp"
#include "ityr/common/topology.hpp"
#include "ityr/ito/ito.hpp"
#include "ityr/ori/ori.hpp"
#include "ityr/pattern/count_iterator.hpp"
#include "ityr/pattern/global_iterator.hpp"
#include "ityr/pattern/serial_loop.hpp"
#include "ityr/pattern/parallel_loop.hpp"
#include "ityr/pattern/reducer.hpp"
 
namespace ityr {
 
namespace internal {
 
template <typename W, typename AccumulateOp, typename CombineOp, typename Reducer,
          typename ReleaseHandler, typename ForwardIterator, typename... ForwardIterators>
inline typename Reducer::accumulator_type
parallel_reduce_generic(const execution::parallel_policy<W>& policy,
                        AccumulateOp                         accumulate_op,
                        CombineOp                            combine_op,
                        Reducer                              reducer,
                        typename Reducer::accumulator_type&& acc,
                        ReleaseHandler                       rh,
                        ForwardIterator                      first,
                        ForwardIterator                      last,
                        ForwardIterators...                  firsts) {
  using acc_t = typename Reducer::accumulator_type;
 
  ori::poll();
 
  // for immediately executing cross-worker tasks in ADWS
  ito::poll([] { return ori::release_lazy(); },
            [&](ori::release_handler rh_) { ori::acquire(rh); ori::acquire(rh_); });
 
  std::size_t d = std::distance(first, last);
  if (d <= policy.cutoff_count) {
    for_each_aux(
        execution::internal::to_sequenced_policy(policy),
        [&](auto&&... refs) {
          accumulate_op(acc, std::forward<decltype(refs)>(refs)...);
        },
        first, last, firsts...);
    return std::move(acc);
  }
 
  auto mid = std::next(first, d / 2);
 
  ito::task_group_data tgdata;
  ito::task_group_begin(&tgdata);
 
  auto&& [p1, p2] = execution::internal::get_child_policies(policy);
 
  ito::thread<acc_t> th(
      ito::with_callback, [=] { ori::acquire(rh); }, [] { ori::release(); },
      execution::internal::get_workhint(policy),
      [=, p1 = p1, acc = std::move(acc)]() mutable {
        return parallel_reduce_generic(p1, accumulate_op, combine_op, reducer,
                                       std::move(acc), rh, first, mid, firsts...);
      });
 
  if (th.serialized()) {
    acc_t acc_r = parallel_reduce_generic(p2, accumulate_op, combine_op, reducer,
                                          th.join(), rh, mid, last, std::next(firsts, d / 2)...);
 
    ito::task_group_end([] { ori::release(); }, [] { ori::acquire(); });
 
    return acc_r;
 
  } else {
    acc_t new_acc = reducer();
    rh = ori::release_lazy();
 
    acc_t acc_r = parallel_reduce_generic(p2, accumulate_op, combine_op, reducer,
                                          std::move(new_acc), rh, mid, last, std::next(firsts, d / 2)...);
 
    ori::release();
 
    acc_t acc_l = th.join();
 
    ito::task_group_end([] { ori::release(); }, [] { ori::acquire(); });
 
    ori::acquire();
 
    combine_op(acc_l, std::move(acc_r), first, mid, last, firsts...);
    return acc_l;
  }
}
 
template <typename AccumulateOp, typename CombineOp, typename Reducer,
          typename ForwardIterator, typename... ForwardIterators>
inline typename Reducer::accumulator_type
reduce_generic(const execution::sequenced_policy&   policy,
               AccumulateOp                         accumulate_op,
               CombineOp                            combine_op [[maybe_unused]],
               Reducer                              reducer [[maybe_unused]],
               typename Reducer::accumulator_type&& acc,
               ForwardIterator                      first,
               ForwardIterator                      last,
               ForwardIterators...                  firsts) {
  execution::internal::assert_policy(policy);
  for_each_aux(
      execution::internal::to_sequenced_policy(policy),
      [&](auto&&... refs) {
        accumulate_op(acc, std::forward<decltype(refs)>(refs)...);
      },
      first, last, firsts...);
  return std::move(acc);
}
 
template <typename W, typename AccumulateOp, typename CombineOp, typename Reducer,
          typename ForwardIterator, typename... ForwardIterators>
inline typename Reducer::accumulator_type
reduce_generic(const execution::parallel_policy<W>& policy,
               AccumulateOp                         accumulate_op,
               CombineOp                            combine_op,
               Reducer                              reducer,
               typename Reducer::accumulator_type&& acc,
               ForwardIterator                      first,
               ForwardIterator                      last,
               ForwardIterators...                  firsts) {
  execution::internal::assert_policy(policy);
  auto rh = ori::release_lazy();
  return parallel_reduce_generic(policy, accumulate_op, combine_op, reducer, std::move(acc),
                                 rh, first, last, firsts...);
}
 
}
 
template <typename ExecutionPolicy, typename ForwardIterator,
          typename Reducer, typename UnaryTransformOp>
inline typename Reducer::accumulator_type
transform_reduce(const ExecutionPolicy& policy,
                 ForwardIterator        first,
                 ForwardIterator        last,
                 Reducer                reducer,
                 UnaryTransformOp       unary_transform_op) {
  if constexpr (ori::is_global_ptr_v<ForwardIterator>) {
    return transform_reduce(
        policy,
        internal::convert_to_global_iterator(first, checkout_mode::read),
        internal::convert_to_global_iterator(last , checkout_mode::read),
        reducer,
        unary_transform_op);
 
  } else {
    auto accumulate_op = [=](auto&& acc, const auto& r) {
      reducer(std::forward<decltype(acc)>(acc), unary_transform_op(r));
    };
 
    auto combine_op = [=](auto&& acc1, auto&& acc2,
                          ForwardIterator, ForwardIterator, ForwardIterator) {
      reducer(std::forward<decltype(acc1)>(acc1), std::forward<decltype(acc2)>(acc2));
    };
 
    return internal::reduce_generic(policy, accumulate_op, combine_op, reducer,
                                    reducer(), first, last);
  }
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIterator2,
          typename Reducer, typename BinaryTransformOp>
inline typename Reducer::accumulator_type
transform_reduce(const ExecutionPolicy& policy,
                 ForwardIterator1       first1,
                 ForwardIterator1       last1,
                 ForwardIterator2       first2,
                 Reducer                reducer,
                 BinaryTransformOp      binary_transform_op) {
  if constexpr (ori::is_global_ptr_v<ForwardIterator1> ||
                ori::is_global_ptr_v<ForwardIterator2>) {
    return transform_reduce(
        policy,
        internal::convert_to_global_iterator(first1, checkout_mode::read),
        internal::convert_to_global_iterator(last1 , checkout_mode::read),
        internal::convert_to_global_iterator(first2, checkout_mode::read),
        reducer,
        binary_transform_op);
 
  } else {
    auto accumulate_op = [=](auto&& acc, const auto& r1, const auto& r2) {
      reducer(std::forward<decltype(acc)>(acc), binary_transform_op(r1, r2));
    };
 
    auto combine_op = [=](auto&& acc1, auto&& acc2,
                          ForwardIterator1, ForwardIterator1, ForwardIterator1, ForwardIterator2) {
      reducer(std::forward<decltype(acc1)>(acc1), std::forward<decltype(acc2)>(acc2));
    };
 
    return internal::reduce_generic(policy, accumulate_op, combine_op, reducer,
                                    reducer(), first1, last1, first2);
  }
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIterator2>
inline auto transform_reduce(const ExecutionPolicy& policy,
                             ForwardIterator1       first1,
                             ForwardIterator1       last1,
                             ForwardIterator2       first2) {
  using T = decltype((*first1) * (*first2));
  return transform_reduce(policy, first1, last1, first2, reducer::plus<T>{}, std::multiplies<>{});
}
 
template <typename ExecutionPolicy, typename ForwardIterator, typename Reducer>
inline typename Reducer::accumulator_type
reduce(const ExecutionPolicy& policy,
       ForwardIterator        first,
       ForwardIterator        last,
       Reducer                reducer) {
  return transform_reduce(policy, first, last, reducer,
      [](auto&& r) -> decltype(auto) { return std::forward<decltype(r)>(r); });
}
 
template <typename ExecutionPolicy, typename ForwardIterator>
inline typename std::iterator_traits<ForwardIterator>::value_type
reduce(const ExecutionPolicy& policy,
       ForwardIterator        first,
       ForwardIterator        last) {
  using T = typename std::iterator_traits<ForwardIterator>::value_type;
  return reduce(policy, first, last, reducer::plus<T>{});
}
 
ITYR_TEST_CASE("[ityr::pattern::parallel_reduce] reduce and transform_reduce") {
  ito::init();
  ori::init();
 
  ITYR_SUBCASE("default cutoff") {
    long n = 10000;
    long r = ito::root_exec([=] {
      return reduce(
          execution::par,
          count_iterator<long>(0),
          count_iterator<long>(n));
    });
    ITYR_CHECK(r == n * (n - 1) / 2);
  }
 
  ITYR_SUBCASE("custom cutoff") {
    long n = 100000;
    long r = ito::root_exec([=] {
      return reduce(
          execution::parallel_policy(100),
          count_iterator<long>(0),
          count_iterator<long>(n));
    });
    ITYR_CHECK(r == n * (n - 1) / 2);
  }
 
  ITYR_SUBCASE("transform unary") {
    long n = 100000;
    long r = ito::root_exec([=] {
      return transform_reduce(
          execution::parallel_policy(100),
          count_iterator<long>(0),
          count_iterator<long>(n),
          reducer::plus<long>{},
          [](long x) { return x * x; });
    });
    ITYR_CHECK(r == n * (n - 1) * (2 * n - 1) / 6);
  }
 
  ITYR_SUBCASE("transform binary") {
    long n = 100000;
    long r = ito::root_exec([=] {
      return transform_reduce(
          execution::parallel_policy(100),
          count_iterator<long>(0),
          count_iterator<long>(n),
          count_iterator<long>(0),
          reducer::plus<long>{},
          [](long x, long y) { return x * y; });
    });
    ITYR_CHECK(r == n * (n - 1) * (2 * n - 1) / 6);
  }
 
  ITYR_SUBCASE("zero elements") {
    long r = ito::root_exec([=] {
      return reduce(
          execution::parallel_policy(100),
          count_iterator<long>(0),
          count_iterator<long>(0));
    });
    ITYR_CHECK(r == 0);
  }
 
  ori::fini();
  ito::fini();
}
 
ITYR_TEST_CASE("[ityr::pattern::parallel_reduce] parallel reduce with global_ptr") {
  ito::init();
  ori::init();
 
  long n = 100000;
  ori::global_ptr<long> p = ori::malloc_coll<long>(n);
 
  ito::root_exec([=] {
    long count = 0;
    for_each(
        execution::sequenced_policy(100),
        make_global_iterator(p    , checkout_mode::write),
        make_global_iterator(p + n, checkout_mode::write),
        [&](long& v) { v = count++; });
  });
 
  ITYR_SUBCASE("default cutoff") {
    long r = ito::root_exec([=] {
      return reduce(
          execution::par,
          p, p + n);
    });
    ITYR_CHECK(r == n * (n - 1) / 2);
  }
 
  ITYR_SUBCASE("custom cutoff and checkout count") {
    long r = ito::root_exec([=] {
      return reduce(
          execution::parallel_policy(100),
          p, p + n);
    });
    ITYR_CHECK(r == n * (n - 1) / 2);
  }
 
  ITYR_SUBCASE("without auto checkout") {
    long r = ito::root_exec([=] {
      return transform_reduce(
          execution::par,
          make_global_iterator(p    , checkout_mode::no_access),
          make_global_iterator(p + n, checkout_mode::no_access),
          reducer::plus<long>{},
          [](ori::global_ref<long> gref) {
            return gref.get();
          });
    });
    ITYR_CHECK(r == n * (n - 1) / 2);
  }
 
  ITYR_SUBCASE("serial") {
    long r = ito::root_exec([=] {
      return reduce(
          execution::sequenced_policy(100),
          p, p + n);
    });
    ITYR_CHECK(r == n * (n - 1) / 2);
  }
 
  ITYR_SUBCASE("move only") {
    ori::global_ptr<common::move_only_t> p_mo = ori::malloc_coll<common::move_only_t>(n);
 
    ito::root_exec([=] {
      for_each(
          execution::parallel_policy(100),
          count_iterator<long>(0),
          count_iterator<long>(n),
          make_global_iterator(p_mo, checkout_mode::write),
          [&](long i, common::move_only_t& v) { v = common::move_only_t(i); });
 
      common::move_only_t r = reduce(
          execution::par,
          p_mo, p_mo + n);
 
      ITYR_CHECK(r.value() == n * (n - 1) / 2);
    });
 
    ori::free_coll(p_mo);
  }
 
  ori::free_coll(p);
 
  ori::fini();
  ito::fini();
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIteratorD,
          typename Reducer, typename UnaryTransformOp>
inline ForwardIteratorD
transform_inclusive_scan(const ExecutionPolicy&               policy,
                         ForwardIterator1                     first1,
                         ForwardIterator1                     last1,
                         ForwardIteratorD                     first_d,
                         Reducer                              reducer,
                         UnaryTransformOp                     unary_transform_op,
                         typename Reducer::accumulator_type&& init) {
  if constexpr (ori::is_global_ptr_v<ForwardIterator1> ||
                ori::is_global_ptr_v<ForwardIteratorD>) {
    using value_type_d = typename std::iterator_traits<ForwardIteratorD>::value_type;
    return transform_inclusive_scan(
        policy,
        internal::convert_to_global_iterator(first1 , checkout_mode::read),
        internal::convert_to_global_iterator(last1  , checkout_mode::read),
        internal::convert_to_global_iterator(first_d, internal::dest_checkout_mode_t<value_type_d>{}),
        reducer,
        unary_transform_op,
        std::move(init));
 
  } else {
    auto accumulate_op = [=](auto&& acc, const auto& r1, auto&& d) {
      reducer(acc, unary_transform_op(r1));
      d = acc;
    };
 
    // TODO: more efficient scan implementation
    auto combine_op = [=](auto&&           acc1,
                          auto&&           acc2,
                          ForwardIterator1 first_,
                          ForwardIterator1 mid_,
                          ForwardIterator1 last_,
                          ForwardIteratorD first_d_) {
      // Add the left accumulator `acc1` to the right half of the region
      auto dm = std::distance(first_, mid_);
      auto dl = std::distance(first_, last_);
      if constexpr (!is_global_iterator_v<ForwardIteratorD>) {
        for_each(policy, std::next(first_d_, dm), std::next(first_d_, dl),
                 [=](auto&& acc_r) { reducer(acc1, acc_r); });
      } else if constexpr (std::is_same_v<typename ForwardIteratorD::mode, checkout_mode::no_access_t>) {
        for_each(policy, std::next(first_d_, dm), std::next(first_d_, dl),
                 [=](auto&& acc_r) { reducer(acc1, acc_r); });
      } else {
        // &*: convert global_iterator -> global_ref -> global_ptr
        auto fd = make_global_iterator(&*first_d_, checkout_mode::read_write);
        for_each(policy, std::next(fd, dm), std::next(fd, dl),
                 [=](auto&& acc_r) { reducer(acc1, acc_r); });
      }
      reducer(std::forward<decltype(acc1)>(acc1), std::forward<decltype(acc2)>(acc2));
    };
 
    internal::reduce_generic(policy, accumulate_op, combine_op, reducer,
                             std::move(init), first1, last1, first_d);
 
    return std::next(first_d, std::distance(first1, last1));
  }
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIteratorD,
          typename Reducer, typename UnaryTransformOp>
inline ForwardIteratorD transform_inclusive_scan(const ExecutionPolicy& policy,
                                                 ForwardIterator1       first1,
                                                 ForwardIterator1       last1,
                                                 ForwardIteratorD       first_d,
                                                 Reducer                reducer,
                                                 UnaryTransformOp       unary_transform_op) {
  return transform_inclusive_scan(policy, first1, last1, first_d, reducer,
                                  unary_transform_op, reducer());
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIteratorD,
          typename Reducer>
inline ForwardIteratorD
inclusive_scan(const ExecutionPolicy&               policy,
               ForwardIterator1                     first1,
               ForwardIterator1                     last1,
               ForwardIteratorD                     first_d,
               Reducer                              reducer,
               typename Reducer::accumulator_type&& init) {
  return transform_inclusive_scan(policy, first1, last1, first_d, reducer,
      [](auto&& r) -> decltype(auto) { return std::forward<decltype(r)>(r); }, std::move(init));
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIteratorD,
          typename Reducer>
inline ForwardIteratorD inclusive_scan(const ExecutionPolicy& policy,
                                       ForwardIterator1       first1,
                                       ForwardIterator1       last1,
                                       ForwardIteratorD       first_d,
                                       Reducer                reducer) {
  return inclusive_scan(policy, first1, last1, first_d, reducer, reducer());
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIteratorD>
inline ForwardIteratorD inclusive_scan(const ExecutionPolicy& policy,
                                       ForwardIterator1       first1,
                                       ForwardIterator1       last1,
                                       ForwardIteratorD       first_d) {
  using T = typename std::iterator_traits<ForwardIterator1>::value_type;
  return inclusive_scan(policy, first1, last1, first_d, reducer::plus<T>{});
}
 
ITYR_TEST_CASE("[ityr::pattern::parallel_reduce] inclusive scan") {
  ito::init();
  ori::init();
 
  long n = 100000;
  ori::global_ptr<long> p1 = ori::malloc_coll<long>(n);
  ori::global_ptr<long> p2 = ori::malloc_coll<long>(n);
 
  ito::root_exec([=] {
    fill(execution::parallel_policy(100),
         p1, p1 + n, 1);
 
    inclusive_scan(
        execution::parallel_policy(100),
        p1, p1 + n, p2);
 
    ITYR_CHECK(p2[0].get() == 1);
    ITYR_CHECK(p2[n - 1].get() == n);
 
    auto sum = reduce(
        execution::parallel_policy(100),
        p2, p2 + n);
 
    ITYR_CHECK(sum == n * (n + 1) / 2);
 
    inclusive_scan(
        execution::parallel_policy(100),
        p1, p1 + n, p2, reducer::multiplies<long>{}, 10);
 
    ITYR_CHECK(p2[0].get() == 10);
    ITYR_CHECK(p2[n - 1].get() == 10);
 
    transform_inclusive_scan(
        execution::parallel_policy(100),
        p1, p1 + n, p2, reducer::plus<long>{}, [](long x) { return x + 1; }, 10);
 
    ITYR_CHECK(p2[0].get() == 12);
    ITYR_CHECK(p2[n - 1].get() == 10 + n * 2);
  });
 
  ori::free_coll(p1);
  ori::free_coll(p2);
 
  ori::fini();
  ito::fini();
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIterator2,
          typename BinaryPredicate>
inline bool equal(const ExecutionPolicy& policy,
                  ForwardIterator1       first1,
                  ForwardIterator1       last1,
                  ForwardIterator2       first2,
                  BinaryPredicate        pred) {
  return transform_reduce(policy, first1, last1, first2, reducer::logical_and{}, pred);
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIterator2,
          typename BinaryPredicate>
inline bool equal(const ExecutionPolicy& policy,
                  ForwardIterator1       first1,
                  ForwardIterator1       last1,
                  ForwardIterator2       first2,
                  ForwardIterator2       last2,
                  BinaryPredicate        pred) {
  return std::distance(first1, last1) == std::distance(first2, last2) &&
         equal(policy, first1, last1, first2, pred);
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIterator2>
inline bool equal(const ExecutionPolicy& policy,
                  ForwardIterator1       first1,
                  ForwardIterator1       last1,
                  ForwardIterator2       first2) {
  return equal(policy, first1, last1, first2, std::equal_to<>{});
}
 
template <typename ExecutionPolicy, typename ForwardIterator1, typename ForwardIterator2>
inline bool equal(const ExecutionPolicy& policy,
                  ForwardIterator1       first1,
                  ForwardIterator1       last1,
                  ForwardIterator2       first2,
                  ForwardIterator2       last2) {
  return equal(policy, first1, last1, first2, last2, std::equal_to<>{});
}
 
ITYR_TEST_CASE("[ityr::pattern::parallel_reduce] equal") {
  ito::init();
  ori::init();
 
  long n = 100000;
  ori::global_ptr<long> p1 = ori::malloc_coll<long>(n);
  ori::global_ptr<long> p2 = ori::malloc_coll<long>(n);
 
  ito::root_exec([=] {
    for_each(
        execution::parallel_policy(100),
        make_global_iterator(p1    , checkout_mode::write),
        make_global_iterator(p1 + n, checkout_mode::write),
        count_iterator<long>(0),
        [=](long& v, long i) { v = i * 2; });
 
    copy(execution::parallel_policy(100), p1, p1 + n, p2);
 
    ITYR_CHECK(equal(execution::parallel_policy(100),
                     p1, p1 + n, p2) == true);
 
    ITYR_CHECK(equal(execution::parallel_policy(100),
                     p1, p1 + n, p2, p2 + n) == true);
 
    ITYR_CHECK(equal(execution::parallel_policy(100),
                     p1, p1 + n, p2, p2 + n - 1) == false);
 
    p2[n / 2].put(0);
 
    ITYR_CHECK(equal(execution::parallel_policy(100), p1, p1 + n, p2) == false);
  });
 
  ori::free_coll(p1);
  ori::free_coll(p2);
 
  ori::fini();
  ito::fini();
}
 
template <typename ExecutionPolicy, typename ForwardIterator, typename Compare>
inline bool is_sorted(const ExecutionPolicy& policy,
                      ForwardIterator        first,
                      ForwardIterator        last,
                      Compare                comp) {
  // Check if comp(a(i+1), a(i)) returns false for all i
  return std::distance(first, last) <= 1 ||
         transform_reduce(policy, std::next(first), last, first,
                          reducer::logical_or{}, comp) == false;
}
 
template <typename ExecutionPolicy, typename ForwardIterator>
inline bool is_sorted(const ExecutionPolicy& policy,
                      ForwardIterator        first,
                      ForwardIterator        last) {
  return is_sorted(policy, first, last, std::less<>{});
}
 
ITYR_TEST_CASE("[ityr::pattern::parallel_reduce] is_sorted") {
  ito::init();
  ori::init();
 
  long n = 100000;
  ori::global_ptr<long> p = ori::malloc_coll<long>(n);
 
  ito::root_exec([=] {
    for_each(
        execution::parallel_policy(100),
        make_global_iterator(p    , checkout_mode::write),
        make_global_iterator(p + n, checkout_mode::write),
        count_iterator<long>(0),
        [=](long& v, long i) { v = i / 3; });
 
    ITYR_CHECK(is_sorted(execution::parallel_policy(100),
                         p, p + n) == true);
 
    ITYR_CHECK(is_sorted(execution::parallel_policy(100),
                         p, p + n, std::greater<>{}) == false);
 
    ITYR_CHECK(is_sorted(execution::parallel_policy(100),
                         make_reverse_iterator(p + n, checkout_mode::read),
                         make_reverse_iterator(p    , checkout_mode::read),
                         std::greater<>{}) == true);
 
    p[n / 4].put(0);
 
    ITYR_CHECK(is_sorted(execution::parallel_policy(100),
                         p, p + n) == false);
  });
 
  ori::free_coll(p);
 
  ori::fini();
  ito::fini();
}
 
}