Intel's Threading Building Blocks (TBB) provides classes for parallel pipeline processing. Each pipeline stage is modelled by deriving the filter class and overriding the operator() function. The stages of the pipeline work around a token that is produced by the first stage and passed between them to the last one. (This is a very brief introduction to these TBB concepts. Please refer to the TBB's tutorial and reference manual for additional information.)

There is a thread about this approach on Intel's TBB forums, discussing shortcomings and possible solutions.

The operator() is defined as follows:

class filter{
   ...
   virtual void *operator() (void *token) = 0;
}

This implementation lacks type safety. A cast is needed inside the operator() to convert the input void * to the data type used as a token between pipeline stages. Another cast is needed for passing the output result to the next pipeline stage.

A header only, type safe template wrapper class around TBB's filter and pipeline is proposed. Download tpipeline.h (preview).

Type safe pipeline: tfilter and pipeline_run

The templated filter, tfilter, has the input and output types explicity defined:

namespace tbb{
   template <typename Input, typename Output>
   class tfilter {
   public;
      typedef enum { serial, parallel } filterState;
      tfilter(filterState state);
      ~tfilter();

      virtual Output* operator() (Input *item) = 0;
   };
}

Several tfilter instances can be composed using the operator>>, that returns a tfilter descendant with the input of the first filter and the output of the latter. This allows a simple syntax for the pipeline_run function and compile time type safety:

namespace tbb{
   void pipeline_run(size_t max_tokens, const tfilter<void, void> &t);
}

Note that a complete pipeline has a void input type (the first stage doesn't use the input) and also void output type (as the last stage's output is discarded). Please check tpipeline.h for implementation details, and also the usage example below.

Example:

A synthetic example follows, a pipeline definition with three stages: Input, Process and Output. To illustrate the type safety, the input stage in this example produces a char *, and the process stage outputs a int *.

The filters definitions are:

Input filter

The input filter has void as input type.

class InputFilter : public tbb::tfilter<void, char> {
public:
   InputFilter() : tbb::tfilter<void, char>(serial) { }
   char *operator()(void *) {
      // ....
      return char*;
   }
};

Process filter

class ProcessFilter : public tbb::tfilter<char, int> {
public:
   ProcessFilter() : tbb::tfilter<char, int>(parallel) { }
   int *operator()(char *) {
      // ....
      return int*;
   }
};

Output filter

The output filter has void as output type.

class OutputFilter : public tbb::tfilter<int, void> {
public:
   ProcessFilter() : tbb::tfilter<int, void>(serial) { }
   void *operator()(int *) {
      // ....
      return 0;
   }
};

Execution of the pipeline:

Finally, the pipeline is run with pipeline_run and composing the filter instances with the >> operator. As this operator is also templated, the pipeline construction is done at compile time, thus providing compile time type safety.

tbb::pipeline_run(numTokens, InputFilter() >> ProcessFilter() >> OutputFilter());

Current shortcomings:

Currently, this approach has some shortcomings, as some Intel TBB developers have pointed out.

• The wrapper class works around some temporary objects being created while calling pipeline_run(). If the user constructs the pipeline before running it, these temporary objects would be already destroyed.

  tbb::tfilter filter_set =
          InputFilter() >> ProcessFilter() >> OutputFilter();
  pipeline_run(numTokens, filter_set);

• Runtime created pipelines are not supported.