native/map.h

Go to the documentation of this file.

#ifndef GRPPI_NATIVE_MAP_H
#define GRPPI_NATIVE_MAP_H

#include "parallel_execution_native.h"
#include "../common/iterator.h"

namespace grppi {

template <typename InputIt, typename OutputIt, typename Transformer>
void map(parallel_execution_native & ex, 
         InputIt first, InputIt last, OutputIt first_out, 
         Transformer && transf_op)
{
  std::vector<std::thread> tasks;
  int numElements = last - first; 
  int elemperthr = numElements / ex.concurrency_degree(); 

  for(int i=1;i<ex.concurrency_degree();i++){
    auto begin = first + (elemperthr * i); 
    auto end = first + (elemperthr * (i+1)); 

    if(i == ex.concurrency_degree()-1 ) end= last;

    auto out = first_out + (elemperthr * i);
    tasks.emplace_back([&](InputIt begin, InputIt end, OutputIt out) {
      auto manager = ex.thread_manager();
          
      while (begin!=end) {
        *out = transf_op(*begin);
        begin++;
        out++;
      }
    }, begin, end, out);
  }
  //Map main threads
  auto end = first+elemperthr;
  while(first!=end) {
    *first_out = transf_op(*first);
    first++;
    first_out++;
  }

  //Join threads
  for(int i=0;i<ex.concurrency_degree()-1;i++){
    tasks[i].join();
  }
}

template <typename InputIt, typename OutputIt, typename Transformer,
          typename ... OtherInputIts> 
void map(parallel_execution_native& ex, 
         InputIt first, InputIt last, OutputIt first_out, 
         Transformer && transf_op, 
         OtherInputIts ... more_inputs)
{
  std::vector<std::thread> tasks;

  //Calculate number of elements per thread
  int numElements = last - first;
  int elemperthr = numElements / ex.concurrency_degree();

  //Create tasks
  for(int i=1;i<ex.concurrency_degree();i++){
    //Calculate local input and output iterator 
    auto begin = first + (elemperthr * i);
    auto end = first + (elemperthr * (i+1));
    if( i == ex.concurrency_degree()-1) end = last;
    auto out = first_out + (elemperthr * i);
    //Begin task
    tasks.emplace_back([&](InputIt begin, InputIt end, OutputIt out, 
      int tid, int nelem, OtherInputIts ... more_inputs) {
        auto manager = ex.thread_manager();
      advance_iterators(nelem*tid, more_inputs ...);
      while (begin!=end) {
        *out = transf_op(*begin, *more_inputs ...);
        advance_iterators(more_inputs ...);
        begin++;
        out++;
      }
    }, begin, end, out, i, elemperthr, more_inputs...);
    //End task
  }

  //Map main thread
  auto end = first + elemperthr;
  while(first!=end) {
    *first_out = transf_op(*first, *more_inputs ...);
    advance_iterators(more_inputs ...);
    first++;
    first_out++;
  }

  //Join threads
  for(int i=0;i<ex.concurrency_degree()-1;i++) {
    tasks[i].join();
  }
}

}

#endif