Stream reduction pattern

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.