iframe-proxy | Sunbelt Computer Software

760 lines (575 loc) · 18 KB
// reference:
// - gomp: https://github.com/gcc-mirror/gcc/blob/master/libgomp/iter.c
// - komp: https://github.com/llvm-mirror/openmp/blob/master/runtime/src/kmp_dispatch.cpp
#pragma once
#include "../core/executor.hpp"
namespace tf {
// ----------------------------------------------------------------------------
// default parallel for
// ----------------------------------------------------------------------------
// Function: for_each
template <typename B, typename E, typename C>
Task FlowBuilder::for_each(B&& beg, E&& end, C&& c) {
  return for_each_guided(
    std::forward<B>(beg), std::forward<E>(end), std::forward<C>(c), 1
// Function: for_each_index
template <typename B, typename E, typename S, typename C>
Task FlowBuilder::for_each_index(B&& beg, E&& end, S&& inc, C&& c){
  return for_each_index_guided(
    std::forward<B>(beg), 
    std::forward<E>(end), 
    std::forward<S>(inc), 
    std::forward<C>(c),
// ----------------------------------------------------------------------------
// parallel for using the guided partition algorithm
// - Polychronopoulos, C. D. and Kuck, D. J. 
//   "Guided Self-Scheduling: A Practical Scheduling Scheme 
//    for Parallel Supercomputers," 
//   IEEE Transactions on Computers, C-36(12):1425–1439 (1987).
// ----------------------------------------------------------------------------
// Function: for_each_guided
template <typename B, typename E, typename C, typename H>
Task FlowBuilder::for_each_guided(B&& beg, E&& end, C&& c, H&& chunk_size){
  using I = stateful_iterator_t<B, E>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg),
   e=std::forward<E>(end), 
   c=std::forward<C>(c),
   h=std::forward<H>(chunk_size)] (Subflow& sf) mutable {
    // fetch the stateful values
    I beg = b;
    I end = e;
    if(beg == end) {
      return;
    size_t chunk_size = (h == 0) ? 1 : h;
    size_t W = sf._executor.num_workers();
    size_t N = std::distance(beg, end);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= chunk_size) {
      std::for_each(beg, end, c);
      return;
    if(N < W) {
      W = N;
    std::atomic<size_t> next(0);
    for(size_t w=0; w<W; w++) {
      sf.emplace([&next, beg, N, chunk_size, W, &c] () mutable {
        size_t z = 0;
        size_t p1 = 2 * W * (chunk_size + 1);
        double p2 = 0.5 / W;
        size_t s0 = next.load(std::memory_order_relaxed);
        while(s0 < N) {
          size_t r = N - s0;
          // fine-grained
          if(r < p1) {
            while(1) {
              s0 = next.fetch_add(chunk_size, std::memory_order_relaxed);
              if(s0 >= N) {
                return;
              size_t e0 = (chunk_size <= (N - s0)) ? s0 + chunk_size : N;
              std::advance(beg, s0-z);
              for(size_t x=s0; x<e0; x++) {
                c(*beg++);
            break;
          // coarse-grained
          else {
            size_t q = static_cast<size_t>(p2 * r);
            if(q < chunk_size) {
              q = chunk_size;
            size_t e0 = (q <= r) ? s0 + q : N;
            if(next.compare_exchange_strong(s0, e0, std::memory_order_relaxed,
                                                    std::memory_order_relaxed)) {
              std::advance(beg, s0-z);
              for(size_t x = s0; x< e0; x++) {
                c(*beg++);
              s0 = next.load(std::memory_order_relaxed);
      }).name("pfg_"s + std::to_string(w));
    sf.join();
  return task;
// Function: for_each_index_guided
template <typename B, typename E, typename S, typename C, typename H>
Task FlowBuilder::for_each_index_guided(
  B&& beg, E&& end, S&& inc, C&& c, H&& chunk_size
  using I = stateful_index_t<B, E, S>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg), 
   e=std::forward<E>(end), 
   a=std::forward<S>(inc), 
   c=std::forward<C>(c),
   h=std::forward<H>(chunk_size)] (Subflow& sf) mutable {
    // fetch the iterator values
    I beg = b;
    I end = e;
    I inc = a;
    if(is_range_invalid(beg, end, inc)) {
      TF_THROW("invalid range [", beg, ", ", end, ") with step size ", inc);
    size_t chunk_size = (h == 0) ? 1 : h;
    size_t W = sf._executor.num_workers();
    size_t N = distance(beg, end, inc);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= chunk_size) {
      for(size_t x=0; x<N; x++, beg+=inc) {
        c(beg);
      return;
    if(N < W) {
      W = N;
    std::atomic<size_t> next(0);
    for(size_t w=0; w<W; w++) {
      sf.emplace([&next, beg, inc, N, chunk_size, W, &c] () mutable {
        size_t p1 = 2 * W * (chunk_size + 1);
        double p2 = 0.5 / W;
        size_t s0 = next.load(std::memory_order_relaxed);
        while(s0 < N) {
          size_t r = N - s0;
          // find-grained
          if(r < p1) {
            while(1) { 
              s0 = next.fetch_add(chunk_size, std::memory_order_relaxed);
              if(s0 >= N) {
                return;
              size_t e0 = (chunk_size <= (N - s0)) ? s0 + chunk_size : N;
              auto s = static_cast<I>(s0) * inc + beg;
              for(size_t x=s0; x<e0; x++, s+=inc) {
                c(s);
            break;
          // coarse-grained
          else {
            size_t q = static_cast<size_t>(p2 * r);
            if(q < chunk_size) {
              q = chunk_size;
            size_t e0 = (q <= r) ? s0 + q : N;
            if(next.compare_exchange_strong(s0, e0, std::memory_order_relaxed,
                                                    std::memory_order_relaxed)) {
              auto s = static_cast<I>(s0) * inc + beg;
              for(size_t x=s0; x<e0; x++, s+= inc) {
                c(s);
              s0 = next.load(std::memory_order_relaxed); 
      }).name("pfg_"s + std::to_string(w));
    sf.join();
  return task;
// ----------------------------------------------------------------------------
// Factoring algorithm
// - Hummel, S. F., Schonberg, E., and Flynn, L. E., 
//   "Factoring: a practical and robust method for scheduling parallel loops," 
//   IEEE/ACM SC, 1991
// ----------------------------------------------------------------------------
/*// Function: for_each_factoring
template <typename B, typename E, typename C>
Task FlowBuilder::for_each_factoring(B&& beg, E&& end, C&& c){
  using I = stateful_iterator_t<B, E>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg),
   e=std::forward<E>(end), 
   c=std::forward<C>(c)] (Subflow& sf) mutable {
    // fetch the iterator values
    I beg = b;
    I end = e;
    if(beg == end) {
      return;
    size_t W = sf._executor.num_workers();
    size_t N = std::distance(beg, end);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= 1) {
      std::for_each(beg, end, c);
      return;
    if(N < W) {
      W = N;
    std::atomic<size_t> batch(0);
    std::atomic<size_t> next(0);
    for(size_t w=0; w<W; w++) {
      sf.emplace([&batch, &next, beg, N, W, &c] () mutable {
        size_t z = 0;
        while(1) {
          size_t c0 = batch.fetch_add(1, std::memory_order_relaxed) + 1;
          size_t b0 = (c0 + W - 1) / W;
          size_t ck = static_cast<size_t>((N >> b0) / (double)W);
          if(ck == 0) {
            ck = 1;
          size_t s0 = next.fetch_add(ck, std::memory_order_relaxed);
          if(s0 >= N) {
            return;
          size_t e0 = (ck <= (N - s0)) ? s0 + ck : N;
          std::advance(beg, s0-z);
          for(size_t x=s0; x<e0; x++) {
            c(*beg++);
          z = e0;
      }).name("pfg_"s + std::to_string(w));
    sf.join();
  return task;
// Function: for_each_factoring
template <typename B, typename E, typename S, typename C>
Task FlowBuilder::for_each_factoring(
  B&& beg, E&& end, S&& inc, C&& c
  using I = stateful_index_t<B, E, S>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg), 
   e=std::forward<E>(end), 
   i=std::forward<S>(inc), 
   c=std::forward<C>(c)] (Subflow& sf) mutable {
    // fetch the iterator values
    I beg = b;
    I end = e;
    I inc = i;
    if(is_range_invalid(beg, end, inc)) {
      TF_THROW("invalid range [", beg, ", ", end, ") with step size ", inc);
    size_t W = sf._executor.num_workers();
    size_t N = distance(beg, end, inc);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= 1) {
      for(size_t x=0; x<N; x++, beg+=inc) {
        c(beg);
      return;
    if(N < W) {
      W = N;
    std::atomic<size_t> batch(0);
    std::atomic<size_t> next(0);
    for(size_t w=0; w<W; w++) {
      sf.emplace([&batch, &next, beg, inc, N, W, &c] () mutable {
        while(1) {
          size_t c0 = batch.fetch_add(1, std::memory_order_relaxed) + 1;
          size_t b0 = (c0 + W - 1) / W;
          size_t ck = static_cast<size_t>((N >> b0) / (double)(W));
          if(ck == 0) {
            ck = 1;
          size_t s0 = next.fetch_add(ck, std::memory_order_relaxed);
          if(s0 >= N) {
            return;
          size_t e0 = (ck <= (N - s0)) ? s0 + ck : N;
          auto s = static_cast<I>(s0) * inc + beg;
          for(size_t x=s0; x<e0; x++, s+=inc) {
            c(s);
      }).name("pff_"s + std::to_string(w));
    sf.join();
  return task;
// ----------------------------------------------------------------------------
// Function: for_each_dynamic
// ----------------------------------------------------------------------------
// Function: for_each_dynamic
template <typename B, typename E, typename C, typename H>
Task FlowBuilder::for_each_dynamic(
  B&& beg, E&& end, C&& c, H&& chunk_size
  using I = stateful_iterator_t<B, E>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg), 
   e=std::forward<E>(end), 
   c=std::forward<C>(c),
   h=std::forward<H>(chunk_size)] (Subflow& sf) mutable {
    I beg = b;
    I end = e;
    if(beg == end) {
      return;
    size_t chunk_size = (h == 0) ? 1 : h;
    size_t W = sf._executor.num_workers();
    size_t N = std::distance(beg, end);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= chunk_size) {
      std::for_each(beg, end, c);
      return;
    if(N < W) {
      W = N;
    std::atomic<size_t> next(0);
    for(size_t w=0; w<W; w++) {
      sf.emplace([&next, beg, N, chunk_size, &c] () mutable {
        size_t z = 0;
        while(1) {
          size_t s0 = next.fetch_add(chunk_size, std::memory_order_relaxed);
          if(s0 >= N) {
            break;
          size_t e0 = (chunk_size <= (N - s0)) ? s0 + chunk_size : N;
          std::advance(beg, s0-z);
          for(size_t x=s0; x<e0; x++) {
            c(*beg++);
          z = e0;
      }).name("pfd_"s + std::to_string(w));
    sf.join();
  return task;
template <typename B, typename E, typename S, typename C, typename H>
Task FlowBuilder::for_each_index_dynamic(
  B&& beg, E&& end, S&& inc, C&& c, H&& chunk_size
  using I = stateful_index_t<B, E, S>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg), 
   e=std::forward<E>(end),
   a=std::forward<S>(inc),
   c=std::forward<C>(c),
   h=std::forward<H>(chunk_size)] (Subflow& sf) mutable {
    I beg = b;
    I end = e;
    I inc = a;
    if(is_range_invalid(beg, end, inc)) {
      TF_THROW("invalid range [", beg, ", ", end, ") with step size ", inc);
    size_t chunk_size = (h == 0) ? 1 : h;
    size_t W = sf._executor.num_workers();
    size_t N = distance(beg, end, inc);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= chunk_size) {
      for(size_t x=0; x<N; x++, beg+=inc) {
        c(beg);
      return;
    if(N < W) {
      W = N;
    std::atomic<size_t> next(0);
    for(size_t w=0; w<W; w++) {
      sf.emplace([&next, beg, inc, N, chunk_size, &c] () mutable {
        while(1) {
          size_t s0 = next.fetch_add(chunk_size, std::memory_order_relaxed);
          if(s0 >= N) {
            break;
          size_t e0 = (chunk_size <= (N - s0)) ? s0 + chunk_size : N;
          I s = static_cast<I>(s0) * inc + beg;
          for(size_t x=s0; x<e0; x++, s+=inc) {
            c(s);
      }).name("pfd_"s + std::to_string(w));
    sf.join();
  return task;
// ----------------------------------------------------------------------------
// parallel for using the static partition algorithm
// ----------------------------------------------------------------------------
// Function: for_each_static
// static scheduling with chunk size
template <typename B, typename E, typename C, typename H>
Task FlowBuilder::for_each_static(
  B&& beg, E&& end, C&& c, H&& chunk_size
  using I = stateful_iterator_t<B, E>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg), 
   e=std::forward<E>(end), 
   c=std::forward<C>(c),
   h=std::forward<H>(chunk_size)] (Subflow& sf) mutable {
    // fetch the iterator
    I beg = b;
    I end = e;
    if(beg == end) {
      return;
    size_t chunk_size = h;
    const size_t W = sf._executor.num_workers();
    const size_t N = std::distance(beg, end);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= chunk_size) {
      std::for_each(beg, end, c);
      return;
    std::atomic<size_t> next(0);
    // even partition
    if(chunk_size == 0){
      // zero-based start and end points
      const size_t q0 = N / W;
      const size_t t0 = N % W;
      for(size_t i=0; i<W; ++i) {
        size_t items = i < t0 ? q0 + 1 : q0;
        if(items == 0) {
          break;
        sf.emplace([&next, beg, items, &c] () mutable {
          size_t s0 = next.fetch_add(items, std::memory_order_relaxed);
          std::advance(beg, s0);
          for(size_t i=0; i<items; i++) {
            c(*beg++);
        }).name("pfs_"s + std::to_string(i));
    // chunk-by-chunk partition
      for(size_t i=0; i<W; ++i) {
        // initial
        if(i*chunk_size >= N) {
          break;
        sf.emplace([&next, beg, end, chunk_size, N, W, &c] () mutable {
          size_t trip = W*chunk_size;
          size_t s0 = next.fetch_add(chunk_size, std::memory_order_relaxed);
          std::advance(beg, s0);
          while(1) {
            size_t items;
            I e = beg;
            for(items=0; items<chunk_size && e != end; items++, e++) {
              c(*e); 
            s0 += trip;
            if(items != chunk_size || s0 >= N) {
              break;
            std::advance(beg, trip);
        }).name("pfs_"s + std::to_string(i));
    sf.join();
  return task;
// Function: for_each_index_static
// static scheduling with chunk size
template <typename B, typename E, typename S, typename C, typename H>
Task FlowBuilder::for_each_index_static(
  B&& beg, E&& end, S&& inc, C&& c, H&& chunk_size
  using I = stateful_index_t<B, E, S>;
  using namespace std::string_literals;
  Task task = emplace(
  [b=std::forward<B>(beg), 
   e=std::forward<E>(end), 
   a=std::forward<S>(inc), 
   c=std::forward<C>(c),
   h=std::forward<H>(chunk_size)] (Subflow& sf) mutable {
    // fetch the indices
    I beg = b;
    I end = e;
    I inc = a;
    if(is_range_invalid(beg, end, inc)) {
      TF_THROW("invalid range [", beg, ", ", end, ") with step size ", inc);
    size_t chunk_size = h;
    const size_t W = sf._executor.num_workers();
    const size_t N = distance(beg, end, inc);
    // only myself - no need to spawn another graph
    if(W <= 1 || N <= chunk_size) {
      for(size_t x=0; x<N; x++, beg+=inc) {
        c(beg);
      return;
    std::atomic<size_t> next(0);
    if(chunk_size == 0) {
      // zero-based start and end points
      const size_t q0 = N / W;
      const size_t t0 = N % W;
      for(size_t i=0; i<W; ++i) {
        size_t items = i < t0 ? q0 + 1 : q0;
        if(items == 0) {
          break;
        sf.emplace([&next, beg, &inc, items, &c] () mutable {
          size_t s0 = next.fetch_add(items, std::memory_order_relaxed);
          I s = static_cast<I>(s0) * inc + beg;
          for(size_t x=0; x<items; x++, s+=inc) {
            c(s);
        }).name("pfs_"s + std::to_string(i));
      for(size_t i=0; i<W; ++i) {
        // initial
        if(i*chunk_size >= N) {
          break;
        sf.emplace([&next, beg, inc, chunk_size, N, W, &c] () mutable {
          size_t trip = W * chunk_size;
          size_t s0 = next.fetch_add(chunk_size, std::memory_order_relaxed);
          while(1) {
            size_t e0 = s0 + chunk_size;
            if(e0 > N) {
            I s = static_cast<I>(s0) * inc + beg;
            for(size_t x=s0; x<e0; x++, s+=inc) {
              c(s);
            if(e0 == N) {
              break;
            s0 += trip;
            if(s0 >= N) {
              break;
        }).name("pfs_"s + std::to_string(i));
    sf.join();
  return task;
}  // end of namespace tf -----------------------------------------------------
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

for_each.hpp

Latest commit

History

for_each.hpp

File metadata and controls
