omp/stream_reduce.h

Go to the documentation of this file.

#ifndef GRPPI_OMP_STREAM_REDUCE_H
#define GRPPI_OMP_STREAM_REDUCE_H

#ifdef GRPPI_OMP

#include "parallel_execution_omp.h"
#include "reduce.h"

namespace grppi {

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

  // TODO: Set generator and consumer in separate threads
  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

#endif