sequential_execution.h

Go to the documentation of this file.

#ifndef GRPPI_SEQ_SEQUENTIAL_EXECUTION_H
#define GRPPI_SEQ_SEQUENTIAL_EXECUTION_H

#include "../common/mpmc_queue.h"
#include "../common/iterator.h"
#include "../common/callable_traits.h"
#include "../common/execution_traits.h"
#include "../common/patterns.h"
#include "../common/pack_traits.h"

#include <type_traits>
#include <tuple>
#include <iterator>
//#include <experimental/optional>

namespace grppi {

class sequential_execution {

public:

  constexpr sequential_execution() noexcept = default;

  constexpr void set_concurrency_degree(int n) const noexcept {}

  constexpr int concurrency_degree() const noexcept { return 1; }

  constexpr void enable_ordering() const noexcept {}

  constexpr void disable_ordering() const noexcept {}

  constexpr bool is_ordered() const noexcept { return true; }

  template <typename ... InputIterators, typename OutputIterator, 
            typename Transformer>
  constexpr void map(std::tuple<InputIterators...> firsts,
      OutputIterator first_out, std::size_t sequence_size, 
      Transformer && transform_op) const;
  
  template <typename InputIterator, typename Identity, typename Combiner>
  constexpr auto reduce(InputIterator first, std::size_t sequence_size,
              Identity && identity,
              Combiner && combine_op) const;

  template <typename ... InputIterators, typename Identity, 
            typename Transformer, typename Combiner>
  constexpr auto map_reduce(std::tuple<InputIterators...> firsts, 
                  std::size_t sequence_size,
                  Identity && identity,
                  Transformer && transform_op, Combiner && combine_op) const;

  template <typename ... InputIterators, typename OutputIterator,
            typename StencilTransformer, typename Neighbourhood>
  constexpr void stencil(std::tuple<InputIterators...> firsts, OutputIterator first_out,
               std::size_t sequence_size,
               StencilTransformer && transform_op,
               Neighbourhood && neighbour_op) const;

  template <typename Input, typename Divider, typename Solver, typename Combiner>
  [[deprecated("Use new interface with predicate argument")]]
  auto divide_conquer(Input && input, 
                      Divider && divide_op, 
                      Solver && solve_op, 
                      Combiner && combine_op) const; 

  template <typename Input, typename Divider,typename Predicate, typename Solver, typename Combiner>
  auto divide_conquer(Input && input,
                      Divider && divide_op,
                      Predicate && predicate_op,
                      Solver && solve_op,
                      Combiner && combine_op) const;


  template <typename Generator, typename ... Transformers>
  void pipeline(Generator && generate_op, 
                Transformers && ... transform_op) const;

  template <typename InputType, typename Transformer, typename OutputType>
  void pipeline(mpmc_queue<InputType> & input_queue, Transformer && transform_op,
                mpmc_queue<OutputType> & output_queue) const
  {
    using namespace std;
    using optional_output_type = typename OutputType::first_type;
    for(;;){
      auto item = input_queue.pop();
      if(!item.first) break;
      do_pipeline(*item.first, std::forward<Transformer>(transform_op),
        [&](auto output_item) {
          output_queue.push( make_pair(optional_output_type{output_item}, item.second) );
        }
      );
    }
    output_queue.push( make_pair(optional_output_type{}, -1) );
  }


private:

  template <typename Item, typename Consumer,
            requires_no_pattern<Consumer> = 0>
  void do_pipeline(Item && item, Consumer && consume_op) const;

  template <typename Item, typename Transformer, typename ... OtherTransformers,
            requires_no_pattern<Transformer> = 0>
  void do_pipeline(Item && item, Transformer && transform_op,
          OtherTransformers && ... other_ops) const;

  template <typename Item, typename FarmTransformer,
            template <typename> class Farm,
            requires_farm<Farm<FarmTransformer>> = 0>
  void do_pipeline(Item && item, Farm<FarmTransformer> & farm_obj) const
  {
    do_pipeline(std::forward<Item>(item), std::move(farm_obj));
  }

