tbb/farm.h

Go to the documentation of this file.

#ifndef GRPPI_TBB_FARM_H
#define GRPPI_TBB_FARM_H

#ifdef GRPPI_TBB

#include "parallel_execution_tbb.h"

#include <experimental/optional>

#include <tbb/tbb.h>

namespace grppi {

template <typename Generator, typename Consumer>
void farm(parallel_execution_tbb & ex, Generator generate_op, 
          Consumer consume_op) 
{
  using namespace std;

  using generated_type = typename result_of<Generator()>::type;
  auto queue = ex.make_queue<generated_type>();

  tbb::task_group g;
  for (int i=0; i<ex.concurrency_degree(); ++i) {
    g.run([&](){
      auto item{queue.pop()};
      while (item) {
        consume_op(*item);
        item = queue.pop();
      }
    });
  }

  //Generate elements
  for (;;) {
    auto item{generate_op()};
    queue.push(item);
      if (!item) {
        for (int i=1; i<ex.concurrency_degree(); i++) {
          queue.push(item);
        }
        break;
      }
    }

    //Join threads
    g.wait();
}

template <typename Generator, typename Transformer, typename Consumer>
void farm(parallel_execution_tbb & 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};
  tbb::task_group generators;
  for (int i=0; i<ex.concurrency_degree(); ++i) {
    generators.run([&](){
      auto item{generated_queue.pop()};
      while (item) {
        auto result = transform_op(*item);
        transformed_queue.push(transformed_type{result});
        item = generated_queue.pop();
      }
      done_threads++;
      if (done_threads==ex.concurrency_degree()) {
        transformed_queue.push(transformed_type{});
      }
    });
  }

  thread consumer_thread([&](){
    auto item {transformed_queue.pop()};
    while (item) {
      consume_op(*item);
      item = transformed_queue.pop(  );
    }
  });

   //Generate elements
  for (;;) {
    auto item = generate_op();
    generated_queue.push(item) ;
    if(!item) {
      for (int i=1; i<ex.concurrency_degree(); ++i) {
        generated_queue.push(item) ;
      }
      break;
    }
  }

  generators.wait();
  consumer_thread.join();
}

}
#endif

#endif