omp/mapreduce.h

Go to the documentation of this file.

#ifndef GRPPI_OMP_MAPREDUCE_H
#define GRPPI_OMP_MAPREDUCE_H

#ifdef GRPPI_OMP

#include "parallel_execution_omp.h"

namespace grppi {

template <typename InputIt, typename Transformer, typename Result, 
          typename Combiner>
Result map_reduce(parallel_execution_omp & ex, 
                    InputIt first, InputIt last, Result identity, 
                    Transformer &&  transform_op,  
                    Combiner && combine_op)
{
  using namespace std;
  Result result{identity};

  std::vector<Result> partial_results(ex.concurrency_degree());
  #pragma omp parallel
  {
    #pragma omp single nowait
    {
      int num_elements = distance(first,last);
      int elements_per_thread = num_elements/ex.concurrency_degree();
      sequential_execution seq{};

      for (int i=1;i<ex.concurrency_degree();i++) {    
        #pragma omp task firstprivate(i)
        {
          auto begin = next(first, elements_per_thread * i);
          auto end = (i==ex.concurrency_degree()-1) ? last :
              next(first, elements_per_thread * (i+1));
          partial_results[i] = map_reduce(seq, 
              begin, end, partial_results[i], 
              std::forward<Transformer>(transform_op), 
              std::forward<Combiner>(combine_op));
        }
      }

      partial_results[0] = map_reduce(seq, 
          first, first+elements_per_thread, partial_results[0], 
          std::forward<Transformer>(transform_op), 
          std::forward<Combiner>(combine_op));
      #pragma omp taskwait
    }
  }

  for (auto && p : partial_results){
    result = combine_op(result, p);
  } 
  return result;
}

}
#endif

#endif