<commit_before>/**
* @version		GrPPI v0.2
* @copyright		Copyright (C) 2017 Universidad Carlos III de Madrid. All rights reserved.
* @license		GNU/GPL, see LICENSE.txt
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You have received a copy of the GNU General Public License in LICENSE.txt
* also available in <http://www.gnu.org/licenses/gpl.html>.
*
* See COPYRIGHT.txt for copyright notices and details.
*/

#ifndef GRPPI_TBB_FARM_H
#define GRPPI_TBB_FARM_H

#ifdef GRPPI_TBB
#include <experimental/optional>

#include <tbb/tbb.h>
namespace grppi{

/**
\addtogroup farm_pattern
@{
*/

/**
\addtogroup farm_pattern_tbb TBB parallel farm pattern
TBB parallel implementation of the \ref md_farm.
@{
*/

/**
\brief Invoke [farm pattern](@ref md_farm) on a data stream with TBB parallel 
execution with a generator and a consumer.
\tparam Generator Callable type for the generation operation.
\tparam Consumer Callable type for the consume operation.
\param ex TBB parallel execution policy object.
\param generate_op Generator operation.
\param consume_op Consumer operation.
*/
template <typename Generator, typename Consumer>
void farm(parallel_execution_tbb & ex, Generator generate_op, 
          Consumer && consume_op) 
{
  using namespace std;

  using generated_type = typename std::result_of<Generator()>::type;
  mpmc_queue<generated_type> queue{ex.queue_size, ex.lockfree};

  tbb::task_group g;
  for (int i=0; i<ex.num_threads; ++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.num_threads; i++) {
          queue.push(item);
        }
        break;
      }
    }

    //Join threads
    g.wait();
}

/**
\brief Invoke [farm pattern](@ref md_farm) on a data stream with TBB parallel 
execution with a generator and a consumer.
\tparam Generator Callable type for the generation operation.
\tparam Tranformer Callable type for the tranformation operation.
\tparam Consumer Callable type for the consume operation.
\param ex TBB parallel execution policy object.
\param generate_op Generator operation.
\param transform_op Transformer operation.
\param consume_op Consumer operation.
*/
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>;

  mpmc_queue<generated_type> generated_queue{ex.queue_size,ex.lockfree};
  mpmc_queue<transformed_type> transformed_queue(ex.queue_size,ex.lockfree);

  atomic<int>done_threads{0};
  tbb::task_group g;
  for (int i=0; i<ex.num_threads; ++i) {
    g.run([&](){
      auto item{generated_queue.pop()};
      while (item) {
        auto out = transform_op( item.value() );
        transformed_queue.push(transformed_type{transform_op(*item)});
        item = generated_queue.pop();
      }
      done_threads++;
      if (done_threads== ex.num_threads) {
        transformed_queue.push(transformed_type{});
      }
    });
  }

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

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

  //Join threads
  g.wait();
  consume.join();
}

}

#endif

#endif<commit_msg>solved regresion bug in tbb::farm<commit_after>/**
* @version		GrPPI v0.2
* @copyright		Copyright (C) 2017 Universidad Carlos III de Madrid. All rights reserved.
* @license		GNU/GPL, see LICENSE.txt
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You have received a copy of the GNU General Public License in LICENSE.txt
* also available in <http://www.gnu.org/licenses/gpl.html>.
*
* See COPYRIGHT.txt for copyright notices and details.
*/

#ifndef GRPPI_TBB_FARM_H
#define GRPPI_TBB_FARM_H

#ifdef GRPPI_TBB

#include <experimental/optional>

#include <tbb/tbb.h>

namespace grppi{

/**
\addtogroup farm_pattern
@{
*/

/**
\addtogroup farm_pattern_tbb TBB parallel farm pattern
TBB parallel implementation of the \ref md_farm.
@{
*/

/**
\brief Invoke [farm pattern](@ref md_farm) on a data stream with TBB parallel 
execution with a generator and a consumer.
\tparam Generator Callable type for the generation operation.
\tparam Consumer Callable type for the consume operation.
\param ex TBB parallel execution policy object.
\param generate_op Generator operation.
\param consume_op Consumer operation.
*/
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;
  mpmc_queue<generated_type> queue{ex.queue_size, ex.lockfree};

  tbb::task_group g;
  for (int i=0; i<ex.num_threads; ++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.num_threads; i++) {
          queue.push(item);
        }
        break;
      }
    }

    //Join threads
    g.wait();
}

/**
\brief Invoke [farm pattern](@ref md_farm) on a data stream with TBB parallel 
execution with a generator and a consumer.
\tparam Generator Callable type for the generation operation.
\tparam Tranformer Callable type for the tranformation operation.
\tparam Consumer Callable type for the consume operation.
\param ex TBB parallel execution policy object.
\param generate_op Generator operation.
\param transform_op Transformer operation.
\param consume_op Consumer operation.
*/
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>;

  mpmc_queue<generated_type> generated_queue(ex.queue_size,ex.lockfree);
  mpmc_queue<transformed_type> transformed_queue(ex.queue_size,ex.lockfree);

  atomic<int>done_threads{0};
  tbb::task_group generators;
  for (int i=0; i<ex.num_threads; ++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.num_threads) {
        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.num_threads; ++i) {
        generated_queue.push(item) ;
      }
      break;
    }
  }

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

/**
@}
@}
*/

}
#endif

#endif
{'fraction_non_alphanumeric': 0.1060981098109811, 'fraction_numerical': 0.0027002700270027003, 'mean_word_length': 3.627277459656429, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4467745', 'n_tokens_mistral': 2805, 'n_tokens_neox': 2713, 'n_words': 827}