native/farm.h

Go to the documentation of this file.

#ifndef GRPPI_NATIVE_FARM_H
#define GRPPI_NATIVE_FARM_H

#include "parallel_execution_native.h"

#include <thread>
#include <utility>
#include <memory>
#include <experimental/optional>

namespace grppi {

template <typename Generator, typename Consumer>
void farm(parallel_execution_native & ex, Generator generate_op, 
          Consumer consume_op) 
{
  using namespace std;
  using result_type = typename result_of<Generator()>::type;
  auto queue = ex.make_queue<result_type>();

  vector<thread> tasks;
  for (int i=0; i<ex.concurrency_degree(); ++i) {
    tasks.emplace_back([&](){
      auto manager = ex.thread_manager();

      auto item{queue.pop()};
      while(item) {
        consume_op(*item);
        item = queue.pop();
      }
      queue.push(item);
    });
  }

  for (;;) {
    auto item{generate_op()};
    queue.push(item);
    if (!item) break;
  }

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

template <typename Generator, typename Transformer, typename Consumer>
void farm(parallel_execution_native & ex, Generator generate_op, 
          Transformer transform_op , Consumer consume_op) 
{
  using namespace std;
  using namespace experimental;
  using generated_type = typename result_of<Generator()>::type;
  using generated_value_type = typename generated_type::value_type;
  using transformed_value_type = 
      typename result_of<Transformer(generated_value_type)>::type;
  using transformed_type = optional<transformed_value_type>;

  auto generated_queue = ex.make_queue<generated_type>();
  auto transformed_queue = ex.make_queue<transformed_type>();

  atomic<int> done_threads(0);
  vector<thread> tasks;

  for (int i=0; i<ex.concurrency_degree(); ++i) {
    tasks.emplace_back([&](){
      auto manager = ex.thread_manager();

      auto item{generated_queue.pop()};
      while (item) {
        transformed_queue.push(transformed_type{transform_op(*item)});
        item = generated_queue.pop();
      }
      generated_queue.push(item);
      done_threads++;
      if (done_threads==ex.concurrency_degree()) {
        transformed_queue.push(transformed_type{});
      }
    });
  }

  tasks.emplace_back([&](){
    auto manager = ex.thread_manager();

    auto item{transformed_queue.pop()};
    while (item) {
      consume_op( item.value() );
      item = transformed_queue.pop( );
    }
  });

  for (;;) {
    auto  item{generate_op()};
    generated_queue.push(item);
    if(!item) break;
  }

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

}

#endif