Pattern for composing streaming operations without threads?

Question

I'm working on a library that lets me write operation on an input "stream" of data (I don't call them that, but it's a potentially unbounded input regardless, think data coming from a socket).

I might have one operation that eg: applies a frequency shift to the incoming data, and then another that applies a frequency-selective filter to that result, I'm writing in C++ so my syntax might look something like this:

input >> tune >> filter >> output;

My problem is that different operations might require an unknown number of data points to compute the output. So eg: tune perhaps (and can) work with an arbitrary number of inputs at a time, but filter requires some minimum number of samples before it can produce output.

The easiest answer is to run each filter in a thread, and connect them with some sort of thread-safe pipe or equivalent. If possible though, I'd like to avoid threading if I can.

Is anyone aware of an alternative pattern or research on composing streaming/batch operations on a stream without resorting to threads and blocking I/O?

Answer 1

Streaming frameworks (such as the built-in streams in Java 8) generally work by setting up a pipeline structure where each node can pull data from the previous node. This requires each node to support a method that can produce a single data point, as well as a reference to the previous node.

For example (pseudo-code):

tune.produce() {
  // call previous.produce() once
  // return the processed output
}

filter.produce() {
  // call previous.produce() as many times as necessary
  // return the next data point of processed output
  // you might keep some state about the last N points
}

When you start consuming the output of the pipeline, it begins a kind of chain reaction that bubbles all the way to the input, and each time you call produce() on the output, it will consume just enough input data points to produce one output data point.

Edit: If performance is a concern, then you can batch samples as Erik Eidt mentioned, so each data point can be a list of samples. Then your nodes might look like this:

input.produce() {
  // return as many samples as available in the input buffer (file/socket/etc.)
}

filter.produce() {
  while (buffer.length < N) {
    buffer.append(previous.produce());
  }
  // consume and process first N samples from buffer
  // return M output samples
}

Answer 2

The general approach that I favor for this type of work is called Functional Reactive Programming or Reactive Programming. A program is built up from observables (things to observer, such as data or events) using operators to compose functions together into an observer chain.

There are a number of libraries that implement reactive extensions to a number of languages. For instance, RxJava is an extension library for Java. Many of these libraries are modelled after the reactive extensions for CLR languages.

The composition operators can process streams of data. More importantly, they can also handle processing across thread boundaries, allowing an observer chain to move data from one thread to another as processing requires it. This, in my mind, is the biggest advantage of the extension libraries: they allow you to connect together smaller functions into larger functions in a thread-safe manner.