omp/farm.h

Go to the documentation of this file.

#ifndef GRPPI_OMP_FARM_H
#define GRPPI_OMP_FARM_H

#ifdef GRPPI_OMP

#include "parallel_execution_omp.h"

#include <experimental/optional>

namespace grppi {

template <typename Generator, typename Consumer>
void farm(parallel_execution_omp & 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>();

  #pragma omp parallel
  {
    #pragma omp single nowait
    {
      for (int i = 0; i<ex.concurrency_degree(); i++) {
        #pragma omp task shared(queue)
        {
          auto item{queue.pop()};
          while (item) {
            consume_op(*item);
            item = queue.pop() ;
          }
        }
      }
    
      for (;;) {
        auto item = generate_op();
        queue.push(item) ;
        if (!item) {
          for (int i=1; i<ex.concurrency_degree(); ++i) {
            queue.push(item);
          }
          break;
        }
      }

      #pragma omp taskwait
    }
  } 
}

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

  auto generated_queue = ex.make_queue<result_type>();
  auto transformed_queue = ex.make_queue<transformed_type>();
  atomic<int> done_threads{0};

  #pragma omp parallel
  {
    #pragma omp single nowait
    {
      for (int i=0; i<ex.concurrency_degree(); ++i) {
        #pragma omp task shared(generated_queue, transformed_queue, transform_op)
        {
          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{});
        }
      }

      #pragma omp task shared(transformed_queue,consume_op)
      { 
        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;
      }

      #pragma omp taskwait
    }
  }
}

}
#endif

#endif