native/mapreduce.h

Go to the documentation of this file.

#ifndef GRPPI_NATIVE_MAPREDUCE_H
#define GRPPI_NATIVE_MAPREDUCE_H

#include "parallel_execution_native.h"

namespace grppi {

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

  vector<Result> partial_results(ex.concurrency_degree());

  const int num_elements = last - first;
  const int elements_per_thread = num_elements/ex.concurrency_degree();
  sequential_execution seq{};

  vector<thread> tasks;
  for(int i=1;i<ex.concurrency_degree();i++){    
    const auto begin = first + (elements_per_thread * i);
    const auto end = (i==ex.concurrency_degree()-1) ? 
        last : 
        (first + elements_per_thread * (i+1));

    tasks.emplace_back([&,begin,end,i](){
        auto manager = ex.thread_manager();
        partial_results[i] = map_reduce(seq, begin, end, partial_results[i], 
            forward<Transformer>(transform_op), forward<Combiner>(combine_op));
    });
  }

    partial_results[0] = map_reduce(seq, 
        first,( first+elements_per_thread ), partial_results[0], 
        forward<Transformer>(transform_op), 
        forward<Combiner>(combine_op));

    for (auto && t : tasks) { t.join(); }

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

    return result;
}

}

#endif