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. This streaming pattern can only be used inside another pattern and consequently does not take an execution policy itself, but uses the execution policy of its enclosing pattern.

The interface to the stream reduction pattern is provided by function grppi::stream_reduce().

grppi::pipeline(ex
  stage1,
  grppi::stream_reduce(arguments...),
  stage2,
  stage3,
  ...)

Stream reduction variants

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.

Details on stream reduction variants

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(
    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(
  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.