<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_NATIVE_PIPELINE_H
#define GRPPI_NATIVE_PIPELINE_H

#include "common/pack_traits.h"

#include <experimental/optional>

#include <thread>


namespace grppi{

template <typename InQueue, typename OutQueue, int Index, 
          typename ... MoreTransformers,
          internal::requires_index_last<Index,MoreTransformers...> = 0>
void composed_pipeline(InQueue & input_queue, 
                       const pipeline_info<parallel_execution_native, MoreTransformers...> & pipe, 
                       OutQueue & output_queue, std::vector<std::thread> & tasks)
{
  composed_pipeline(pipe.exectype, input_queue, 
      std::get<Index>(pipe.stages), output_queue, tasks);
}


template <typename InQueue, typename OutQueue, int Index, 
          typename ... MoreTransformers,
          internal::requires_index_not_last<Index,MoreTransformers...> = 0>
void composed_pipeline(InQueue & input_queue, 
                       const pipeline_info<parallel_execution_native,MoreTransformers...> & pipeline_obj, 
                       OutQueue & output_queue, std::vector<std::thread> & tasks)
{
  using stage_type = 
      typename std::tuple_element<Index,decltype(pipeline_obj.stages)>::type;
  using input_type = typename InQueue::value_type;
  using input_value_type = typename input_type::value_type;
  using result_type = 
      typename std::result_of<stage_type(input_value_type)>::type;

  // TODO: Why static?
  static mpmc_queue<std::experimental::optional<result_type>> tmp_queue{
      pipeline_obj.exectype.queue_size, pipeline_obj.exectype.lockfree}; 

  composed_pipeline(pipeline_obj.exectype, input_queue, 
      std::get<Index>(pipeline_obj.stages), tmp_queue, tasks);
  composed_pipeline<mpmc_queue<std::experimental::optional<result_type>>, 
      OutQueue, Index+1, MoreTransformers ...>(
          tmp_queue,pipeline_obj,output_queue,tasks);
}

template <typename InQueue, typename Transformer, typename OutQueue>
void composed_pipeline(parallel_execution_native & ex, InQueue & input_queue,
                       Transformer && transform_op, OutQueue & output_queue, 
                       std::vector<std::thread> & tasks)
{
  using namespace std;
  tasks.push_back(
    thread{[&](){<commit_msg>Simplfied typing in composed pipeline<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_NATIVE_PIPELINE_H
#define GRPPI_NATIVE_PIPELINE_H

#include "common/pack_traits.h"

#include <experimental/optional>

#include <thread>


namespace grppi{

template <typename InQueue, typename OutQueue, int Index, 
          typename ... MoreTransformers,
          internal::requires_index_last<Index,MoreTransformers...> = 0>
void composed_pipeline(InQueue & input_queue, 
                       const pipeline_info<parallel_execution_native, MoreTransformers...> & pipe, 
                       OutQueue & output_queue, std::vector<std::thread> & tasks)
{
  composed_pipeline(pipe.exectype, input_queue, 
      std::get<Index>(pipe.stages), output_queue, tasks);
}


template <typename InQueue, typename OutQueue, int Index, 
          typename ... MoreTransformers,
          internal::requires_index_not_last<Index,MoreTransformers...> = 0>
void composed_pipeline(InQueue & input_queue, 
                       const pipeline_info<parallel_execution_native,MoreTransformers...> & pipeline_obj, 
                       OutQueue & output_queue, std::vector<std::thread> & tasks)
{
  using namespace std;

  using stage_type = 
      typename tuple_element<Index,decltype(pipeline_obj.stages)>::type;
  using input_type = typename InQueue::value_type;
  using input_value_type = typename input_type::value_type;
  using result_value_type = 
      typename result_of<stage_type(input_value_type)>::type;
  using result_type = experimental::optional<result_value_type>;

  // TODO: Why static?
  static mpmc_queue<result_type> tmp_queue{
      pipeline_obj.exectype.queue_size, pipeline_obj.exectype.lockfree}; 

  composed_pipeline(pipeline_obj.exectype, input_queue, 
      get<Index>(pipeline_obj.stages), tmp_queue, tasks);
  composed_pipeline<mpmc_queue<result_type>, 
      OutQueue, Index+1, MoreTransformers ...>(
          tmp_queue,pipeline_obj,output_queue,tasks);
}

template <typename InQueue, typename Transformer, typename OutQueue>
void composed_pipeline(parallel_execution_native & ex, InQueue & input_queue,
                       Transformer && transform_op, OutQueue & output_queue, 
                       std::vector<std::thread> & tasks)
{
  using namespace std;
  tasks.push_back(
    thread{[&](){
{'fraction_non_alphanumeric': 0.09285484390086901, 'fraction_numerical': 0.0032185387833923397, 'mean_word_length': 3.9286280729579697, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 46, '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': '4300469', 'n_tokens_mistral': 1841, 'n_tokens_neox': 1767, 'n_words': 489}