The stream reduction pattern consists on the use of the reduce pattern over a stream of data. Each element is processed using a reduce operator, that generates a new output element. The output elements are sent to an output stream.

The interface to the stream reduction pattern is provided by function grppi::stream_reduce(). As all functions in GrPPI, this function takes as its first argument an execution policy.

grppi::stream_reduce(exec, other_arguments...);

Stream reduction variants

Stand-alone stream reduce: Applies a reduction operation to a window of stream values produced by a generator and invokes a consumer for each resulting value.
Composable stream reduce: Defines a stream reduce that can be used as a building block by another pattern (e.g. a pipeline).

Key elements in stream reduction

The key elements in a stream reduction is the Combiner operation, the window size and the offset.

A Combiner is any C++ callable entity, that is able to combine two values into a single value. A Combiner cmb is any operation taking two values x and y of types T and U and returning a combined value of type T, making valid the following:

T x;
U y;
T res = cmb(x,y);

A window size is an integer value that defines the number of elements that should be collapsed to produce a resulting value.

An offset is an integer value that defines the overlapping degree among windows.

A stand-alone stream reduction also has a Generator. The generator may be any C++ callable entity that produces values in a way that allows to signal an indication of end-of stream. For this purpose, the stream reduction requires that the generator produces values of any type that is compatible with a subset of an optional interface:

auto r = generator(); // r is a value generated by a generator.
if (r) {       // r is convertible to bool
  auto y = *r; // r can be dereferenced.
}

Finally, a stand-alone stream reduction has a Consumer. The consumer may be any C++ callable entity that takes values of the result type of the Combiner.

Details on stream reduction variants

Stand-alone stream reduction

A stand alone stream reduction has six elements:

A window size.
An offset.
An identity value provided as an input value.
A Generator of values.
A Combiner combining two values into a single value.
A Consumer of generated values.
Example: Generate a stream of integers of collapsed windows.
int i = 0;
grppi::stream_reduce(e,
window_size, offset, 0,
[&i,n]() -> optional<int> {
if (i<n) return i++;
else return {};
},
[](int x, int y) { return x+y; },
[](int x) { cout << x << endl; }
);

Composable stream reduction

A composable stream reduction has four elements:

A window size.
An offset.
An identity value provided as an input value.
A Combiner combining two values into a single value.

The input values will be generated by the upper level pattern, which will be also responsible for consuming the output values.

Example: A stream reduction stage in a pipeline.

grppi::pipeline(exec,
  stageA,
  stageB,
  grppi::stream_reduce(exec, 
    window_size, offset, 0,
    [](int x, int y) { return x+y; }),
  stageC
  );

Note: For brevity we do not show here the details of other stages.

For composing complex patterns, the stream_reduce() function may be used to create an object that may be supplied to another pattern to build a composed pattern.

Example: A composable stream reduction stage in a pipeline.

auto chunk_sum = grppi::stream_reduce(exec, 
  window_size, offset, 0,
  [](int x, int y) { return x+y; });
grppi::pipeline(exec,
  stageA,
  stageB,
  chunk_sum,
  stageC
  );

Note: For brevity we do not show here the details of other stages.