  template <typename Item, typename FarmTransformer,
            template <typename> class Farm,
            requires_farm<Farm<FarmTransformer>> = 0>
  void do_pipeline(Item && item, Farm<FarmTransformer> && farm_obj) const;

  template <typename Item, typename Execution, typename Transformer,
            template <typename, typename> class Context,
            typename ... OtherTransformers,
            requires_context<Context<Execution,Transformer>> = 0>
  void do_pipeline(Item && item, Context<Execution,Transformer> && context_op,
       OtherTransformers &&... other_ops) const
  {
     do_pipeline(item, std::forward<Transformer>(context_op.transformer()), 
       std::forward<OtherTransformers>(other_ops)...);
  }

  template <typename Item, typename Execution, typename Transformer,
            template <typename, typename> class Context,
            typename ... OtherTransformers,
            requires_context<Context<Execution,Transformer>> = 0>
  void do_pipeline(Item && item, Context<Execution,Transformer> & context_op,
       OtherTransformers &&... other_ops) const
  {
    do_pipeline(item, std::move(context_op),
      std::forward<OtherTransformers>(other_ops)...);
  }


  template <typename Item, typename FarmTransformer,
            template <typename> class Farm,
            typename... OtherTransformers,
            requires_farm<Farm<FarmTransformer>> = 0>
  void do_pipeline(Item && item, Farm<FarmTransformer> & farm_obj,
                   OtherTransformers && ... other_transform_ops) const
  {
    do_pipeline(std::forward<Item>(item), std::move(farm_obj),
        std::forward<OtherTransformers>(other_transform_ops)...);
  }

  template <typename Item, typename FarmTransformer,
            template <typename> class Farm,
            typename... OtherTransformers,
            requires_farm<Farm<FarmTransformer>> = 0>
  void do_pipeline(Item && item, Farm<FarmTransformer> && farm_obj,
                   OtherTransformers && ... other_transform_ops) const;

  template <typename Item, typename Predicate,
            template <typename> class Filter,
            typename ... OtherTransformers,
            requires_filter<Filter<Predicate>> = 0>
  void do_pipeline(Item && item, Filter<Predicate> & filter_obj,
                   OtherTransformers && ... other_transform_ops) const
  {
    do_pipeline(std::forward<Item>(item), std::move(filter_obj),
        std::forward<OtherTransformers>(other_transform_ops)...);
  }

  template <typename Item, typename Predicate,
            template <typename> class Filter,
            typename ... OtherTransformers,
            requires_filter<Filter<Predicate>> = 0>
  void do_pipeline(Item && item, Filter<Predicate> && filter_obj,
                   OtherTransformers && ... other_transform_ops) const;

  template <typename Item, typename Combiner, typename Identity,
            template <typename C, typename I> class Reduce,
            typename ... OtherTransformers,
            requires_reduce<Reduce<Combiner,Identity>> = 0>
  void do_pipeline(Item && item, Reduce<Combiner,Identity> & reduce_obj,
                   OtherTransformers && ... other_transform_ops) const
  {
    do_pipeline(std::forward<Item>(item), std::move(reduce_obj),
        std::forward<OtherTransformers>(other_transform_ops)...);
  };
    

  template <typename Item, typename Combiner, typename Identity,
            template <typename C, typename I> class Reduce,
            typename ... OtherTransformers,
            requires_reduce<Reduce<Combiner,Identity>> = 0>
  void do_pipeline(Item && item, Reduce<Combiner,Identity> && reduce_obj,
                   OtherTransformers && ... other_transform_ops) const;

  template <typename Item, typename Transformer, typename Predicate,
            template <typename T, typename P> class Iteration,
            typename ... OtherTransformers,
            requires_iteration<Iteration<Transformer,Predicate>> = 0>
  void do_pipeline(Item && item, 
          Iteration<Transformer,Predicate> & iteration_obj, 
          OtherTransformers && ... other_transform_ops) const
  {
    do_pipeline(std::forward<Item>(item), std::move(iteration_obj),
        std::forward<OtherTransformers>(other_transform_ops)...);
  }

