seq/stream_reduce.h

Go to the documentation of this file.

#ifndef GRPPI_SEQ_STREAM_REDUCE_H
#define GRPPI_SEQ_STREAM_REDUCE_H

#include "sequential_execution.h"
#include "reduce.h"

#include <vector>

namespace grppi {

template <typename Identity, typename Combiner, typename Consumer, 
          typename Generator>
void stream_reduce(sequential_execution & ex,
                   int window_size, int offset, Identity identity,
                   Generator && generate_op, Combiner && combine_op, 
                   Consumer && consume_op)
{
  using namespace std;
  using generated_type = typename result_of<Generator()>::type;
  using generated_value_type = typename generated_type::value_type;
  // TODO: Evaluate better structure than vector
  vector<generated_value_type> values;
  values.reserve(window_size);
     
  auto item{generate_op()};
  for (;;) {
    while (item && values.size()!=window_size) {
      values.push_back(*item);
      item = generate_op();
    }
    if (values.size()>0) {
      auto reduced_value = reduce(ex, values.begin(), values.end(), identity,
          std::forward<Combiner>(combine_op));
      consume_op(reduced_value);
      if(item){
        if (offset <= window_size) {
          values.erase(values.begin(), values.begin() + offset);
        }
        else {
          values.erase(values.begin(), values.end());
          auto diff = offset - window_size;
          while (diff > 0 && item) {
            item = generate_op();
            diff--;
          }
        }
      }
    }
    if (!item ) break;
  } 
}

}

#endif