  template <typename Item, typename Transformer, typename Predicate,
            template <typename T, typename P> class Iteration,
            typename ...OtherTransformers,
            requires_iteration<Iteration<Transformer,Predicate>> = 0,
            requires_no_pattern<Transformer> = 0>
  void do_pipeline(Item && item, 
          Iteration<Transformer,Predicate> && iteration_obj, 
          OtherTransformers && ... other_transform_ops) const;

  template <typename Item, typename Transformer, typename Predicate,
            template <typename T, typename P> class Iteration,
            typename ...OtherTransformers,
            requires_iteration<Iteration<Transformer,Predicate>> = 0,
            requires_pipeline<Transformer> = 0>
  void do_pipeline(Item && item, 
          Iteration<Transformer,Predicate> && iteration_obj, 
          OtherTransformers && ... other_transform_ops) const;

  template <typename Item, typename ... Transformers,
            template <typename...> class Pipeline,
            typename ... OtherTransformers,
            requires_pipeline<Pipeline<Transformers...>> = 0>
  void do_pipeline(Item && item, Pipeline<Transformers...> & pipeline_obj,
                   OtherTransformers && ... other_transform_ops) const
  {
    do_pipeline(std::forward<Item>(item), std::move(pipeline_obj),
        std::forward<OtherTransformers>(other_transform_ops)...);
  }

  template <typename Item, typename ... Transformers,
            template <typename...> class Pipeline,
            typename ... OtherTransformers,
            requires_pipeline<Pipeline<Transformers...>> = 0>
  void do_pipeline(Item && item, Pipeline<Transformers...> && pipeline_obj,
                   OtherTransformers && ... other_transform_ops) const;

  template <typename Item, typename ... Transformers, std::size_t ... I>
  void do_pipeline_nested(Item && item, 
          std::tuple<Transformers...> && transform_ops,
          std::index_sequence<I...>) const;

};

template <typename E>
constexpr bool is_sequential_execution() {
  return std::is_same<E, sequential_execution>::value;
}

template <>
constexpr bool is_supported<sequential_execution>() { return true; }

template <>
constexpr bool supports_map<sequential_execution>() { return true; }

template <>
constexpr bool supports_reduce<sequential_execution>() { return true; }

template <>
constexpr bool supports_map_reduce<sequential_execution>() { return true; }

template <>
constexpr bool supports_stencil<sequential_execution>() { return true; }

template <>
constexpr bool supports_divide_conquer<sequential_execution>() { return true; }

template <>
constexpr bool supports_pipeline<sequential_execution>() { return true; }

template <typename ... InputIterators, typename OutputIterator,
          typename Transformer>
constexpr void sequential_execution::map(
    std::tuple<InputIterators...> firsts,
    OutputIterator first_out, 
    std::size_t sequence_size, 
    Transformer && transform_op) const
{
  const auto last = std::next(std::get<0>(firsts), sequence_size);
  while (std::get<0>(firsts) != last) {
    *first_out++ = apply_deref_increment(
        std::forward<Transformer>(transform_op), firsts);
  }
}

template <typename InputIterator, typename Identity, typename Combiner>
constexpr auto sequential_execution::reduce(
    InputIterator first, 
    std::size_t sequence_size,
    Identity && identity,
    Combiner && combine_op) const
{
  const auto last = std::next(first, sequence_size);
  auto result{identity};
  while (first != last) {
    result = combine_op(result, *first++);
  }
  return result;
}

template <typename ... InputIterators, typename Identity, 
          typename Transformer, typename Combiner>
constexpr auto sequential_execution::map_reduce(
    std::tuple<InputIterators...> firsts,
    std::size_t sequence_size, 
    Identity && identity,
    Transformer && transform_op, Combiner && combine_op) const
{
  const auto last = std::next(std::get<0>(firsts), sequence_size);
  auto result{identity};
  while (std::get<0>(firsts) != last) {
    result = combine_op(result, apply_deref_increment(
        std::forward<Transformer>(transform_op), firsts));
  }
  return result;
}

template <typename ... InputIterators, typename OutputIterator,
          typename StencilTransformer, typename Neighbourhood>
constexpr void sequential_execution::stencil(
    std::tuple<InputIterators...> firsts, OutputIterator first_out,
    std::size_t sequence_size,
    StencilTransformer && transform_op,
    Neighbourhood && neighbour_op) const
{
  const auto last = std::next(std::get<0>(firsts), sequence_size);
  while (std::get<0>(firsts) != last) {
    const auto f = std::get<0>(firsts);
    *first_out++ = transform_op(f, 
        apply_increment(std::forward<Neighbourhood>(neighbour_op), firsts));
  }
}


template <typename Input, typename Divider, typename Predicate, typename Solver, typename Combiner>
auto sequential_execution::divide_conquer(
    Input && input,
    Divider && divide_op,
    Predicate && predicate_op,
    Solver && solve_op,
    Combiner && combine_op) const
{

  if (predicate_op(input)) { return solve_op(std::forward<Input>(input)); }
  auto subproblems = divide_op(std::forward<Input>(input));

  using subproblem_type =
      std::decay_t<typename std::result_of<Solver(Input)>::type>;
  std::vector<subproblem_type> solutions;
  for (auto && sp : subproblems) {
    solutions.push_back(divide_conquer(sp,
        std::forward<Divider>(divide_op), std::forward<Predicate>(predicate_op),std::forward<Solver>(solve_op),
        std::forward<Combiner>(combine_op)));
  }
  return reduce(std::next(solutions.begin()), solutions.size()-1, solutions[0],
      std::forward<Combiner>(combine_op));
}


template <typename Input, typename Divider, typename Solver, typename Combiner>
auto sequential_execution::divide_conquer(
    Input && input, 
    Divider && divide_op, 
    Solver && solve_op, 
    Combiner && combine_op) const
{

  auto subproblems = divide_op(std::forward<Input>(input));
  if (subproblems.size()<=1) { return solve_op(std::forward<Input>(input)); }

  using subproblem_type = 
      std::decay_t<typename std::result_of<Solver(Input)>::type>;
  std::vector<subproblem_type> solutions;
  for (auto && sp : subproblems) {
    solutions.push_back(divide_conquer(sp, 
        std::forward<Divider>(divide_op), std::forward<Solver>(solve_op), 
        std::forward<Combiner>(combine_op)));
  }
  return reduce(std::next(solutions.begin()), solutions.size()-1, solutions[0],
      std::forward<Combiner>(combine_op));
}

template <typename Generator, typename ... Transformers>
void sequential_execution::pipeline(
    Generator && generate_op,
    Transformers && ... transform_ops) const
{
  static_assert(is_generator<Generator>,
    "First pipeline stage must be a generator");

  for (;;) {
    auto x = generate_op();
    if (!x) break;
    do_pipeline(*x, std::forward<Transformers>(transform_ops)...);
  }
}

template <typename Item, typename Consumer,
          requires_no_pattern<Consumer>>
void sequential_execution::do_pipeline(
    Item && item,
    Consumer && consume_op) const
{
  consume_op(std::forward<Item>(item));
}

template <typename Item, typename Transformer, typename ... OtherTransformers,
            requires_no_pattern<Transformer>>
void sequential_execution::do_pipeline(
    Item && item,
    Transformer && transform_op,
    OtherTransformers && ... other_ops) const
{
  static_assert(!is_consumer<Transformer,Item>,
    "Itermediate pipeline stage cannot be a consumer");

  do_pipeline(transform_op(std::forward<Item>(item)), 
      std::forward<OtherTransformers>(other_ops)...);
}

template <typename Item, typename FarmTransformer,
          template <typename> class Farm,
          requires_farm<Farm<FarmTransformer>>>
void sequential_execution::do_pipeline(
    Item && item, 
    Farm<FarmTransformer> && farm_obj) const
{
  farm_obj(std::forward<Item>(item));
}

template <typename Item, typename FarmTransformer,
          template <typename> class Farm,
          typename... OtherTransformers,
          requires_farm<Farm<FarmTransformer>>>
void sequential_execution::do_pipeline(
    Item && item, 
    Farm<FarmTransformer> && farm_obj,
    OtherTransformers && ... other_transform_ops) const
{
  static_assert(!is_consumer<Farm<FarmTransformer>,Item>,
    "Itermediate pipeline stage cannot be a consumer");
  do_pipeline(farm_obj(std::forward<Item>(item)), 
      std::forward<OtherTransformers>(other_transform_ops)...);
}

template <typename Item, typename Predicate,
          template <typename> class Filter,
          typename ... OtherTransformers,
          requires_filter<Filter<Predicate>>>
void sequential_execution::do_pipeline(
    Item && item, 
    Filter<Predicate> && filter_obj,
    OtherTransformers && ... other_transform_ops) const
{
  if (filter_obj(std::forward<Item>(item))) {
    do_pipeline(std::forward<Item>(item),
        std::forward<OtherTransformers>(other_transform_ops)...);
  }
}

template <typename Item, typename Combiner, typename Identity,
          template <typename C, typename I> class Reduce,
          typename ... OtherTransformers,
          requires_reduce<Reduce<Combiner,Identity>>>
void sequential_execution::do_pipeline(
    Item && item, 
    Reduce<Combiner,Identity> && reduce_obj,
    OtherTransformers && ... other_transform_ops) const
{
  reduce_obj.add_item(std::forward<Identity>(item));
  if (reduce_obj.reduction_needed()) {
    auto red = reduce_obj.reduce_window(*this);
    do_pipeline(red,
        std::forward<OtherTransformers...>(other_transform_ops)...);
  }
}

template <typename Item, typename Transformer, typename Predicate,
          template <typename T, typename P> class Iteration,
          typename ... OtherTransformers,
          requires_iteration<Iteration<Transformer,Predicate>>,
          requires_no_pattern<Transformer>>
void sequential_execution::do_pipeline(
    Item && item, 
    Iteration<Transformer,Predicate> && iteration_obj, 
    OtherTransformers && ... other_transform_ops) const
{
  auto new_item = iteration_obj.transform(std::forward<Item>(item));
  while (!iteration_obj.predicate(new_item)) {
    new_item = iteration_obj.transform(new_item);
  }
  do_pipeline(new_item,
      std::forward<OtherTransformers...>(other_transform_ops)...);
}

template <typename Item, typename Transformer, typename Predicate,
          template <typename T, typename P> class Iteration,
          typename ... OtherTransformers,
          requires_iteration<Iteration<Transformer,Predicate>>,
          requires_pipeline<Transformer>>
void sequential_execution::do_pipeline(
    Item && item, 
    Iteration<Transformer,Predicate> && iteration_obj, 
    OtherTransformers && ... other_transform_ops) const
{
  static_assert(!is_pipeline<Transformer>, "Not implemented");
}

template <typename Item, typename ... Transformers,
          template <typename...> class Pipeline,
          typename ... OtherTransformers,
          requires_pipeline<Pipeline<Transformers...>>>
void sequential_execution::do_pipeline(
    Item && item, 
    Pipeline<Transformers...> && pipeline_obj,
    OtherTransformers && ... other_transform_ops) const
{
  do_pipeline_nested(
      std::forward<Item>(item),
      std::tuple_cat(pipeline_obj.transformers(), 
          std::forward_as_tuple(other_transform_ops...)),
      std::make_index_sequence<sizeof...(Transformers)+sizeof...(OtherTransformers)>());
}

template <typename Item, typename ... Transformers, std::size_t ... I>
void sequential_execution::do_pipeline_nested(
    Item && item, 
    std::tuple<Transformers...> && transform_ops,
    std::index_sequence<I...>) const
{
  do_pipeline(
      std::forward<Item>(item),
      std::forward<Transformers>(std::get<I>(transform_ops))...);
}

} // end namespace grppi

#endif